"""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 given, settings, assume from hypothesis import strategies as st from services.trading.tax_lots import ClosedLot, 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