yuqei
409 lines
16 KiB
Python
409 lines
16 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 (
|
|
BacktestCostConfig,
|
|
BacktestOutcome,
|
|
BacktestQualityGate,
|
|
BacktestService,
|
|
BacktestSignal,
|
|
build_cumulative_r_curve,
|
|
compute_max_consecutive_losses,
|
|
compute_max_drawdown_r,
|
|
compute_outcome_r_multiple,
|
|
diagnose_backtest_quality,
|
|
estimate_trade_cost_r,
|
|
evaluate_backtest_quality,
|
|
evaluate_signal_outcome,
|
|
summarize_backtest_outcomes,
|
|
)
|
|
from src.repositories.backtest_repository import BacktestRepository
|
|
|
|
|
|
class BacktestOutcomeTests(unittest.TestCase):
|
|
def test_buy_signal_uses_stop_first_when_entry_and_target_are_ambiguous(self) -> None:
|
|
signal = BacktestSignal(
|
|
signal_id=1,
|
|
side="buy",
|
|
entry_low=100.0,
|
|
entry_high=101.0,
|
|
stop_loss=99.0,
|
|
tp1=102.0,
|
|
tp2=104.0,
|
|
)
|
|
outcome = evaluate_signal_outcome(
|
|
signal,
|
|
[{"high": 105.0, "low": 98.0}],
|
|
)
|
|
|
|
self.assertEqual(outcome.outcome, "loss")
|
|
self.assertFalse(outcome.invalidated_before_entry)
|
|
self.assertIn("stopped_out", outcome.error_tags)
|
|
self.assertIn("same_candle_entry_exit_ambiguous", outcome.error_tags)
|
|
self.assertTrue(outcome.meta["same_candle_ambiguous"])
|
|
self.assertEqual(outcome.meta["same_candle_priority"], "stop_first")
|
|
|
|
def test_untriggered_entry_is_missed_without_invalidating_before_entry(self) -> None:
|
|
signal = BacktestSignal(
|
|
signal_id=2,
|
|
side="sell",
|
|
entry_low=100.0,
|
|
entry_high=101.0,
|
|
stop_loss=102.0,
|
|
tp1=99.0,
|
|
tp2=97.0,
|
|
)
|
|
outcome = evaluate_signal_outcome(
|
|
signal,
|
|
[{"high": 99.5, "low": 98.0}],
|
|
)
|
|
|
|
self.assertEqual(outcome.outcome, "missed")
|
|
self.assertFalse(outcome.invalidated_before_entry)
|
|
self.assertIn("entry_not_triggered", outcome.error_tags)
|
|
self.assertFalse(outcome.meta["entry_triggered"])
|
|
|
|
def test_sell_signal_can_reach_tp2(self) -> None:
|
|
signal = BacktestSignal(
|
|
signal_id=3,
|
|
side="sell",
|
|
entry_low=100.0,
|
|
entry_high=101.0,
|
|
stop_loss=103.0,
|
|
tp1=99.0,
|
|
tp2=97.0,
|
|
rr_tp1=0.75,
|
|
rr_tp2=2.5,
|
|
)
|
|
outcome = evaluate_signal_outcome(
|
|
signal,
|
|
[
|
|
{"high": 100.5, "low": 99.5},
|
|
{"high": 100.0, "low": 96.5},
|
|
],
|
|
)
|
|
|
|
self.assertEqual(outcome.outcome, "win")
|
|
self.assertTrue(outcome.tp1_hit)
|
|
self.assertTrue(outcome.tp2_hit)
|
|
self.assertAlmostEqual(outcome.meta["mfe_r"], 1.6)
|
|
self.assertAlmostEqual(outcome.meta["mae_r"], 0.0)
|
|
self.assertEqual(outcome.meta["r_multiple"], 2.5)
|
|
self.assertEqual(outcome.error_tags, [])
|
|
|
|
def test_outcome_r_multiple_applies_cost_to_executed_results(self) -> None:
|
|
self.assertAlmostEqual(compute_outcome_r_multiple("win", rr_tp2=2.5, cost_r=0.1), 2.4)
|
|
self.assertAlmostEqual(compute_outcome_r_multiple("loss", cost_r=0.1), -1.1)
|
|
self.assertEqual(compute_outcome_r_multiple("missed", cost_r=0.1), 0.0)
|
|
|
|
def test_cost_model_reduces_executed_r_multiple(self) -> None:
|
|
signal = BacktestSignal(
|
|
signal_id=4,
|
|
side="sell",
|
|
entry_low=100.0,
|
|
entry_high=101.0,
|
|
stop_loss=103.0,
|
|
tp1=99.0,
|
|
tp2=97.0,
|
|
rr_tp1=0.75,
|
|
rr_tp2=2.5,
|
|
)
|
|
cost_config = BacktestCostConfig(fee_rate=0.0005, slippage_rate=0.0005, fixed_cost_r=0.05)
|
|
expected_cost_r = estimate_trade_cost_r(
|
|
signal=signal,
|
|
outcome="win",
|
|
entry_price=100.5,
|
|
entry_triggered=True,
|
|
cost_config=cost_config,
|
|
)
|
|
|
|
outcome = evaluate_signal_outcome(
|
|
signal,
|
|
[
|
|
{"high": 100.5, "low": 99.5},
|
|
{"high": 100.0, "low": 96.5},
|
|
],
|
|
cost_config=cost_config,
|
|
)
|
|
|
|
self.assertAlmostEqual(expected_cost_r, 0.129)
|
|
self.assertAlmostEqual(outcome.meta["cost_r"], expected_cost_r)
|
|
self.assertAlmostEqual(outcome.meta["r_multiple"], 2.5 - expected_cost_r)
|
|
self.assertEqual(outcome.meta["cost_model"]["fee_rate"], 0.0005)
|
|
|
|
def test_summary_includes_expectancy_and_profit_factor_r(self) -> None:
|
|
outcomes = [
|
|
BacktestOutcome(
|
|
signal_id=1,
|
|
outcome="win",
|
|
mfe=3.0,
|
|
mae=0.5,
|
|
tp1_hit=True,
|
|
tp2_hit=True,
|
|
invalidated_before_entry=False,
|
|
error_tags=[],
|
|
meta={"signal_side": "buy", "session_codes": ["LONDON"], "weekday": "Monday", "mfe_r": 2.6, "mae_r": 0.4, "cost_r": 0.1, "r_multiple": 2.4},
|
|
),
|
|
BacktestOutcome(
|
|
signal_id=2,
|
|
outcome="loss",
|
|
mfe=0.2,
|
|
mae=1.0,
|
|
tp1_hit=False,
|
|
tp2_hit=False,
|
|
invalidated_before_entry=False,
|
|
error_tags=["stopped_out"],
|
|
meta={"signal_side": "buy", "session_codes": ["LONDON"], "weekday": "Monday", "mfe_r": 0.2, "mae_r": 1.2, "cost_r": 0.1, "r_multiple": -1.1},
|
|
),
|
|
BacktestOutcome(
|
|
signal_id=3,
|
|
outcome="loss",
|
|
mfe=0.0,
|
|
mae=0.0,
|
|
tp1_hit=False,
|
|
tp2_hit=False,
|
|
invalidated_before_entry=False,
|
|
error_tags=["stopped_out"],
|
|
meta={"signal_side": "buy", "session_codes": ["LONDON"], "weekday": "Monday", "mfe_r": 0.1, "mae_r": 1.1, "cost_r": 0.1, "r_multiple": -1.1},
|
|
),
|
|
BacktestOutcome(
|
|
signal_id=4,
|
|
outcome="missed",
|
|
mfe=0.0,
|
|
mae=0.0,
|
|
tp1_hit=False,
|
|
tp2_hit=False,
|
|
invalidated_before_entry=False,
|
|
error_tags=["entry_not_triggered"],
|
|
meta={"signal_side": "sell", "session_codes": ["NY_AM"], "weekday": "Tuesday", "mfe_r": 0.0, "mae_r": 0.0, "cost_r": 0.0, "r_multiple": 0.0},
|
|
),
|
|
]
|
|
|
|
summary = summarize_backtest_outcomes(outcomes)
|
|
|
|
self.assertAlmostEqual(summary["total_r"], 0.2)
|
|
self.assertAlmostEqual(summary["expectancy_r"], 0.05)
|
|
self.assertAlmostEqual(summary["profit_factor_r"], 2.4 / 2.2)
|
|
self.assertEqual(summary["executed_signals"], 3)
|
|
self.assertAlmostEqual(summary["executed_rate"], 0.75)
|
|
self.assertAlmostEqual(summary["executed_expectancy_r"], 0.2 / 3)
|
|
self.assertAlmostEqual(summary["executed_profit_factor_r"], 2.4 / 2.2)
|
|
self.assertAlmostEqual(summary["avg_mfe_r"], 0.725)
|
|
self.assertAlmostEqual(summary["avg_mae_r"], 0.675)
|
|
self.assertAlmostEqual(summary["max_drawdown_r"], 2.2)
|
|
self.assertEqual(summary["max_consecutive_losses"], 2)
|
|
self.assertEqual(summary["cumulative_r_curve"], [2.4, 1.2999999999999998, 0.19999999999999973, 0.19999999999999973])
|
|
self.assertAlmostEqual(summary["total_cost_r"], 0.3)
|
|
self.assertAlmostEqual(summary["avg_cost_r"], 0.075)
|
|
self.assertAlmostEqual(summary["by_session"]["LONDON"]["expectancy_r"], 0.2 / 3)
|
|
self.assertAlmostEqual(summary["by_session"]["LONDON"]["avg_mfe_r"], 2.9 / 3)
|
|
self.assertAlmostEqual(summary["by_session"]["LONDON"]["avg_mae_r"], 2.7 / 3)
|
|
self.assertAlmostEqual(summary["by_session"]["LONDON"]["avg_cost_r"], 0.1)
|
|
self.assertAlmostEqual(summary["by_session"]["LONDON"]["max_drawdown_r"], 2.2)
|
|
self.assertEqual(summary["by_session"]["LONDON"]["max_consecutive_losses"], 2)
|
|
self.assertEqual(summary["by_session"]["LONDON"]["executed"], 3)
|
|
self.assertAlmostEqual(summary["by_session"]["LONDON"]["executed_expectancy_r"], 0.2 / 3)
|
|
self.assertAlmostEqual(summary["by_side"]["buy"]["total_r"], 0.2)
|
|
|
|
def test_drawdown_helpers_track_cumulative_r_peak_to_trough(self) -> None:
|
|
values = [1.0, -0.5, 2.0, -3.0, 0.5]
|
|
|
|
self.assertEqual(build_cumulative_r_curve(values), [1.0, 0.5, 2.5, -0.5, 0.0])
|
|
self.assertEqual(compute_max_drawdown_r(values), 3.0)
|
|
|
|
def test_counts_max_consecutive_losses(self) -> None:
|
|
outcomes = [
|
|
BacktestOutcome(1, "loss", 0.0, 1.0, False, False, False, [], {}),
|
|
BacktestOutcome(2, "loss", 0.0, 1.0, False, False, False, [], {}),
|
|
BacktestOutcome(3, "breakeven", 0.0, 0.0, True, False, False, [], {}),
|
|
BacktestOutcome(4, "loss", 0.0, 1.0, False, False, False, [], {}),
|
|
]
|
|
|
|
self.assertEqual(compute_max_consecutive_losses(outcomes), 2)
|
|
|
|
def test_backtest_quality_gate_passes_when_metrics_are_healthy(self) -> None:
|
|
result = evaluate_backtest_quality(
|
|
{
|
|
"total_signals": 120,
|
|
"executed_signals": 80,
|
|
"expectancy_r": 0.15,
|
|
"profit_factor_r": 1.4,
|
|
"max_drawdown_r": 4.5,
|
|
"max_consecutive_losses": 3,
|
|
}
|
|
)
|
|
|
|
self.assertTrue(result["passed"])
|
|
self.assertEqual(result["reasons"], [])
|
|
|
|
def test_backtest_quality_gate_rejects_weak_metrics(self) -> None:
|
|
result = evaluate_backtest_quality(
|
|
{
|
|
"total_signals": 12,
|
|
"executed_signals": 3,
|
|
"expectancy_r": -0.05,
|
|
"profit_factor_r": 0.9,
|
|
"max_drawdown_r": 12.0,
|
|
"max_consecutive_losses": 8,
|
|
},
|
|
gate=BacktestQualityGate(min_total_signals=50, max_drawdown_r=6.0, max_consecutive_losses=4),
|
|
)
|
|
|
|
self.assertFalse(result["passed"])
|
|
self.assertEqual(
|
|
result["reasons"],
|
|
[
|
|
"sample_size_too_small",
|
|
"executed_sample_too_small",
|
|
"expectancy_below_threshold",
|
|
"profit_factor_below_threshold",
|
|
"drawdown_above_threshold",
|
|
"consecutive_losses_above_threshold",
|
|
],
|
|
)
|
|
|
|
def test_quality_diagnostics_identifies_weak_segments_and_actions(self) -> None:
|
|
summary = {
|
|
"total_signals": 100,
|
|
"executed_signals": 80,
|
|
"expectancy_r": -0.08,
|
|
"profit_factor_r": 0.9,
|
|
"max_drawdown_r": 12.0,
|
|
"max_consecutive_losses": 8,
|
|
"by_session": {
|
|
"NY_AM": {"count": 12, "executed": 11, "expectancy_r": -0.9, "max_drawdown_r": 11.0, "max_consecutive_losses": 8},
|
|
"LONDON": {"count": 30, "executed": 25, "expectancy_r": 0.05, "max_drawdown_r": 4.0, "max_consecutive_losses": 2},
|
|
},
|
|
"by_side": {
|
|
"buy": {"count": 80, "executed": 70, "expectancy_r": -0.1, "max_drawdown_r": 13.0, "max_consecutive_losses": 8},
|
|
},
|
|
"by_weekday": {},
|
|
}
|
|
gate = evaluate_backtest_quality(summary)
|
|
|
|
diagnostics = diagnose_backtest_quality(summary, gate)
|
|
|
|
self.assertEqual(diagnostics["quality_state"], "blocked")
|
|
self.assertEqual(diagnostics["weak_segments"]["session"][0]["name"], "NY_AM")
|
|
self.assertIn("tighten_entry_filters_around_negative_expectancy_segments", diagnostics["recommended_actions"])
|
|
self.assertIn("add_session_side_or_weekday_risk_filters_before_execution", diagnostics["recommended_actions"])
|
|
|
|
|
|
class BacktestServiceRunTests(unittest.TestCase):
|
|
def test_run_backtest_persists_quality_gate_in_summary(self) -> None:
|
|
case = self
|
|
|
|
class FakeBacktestRepository:
|
|
def __init__(self) -> None:
|
|
self.summary = {}
|
|
self.results = []
|
|
self.run_config = {}
|
|
|
|
def fetch_candidate_signals(self, instrument_id: int, limit: int):
|
|
case.assertEqual(instrument_id, 1)
|
|
case.assertEqual(limit, 10)
|
|
return [
|
|
{
|
|
"id": 7,
|
|
"created_ts": datetime(2026, 4, 29, 9, 0, tzinfo=timezone.utc),
|
|
"expires_ts": datetime(2026, 4, 29, 10, 0, tzinfo=timezone.utc),
|
|
"side": "buy",
|
|
"entry_low": 100.0,
|
|
"entry_high": 101.0,
|
|
"stop_loss": 99.0,
|
|
"tp1": 102.0,
|
|
"tp2": 104.0,
|
|
"rr_tp1": 1.0,
|
|
"rr_tp2": 2.0,
|
|
}
|
|
]
|
|
|
|
def insert_backtest_run(self, **kwargs):
|
|
self.run_config = kwargs["config"]
|
|
return 42
|
|
|
|
def fetch_sessions_covering_window(self, **kwargs):
|
|
return [{"session_code": "LONDON"}]
|
|
|
|
def fetch_future_candles(self, **kwargs):
|
|
return [{"high": 104.5, "low": 100.0}]
|
|
|
|
def insert_backtest_result(self, **kwargs):
|
|
self.results.append(kwargs)
|
|
|
|
def update_backtest_run_summary(self, backtest_run_id: int, summary: dict) -> None:
|
|
case.assertEqual(backtest_run_id, 42)
|
|
self.summary = summary
|
|
|
|
repository = FakeBacktestRepository()
|
|
service = BacktestService(repository=repository)
|
|
|
|
run_id, outcomes = service.run_backtest(instrument_id=1, limit=10)
|
|
|
|
self.assertEqual(run_id, 42)
|
|
self.assertEqual(len(outcomes), 1)
|
|
self.assertEqual(len(repository.results), 1)
|
|
self.assertIn("quality_gate", repository.summary)
|
|
self.assertIn("quality_diagnostics", repository.summary)
|
|
self.assertFalse(repository.summary["quality_gate"]["passed"])
|
|
self.assertEqual(repository.summary["quality_diagnostics"]["quality_state"], "blocked")
|
|
self.assertIn("sample_size_too_small", repository.summary["quality_gate"]["reasons"])
|
|
self.assertIn("executed_sample_too_small", repository.summary["quality_gate"]["reasons"])
|
|
self.assertEqual(repository.run_config["quality_gate"]["min_total_signals"], 100)
|
|
self.assertEqual(repository.run_config["quality_gate"]["min_executed_signals"], 30)
|
|
|
|
|
|
class BacktestRepositorySerializationTests(unittest.TestCase):
|
|
def test_insert_backtest_result_serializes_datetime_in_meta(self) -> None:
|
|
payload = {
|
|
"signal_created_ts": datetime(2026, 4, 24, 12, 0, tzinfo=timezone.utc),
|
|
"weekday": "Friday",
|
|
}
|
|
|
|
class StubSession:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def execute(self, _sql, params):
|
|
captured.update(params)
|
|
|
|
class StubBegin:
|
|
def __enter__(self):
|
|
return StubSession()
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
captured: dict = {}
|
|
original_begin = BacktestRepository.insert_backtest_result.__globals__["SessionLocal"].begin
|
|
try:
|
|
BacktestRepository.insert_backtest_result.__globals__["SessionLocal"].begin = lambda: StubBegin()
|
|
repo = BacktestRepository()
|
|
repo.insert_backtest_result(
|
|
backtest_run_id=1,
|
|
signal_id=1,
|
|
outcome="missed",
|
|
mfe=0.0,
|
|
mae=0.0,
|
|
tp1_hit=False,
|
|
tp2_hit=False,
|
|
invalidated_before_entry=False,
|
|
error_tags=[],
|
|
meta=payload,
|
|
)
|
|
finally:
|
|
BacktestRepository.insert_backtest_result.__globals__["SessionLocal"].begin = original_begin
|
|
|
|
self.assertIn("meta", captured)
|
|
self.assertIn("2026-04-24 12:00:00+00:00", captured["meta"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|