"""Property-based tests for the Performance Tracker. Feature: autonomous-trading-engine Property 26: Performance metrics computation. Property 33: Micro-trade metrics tracked separately. """ from __future__ import annotations import math from datetime import timedelta from hypothesis import given, settings from hypothesis import strategies as st from services.trading.models import ClosedTrade from services.trading.performance_tracker import PerformanceComputer # --------------------------------------------------------------------------- # Hypothesis strategies # --------------------------------------------------------------------------- def _closed_trade_strategy( is_micro: st.SearchStrategy[bool] | None = None, ) -> st.SearchStrategy[ClosedTrade]: """Generate random ClosedTrade objects with consistent pnl fields.""" return st.builds( _make_closed_trade, ticker=st.from_regex(r"[A-Z]{3,5}", fullmatch=True), entry_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False), exit_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False), quantity=st.integers(min_value=1, max_value=100), hold_days=st.integers(min_value=1, max_value=90), is_micro_trade=is_micro if is_micro is not None else st.booleans(), ) def _make_closed_trade( ticker: str, entry_price: float, exit_price: float, quantity: int, hold_days: int, is_micro_trade: bool, ) -> ClosedTrade: """Create a ClosedTrade with consistent pnl fields.""" pnl = (exit_price - entry_price) * quantity pnl_pct = (exit_price - entry_price) / entry_price if entry_price > 0 else 0.0 return ClosedTrade( ticker=ticker, entry_price=entry_price, exit_price=exit_price, quantity=quantity, pnl=pnl, pnl_pct=pnl_pct, hold_duration=timedelta(days=hold_days), recommendation_id=None, is_micro_trade=is_micro_trade, ) def _daily_returns_strategy( min_size: int = 2, max_size: int = 60, ) -> st.SearchStrategy[list[float]]: """Generate random daily return sequences.""" return st.lists( st.floats(min_value=-0.10, max_value=0.10, allow_nan=False, allow_infinity=False), min_size=min_size, max_size=max_size, ) # --------------------------------------------------------------------------- # Property 26: Performance metrics computation # **Validates: Requirements 14.1, 14.2** # --------------------------------------------------------------------------- class TestProperty26PerformanceMetrics: """Property 26: Performance metrics computation. **Validates: Requirements 14.1, 14.2** """ computer = PerformanceComputer() @settings(max_examples=100) @given( trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=30, ), ) def test_win_rate_equals_wins_over_total(self, trades: list[ClosedTrade]) -> None: """win_rate = wins / total_trades.""" metrics = self.computer.compute_metrics( closed_trades=trades, portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) total = len(trades) wins = sum(1 for t in trades if t.pnl > 0) expected_win_rate = wins / total if total > 0 else 0.0 assert abs(metrics.win_rate - expected_win_rate) < 1e-9, ( f"Expected win_rate={expected_win_rate}, got {metrics.win_rate}" ) assert metrics.win_count == wins assert metrics.loss_count == total - wins @settings(max_examples=100) @given( trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=30, ), ) def test_profit_factor_equals_gross_profits_over_losses( self, trades: list[ClosedTrade], ) -> None: """profit_factor = gross_profits / gross_losses (inf if no losses).""" metrics = self.computer.compute_metrics( closed_trades=trades, portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) gross_profits = sum(t.pnl for t in trades if t.pnl > 0) gross_losses = abs(sum(t.pnl for t in trades if t.pnl <= 0)) if gross_losses > 0: expected_pf = gross_profits / gross_losses assert abs(metrics.profit_factor - expected_pf) < 1e-6, ( f"Expected profit_factor={expected_pf}, got {metrics.profit_factor}" ) else: if gross_profits > 0: assert metrics.profit_factor == float("inf") else: assert metrics.profit_factor == 0.0 @settings(max_examples=100) @given( daily_returns=_daily_returns_strategy(min_size=2, max_size=60), ) def test_sharpe_ratio_formula_consistency( self, daily_returns: list[float], ) -> None: """Sharpe ratio = (mean / std) * sqrt(252), 0 if std=0 or <2 points.""" metrics = self.computer.compute_metrics( closed_trades=[], portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=daily_returns, ) n = len(daily_returns) mean_r = sum(daily_returns) / n variance = sum((r - mean_r) ** 2 for r in daily_returns) / (n - 1) std_r = math.sqrt(variance) if std_r < 1e-12: assert metrics.sharpe_ratio == 0.0 else: expected = (mean_r / std_r) * math.sqrt(252) assert abs(metrics.sharpe_ratio - expected) < 1e-6, ( f"Expected sharpe={expected}, got {metrics.sharpe_ratio}" ) @settings(max_examples=100) @given( daily_returns=st.just([0.05, 0.05, 0.05]), ) def test_sharpe_zero_when_std_is_zero(self, daily_returns: list[float]) -> None: """Sharpe ratio is 0.0 when all daily returns are identical (std=0).""" metrics = self.computer.compute_metrics( closed_trades=[], portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=daily_returns, ) assert metrics.sharpe_ratio == 0.0 @settings(max_examples=100) @given( daily_returns=st.just([0.01]), ) def test_sharpe_zero_when_fewer_than_2_points( self, daily_returns: list[float], ) -> None: """Sharpe ratio is 0.0 when fewer than 2 data points.""" metrics = self.computer.compute_metrics( closed_trades=[], portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=daily_returns, ) assert metrics.sharpe_ratio == 0.0 @settings(max_examples=100) @given( trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=0, max_size=20, ), ) def test_empty_trades_gives_zero_metrics(self, trades: list[ClosedTrade]) -> None: """With no trades, win_rate=0, profit_factor=0, counts=0.""" if len(trades) > 0: return # Only test empty case metrics = self.computer.compute_metrics( closed_trades=[], portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) assert metrics.win_count == 0 assert metrics.loss_count == 0 assert metrics.win_rate == 0.0 assert metrics.profit_factor == 0.0 @settings(max_examples=100) @given( trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=30, ), ) def test_realized_pnl_equals_sum_of_trade_pnls( self, trades: list[ClosedTrade], ) -> None: """realized_pnl equals the sum of all trade P&Ls.""" metrics = self.computer.compute_metrics( closed_trades=trades, portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) expected_pnl = sum(t.pnl for t in trades) assert abs(metrics.realized_pnl - expected_pnl) < 1e-6 @settings(max_examples=100) @given( daily_returns=_daily_returns_strategy(min_size=3, max_size=60), ) def test_max_drawdown_non_negative(self, daily_returns: list[float]) -> None: """Max drawdown is always non-negative.""" metrics = self.computer.compute_metrics( closed_trades=[], portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=daily_returns, ) assert metrics.max_drawdown >= 0.0 assert metrics.current_drawdown_pct >= 0.0 # --------------------------------------------------------------------------- # Property 33: Micro-trade metrics tracked separately # **Validates: Requirements 20.7** # --------------------------------------------------------------------------- class TestProperty33MicroTradeMetrics: """Property 33: Micro-trade metrics tracked separately. **Validates: Requirements 20.7** """ computer = PerformanceComputer() @settings(max_examples=100) @given( standard_trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=15, ), micro_trades=st.lists( _closed_trade_strategy(is_micro=st.just(True)), min_size=1, max_size=15, ), ) def test_micro_trade_filter_returns_only_micro( self, standard_trades: list[ClosedTrade], micro_trades: list[ClosedTrade], ) -> None: """filter_by_micro_trade(is_micro=True) returns only micro-trades.""" all_trades = standard_trades + micro_trades filtered = self.computer.filter_by_micro_trade(all_trades, is_micro=True) assert len(filtered) == len(micro_trades) for t in filtered: assert t.is_micro_trade is True @settings(max_examples=100) @given( standard_trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=15, ), micro_trades=st.lists( _closed_trade_strategy(is_micro=st.just(True)), min_size=1, max_size=15, ), ) def test_standard_trade_filter_returns_only_standard( self, standard_trades: list[ClosedTrade], micro_trades: list[ClosedTrade], ) -> None: """filter_by_micro_trade(is_micro=False) returns only standard trades.""" all_trades = standard_trades + micro_trades filtered = self.computer.filter_by_micro_trade(all_trades, is_micro=False) assert len(filtered) == len(standard_trades) for t in filtered: assert t.is_micro_trade is False @settings(max_examples=100) @given( standard_trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=15, ), micro_trades=st.lists( _closed_trade_strategy(is_micro=st.just(True)), min_size=1, max_size=15, ), ) def test_micro_metrics_independent_from_standard( self, standard_trades: list[ClosedTrade], micro_trades: list[ClosedTrade], ) -> None: """Micro-trade metrics are computed independently from standard metrics.""" all_trades = standard_trades + micro_trades # Compute metrics for micro-trades only micro_only = self.computer.filter_by_micro_trade(all_trades, is_micro=True) micro_metrics = self.computer.compute_metrics( closed_trades=micro_only, portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) # Compute metrics for standard trades only standard_only = self.computer.filter_by_micro_trade(all_trades, is_micro=False) standard_metrics = self.computer.compute_metrics( closed_trades=standard_only, portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) # Verify counts are independent assert micro_metrics.win_count + micro_metrics.loss_count == len(micro_trades) assert standard_metrics.win_count + standard_metrics.loss_count == len(standard_trades) # Verify win rates are computed independently micro_wins = sum(1 for t in micro_trades if t.pnl > 0) expected_micro_wr = micro_wins / len(micro_trades) if micro_trades else 0.0 assert abs(micro_metrics.win_rate - expected_micro_wr) < 1e-9 standard_wins = sum(1 for t in standard_trades if t.pnl > 0) expected_std_wr = standard_wins / len(standard_trades) if standard_trades else 0.0 assert abs(standard_metrics.win_rate - expected_std_wr) < 1e-9 @settings(max_examples=100) @given( standard_trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=15, ), micro_trades=st.lists( _closed_trade_strategy(is_micro=st.just(True)), min_size=1, max_size=15, ), ) def test_standard_metrics_not_contaminated_by_micro( self, standard_trades: list[ClosedTrade], micro_trades: list[ClosedTrade], ) -> None: """Standard trade metrics are not contaminated by micro-trades.""" # Compute standard metrics from standard trades only standard_metrics = self.computer.compute_metrics( closed_trades=standard_trades, portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) # Compute standard metrics after filtering from mixed set all_trades = standard_trades + micro_trades filtered_standard = self.computer.filter_by_micro_trade(all_trades, is_micro=False) filtered_metrics = self.computer.compute_metrics( closed_trades=filtered_standard, portfolio_value=10000.0, active_pool=8000.0, reserve_pool=2000.0, daily_pnl=0.0, unrealized_pnl=0.0, portfolio_heat=0.0, daily_returns=[], ) # Metrics should be identical assert standard_metrics.win_count == filtered_metrics.win_count assert standard_metrics.loss_count == filtered_metrics.loss_count assert abs(standard_metrics.win_rate - filtered_metrics.win_rate) < 1e-9 assert abs(standard_metrics.realized_pnl - filtered_metrics.realized_pnl) < 1e-6 # Handle inf == inf case (inf - inf = nan) if math.isinf(standard_metrics.profit_factor) and math.isinf(filtered_metrics.profit_factor): pass # Both infinity — equal else: assert abs(standard_metrics.profit_factor - filtered_metrics.profit_factor) < 1e-6 @settings(max_examples=100) @given( standard_trades=st.lists( _closed_trade_strategy(is_micro=st.just(False)), min_size=1, max_size=15, ), micro_trades=st.lists( _closed_trade_strategy(is_micro=st.just(True)), min_size=1, max_size=15, ), ) def test_filter_preserves_all_trades( self, standard_trades: list[ClosedTrade], micro_trades: list[ClosedTrade], ) -> None: """Filtering micro + standard covers all trades with no overlap.""" all_trades = standard_trades + micro_trades micro_filtered = self.computer.filter_by_micro_trade(all_trades, is_micro=True) standard_filtered = self.computer.filter_by_micro_trade(all_trades, is_micro=False) assert len(micro_filtered) + len(standard_filtered) == len(all_trades)