Celes Renata
c85c0068a2
- Replace all datetime.utcnow() with datetime.now(tz=timezone.utc) across 8 files - Fix 12 failing tests to match current implementation behavior - Fix pytest_plugins in non-top-level conftest (moved to root conftest.py) - Auto-fix 189 lint issues (import sorting, unused imports) - Add CI/CD pipeline infrastructure (ARC, ArgoCD, Kargo manifests) - Add values-beta.yaml and values-paper.yaml for staged deployments - Update GitHub Actions workflow to use self-hosted-gremlin runners - Add integration-test job to CI pipeline Result: 1596 passed, 0 failed, 0 warnings
291 lines
8.7 KiB
Python
291 lines
8.7 KiB
Python
"""Property-based tests for the Tax Lot Tracker.
|
|
|
|
Feature: autonomous-trading-engine
|
|
|
|
Tests properties 22 and 23 from the design specification,
|
|
covering FIFO lot ordering and wash sale detection.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, timedelta
|
|
|
|
from hypothesis import assume, given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from services.trading.tax_lots import TaxLot, TaxLotTracker
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hypothesis strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _tax_lot_strategy(
|
|
ticker: str = "AAPL",
|
|
status: str = "open",
|
|
) -> st.SearchStrategy[TaxLot]:
|
|
"""Generate a random TaxLot with a fixed ticker and status."""
|
|
return st.builds(
|
|
TaxLot,
|
|
ticker=st.just(ticker),
|
|
quantity=st.integers(min_value=1, max_value=100),
|
|
cost_basis_per_share=st.floats(
|
|
min_value=1.0, max_value=500.0,
|
|
allow_nan=False, allow_infinity=False,
|
|
),
|
|
acquisition_date=st.dates(
|
|
min_value=date(2023, 1, 1),
|
|
max_value=date(2025, 6, 1),
|
|
),
|
|
status=st.just(status),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 22: Tax lot FIFO ordering
|
|
# **Validates: Requirements 12.4**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty22TaxLotFIFOOrdering:
|
|
"""Property 22: Tax lot FIFO ordering.
|
|
|
|
For any sequence of buy transactions, closing lots SHALL follow
|
|
FIFO order — the earliest-acquired open lot is closed first.
|
|
The realized P&L for each closed lot SHALL equal
|
|
(exit_price - cost_basis_per_share) * quantity.
|
|
|
|
**Validates: Requirements 12.4**
|
|
"""
|
|
|
|
@given(
|
|
lots=st.lists(
|
|
_tax_lot_strategy(),
|
|
min_size=1,
|
|
max_size=10,
|
|
),
|
|
exit_price=st.floats(
|
|
min_value=1.0, max_value=500.0,
|
|
allow_nan=False, allow_infinity=False,
|
|
),
|
|
exit_date=st.dates(
|
|
min_value=date(2025, 1, 1),
|
|
max_value=date(2025, 12, 31),
|
|
),
|
|
)
|
|
@settings(max_examples=100)
|
|
def test_lots_closed_in_fifo_order(
|
|
self,
|
|
lots: list[TaxLot],
|
|
exit_price: float,
|
|
exit_date: date,
|
|
) -> None:
|
|
"""Closed lots are ordered by acquisition_date ascending (FIFO)."""
|
|
total_available = sum(lot.quantity for lot in lots)
|
|
assume(total_available > 0)
|
|
|
|
tracker = TaxLotTracker()
|
|
closed = tracker.close_lots_fifo(
|
|
lots=lots,
|
|
quantity=total_available,
|
|
exit_price=exit_price,
|
|
exit_date=exit_date,
|
|
)
|
|
|
|
# Verify FIFO: each closed lot's acquisition_date is <= the next
|
|
for i in range(len(closed) - 1):
|
|
assert closed[i].acquisition_date <= closed[i + 1].acquisition_date, (
|
|
f"FIFO violated: lot {i} acquired {closed[i].acquisition_date} "
|
|
f"but lot {i+1} acquired {closed[i+1].acquisition_date}"
|
|
)
|
|
|
|
@given(
|
|
lots=st.lists(
|
|
_tax_lot_strategy(),
|
|
min_size=1,
|
|
max_size=10,
|
|
),
|
|
exit_price=st.floats(
|
|
min_value=1.0, max_value=500.0,
|
|
allow_nan=False, allow_infinity=False,
|
|
),
|
|
exit_date=st.dates(
|
|
min_value=date(2025, 1, 1),
|
|
max_value=date(2025, 12, 31),
|
|
),
|
|
)
|
|
@settings(max_examples=100)
|
|
def test_realized_pnl_formula(
|
|
self,
|
|
lots: list[TaxLot],
|
|
exit_price: float,
|
|
exit_date: date,
|
|
) -> None:
|
|
"""Realized P&L = (exit_price - cost_basis_per_share) * quantity per lot."""
|
|
total_available = sum(lot.quantity for lot in lots)
|
|
assume(total_available > 0)
|
|
|
|
tracker = TaxLotTracker()
|
|
closed = tracker.close_lots_fifo(
|
|
lots=lots,
|
|
quantity=total_available,
|
|
exit_price=exit_price,
|
|
exit_date=exit_date,
|
|
)
|
|
|
|
for cl in closed:
|
|
expected_pnl = (cl.exit_price - cl.cost_basis_per_share) * cl.quantity
|
|
assert abs(cl.realized_pnl - expected_pnl) < 1e-6, (
|
|
f"P&L mismatch: got {cl.realized_pnl}, "
|
|
f"expected {expected_pnl} for lot acquired {cl.acquisition_date}"
|
|
)
|
|
|
|
@given(
|
|
lots=st.lists(
|
|
_tax_lot_strategy(),
|
|
min_size=2,
|
|
max_size=10,
|
|
),
|
|
exit_price=st.floats(
|
|
min_value=1.0, max_value=500.0,
|
|
allow_nan=False, allow_infinity=False,
|
|
),
|
|
exit_date=st.dates(
|
|
min_value=date(2025, 1, 1),
|
|
max_value=date(2025, 12, 31),
|
|
),
|
|
)
|
|
@settings(max_examples=100)
|
|
def test_partial_close_respects_fifo(
|
|
self,
|
|
lots: list[TaxLot],
|
|
exit_price: float,
|
|
exit_date: date,
|
|
) -> None:
|
|
"""Closing fewer shares than available still follows FIFO order."""
|
|
total_available = sum(lot.quantity for lot in lots)
|
|
assume(total_available >= 2)
|
|
|
|
# Close only half the shares
|
|
close_qty = total_available // 2
|
|
assume(close_qty > 0)
|
|
|
|
tracker = TaxLotTracker()
|
|
closed = tracker.close_lots_fifo(
|
|
lots=lots,
|
|
quantity=close_qty,
|
|
exit_price=exit_price,
|
|
exit_date=exit_date,
|
|
)
|
|
|
|
# Total closed quantity should not exceed requested
|
|
total_closed = sum(cl.quantity for cl in closed)
|
|
assert total_closed <= close_qty
|
|
|
|
# FIFO order maintained
|
|
for i in range(len(closed) - 1):
|
|
assert closed[i].acquisition_date <= closed[i + 1].acquisition_date
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 23: Wash sale detection within 30-day window
|
|
# **Validates: Requirements 12.2, 12.3**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty23WashSaleDetection:
|
|
"""Property 23: Wash sale detection within 30-day window.
|
|
|
|
For any position closed at a loss, the Tax Lot Tracker SHALL flag
|
|
it as a potential wash sale if the same ticker was purchased within
|
|
30 days before or after the loss date.
|
|
|
|
**Validates: Requirements 12.2, 12.3**
|
|
"""
|
|
|
|
@given(
|
|
loss_date=st.dates(
|
|
min_value=date(2024, 6, 1),
|
|
max_value=date(2025, 6, 1),
|
|
),
|
|
days_offset=st.integers(min_value=-30, max_value=30),
|
|
)
|
|
@settings(max_examples=100)
|
|
def test_wash_sale_flagged_within_30_days(
|
|
self,
|
|
loss_date: date,
|
|
days_offset: int,
|
|
) -> None:
|
|
"""Purchase within 30 days of loss triggers wash sale flag."""
|
|
purchase_date = loss_date + timedelta(days=days_offset)
|
|
purchase = TaxLot(
|
|
ticker="AAPL",
|
|
quantity=10,
|
|
cost_basis_per_share=100.0,
|
|
acquisition_date=purchase_date,
|
|
)
|
|
|
|
tracker = TaxLotTracker()
|
|
result = tracker.check_wash_sale(
|
|
loss_date=loss_date,
|
|
purchases=[purchase],
|
|
)
|
|
|
|
assert result is True, (
|
|
f"Expected wash sale flag for purchase {days_offset} days "
|
|
f"from loss date {loss_date}"
|
|
)
|
|
|
|
@given(
|
|
loss_date=st.dates(
|
|
min_value=date(2024, 6, 1),
|
|
max_value=date(2025, 6, 1),
|
|
),
|
|
days_offset=st.integers(min_value=31, max_value=365),
|
|
direction=st.sampled_from([-1, 1]),
|
|
)
|
|
@settings(max_examples=100)
|
|
def test_no_wash_sale_outside_30_days(
|
|
self,
|
|
loss_date: date,
|
|
days_offset: int,
|
|
direction: int,
|
|
) -> None:
|
|
"""Purchase outside 30-day window does not trigger wash sale."""
|
|
purchase_date = loss_date + timedelta(days=days_offset * direction)
|
|
purchase = TaxLot(
|
|
ticker="AAPL",
|
|
quantity=10,
|
|
cost_basis_per_share=100.0,
|
|
acquisition_date=purchase_date,
|
|
)
|
|
|
|
tracker = TaxLotTracker()
|
|
result = tracker.check_wash_sale(
|
|
loss_date=loss_date,
|
|
purchases=[purchase],
|
|
)
|
|
|
|
assert result is False, (
|
|
f"Unexpected wash sale flag for purchase {days_offset * direction} "
|
|
f"days from loss date {loss_date}"
|
|
)
|
|
|
|
@given(
|
|
loss_date=st.dates(
|
|
min_value=date(2024, 6, 1),
|
|
max_value=date(2025, 6, 1),
|
|
),
|
|
)
|
|
@settings(max_examples=100)
|
|
def test_no_wash_sale_with_no_purchases(
|
|
self,
|
|
loss_date: date,
|
|
) -> None:
|
|
"""No purchases means no wash sale."""
|
|
tracker = TaxLotTracker()
|
|
result = tracker.check_wash_sale(
|
|
loss_date=loss_date,
|
|
purchases=[],
|
|
)
|
|
|
|
assert result is False
|