Codex
1003 lines
45 KiB
Python
1003 lines
45 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,
|
|
BACKTEST_SUBJECT_PLAN_REPLAY,
|
|
build_timeframe_bucket,
|
|
classify_backtest_lifecycle_state,
|
|
BacktestOutcome,
|
|
BacktestQualityGate,
|
|
BacktestReplayConfig,
|
|
BacktestService,
|
|
BacktestSignal,
|
|
build_cumulative_r_curve,
|
|
compute_max_consecutive_losses,
|
|
compute_max_drawdown_r,
|
|
compute_outcome_r_multiple,
|
|
build_plan_replay_units,
|
|
estimate_trade_cost_r,
|
|
evaluate_backtest_quality,
|
|
evaluate_signal_outcome,
|
|
evaluate_signal_outcome_with_replay,
|
|
normalize_timeframe_set,
|
|
summarize_backtest_outcomes,
|
|
)
|
|
from src.repositories.backtest_repository import BacktestRepository
|
|
|
|
|
|
class BacktestOutcomeTests(unittest.TestCase):
|
|
def test_classifies_backtest_lifecycle_states(self) -> None:
|
|
self.assertEqual(
|
|
classify_backtest_lifecycle_state(outcome="win", tp1_hit=True, tp2_hit=True, invalidated_before_entry=False),
|
|
"tp2_closed",
|
|
)
|
|
self.assertEqual(
|
|
classify_backtest_lifecycle_state(outcome="breakeven", tp1_hit=True, tp2_hit=False, invalidated_before_entry=False),
|
|
"tp1_reached",
|
|
)
|
|
self.assertEqual(
|
|
classify_backtest_lifecycle_state(outcome="loss", tp1_hit=False, tp2_hit=False, invalidated_before_entry=False),
|
|
"stopped_out",
|
|
)
|
|
self.assertEqual(
|
|
classify_backtest_lifecycle_state(outcome="missed", tp1_hit=False, tp2_hit=False, invalidated_before_entry=False),
|
|
"missed",
|
|
)
|
|
self.assertEqual(
|
|
classify_backtest_lifecycle_state(outcome="missed", tp1_hit=False, tp2_hit=False, invalidated_before_entry=True),
|
|
"invalidated_before_entry",
|
|
)
|
|
|
|
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)
|
|
self.assertAlmostEqual(
|
|
compute_outcome_r_multiple(
|
|
"breakeven",
|
|
rr_tp1=1.0,
|
|
cost_r=0.1,
|
|
tp1_hit=True,
|
|
tp1_partial_fraction=0.5,
|
|
stopped_after_tp1=True,
|
|
move_stop_applied=True,
|
|
),
|
|
0.4,
|
|
)
|
|
|
|
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_tp1_partial_and_move_stop_affect_backtest_r_multiple(self) -> None:
|
|
signal = BacktestSignal(
|
|
signal_id=40,
|
|
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.3333333333,
|
|
management_rules=[
|
|
{"code": "tp1_partial", "fraction": 0.5},
|
|
{"code": "move_stop", "to": "breakeven"},
|
|
],
|
|
)
|
|
|
|
outcome = evaluate_signal_outcome(
|
|
signal,
|
|
[
|
|
{"high": 102.1, "low": 100.2},
|
|
{"high": 101.0, "low": 100.4},
|
|
],
|
|
)
|
|
|
|
self.assertEqual(outcome.outcome, "breakeven")
|
|
self.assertTrue(outcome.tp1_hit)
|
|
self.assertFalse(outcome.tp2_hit)
|
|
self.assertAlmostEqual(outcome.meta["r_multiple"], 0.5)
|
|
self.assertTrue(outcome.meta["management_applied"]["tp1_partial"])
|
|
self.assertTrue(outcome.meta["management_applied"]["move_stop"])
|
|
self.assertTrue(outcome.meta["management_applied"]["remaining_position_closed"])
|
|
|
|
def test_replay_no_lookahead_tracks_visible_candles_and_confirmed_swings_at_entry(self) -> None:
|
|
signal = BacktestSignal(
|
|
signal_id=5,
|
|
side="buy",
|
|
entry_low=96.0,
|
|
entry_high=97.0,
|
|
stop_loss=95.0,
|
|
tp1=100.0,
|
|
tp2=103.0,
|
|
rr_tp1=1.0,
|
|
rr_tp2=3.0,
|
|
)
|
|
|
|
outcome = evaluate_signal_outcome_with_replay(
|
|
signal,
|
|
[
|
|
{"ts_open": 1, "ts_close": 2, "open": 97.7, "high": 98.0, "low": 97.5, "close": 97.8},
|
|
{"ts_open": 2, "ts_close": 3, "open": 97.0, "high": 99.0, "low": 96.2, "close": 98.0},
|
|
{"ts_open": 3, "ts_close": 4, "open": 97.2, "high": 98.0, "low": 96.5, "close": 97.4},
|
|
{"ts_open": 4, "ts_close": 5, "open": 98.5, "high": 103.5, "low": 97.0, "close": 103.0},
|
|
],
|
|
replay_config=BacktestReplayConfig(left_bars=1, right_bars=1),
|
|
)
|
|
|
|
self.assertEqual(outcome.outcome, "win")
|
|
self.assertEqual(outcome.meta["backtest_mode"], "replay_no_lookahead")
|
|
self.assertEqual(outcome.meta["replay_entry_frame_index"], 1)
|
|
self.assertEqual(outcome.meta["replay_known_candles_at_entry"], 2)
|
|
self.assertEqual(outcome.meta["replay_confirmed_swing_count_at_entry"], 0)
|
|
self.assertEqual(outcome.meta["replay_confirmed_swing_count_total"], 2)
|
|
self.assertEqual(outcome.meta["replay_last_confirmed_swing_at_entry"], None)
|
|
self.assertEqual(outcome.meta["replay_exit_frame_index"], 3)
|
|
self.assertEqual(outcome.meta["replay_left_bars"], 1)
|
|
self.assertEqual(outcome.meta["replay_right_bars"], 1)
|
|
|
|
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", "signal_symbol": "BTC-USDT", "signal_timeframe_bucket": "15m/5m/1m", "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", "signal_symbol": "BTC-USDT", "signal_timeframe_bucket": "15m/5m/1m", "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", "signal_symbol": "BTC-USDT", "signal_timeframe_bucket": "15m/5m/1m", "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", "signal_symbol": "ETH-USDT", "signal_timeframe_bucket": "1h/15m/5m", "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.assertEqual(summary["lifecycle_states"]["tp2_closed"], 1)
|
|
self.assertEqual(summary["lifecycle_states"]["stopped_out"], 2)
|
|
self.assertEqual(summary["lifecycle_states"]["missed"], 1)
|
|
self.assertEqual(summary["by_symbol"]["BTC-USDT"]["count"], 3)
|
|
self.assertEqual(summary["by_symbol"]["ETH-USDT"]["misses"], 1)
|
|
self.assertEqual(summary["by_timeframe"]["15m/5m/1m"]["count"], 3)
|
|
self.assertEqual(summary["by_timeframe"]["1h/15m/5m"]["misses"], 1)
|
|
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["pd_array_reaction_status_counts"]["study_only"], 4)
|
|
self.assertEqual(summary["pd_array_reaction_reason_counts"]["missing_pd_array"], 4)
|
|
self.assertEqual(summary["by_pd_array_type"]["missing_pd_array"]["count"], 4)
|
|
self.assertEqual(summary["time_session_kernel_status_counts"]["study_only"], 4)
|
|
self.assertEqual(summary["time_session_kernel_reason_counts"]["missing_session_window"], 4)
|
|
self.assertEqual(summary["by_time_session_code"]["LONDON"]["count"], 3)
|
|
self.assertEqual(summary["model_playbook_kernel_status_counts"]["study_only"], 4)
|
|
self.assertEqual(summary["model_playbook_kernel_reason_counts"]["missing_model_playbook"], 4)
|
|
self.assertEqual(summary["by_model_playbook"]["missing_model_playbook"]["count"], 4)
|
|
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_plan_replay_quality_gate_uses_executed_sample_basis(self) -> None:
|
|
result = evaluate_backtest_quality(
|
|
{
|
|
"subject_type": BACKTEST_SUBJECT_PLAN_REPLAY,
|
|
"total_signals": 2,
|
|
"executed_signals": 35,
|
|
"expectancy_r": 0.2,
|
|
"profit_factor_r": 1.6,
|
|
"max_drawdown_r": 4.0,
|
|
"max_consecutive_losses": 2,
|
|
},
|
|
gate=BacktestQualityGate(min_total_signals=100, min_executed_signals=30),
|
|
sample_basis="executed",
|
|
)
|
|
|
|
self.assertTrue(result["passed"])
|
|
self.assertEqual(result["reasons"], [])
|
|
self.assertEqual(result["sample_basis"], "executed")
|
|
|
|
def test_build_plan_replay_units_tracks_appearance_and_confirmation_timeline(self) -> None:
|
|
rows = [
|
|
{
|
|
"candidate_id": 1,
|
|
"trading_plan_id": 77,
|
|
"candidate_model_code": "OSOK",
|
|
"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
|
|
"side": "buy",
|
|
"setup_timeframe": "15m",
|
|
"trigger_timeframe": "5m",
|
|
"created_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc),
|
|
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
|
|
"confirmation_state": {"executable": False},
|
|
"missing_confirmations": ["sweep_confirmed"],
|
|
"entry": {},
|
|
"candidate_meta": {},
|
|
"candidate_status": "awaiting_confirmation",
|
|
"session_code": "LONDON",
|
|
"allowed_sessions": ["LONDON"],
|
|
},
|
|
{
|
|
"candidate_id": 2,
|
|
"trading_plan_id": 77,
|
|
"candidate_model_code": "OSOK",
|
|
"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
|
|
"side": "buy",
|
|
"setup_timeframe": "15m",
|
|
"trigger_timeframe": "5m",
|
|
"created_ts": datetime(2026, 5, 6, 0, 5, tzinfo=timezone.utc),
|
|
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
|
|
"confirmation_state": {"executable": True},
|
|
"missing_confirmations": [],
|
|
"liquidity_target_side": "buy_side_liquidity",
|
|
"working_draw": "upper_external_pool",
|
|
"liquidity_pool_id": "bsl-old-high",
|
|
"liquidity_pool_rank": "primary",
|
|
"liquidity_event": "sweep",
|
|
"swept_level": 103.0,
|
|
"sweep_confirmed": True,
|
|
"post_take_reaction": "acceptance_displacement",
|
|
"pd_array_type": "fvg",
|
|
"pd_array_id": "fvg-1",
|
|
"pd_array_role": "entry",
|
|
"pd_array_rank": "primary",
|
|
"array_location": "discount",
|
|
"array_timeframe": "5m",
|
|
"array_freshness": "fresh",
|
|
"formation_event": "post_sweep_displacement",
|
|
"displacement_origin": "post_sweep_displacement",
|
|
"fvg_role": "entry",
|
|
"fvg_retrace_confirmed": True,
|
|
"reaction_quality": "rejection_confirmed",
|
|
"retest_confirmed": True,
|
|
"pd_array_invalidation": "fvg_accepted_other_side",
|
|
"entry_low": 100.0,
|
|
"entry_high": 100.2,
|
|
"stop_loss": 99.0,
|
|
"tp1": 101.0,
|
|
"tp2": 103.0,
|
|
"management_rules": [{"code": "tp1_partial", "fraction": 0.5}],
|
|
"entry": {
|
|
"entry_low": 100.0,
|
|
"entry_high": 100.2,
|
|
"stop_loss": 99.0,
|
|
"tp1": 101.0,
|
|
"tp2": 103.0,
|
|
"target_plan": {
|
|
"status": "ready",
|
|
"tp1": {"price": 101.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
|
|
"tp2": {"price": 103.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
|
|
},
|
|
"management_rules": [{"code": "tp1_partial", "fraction": 0.5}],
|
|
},
|
|
"candidate_status": "executable",
|
|
"session_code": "LONDON",
|
|
"session_window_state": "inside",
|
|
"allowed_sessions": ["LONDON"],
|
|
"killzone_name": "LONDON_KILLZONE",
|
|
"killzone_state": "active",
|
|
"judas_window_state": "cleared",
|
|
"opening_range_state": "defined",
|
|
"opening_range_id": "or-london",
|
|
"opening_range_high": 101.0,
|
|
"opening_range_low": 99.0,
|
|
"weekly_profile": "midweek_expansion",
|
|
"weekly_bias": "bullish_draw_to_bsl",
|
|
"news_window_state": "clear",
|
|
"high_impact_news": False,
|
|
"low_liquidity_window": False,
|
|
"session_invalidation": "london_close_or_reclaim",
|
|
},
|
|
]
|
|
|
|
units = build_plan_replay_units(rows)
|
|
|
|
self.assertEqual(len(units), 1)
|
|
self.assertEqual(units[0]["candidate_id"], 2)
|
|
self.assertEqual(units[0]["appeared_ts"], rows[0]["created_ts"])
|
|
self.assertEqual(units[0]["confirmed_ts"], rows[1]["created_ts"])
|
|
self.assertAlmostEqual(units[0]["rr_tp1"], 0.8181818181)
|
|
self.assertEqual(units[0]["target_plan"]["tp2"]["objective_layer"], "ERL")
|
|
self.assertEqual(units[0]["liquidity_backtest_features"]["liquidity_target_side"], "buy_side_liquidity")
|
|
self.assertEqual(units[0]["liquidity_kernel_status"]["status"], "liquidity_mapped")
|
|
self.assertEqual(units[0]["pd_array_backtest_features"]["pd_array_type"], "fvg")
|
|
self.assertEqual(units[0]["pd_array_reaction_status"]["status"], "pd_array_mapped")
|
|
self.assertEqual(units[0]["time_session_backtest_features"]["session_code"], "LONDON")
|
|
self.assertEqual(units[0]["time_session_kernel_status"]["status"], "time_session_mapped")
|
|
self.assertEqual(units[0]["model_playbook_backtest_features"]["model_playbook_id"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL")
|
|
self.assertEqual(units[0]["model_playbook_kernel_status"]["status"], "model_playbook_mapped")
|
|
|
|
|
|
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, timeframe_set=None, session_codes=None):
|
|
case.assertEqual(instrument_id, 1)
|
|
case.assertEqual(limit, 10)
|
|
case.assertEqual(timeframe_set, {"bias_tf": "1m", "setup_tf": "1m", "trigger_tf": "1m"})
|
|
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,
|
|
"symbol": "BTC-USDT",
|
|
"bias_tf": "1m",
|
|
"setup_tf": "1m",
|
|
"trigger_tf": "1m",
|
|
"target_plan": {
|
|
"status": "ready",
|
|
"tp1": {"price": 102.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
|
|
"tp2": {"price": 104.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
|
|
"management_rules": [{"code": "tp1_partial"}],
|
|
},
|
|
"signal_meta": {
|
|
"target_plan": {
|
|
"status": "ready",
|
|
"tp1": {"price": 102.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
|
|
"tp2": {"price": 104.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
|
|
"management_rules": [{"code": "tp1_partial"}],
|
|
},
|
|
"management_rules": [{"code": "tp1_partial"}],
|
|
},
|
|
}
|
|
]
|
|
|
|
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.assertEqual(repository.results[0]["meta"]["signal_symbol"], "BTC-USDT")
|
|
self.assertEqual(repository.results[0]["meta"]["signal_timeframe_bucket"], "1m/1m/1m")
|
|
self.assertEqual(repository.results[0]["meta"]["target_plan"]["tp1"]["objective_layer"], "IRL")
|
|
self.assertEqual(repository.results[0]["meta"]["management_rules"][0]["code"], "tp1_partial")
|
|
self.assertEqual(repository.summary["by_symbol"]["BTC-USDT"]["count"], 1)
|
|
self.assertEqual(repository.summary["by_timeframe"]["1m/1m/1m"]["count"], 1)
|
|
self.assertEqual(repository.results[0]["meta"]["session_codes"], ["LONDON"])
|
|
self.assertIn("quality_gate", repository.summary)
|
|
self.assertFalse(repository.summary["quality_gate"]["passed"])
|
|
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)
|
|
|
|
def test_run_backtest_preserves_signal_session_code_when_no_session_window_matches(self) -> None:
|
|
case = self
|
|
|
|
class FakeBacktestRepository:
|
|
def __init__(self) -> None:
|
|
self.results = []
|
|
|
|
def fetch_candidate_signals(self, instrument_id: int, limit: int, timeframe_set=None, session_codes=None):
|
|
return [
|
|
{
|
|
"id": 7,
|
|
"created_ts": datetime(2026, 4, 29, 21, 0, tzinfo=timezone.utc),
|
|
"expires_ts": datetime(2026, 4, 29, 22, 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,
|
|
"symbol": "BTC-USDT",
|
|
"bias_tf": "1m",
|
|
"setup_tf": "1m",
|
|
"trigger_tf": "1m",
|
|
"session_code": "OFF_HOURS",
|
|
"target_plan": {},
|
|
"signal_meta": {},
|
|
}
|
|
]
|
|
|
|
def insert_backtest_run(self, **kwargs):
|
|
return 99
|
|
|
|
def fetch_sessions_covering_window(self, **kwargs):
|
|
return []
|
|
|
|
def fetch_future_candles(self, **kwargs):
|
|
return []
|
|
|
|
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, 99)
|
|
|
|
repository = FakeBacktestRepository()
|
|
service = BacktestService(repository=repository)
|
|
|
|
run_id, outcomes = service.run_backtest(instrument_id=1, limit=1)
|
|
|
|
self.assertEqual(run_id, 99)
|
|
self.assertEqual(len(outcomes), 1)
|
|
self.assertEqual(repository.results[0]["meta"]["session_codes"], ["OFF_HOURS"])
|
|
self.assertEqual(repository.results[0]["meta"]["session_filter"], "outside_window")
|
|
self.assertIn("outside_session_window", repository.results[0]["error_tags"])
|
|
|
|
def test_run_backtest_persists_replay_mode_and_replay_meta(self) -> None:
|
|
case = self
|
|
|
|
class FakeBacktestRepository:
|
|
def __init__(self) -> None:
|
|
self.results = []
|
|
self.run_config = {}
|
|
|
|
def fetch_candidate_signals(self, instrument_id: int, limit: int, timeframe_set=None, session_codes=None):
|
|
case.assertEqual(instrument_id, 1)
|
|
case.assertEqual(limit, 10)
|
|
return [
|
|
{
|
|
"id": 8,
|
|
"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": 96.0,
|
|
"entry_high": 97.0,
|
|
"stop_loss": 95.0,
|
|
"tp1": 100.0,
|
|
"tp2": 103.0,
|
|
"rr_tp1": 1.0,
|
|
"rr_tp2": 3.0,
|
|
"symbol": "BTC-USDT",
|
|
"bias_tf": "1m",
|
|
"setup_tf": "1m",
|
|
"trigger_tf": "1m",
|
|
}
|
|
]
|
|
|
|
def insert_backtest_run(self, **kwargs):
|
|
self.run_config = kwargs["config"]
|
|
return 77
|
|
|
|
def fetch_sessions_covering_window(self, **kwargs):
|
|
return [{"session_code": "LONDON"}]
|
|
|
|
def fetch_future_candles(self, **kwargs):
|
|
return [
|
|
{"ts_open": 1, "ts_close": 2, "open": 97.7, "high": 98.0, "low": 97.5, "close": 97.8},
|
|
{"ts_open": 2, "ts_close": 3, "open": 97.0, "high": 99.0, "low": 96.2, "close": 98.0},
|
|
{"ts_open": 3, "ts_close": 4, "open": 97.2, "high": 98.0, "low": 96.5, "close": 97.4},
|
|
{"ts_open": 4, "ts_close": 5, "open": 98.5, "high": 103.5, "low": 97.0, "close": 103.0},
|
|
]
|
|
|
|
def insert_backtest_result(self, **kwargs):
|
|
self.results.append(kwargs)
|
|
|
|
def update_backtest_run_summary(self, backtest_run_id: int, summary: dict) -> None:
|
|
return None
|
|
|
|
repository = FakeBacktestRepository()
|
|
service = BacktestService(repository=repository)
|
|
|
|
run_id, outcomes = service.run_backtest(
|
|
instrument_id=1,
|
|
limit=10,
|
|
mode="replay_no_lookahead",
|
|
replay_config=BacktestReplayConfig(left_bars=1, right_bars=1),
|
|
)
|
|
|
|
self.assertEqual(run_id, 77)
|
|
self.assertEqual(len(outcomes), 1)
|
|
self.assertEqual(repository.run_config["mode"], "replay_no_lookahead")
|
|
self.assertEqual(repository.run_config["replay"]["left_bars"], 1)
|
|
self.assertEqual(repository.run_config["replay"]["right_bars"], 1)
|
|
self.assertEqual(repository.results[0]["meta"]["backtest_mode"], "replay_no_lookahead")
|
|
self.assertEqual(repository.results[0]["meta"]["replay_confirmed_swing_count_total"], 2)
|
|
|
|
def test_run_backtest_supports_plan_replay_subject_and_plan_metrics(self) -> None:
|
|
case = self
|
|
|
|
class FakePlanReplayRepository:
|
|
def __init__(self) -> None:
|
|
self.summary = {}
|
|
self.results = []
|
|
self.run_config = {}
|
|
|
|
def fetch_active_trading_plan(self, instrument_id: int, trading_plan_id: int = None):
|
|
return {"id": 77}
|
|
|
|
def insert_backtest_run(self, **kwargs):
|
|
self.run_config = kwargs
|
|
return 501
|
|
|
|
def fetch_plan_replay_candidate_rows(self, **kwargs):
|
|
case.assertEqual(kwargs["trading_plan_id"], 77)
|
|
return [
|
|
{
|
|
"candidate_id": 11,
|
|
"trading_plan_id": 77,
|
|
"candidate_model_code": "OSOK",
|
|
"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
|
|
"candidate_status": "awaiting_confirmation",
|
|
"side": "buy",
|
|
"setup_timeframe": "15m",
|
|
"trigger_timeframe": "5m",
|
|
"created_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc),
|
|
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
|
|
"confirmation_state": {"executable": False},
|
|
"missing_confirmations": ["sweep_confirmed"],
|
|
"entry": {},
|
|
"invalidations": [],
|
|
"candidate_meta": {},
|
|
"session_code": "LONDON",
|
|
"allowed_sessions": ["LONDON"],
|
|
"primary_draw": {"side": "buy_side"},
|
|
},
|
|
{
|
|
"candidate_id": 12,
|
|
"trading_plan_id": 77,
|
|
"candidate_model_code": "OSOK",
|
|
"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
|
|
"candidate_status": "executable",
|
|
"side": "buy",
|
|
"setup_timeframe": "15m",
|
|
"trigger_timeframe": "5m",
|
|
"created_ts": datetime(2026, 5, 6, 0, 5, tzinfo=timezone.utc),
|
|
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
|
|
"confirmation_state": {"executable": True},
|
|
"missing_confirmations": [],
|
|
"pd_array_type": "fvg",
|
|
"pd_array_id": "fvg-12",
|
|
"pd_array_role": "entry",
|
|
"pd_array_rank": "primary",
|
|
"array_location": "discount",
|
|
"array_timeframe": "5m",
|
|
"array_freshness": "fresh",
|
|
"formation_event": "post_sweep_displacement",
|
|
"displacement_origin": "post_sweep_displacement",
|
|
"fvg_role": "entry",
|
|
"fvg_retrace_confirmed": True,
|
|
"reaction_quality": "rejection_confirmed",
|
|
"retest_confirmed": True,
|
|
"pd_array_invalidation": "fvg_accepted_other_side",
|
|
"entry": {
|
|
"entry_low": 100.0,
|
|
"entry_high": 100.2,
|
|
"stop_loss": 99.0,
|
|
"tp1": 101.0,
|
|
"tp2": 103.0,
|
|
"target_plan": {
|
|
"status": "ready",
|
|
"tp1": {"price": 101.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
|
|
"tp2": {"price": 103.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
|
|
},
|
|
"management_rules": [
|
|
{"code": "tp1_partial", "fraction": 0.5},
|
|
{"code": "move_stop", "to": "breakeven"},
|
|
],
|
|
},
|
|
"invalidations": [],
|
|
"session_code": "LONDON",
|
|
"session_window_state": "inside",
|
|
"candidate_meta": {
|
|
"management_rules": [
|
|
{"code": "tp1_partial", "fraction": 0.5},
|
|
{"code": "move_stop", "to": "breakeven"},
|
|
]
|
|
},
|
|
"allowed_sessions": ["LONDON"],
|
|
"killzone_name": "LONDON_KILLZONE",
|
|
"killzone_state": "active",
|
|
"judas_window_state": "cleared",
|
|
"opening_range_state": "defined",
|
|
"opening_range_id": "or-london",
|
|
"opening_range_high": 101.0,
|
|
"opening_range_low": 99.0,
|
|
"weekly_profile": "midweek_expansion",
|
|
"weekly_bias": "bullish_draw_to_bsl",
|
|
"news_window_state": "clear",
|
|
"high_impact_news": False,
|
|
"low_liquidity_window": False,
|
|
"session_invalidation": "london_close_or_reclaim",
|
|
"primary_draw": {"side": "buy_side"},
|
|
},
|
|
]
|
|
|
|
def fetch_sessions_covering_window(self, **kwargs):
|
|
return [{"session_code": "LONDON"}]
|
|
|
|
def fetch_future_candles(self, **kwargs):
|
|
return [
|
|
{"ts_open": 1, "ts_close": 2, "high": 101.2, "low": 100.0, "open": 100.1, "close": 101.0},
|
|
{"ts_open": 2, "ts_close": 3, "high": 101.0, "low": 99.9, "open": 100.9, "close": 100.1},
|
|
]
|
|
|
|
def insert_backtest_result(self, **kwargs):
|
|
self.results.append(kwargs)
|
|
|
|
def update_backtest_run_summary(self, backtest_run_id: int, summary: dict) -> None:
|
|
self.summary = summary
|
|
|
|
repository = FakePlanReplayRepository()
|
|
service = BacktestService(repository=repository)
|
|
|
|
run_id, outcomes = service.run_backtest(
|
|
instrument_id=1,
|
|
subject_type=BACKTEST_SUBJECT_PLAN_REPLAY,
|
|
trading_plan_id=77,
|
|
timeframe_set={"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
|
|
)
|
|
|
|
self.assertEqual(run_id, 501)
|
|
self.assertEqual(len(outcomes), 1)
|
|
self.assertEqual(repository.run_config["subject_type"], BACKTEST_SUBJECT_PLAN_REPLAY)
|
|
self.assertEqual(repository.results[0]["trading_plan_id"], 77)
|
|
self.assertEqual(repository.results[0]["candidate_id"], 12)
|
|
self.assertEqual(repository.summary["subject_type"], BACKTEST_SUBJECT_PLAN_REPLAY)
|
|
self.assertEqual(repository.summary["plan_count"], 1)
|
|
self.assertEqual(repository.summary["confirmed_candidates"], 1)
|
|
self.assertAlmostEqual(repository.summary["candidate_confirmation_rate"], 1.0)
|
|
self.assertIn("draw_target_hit_rate", repository.summary)
|
|
self.assertEqual(
|
|
repository.summary["doctrine_backtest_cohort"]["cohort_key"],
|
|
"ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete",
|
|
)
|
|
self.assertEqual(
|
|
repository.summary["by_doctrine_cohort"]["ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete"]["count"],
|
|
1,
|
|
)
|
|
self.assertEqual(repository.summary["pd_array_reaction_status"]["status"], "pd_array_mapped")
|
|
self.assertEqual(repository.summary["by_pd_array_type"]["fvg"]["count"], 1)
|
|
self.assertEqual(repository.summary["time_session_kernel_status"]["status"], "time_session_mapped")
|
|
self.assertEqual(repository.summary["by_time_session_code"]["LONDON"]["count"], 1)
|
|
self.assertEqual(repository.summary["model_playbook_kernel_status"]["status"], "model_playbook_mapped")
|
|
self.assertEqual(
|
|
repository.summary["by_model_playbook"]["ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"]["count"],
|
|
1,
|
|
)
|
|
|
|
def test_normalize_timeframe_set_and_bucket_helpers(self) -> None:
|
|
normalized = normalize_timeframe_set({"bias_tf": "1h", "setup_tf": "15m"})
|
|
|
|
self.assertEqual(normalized, {"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "15m"})
|
|
self.assertEqual(build_timeframe_bucket(**normalized), "1h/15m/15m")
|
|
|
|
|
|
class BacktestRepositoryQueryTests(unittest.TestCase):
|
|
def test_fetch_candidate_signals_keeps_rows_without_bias_snapshot_under_timeframe_filter(self) -> None:
|
|
captured = {}
|
|
|
|
class StubResult:
|
|
def mappings(self):
|
|
return self
|
|
|
|
def all(self):
|
|
return [{"id": 1, "bias_tf": None, "setup_tf": None, "trigger_tf": None}]
|
|
|
|
class StubSession:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def execute(self, sql, params):
|
|
captured["sql"] = str(sql)
|
|
captured["params"] = dict(params)
|
|
return StubResult()
|
|
|
|
original_session_local = BacktestRepository.fetch_candidate_signals.__globals__["SessionLocal"]
|
|
try:
|
|
BacktestRepository.fetch_candidate_signals.__globals__["SessionLocal"] = lambda: StubSession()
|
|
repo = BacktestRepository()
|
|
rows = repo.fetch_candidate_signals(
|
|
instrument_id=1,
|
|
limit=5,
|
|
timeframe_set={"bias_tf": "1m", "setup_tf": "1m", "trigger_tf": "1m"},
|
|
)
|
|
finally:
|
|
BacktestRepository.fetch_candidate_signals.__globals__["SessionLocal"] = original_session_local
|
|
|
|
self.assertEqual(rows[0]["id"], 1)
|
|
self.assertIn("ts.bias_snapshot_id is null or bs.bias_tf = :bias_tf", captured["sql"])
|
|
self.assertIn("ts.bias_snapshot_id is null or bs.setup_tf = :setup_tf", captured["sql"])
|
|
self.assertIn("ts.bias_snapshot_id is null or bs.trigger_tf = :trigger_tf", captured["sql"])
|
|
self.assertIn("ts.status = 'awaiting_confirmation'", captured["sql"])
|
|
self.assertIn("coalesce((ts.meta->>'execution_permission')::boolean, false) = true", captured["sql"])
|
|
self.assertIn("not coalesce(ts.invalidations, '[]'::jsonb) ? 'session_mismatch'", captured["sql"])
|
|
self.assertEqual(captured["params"]["bias_tf"], "1m")
|
|
self.assertEqual(captured["params"]["setup_tf"], "1m")
|
|
self.assertEqual(captured["params"]["trigger_tf"], "1m")
|
|
|
|
|
|
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()
|