Files
ai-exchange/tests/unit/test_backtest_report_service.py
T

138 lines
5.0 KiB
Python

import os
import unittest
from datetime import datetime, timezone
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
from src.services.backtest import BacktestOutcome, BacktestQualityGate, summarize_backtest_outcomes
from src.services.reports import BacktestReportService
class BacktestSummaryTests(unittest.TestCase):
def test_summarizes_outcomes_by_session_side_and_weekday(self) -> None:
outcomes = [
BacktestOutcome(
signal_id=1,
outcome="win",
mfe=4.0,
mae=1.0,
tp1_hit=True,
tp2_hit=True,
invalidated_before_entry=False,
error_tags=[],
meta={
"signal_side": "buy",
"signal_symbol": "BTC-USDT",
"session_codes": ["LONDON"],
"weekday": "Monday",
},
),
BacktestOutcome(
signal_id=2,
outcome="loss",
mfe=1.0,
mae=3.0,
tp1_hit=False,
tp2_hit=False,
invalidated_before_entry=False,
error_tags=["stopped_out"],
meta={
"signal_side": "sell",
"signal_symbol": "ETH-USDT",
"session_codes": ["NY_AM"],
"weekday": "Monday",
},
),
BacktestOutcome(
signal_id=3,
outcome="missed",
mfe=0.0,
mae=0.5,
tp1_hit=False,
tp2_hit=False,
invalidated_before_entry=True,
error_tags=["entry_not_triggered"],
meta={
"signal_side": "buy",
"signal_symbol": "BTC-USDT",
"session_codes": ["LONDON", "NY_AM"],
"weekday": "Tuesday",
},
),
]
summary = summarize_backtest_outcomes(outcomes)
self.assertEqual(summary["total_signals"], 3)
self.assertEqual(summary["outcomes"]["win"], 1)
self.assertEqual(summary["outcomes"]["loss"], 1)
self.assertEqual(summary["outcomes"]["missed"], 1)
self.assertEqual(summary["lifecycle_states"]["tp2_closed"], 1)
self.assertEqual(summary["lifecycle_states"]["stopped_out"], 1)
self.assertEqual(summary["lifecycle_states"]["invalidated_before_entry"], 1)
self.assertAlmostEqual(summary["win_rate"], 1 / 3)
self.assertEqual(summary["by_side"]["buy"]["count"], 2)
self.assertEqual(summary["by_side"]["sell"]["losses"], 1)
self.assertEqual(summary["by_symbol"]["BTC-USDT"]["count"], 2)
self.assertEqual(summary["by_symbol"]["ETH-USDT"]["losses"], 1)
self.assertEqual(summary["by_session"]["LONDON"]["count"], 2)
self.assertEqual(summary["by_session"]["NY_AM"]["count"], 2)
self.assertEqual(summary["by_weekday"]["Monday"]["count"], 2)
self.assertEqual(summary["by_weekday"]["Tuesday"]["misses"], 1)
def test_report_service_reuses_backtest_summary(self) -> None:
outcomes = [
BacktestOutcome(
signal_id=9,
outcome="breakeven",
mfe=2.0,
mae=1.2,
tp1_hit=True,
tp2_hit=False,
invalidated_before_entry=False,
error_tags=[],
meta={"signal_side": "buy", "signal_symbol": "BTC-USDT", "session_codes": ["NY_PM"], "weekday": "Friday"},
)
]
service = BacktestReportService()
summary = service.build_summary(outcomes)
self.assertEqual(summary["by_session"]["NY_PM"]["breakevens"], 1)
self.assertEqual(summary["by_side"]["buy"]["count"], 1)
self.assertEqual(summary["by_symbol"]["BTC-USDT"]["breakevens"], 1)
self.assertIn("quality_gate", summary)
self.assertFalse(summary["quality_gate"]["passed"])
def test_report_service_accepts_custom_quality_gate(self) -> None:
outcomes = [
BacktestOutcome(
signal_id=10,
outcome="win",
mfe=2.0,
mae=0.5,
tp1_hit=True,
tp2_hit=True,
invalidated_before_entry=False,
error_tags=[],
meta={"signal_side": "buy", "signal_symbol": "BTC-USDT", "session_codes": ["LONDON"], "weekday": "Friday", "r_multiple": 2.0},
)
]
service = BacktestReportService()
summary = service.build_summary(
outcomes,
quality_gate=BacktestQualityGate(
min_total_signals=1,
min_executed_signals=1,
min_profit_factor_r=1.0,
max_drawdown_r=1.0,
),
)
self.assertTrue(summary["quality_gate"]["passed"])
if __name__ == "__main__":
unittest.main()