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
510 lines
17 KiB
Python
510 lines
17 KiB
Python
"""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)
|