"""Unit tests for libs/backtest/tracker.py — SQS computation & journal I/O.""" from __future__ import annotations import datetime as dt import json import multiprocessing import time from pathlib import Path import pytest from libs.backtest.domain import ( CommonWindowSummary, JournalEntry, MetricsBundle, RobustnessHorizonSummary, RobustnessMatrixSummary, SplitResult, SQSWeights, WalkForwardAggregate, WalkForwardFoldResult, WalkForwardGapStats, WalkForwardSummary, ) from libs.backtest.tracker import ( _normalize, _normalize_band, _normalize_inverse, _gap_penalty, _fold_variance_penalty, _trade_credibility, _engine_reliability_penalty, attach_out_of_time_robustness_summary, attach_robustness_summary, attach_walk_forward_summary, append_journal_entry, build_split_result, check_duplicate, classify_strategy_family, compute_deployment_score, compute_public_sqs, compute_public_sqs_v2, compute_public_sqs_v3, compute_public_sqs_v4, compute_common_window_score, compute_promotion_score, _compute_oot_positive_rate_63plus, compute_oot_robustness_gate, compute_oot_robustness_quality, compute_robustness_gate, compute_rqs, compute_sqs, compute_sqs_v2, compute_unified_score, compute_unified_split_quality, compute_wfqs, compute_wfqs_v2, filter_registry_entries, get_next_entry_id, journal_lock, load_journal, refresh_public_scores, rebuild_registry, scan_runs_for_experiment, sync_official_manifests, ) def _write_locked_journal_entry(payload: tuple[str, str]) -> str: journal_path_str, experiment_name = payload journal_path = Path(journal_path_str) with journal_lock(journal_path): entry_id = get_next_entry_id(journal_path) time.sleep(0.05) append_journal_entry( journal_path, JournalEntry( entry_id=entry_id, timestamp="2026-03-17T10:10:07+00:00", experiment_name=experiment_name, hypothesis="h", ), ) return entry_id # --------------------------------------------------------------------------- # _normalize / _normalize_inverse # --------------------------------------------------------------------------- class TestNormalize: def test_at_low_boundary(self): assert _normalize(0.8, low=0.8, high=2.0) == 0.0 def test_at_high_boundary(self): assert _normalize(2.0, low=0.8, high=2.0) == 100.0 def test_midpoint(self): assert _normalize(1.4, low=0.8, high=2.0) == pytest.approx(50.0) def test_below_low_clamps(self): assert _normalize(0.0, low=0.8, high=2.0) == 0.0 def test_above_high_clamps(self): assert _normalize(5.0, low=0.8, high=2.0) == 100.0 def test_none_returns_zero(self): assert _normalize(None, low=0.8, high=2.0) == 0.0 class TestNormalizeInverse: def test_dd_at_worst(self): # 10% drawdown = worst (0 pts) assert _normalize_inverse(10.0, low=10.0, high=1.0) == 0.0 def test_dd_at_best(self): # 1% drawdown = best (100 pts) assert _normalize_inverse(1.0, low=10.0, high=1.0) == 100.0 def test_dd_midpoint(self): assert _normalize_inverse(5.5, low=10.0, high=1.0) == pytest.approx(50.0) def test_none_returns_zero(self): assert _normalize_inverse(None, low=10.0, high=1.0) == 0.0 class TestNormalizeBand: def test_band_plateau_scores_max(self): assert _normalize_band(60.0, low_bad=10.0, low_good=40.0, high_good=80.0, high_bad=100.0) == 100.0 def test_below_band_ramps_up(self): assert _normalize_band(25.0, low_bad=10.0, low_good=40.0, high_good=80.0, high_bad=100.0) == pytest.approx(50.0) def test_outside_band_scores_zero(self): assert _normalize_band(5.0, low_bad=10.0, low_good=40.0, high_good=80.0, high_bad=100.0) == 0.0 assert _normalize_band(100.0, low_bad=10.0, low_good=40.0, high_good=80.0, high_bad=100.0) == 0.0 # --------------------------------------------------------------------------- # compute_sqs # --------------------------------------------------------------------------- class TestComputeSQS: def test_perfect_metrics(self): """All metrics at 100-point boundaries => SQS near 100.""" m = MetricsBundle( trade_count=200, profit_factor=2.0, total_return_pct=5.0, max_drawdown_pct=1.0, sharpe_ratio=2.0, win_rate=0.65, monthly_win_rate=0.70, equity_curve_r_squared=0.80, ) sqs, breakdown = compute_sqs(m) assert sqs == pytest.approx(100.0, abs=0.5) assert breakdown["profitability"] == pytest.approx(100.0, abs=0.5) assert breakdown["risk"] == pytest.approx(100.0, abs=0.5) assert breakdown["consistency"] == pytest.approx(100.0, abs=0.5) assert breakdown["robustness"] == pytest.approx(100.0, abs=0.5) def test_worst_metrics(self): """All metrics at 0-point boundaries => SQS = 0.""" m = MetricsBundle( trade_count=5, profit_factor=0.5, total_return_pct=-10.0, max_drawdown_pct=15.0, sharpe_ratio=-2.0, win_rate=0.20, monthly_win_rate=0.10, equity_curve_r_squared=-0.5, ) sqs, _ = compute_sqs(m) assert sqs == 0.0 def test_low_trade_penalty(self): """< 20 trades => SQS * 0.5.""" m = MetricsBundle( trade_count=15, profit_factor=1.5, total_return_pct=2.0, max_drawdown_pct=3.0, sharpe_ratio=1.0, win_rate=0.55, monthly_win_rate=0.55, equity_curve_r_squared=0.5, ) sqs_penalized, _ = compute_sqs(m) m_enough = m.model_copy(update={"trade_count": 100}) sqs_full, _ = compute_sqs(m_enough) # Penalised score should be roughly half (trade_count affects robustness sub-score too) assert sqs_penalized < sqs_full assert sqs_penalized > 0 def test_midrange_metrics(self): """Typical mid-range strategy should score 30-60.""" m = MetricsBundle( trade_count=80, profit_factor=1.1, total_return_pct=0.5, max_drawdown_pct=5.0, sharpe_ratio=0.5, win_rate=0.50, monthly_win_rate=0.50, equity_curve_r_squared=0.30, ) sqs, _ = compute_sqs(m) assert 30 <= sqs <= 65 def test_custom_weights(self): """Custom weights should change the SQS.""" m = MetricsBundle( trade_count=80, profit_factor=2.0, total_return_pct=5.0, max_drawdown_pct=8.0, sharpe_ratio=0.0, win_rate=0.40, monthly_win_rate=0.40, ) w_profit_heavy = SQSWeights(profitability=0.80, risk=0.10, consistency=0.05, robustness=0.05) w_risk_heavy = SQSWeights(profitability=0.10, risk=0.80, consistency=0.05, robustness=0.05) sqs_profit, _ = compute_sqs(m, w_profit_heavy) sqs_risk, _ = compute_sqs(m, w_risk_heavy) # This strategy has great profitability but mediocre risk assert sqs_profit > sqs_risk class TestComputeSQSv2: def test_missing_exposure_returns_none(self): m = MetricsBundle( trade_count=50, profit_factor=1.5, total_return_pct=2.0, max_drawdown_pct=2.0, sharpe_ratio=1.2, win_rate=0.55, monthly_win_rate=0.60, equity_curve_r_squared=0.50, ) sqs_v2, breakdown = compute_sqs_v2(m) assert sqs_v2 is None assert breakdown == {} def test_capital_efficiency_can_break_tie(self): base = dict( trade_count=50, profit_factor=1.5, total_return_pct=2.0, max_drawdown_pct=2.0, sharpe_ratio=1.2, win_rate=0.55, monthly_win_rate=0.60, equity_curve_r_squared=0.50, days_in_market_pct=60.0, ) efficient = MetricsBundle( **base, avg_gross_exposure_pct=4.0, avg_net_exposure_pct=-1.0, ) inefficient = MetricsBundle( **base, avg_gross_exposure_pct=10.0, avg_net_exposure_pct=-1.0, ) efficient_score, _ = compute_sqs_v2(efficient) inefficient_score, _ = compute_sqs_v2(inefficient) assert efficient_score is not None assert inefficient_score is not None assert efficient_score > inefficient_score class TestComputePromotionScore: def test_requires_valid_and_test(self): score, breakdown = compute_promotion_score(None, None) assert score is None assert breakdown == {} def test_penalizes_test_only_outperformance(self): overfit_test = SplitResult( run_id="bt_test", trade_count=56, profit_factor=2.0, total_return_pct=2.2, win_rate=0.61, max_drawdown_pct=0.4, sharpe_ratio=4.2, monthly_win_rate=1.0, equity_curve_r_squared=0.88, avg_gross_exposure_pct=6.2, avg_net_exposure_pct=1.1, days_in_market_pct=76.6, ) overfit_valid = SplitResult( run_id="bt_valid", trade_count=53, profit_factor=1.24, total_return_pct=0.7, win_rate=0.51, max_drawdown_pct=0.9, sharpe_ratio=1.2, monthly_win_rate=0.75, equity_curve_r_squared=0.15, avg_gross_exposure_pct=7.5, avg_net_exposure_pct=1.6, days_in_market_pct=77.2, ) robust_test = SplitResult( run_id="bt_test_robust", trade_count=22, profit_factor=4.19, total_return_pct=1.24, win_rate=0.73, max_drawdown_pct=0.22, sharpe_ratio=3.6, monthly_win_rate=0.67, equity_curve_r_squared=0.92, avg_gross_exposure_pct=2.3, avg_net_exposure_pct=-0.8, days_in_market_pct=42.6, ) robust_valid = SplitResult( run_id="bt_valid_robust", trade_count=28, profit_factor=2.31, total_return_pct=1.3, win_rate=0.68, max_drawdown_pct=0.37, sharpe_ratio=3.05, monthly_win_rate=0.75, equity_curve_r_squared=0.49, avg_gross_exposure_pct=4.1, avg_net_exposure_pct=-1.9, days_in_market_pct=57.9, ) overfit_score, overfit_breakdown = compute_promotion_score(overfit_test, overfit_valid) robust_score, robust_breakdown = compute_promotion_score(robust_test, robust_valid) assert overfit_score is not None assert robust_score is not None assert robust_score > overfit_score assert overfit_breakdown["floor_quality"] < overfit_breakdown["test_quality"] class TestComputeUnifiedScore: def test_requires_valid_and_test(self): score, breakdown = compute_unified_score(None, None) assert score is None assert breakdown == {} def test_split_quality_rewards_efficiency(self): efficient = MetricsBundle( trade_count=30, profit_factor=2.0, total_return_pct=1.2, max_drawdown_pct=0.4, sharpe_ratio=2.5, win_rate=0.60, monthly_win_rate=0.75, equity_curve_r_squared=0.70, avg_gross_exposure_pct=3.0, avg_net_exposure_pct=-1.0, days_in_market_pct=55.0, ) inefficient = efficient.model_copy( update={"avg_gross_exposure_pct": 8.0, "days_in_market_pct": 85.0} ) efficient_score, _ = compute_unified_split_quality(efficient) inefficient_score, _ = compute_unified_split_quality(inefficient) assert efficient_score is not None assert inefficient_score is not None assert efficient_score > inefficient_score def test_overfit_strategy_scores_below_robust_strategy(self): overfit_test = SplitResult( run_id="bt_step35_test", trade_count=56, profit_factor=2.009, total_return_pct=2.181, win_rate=0.607, max_drawdown_pct=0.442, sharpe_ratio=4.217, monthly_win_rate=1.0, equity_curve_r_squared=0.889, avg_gross_exposure_pct=6.228, avg_net_exposure_pct=1.126, days_in_market_pct=76.6, ) overfit_valid = SplitResult( run_id="bt_step35_valid", trade_count=53, profit_factor=1.243, total_return_pct=0.708, win_rate=0.509, max_drawdown_pct=0.885, sharpe_ratio=1.180, monthly_win_rate=0.75, equity_curve_r_squared=0.146, avg_gross_exposure_pct=7.511, avg_net_exposure_pct=1.567, days_in_market_pct=77.2, ) robust_test = SplitResult( run_id="bt_step52_test", trade_count=22, profit_factor=3.352, total_return_pct=1.011, win_rate=0.773, max_drawdown_pct=0.243, sharpe_ratio=3.032, monthly_win_rate=0.667, equity_curve_r_squared=0.847, avg_gross_exposure_pct=2.858, avg_net_exposure_pct=-2.223, days_in_market_pct=53.2, ) robust_valid = SplitResult( run_id="bt_step52_valid", trade_count=25, profit_factor=5.660, total_return_pct=1.819, win_rate=0.72, max_drawdown_pct=0.217, sharpe_ratio=4.756, monthly_win_rate=1.0, equity_curve_r_squared=0.661, avg_gross_exposure_pct=3.584, avg_net_exposure_pct=-2.593, days_in_market_pct=56.1, ) overfit_score, overfit_breakdown = compute_unified_score(overfit_test, overfit_valid) robust_score, robust_breakdown = compute_unified_score(robust_test, robust_valid) assert overfit_score is not None assert robust_score is not None assert robust_score > overfit_score assert overfit_breakdown["gap_quality"] < robust_breakdown["gap_quality"] class TestComputePublicSQS: def test_requires_walk_forward_and_robustness(self): train_result = SplitResult(run_id="bt_train", trade_count=10, total_return_pct=20.0) valid_result = SplitResult(run_id="bt_valid", trade_count=8, total_return_pct=10.0) test_result = SplitResult(run_id="bt_test", trade_count=9, total_return_pct=12.0) public_score, breakdown, source = compute_public_sqs( train_result, valid_result, test_result, deployment_score=55.0, rqs_score=70.0, ) assert public_score is None assert source == "pending_validation" assert breakdown["requires_walk_forward"] == 1.0 assert breakdown["requires_robustness"] == 1.0 assert breakdown["requires_out_of_time_robustness"] == 1.0 def test_v3_removes_stress_quality_penalty_from_primary_rank(self): train = SplitResult(run_id="bt_train", trade_count=20, total_return_pct=20.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=15, total_return_pct=10.0, win_rate=0.6) test = SplitResult( run_id="bt_test", trade_count=24, total_return_pct=12.0, win_rate=0.6, days_in_market_pct=55.0, ) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=12.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=90.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=18.0, median_return_pct=12.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) weak_stress = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=55.0, overall_worst_return_pct=-12.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=2.0, median_return_pct=1.0, worst_return_pct=-5.0, positive_window_rate_pct=54.5, mean_max_drawdown_pct=4.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=8.0, median_return_pct=5.0, worst_return_pct=-12.0, positive_window_rate_pct=61.5, mean_max_drawdown_pct=10.0, ), ], ) legacy_score, legacy_breakdown, _ = compute_public_sqs_v2( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=weak_stress, ) v3_score, v3_breakdown, source = compute_public_sqs_v3( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=weak_stress, ) assert legacy_score is not None assert v3_score is not None assert v3_score > legacy_score assert source == "v3_deployment+robustness+oot_gate" assert "oot_quality" in v3_breakdown assert "oot_quality_factor" not in v3_breakdown def test_requires_both_validations_even_if_robustness_exists(self): train_result = SplitResult(run_id="bt_train", trade_count=10, total_return_pct=20.0) valid_result = SplitResult(run_id="bt_valid", trade_count=8, total_return_pct=10.0) test_result = SplitResult(run_id="bt_test", trade_count=9, total_return_pct=12.0) robustness = RobustnessMatrixSummary( horizons_days=[21, 63, 252], step_days=21, overall_window_count=12, overall_positive_window_rate_pct=70.0, overall_worst_return_pct=-15.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=4, mean_return_pct=5.0, median_return_pct=4.0, worst_return_pct=-2.0, positive_window_rate_pct=75.0, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=3, mean_return_pct=12.0, median_return_pct=9.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) public_score, breakdown, source = compute_public_sqs( train_result, valid_result, test_result, deployment_score=50.0, robustness_matrix_summary=robustness, rqs_score=70.0, ) assert public_score is None assert source == "pending_validation" assert breakdown["requires_walk_forward"] == 1.0 assert breakdown["requires_out_of_time_robustness"] == 1.0 def test_v4_falls_back_to_v3_without_common_window(self): train = SplitResult(run_id="bt_train", trade_count=20, total_return_pct=20.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=15, total_return_pct=10.0, win_rate=0.6) test = SplitResult( run_id="bt_test", trade_count=24, total_return_pct=12.0, win_rate=0.6, days_in_market_pct=55.0, ) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=12.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=90.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=18.0, median_return_pct=12.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) oot = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=70.0, overall_worst_return_pct=-9.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=2.0, median_return_pct=1.5, worst_return_pct=-4.0, positive_window_rate_pct=60.0, mean_max_drawdown_pct=4.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=7.0, median_return_pct=4.5, worst_return_pct=-9.0, positive_window_rate_pct=75.0, mean_max_drawdown_pct=8.0, ), ], ) v3_score, _, _ = compute_public_sqs_v3( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, ) v4_score, _, source = compute_public_sqs_v4( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, common_window_summary=None, ) assert v3_score == v4_score assert source == "v4_fallback_v3_missing_common_window" class TestCommonWindowScore: def test_rewards_higher_return_and_capital_efficiency(self): compounder = CommonWindowSummary( snapshot_id="snap_a", start_date=dt.date(2022, 3, 3), end_date=dt.date(2026, 3, 13), metrics=MetricsBundle( trade_count=180, profit_factor=8.2, total_return_pct=262.9, max_drawdown_pct=4.46, sharpe_ratio=2.68, avg_gross_exposure_pct=25.9, days_in_market_pct=67.1, ), ) slower = CommonWindowSummary( snapshot_id="snap_b", start_date=dt.date(2022, 3, 3), end_date=dt.date(2026, 3, 13), metrics=MetricsBundle( trade_count=146, profit_factor=8.7, total_return_pct=130.1, max_drawdown_pct=5.38, sharpe_ratio=2.38, avg_gross_exposure_pct=36.5, days_in_market_pct=73.7, ), ) compounder_score, compounder_breakdown = compute_common_window_score(compounder) slower_score, _ = compute_common_window_score(slower) assert compounder_score is not None assert slower_score is not None assert compounder_score > slower_score assert compounder_breakdown["cw_return_on_gross"] > 90.0 def test_v4_can_promote_compounder_when_v3_gap_is_small(self): train = SplitResult(run_id="bt_train", trade_count=100, total_return_pct=50.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=25, total_return_pct=40.0, win_rate=0.7) test = SplitResult( run_id="bt_test", trade_count=24, total_return_pct=60.0, win_rate=0.7, days_in_market_pct=75.0, ) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=12.0, median_return_pct=9.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=2.0, mean_max_drawdown_pct=4.5, mean_trade_count=14.0, mean_win_rate=0.58, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=12, overall_positive_window_rate_pct=85.0, overall_worst_return_pct=-3.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=8, mean_return_pct=4.0, median_return_pct=4.0, worst_return_pct=-2.0, positive_window_rate_pct=80.0, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=15.0, median_return_pct=10.0, worst_return_pct=2.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) oot = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=65.0, overall_worst_return_pct=-10.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=1.5, median_return_pct=1.0, worst_return_pct=-4.0, positive_window_rate_pct=60.0, mean_max_drawdown_pct=4.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=6.0, median_return_pct=4.0, worst_return_pct=-10.0, positive_window_rate_pct=75.0, mean_max_drawdown_pct=8.0, ), ], ) weak_common = CommonWindowSummary( snapshot_id="snap_a", start_date=dt.date(2022, 3, 3), end_date=dt.date(2026, 3, 13), metrics=MetricsBundle( trade_count=140, profit_factor=7.5, total_return_pct=130.1, max_drawdown_pct=5.38, sharpe_ratio=2.38, avg_gross_exposure_pct=36.5, days_in_market_pct=73.7, ), ) strong_common = CommonWindowSummary( snapshot_id="snap_b", start_date=dt.date(2022, 3, 3), end_date=dt.date(2026, 3, 13), metrics=MetricsBundle( trade_count=180, profit_factor=8.2, total_return_pct=262.9, max_drawdown_pct=4.46, sharpe_ratio=2.68, avg_gross_exposure_pct=25.9, days_in_market_pct=67.1, ), ) v4_weak, _, _ = compute_public_sqs_v4( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, common_window_summary=weak_common, ) v4_strong, _, _ = compute_public_sqs_v4( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, common_window_summary=strong_common, ) assert v4_weak is not None assert v4_strong is not None assert v4_strong > v4_weak class TestComputeRQS: def test_requires_valid_and_test(self): score, breakdown = compute_rqs(None, None, None) assert score is None assert breakdown == {} def test_rewards_return_max_profile(self): conservative_train = SplitResult( run_id="bt_cons_train", trade_count=30, profit_factor=2.5, total_return_pct=12.0, annualized_return_pct=6.0, win_rate=0.8, max_drawdown_pct=2.0, sharpe_ratio=2.0, monthly_win_rate=0.8, equity_curve_r_squared=0.8, avg_gross_exposure_pct=10.0, avg_net_exposure_pct=10.0, days_in_market_pct=15.0, ) conservative_valid = SplitResult( run_id="bt_cons_valid", trade_count=12, profit_factor=2.0, total_return_pct=10.0, annualized_return_pct=30.0, win_rate=0.75, max_drawdown_pct=3.0, sharpe_ratio=2.2, monthly_win_rate=1.0, equity_curve_r_squared=0.7, avg_gross_exposure_pct=9.0, avg_net_exposure_pct=9.0, days_in_market_pct=18.0, ) conservative_test = SplitResult( run_id="bt_cons_test", trade_count=14, profit_factor=2.1, total_return_pct=18.0, annualized_return_pct=58.0, win_rate=0.8, max_drawdown_pct=3.5, sharpe_ratio=2.0, monthly_win_rate=1.0, equity_curve_r_squared=0.7, avg_gross_exposure_pct=14.0, avg_net_exposure_pct=14.0, days_in_market_pct=20.0, ) return_max_train = SplitResult( run_id="bt_ret_train", trade_count=13, profit_factor=None, total_return_pct=44.1, annualized_return_pct=18.0, win_rate=1.0, max_drawdown_pct=5.4, sharpe_ratio=1.5, monthly_win_rate=1.0, equity_curve_r_squared=0.76, avg_gross_exposure_pct=11.1, avg_net_exposure_pct=11.1, days_in_market_pct=11.8, ) return_max_valid = SplitResult( run_id="bt_ret_valid", trade_count=7, profit_factor=None, total_return_pct=35.3, annualized_return_pct=105.0, win_rate=1.0, max_drawdown_pct=4.1, sharpe_ratio=3.0, monthly_win_rate=1.0, equity_curve_r_squared=0.81, avg_gross_exposure_pct=17.7, avg_net_exposure_pct=17.7, days_in_market_pct=19.4, ) return_max_test = SplitResult( run_id="bt_ret_test", trade_count=15, profit_factor=5.2, total_return_pct=44.7, annualized_return_pct=125.0, win_rate=0.867, max_drawdown_pct=8.3, sharpe_ratio=2.6, monthly_win_rate=1.0, equity_curve_r_squared=0.87, avg_gross_exposure_pct=42.3, avg_net_exposure_pct=42.3, days_in_market_pct=51.1, ) conservative_score, conservative_breakdown = compute_rqs( conservative_train, conservative_valid, conservative_test, ) return_max_score, return_max_breakdown = compute_rqs( return_max_train, return_max_valid, return_max_test, ) assert conservative_score is not None assert return_max_score is not None assert return_max_score > conservative_score assert return_max_breakdown["train_quality"] > 70.0 assert return_max_breakdown["valid_quality"] > 70.0 assert return_max_breakdown["test_quality"] > conservative_breakdown["test_quality"] def test_missing_train_gets_penalized(self): valid = SplitResult( run_id="bt_valid", trade_count=15, profit_factor=2.5, total_return_pct=20.0, win_rate=0.8, max_drawdown_pct=3.0, sharpe_ratio=2.0, monthly_win_rate=1.0, equity_curve_r_squared=0.7, avg_gross_exposure_pct=15.0, avg_net_exposure_pct=15.0, days_in_market_pct=30.0, ) test = SplitResult( run_id="bt_test", trade_count=15, profit_factor=2.5, total_return_pct=25.0, win_rate=0.8, max_drawdown_pct=3.0, sharpe_ratio=2.0, monthly_win_rate=1.0, equity_curve_r_squared=0.7, avg_gross_exposure_pct=20.0, avg_net_exposure_pct=20.0, days_in_market_pct=35.0, ) train = SplitResult( run_id="bt_train", trade_count=15, profit_factor=2.0, total_return_pct=30.0, win_rate=0.7, max_drawdown_pct=4.0, sharpe_ratio=1.5, monthly_win_rate=0.8, equity_curve_r_squared=0.6, avg_gross_exposure_pct=14.0, avg_net_exposure_pct=14.0, days_in_market_pct=25.0, ) full_score, _ = compute_rqs(train, valid, test) penalized_score, _ = compute_rqs(None, valid, test) assert full_score is not None assert penalized_score is not None assert full_score > penalized_score class TestComputeWFQS: def test_requires_summary(self): score, breakdown = compute_wfqs(None) assert score is None assert breakdown == {} def test_rewards_stable_walk_forward_profile(self): robust = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=14.0, median_return_pct=12.0, worst_return_pct=2.0, positive_fold_rate_pct=87.5, mean_profit_factor=2.2, mean_max_drawdown_pct=4.0, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=18.0, worst_train_test_return_gap_pct=30.0, ), ) overfit = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=4.0, worst_return_pct=-8.0, positive_fold_rate_pct=62.5, mean_profit_factor=1.3, mean_max_drawdown_pct=8.5, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=55.0, worst_train_test_return_gap_pct=95.0, ), ) robust_score, robust_breakdown = compute_wfqs(robust) overfit_score, _ = compute_wfqs(overfit) assert robust_score is not None assert overfit_score is not None assert robust_score > overfit_score assert robust_breakdown["train_test_gap"] > 50.0 class TestComputeDeploymentScore: def test_requires_walk_forward(self): score, breakdown = compute_deployment_score(None, None, None, None) assert score is None assert breakdown == {} def test_gate_penalizes_overfit_profile(self): train = SplitResult(run_id="bt_train", trade_count=10, total_return_pct=40.0, win_rate=1.0) valid = SplitResult(run_id="bt_valid", trade_count=10, total_return_pct=20.0, win_rate=0.8) test = SplitResult(run_id="bt_test", trade_count=10, total_return_pct=25.0, win_rate=0.8) robust_wf = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=16.0, median_return_pct=12.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=2.2, mean_max_drawdown_pct=4.0, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, worst_train_test_return_gap_pct=35.0, ), ) overfit_wf = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=16.0, median_return_pct=2.0, worst_return_pct=-7.0, positive_fold_rate_pct=50.0, mean_profit_factor=1.5, mean_max_drawdown_pct=7.0, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=60.0, worst_train_test_return_gap_pct=90.0, ), ) robust_score, robust_breakdown = compute_deployment_score(train, valid, test, robust_wf, rqs_score=70.0) overfit_score, overfit_breakdown = compute_deployment_score(train, valid, test, overfit_wf, rqs_score=70.0) assert robust_score is not None assert overfit_score is not None assert robust_score > overfit_score assert robust_breakdown["gate_factor"] == 1.0 assert overfit_breakdown["gate_factor"] < 1.0 class TestComputeRobustnessGate: @staticmethod def _summary( *, positive_rate: float, median_63d: float, median_252d: float, worst_return: float, ) -> RobustnessMatrixSummary: return RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=12, overall_positive_window_rate_pct=positive_rate, overall_worst_return_pct=worst_return, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=5.0, median_return_pct=median_63d, worst_return_pct=-2.0, positive_window_rate_pct=positive_rate, mean_max_drawdown_pct=4.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=3, mean_return_pct=12.0, median_return_pct=median_252d, worst_return_pct=1.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=6.0, ), ], ) def test_all_four_checks_pass(self): factor, breakdown = compute_robustness_gate( self._summary(positive_rate=70.0, median_63d=4.0, median_252d=9.0, worst_return=-5.0) ) assert factor == 1.0 assert sum(breakdown.values()) == 4.0 def test_three_checks_pass(self): factor, breakdown = compute_robustness_gate( self._summary(positive_rate=70.0, median_63d=4.0, median_252d=9.0, worst_return=-20.0) ) assert factor == 0.85 assert sum(breakdown.values()) == 3.0 def test_two_checks_pass(self): factor, breakdown = compute_robustness_gate( self._summary(positive_rate=70.0, median_63d=1.0, median_252d=9.0, worst_return=-20.0) ) assert factor == 0.65 assert sum(breakdown.values()) == 2.0 def test_one_check_pass(self): factor, breakdown = compute_robustness_gate( self._summary(positive_rate=50.0, median_63d=1.0, median_252d=9.0, worst_return=-20.0) ) assert factor == 0.40 assert sum(breakdown.values()) == 1.0 def test_zero_checks_pass(self): factor, breakdown = compute_robustness_gate( self._summary(positive_rate=50.0, median_63d=1.0, median_252d=5.0, worst_return=-20.0) ) assert factor == 0.20 assert sum(breakdown.values()) == 0.0 # --------------------------------------------------------------------------- # OOT-specific functions # --------------------------------------------------------------------------- class TestComputeOotPositiveRate63Plus: @staticmethod def _summary_with_21d( *, pos_21d: float = 30.0, pos_63d: float = 54.5, pos_252d: float = 61.5, cnt_21d: int = 24, cnt_63d: int = 22, cnt_252d: int = 13, ) -> RobustnessMatrixSummary: return RobustnessMatrixSummary( horizons_days=[21, 63, 252], step_days=21, overall_window_count=cnt_21d + cnt_63d + cnt_252d, overall_positive_window_rate_pct=40.0, overall_worst_return_pct=-12.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=21, window_count=cnt_21d, mean_return_pct=0.5, median_return_pct=0.1, worst_return_pct=-3.0, positive_window_rate_pct=pos_21d, mean_max_drawdown_pct=2.0, ), RobustnessHorizonSummary( horizon_days=63, window_count=cnt_63d, mean_return_pct=2.0, median_return_pct=1.33, worst_return_pct=-5.0, positive_window_rate_pct=pos_63d, mean_max_drawdown_pct=4.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=cnt_252d, mean_return_pct=8.0, median_return_pct=6.72, worst_return_pct=-12.0, positive_window_rate_pct=pos_252d, mean_max_drawdown_pct=10.0, ), ], ) def test_excludes_21d_windows(self): s = self._summary_with_21d(pos_21d=10.0, pos_63d=54.5, pos_252d=61.5) rate = _compute_oot_positive_rate_63plus(s) # (54.5*22 + 61.5*13) / (22+13) = (1199 + 799.5) / 35 ≈ 57.1 assert rate is not None assert 57.0 <= rate <= 57.2 def test_63d_only(self): s = RobustnessMatrixSummary( horizons_days=[63], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=70.0, overall_worst_return_pct=-5.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=10, mean_return_pct=3.0, median_return_pct=2.5, worst_return_pct=-2.0, positive_window_rate_pct=70.0, mean_max_drawdown_pct=3.0, ), ], ) assert _compute_oot_positive_rate_63plus(s) == 70.0 def test_no_qualifying_horizons(self): s = RobustnessMatrixSummary( horizons_days=[21], step_days=21, overall_window_count=5, overall_positive_window_rate_pct=40.0, overall_worst_return_pct=-8.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=21, window_count=5, mean_return_pct=0.5, median_return_pct=0.1, worst_return_pct=-3.0, positive_window_rate_pct=40.0, mean_max_drawdown_pct=2.0, ), ], ) assert _compute_oot_positive_rate_63plus(s) is None class TestComputeOotRobustnessGate: @staticmethod def _summary( *, pos_63d: float, pos_252d: float, median_63d: float, median_252d: float, worst_return: float, ) -> RobustnessMatrixSummary: return RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=50.0, overall_worst_return_pct=worst_return, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=2.0, median_return_pct=median_63d, worst_return_pct=-5.0, positive_window_rate_pct=pos_63d, mean_max_drawdown_pct=4.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=8.0, median_return_pct=median_252d, worst_return_pct=-1.0, positive_window_rate_pct=pos_252d, mean_max_drawdown_pct=6.0, ), ], ) def test_all_four_pass_stress(self): # 63d+ rate: (60*6 + 50*4)/10 = 56% >= 50 ✓ factor, breakdown = compute_oot_robustness_gate( self._summary(pos_63d=60.0, pos_252d=50.0, median_63d=1.0, median_252d=4.0, worst_return=-10.0) ) assert factor == 1.0 assert sum(breakdown.values()) == 4.0 def test_three_pass(self): # 63d+ rate: (60*6 + 50*4)/10 = 56% >= 50 ✓, 63d med 1.0 >= 0.5 ✓, # 252d med 4.0 >= 3.0 ✓, worst -16 < -15 ✗ factor, _ = compute_oot_robustness_gate( self._summary(pos_63d=60.0, pos_252d=50.0, median_63d=1.0, median_252d=4.0, worst_return=-16.0) ) assert factor == 0.85 def test_zero_pass_stress(self): # 63d+ rate: (40*6 + 30*4)/10 = 36% < 50 ✗, 63d med 0.2 < 0.5 ✗, # 252d med 1.0 < 3.0 ✗, worst -20 < -15 ✗ factor, _ = compute_oot_robustness_gate( self._summary(pos_63d=40.0, pos_252d=30.0, median_63d=0.2, median_252d=1.0, worst_return=-20.0) ) assert factor == 0.20 def test_none_summary(self): factor, breakdown = compute_oot_robustness_gate(None) assert factor == 1.0 assert breakdown == {} def test_main_gate_unchanged(self): """Ensure compute_robustness_gate thresholds are not affected.""" s = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=70.0, overall_worst_return_pct=-5.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=4.0, worst_return_pct=-1.0, positive_window_rate_pct=80.0, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=12.0, median_return_pct=9.0, worst_return_pct=1.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) factor, _ = compute_robustness_gate(s) assert factor == 1.0 class TestComputeOotRobustnessQuality: def test_strong_stress_test_survivor(self): s = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=35, overall_positive_window_rate_pct=55.0, overall_worst_return_pct=-12.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=22, mean_return_pct=2.0, median_return_pct=1.33, worst_return_pct=-5.0, positive_window_rate_pct=54.5, mean_max_drawdown_pct=4.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=13, mean_return_pct=8.0, median_return_pct=6.72, worst_return_pct=-12.0, positive_window_rate_pct=61.5, mean_max_drawdown_pct=10.0, ), ], ) quality, breakdown = compute_oot_robustness_quality(s) assert quality is not None assert quality > 75.0 assert "rb_quality_positive_rate" in breakdown def test_none_returns_none(self): quality, breakdown = compute_oot_robustness_quality(None) assert quality is None assert breakdown == {} def test_weak_stress_test(self): s = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=35, overall_positive_window_rate_pct=35.0, overall_worst_return_pct=-22.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=22, mean_return_pct=-1.0, median_return_pct=-0.76, worst_return_pct=-8.0, positive_window_rate_pct=45.5, mean_max_drawdown_pct=6.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=13, mean_return_pct=-2.0, median_return_pct=-0.92, worst_return_pct=-22.0, positive_window_rate_pct=46.2, mean_max_drawdown_pct=18.0, ), ], ) quality, _ = compute_oot_robustness_quality(s) assert quality is not None assert quality < 65.0 def test_refresh_public_scores_recomputes_stale_scores(tmp_path: Path): journal_path = tmp_path / "improvement_journal.jsonl" train = SplitResult(run_id="bt_train", trade_count=20, total_return_pct=20.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=15, total_return_pct=10.0, win_rate=0.6) test = SplitResult( run_id="bt_test", trade_count=24, total_return_pct=12.0, win_rate=0.6, days_in_market_pct=55.0, ) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=12.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=90.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=18.0, median_return_pct=12.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) stale_entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-22T00:00:00+00:00", experiment_name="return_max_long_v1.200", hypothesis="stale", results={"train": train, "valid": valid, "test": test}, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=robustness, sqs_score=999.0, sqs_breakdown={"old": 1.0}, sqs_v2_score=999.0, sqs_v2_breakdown={"old": 1.0}, rqs_score=1.0, wfqs_score=1.0, wfqs_v2_score=1.0, deployment_score=1.0, ) append_journal_entry(journal_path, stale_entry) updated = refresh_public_scores(journal_path, selector=lambda e: e.experiment_name.endswith("v1.200")) assert updated == 1 refreshed = load_journal(journal_path)[0] expected_sqs, expected_breakdown, _ = compute_public_sqs( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=robustness, ) expected_stress_sqs, expected_stress_breakdown, _ = compute_public_sqs_v2( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=robustness, ) assert refreshed.sqs_score == expected_sqs assert refreshed.sqs_breakdown == expected_breakdown assert refreshed.stress_sqs_score == expected_stress_sqs assert refreshed.stress_sqs_breakdown == expected_stress_breakdown assert refreshed.sqs_score != 999.0 assert refreshed.sqs_v2_score != 999.0 # --------------------------------------------------------------------------- # build_split_result # --------------------------------------------------------------------------- def test_build_split_result(): m = MetricsBundle( trade_count=50, profit_factor=1.2, total_return_pct=2.5, win_rate=0.55, max_drawdown_pct=3.0, sharpe_ratio=0.8, monthly_win_rate=0.60, equity_curve_r_squared=0.40, avg_gross_exposure_pct=18.5, avg_net_exposure_pct=-6.5, days_in_market_pct=27.0, ) sr = build_split_result("test", "bt_run123", m) assert sr.run_id == "bt_run123" assert sr.trade_count == 50 assert sr.profit_factor == 1.2 assert sr.total_return_pct == 2.5 assert sr.avg_gross_exposure_pct == 18.5 assert sr.avg_net_exposure_pct == -6.5 assert sr.days_in_market_pct == 27.0 # --------------------------------------------------------------------------- # Journal I/O # --------------------------------------------------------------------------- class TestJournalIO: def test_append_and_load(self, tmp_path): journal_path = tmp_path / "journal.jsonl" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-16T12:00:00", experiment_name="test_exp_1", hypothesis="Test hypothesis", sqs_score=55.0, sqs_breakdown={"profitability": 60.0, "risk": 50.0, "consistency": 55.0, "robustness": 50.0}, verdict="better", ) append_journal_entry(journal_path, entry) entries = load_journal(journal_path) assert len(entries) == 1 assert entries[0].entry_id == "IMP-0001" assert entries[0].experiment_name == "test_exp_1" assert entries[0].sqs_score == 55.0 def test_multiple_entries(self, tmp_path): journal_path = tmp_path / "journal.jsonl" for i in range(3): entry = JournalEntry( entry_id=f"IMP-{i+1:04d}", timestamp=f"2026-03-{16+i}T12:00:00", experiment_name=f"exp_{i}", hypothesis=f"Hypothesis {i}", sqs_score=float(40 + i * 10), ) append_journal_entry(journal_path, entry) entries = load_journal(journal_path) assert len(entries) == 3 assert entries[2].sqs_score == 60.0 def test_get_next_entry_id(self, tmp_path): journal_path = tmp_path / "journal.jsonl" assert get_next_entry_id(journal_path) == "IMP-0001" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-16T12:00:00", experiment_name="exp_1", hypothesis="h", ) append_journal_entry(journal_path, entry) assert get_next_entry_id(journal_path) == "IMP-0002" def test_journal_lock_serializes_concurrent_writers(self, tmp_path): journal_path = tmp_path / "journal.jsonl" ctx = multiprocessing.get_context("spawn") payloads = [ (str(journal_path), "exp_a"), (str(journal_path), "exp_b"), (str(journal_path), "exp_c"), ] with ctx.Pool(processes=3) as pool: ids = pool.map(_write_locked_journal_entry, payloads) assert sorted(ids) == ["IMP-0001", "IMP-0002", "IMP-0003"] assert [entry.entry_id for entry in load_journal(journal_path)] == [ "IMP-0001", "IMP-0002", "IMP-0003", ] def test_load_empty(self, tmp_path): journal_path = tmp_path / "nonexistent.jsonl" entries = load_journal(journal_path) assert entries == [] # --------------------------------------------------------------------------- # check_duplicate # --------------------------------------------------------------------------- class TestCheckDuplicate: def test_finds_duplicates(self, tmp_path): journal_path = tmp_path / "journal.jsonl" for name in ["exp_a", "exp_b", "exp_a"]: entry = JournalEntry( entry_id=get_next_entry_id(journal_path), timestamp="2026-03-16T12:00:00", experiment_name=name, hypothesis="h", ) append_journal_entry(journal_path, entry) dupes = check_duplicate(journal_path, "exp_a") assert len(dupes) == 2 def test_no_duplicates(self, tmp_path): journal_path = tmp_path / "journal.jsonl" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-16T12:00:00", experiment_name="exp_a", hypothesis="h", ) append_journal_entry(journal_path, entry) dupes = check_duplicate(journal_path, "exp_z") assert len(dupes) == 0 # --------------------------------------------------------------------------- # rebuild_registry # --------------------------------------------------------------------------- class TestRebuildRegistry: @staticmethod def _validated_summary_triplet() -> tuple[WalkForwardSummary, RobustnessMatrixSummary, RobustnessMatrixSummary]: wfv = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=12.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.9, mean_max_drawdown_pct=4.0, mean_trade_count=18.0, mean_win_rate=0.56, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=24.0, worst_train_test_return_gap_pct=30.0, fold_return_cv=0.5, ), ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=80.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.5, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=16.0, median_return_pct=12.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) oot = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=72.0, overall_worst_return_pct=-4.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=3.0, median_return_pct=2.5, worst_return_pct=-2.0, positive_window_rate_pct=66.7, mean_max_drawdown_pct=3.2, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.5, positive_window_rate_pct=75.0, mean_max_drawdown_pct=5.2, ), ], ) return wfv, robustness, oot def test_registry_and_leaderboard(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" wfv, robustness, oot = self._validated_summary_triplet() # Create entries with different SQS scores for i, (name, sqs) in enumerate([("exp_low", 30.0), ("exp_high", 70.0), ("exp_mid", 50.0)]): test_result = SplitResult( run_id=f"bt_{name}", trade_count=60, profit_factor=1.0 + i * 0.2, total_return_pct=float(i), win_rate=0.5, monthly_win_rate=0.6, equity_curve_r_squared=0.5, avg_gross_exposure_pct=5.0, avg_net_exposure_pct=-1.0, days_in_market_pct=60.0, ) valid_result = SplitResult( run_id=f"bt_{name}_valid", trade_count=25, profit_factor=1.0 + i * 0.1, total_return_pct=float(i) / 2, win_rate=0.45, monthly_win_rate=0.5, equity_curve_r_squared=0.3, avg_gross_exposure_pct=4.0, avg_net_exposure_pct=-0.5, days_in_market_pct=45.0, ) entry = JournalEntry( entry_id=f"IMP-{i+1:04d}", timestamp=f"2026-03-{16+i}T12:00:00", experiment_name=name, hypothesis=f"h{i}", sqs_score=sqs, promotion_score=sqs - 5, unified_score=sqs - 10, results={"test": test_result, "valid": valid_result}, walk_forward_summary=wfv, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, verdict="better" if sqs > 50 else "worse", ) append_journal_entry(journal_path, entry) registry = rebuild_registry(journal_path, registry_path, leaderboard_path) # Sorted by public SQS descending assert len(registry.entries) == 3 assert registry.entries[0].sqs_score is not None assert registry.entries[-1].sqs_score is not None assert registry.entries[0].sqs_score >= registry.entries[1].sqs_score >= registry.entries[2].sqs_score assert registry.entries[0].sqs_v2_score is not None assert registry.entries[0].promotion_score is not None assert registry.entries[0].sqs_score < 70.0 # Files exist assert registry_path.exists() assert leaderboard_path.exists() # Leaderboard contains table lb_text = leaderboard_path.read_text() assert "exp_high" in lb_text assert "exp_low" in lb_text assert "| # |" in lb_text assert "| # | Experiment | SQS | [Tr]Ret% | [V]Ret% | [T]Ret% | [T]Ann% | [T]DD% | [T]Gross% | [T]DIM% | [T]R/G | Date |" in lb_text def test_registry_skips_legacy_overlay_entries(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-25T22:00:00+00:00", experiment_name="return_book_overlay_v3", hypothesis="overlay", tags=["overlay"], ) append_journal_entry(journal_path, entry) registry = rebuild_registry(journal_path, registry_path, leaderboard_path) visible = filter_registry_entries(registry.entries) assert len(visible) == 0 assert registry.entries == [] lb_text = leaderboard_path.read_text() assert "return_book_overlay_v3" not in lb_text assert "Default view excludes retired legacy PEAD" in lb_text def test_registry_preserves_walk_forward_summary(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-19T10:00:00+00:00", experiment_name="return_max_long_v999_wfv", hypothesis="wfv", results={ "valid": SplitResult(run_id="bt_valid", trade_count=5, total_return_pct=5.0), "test": SplitResult(run_id="bt_test", trade_count=5, total_return_pct=7.0), }, walk_forward_summary=WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=2, folds=[ WalkForwardFoldResult( fold_index=0, train_start="2024-01-01", train_end="2024-12-31", test_start="2025-01-01", test_end="2025-03-31", train_run_id="wf_train_0", test_run_id="wf_test_0", train_metrics=SplitResult(run_id="wf_train_0", trade_count=3, total_return_pct=10.0), test_metrics=SplitResult(run_id="wf_test_0", trade_count=2, total_return_pct=4.0), ), ], train_aggregate=WalkForwardAggregate(mean_return_pct=12.5), test_aggregate=WalkForwardAggregate(mean_return_pct=6.5, median_return_pct=6.5, worst_return_pct=4.0, positive_fold_rate_pct=100.0), gap_stats=WalkForwardGapStats(mean_train_test_return_gap_pct=6.0, worst_train_test_return_gap_pct=8.0), ), ) append_journal_entry(journal_path, entry) registry = rebuild_registry(journal_path, registry_path, leaderboard_path) assert registry.entries[0].walk_forward_summary is not None assert registry.entries[0].walk_forward_summary.fold_count == 2 registry_payload = json.loads(registry_path.read_text()) assert registry_payload["entries"][0]["walk_forward_summary"]["test_aggregate"]["mean_return_pct"] == 6.5 def test_attach_walk_forward_summary_backfills_scores(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-19T10:00:00+00:00", experiment_name="return_max_long_v999_wfv", hypothesis="wfv", results={ "train": SplitResult(run_id="bt_train", trade_count=5, total_return_pct=8.0), "valid": SplitResult(run_id="bt_valid", trade_count=5, total_return_pct=6.0), "test": SplitResult(run_id="bt_test", trade_count=5, total_return_pct=7.0), }, ) append_journal_entry(journal_path, entry) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=2, folds=[ WalkForwardFoldResult( fold_index=0, train_start="2024-01-01", train_end="2024-12-31", test_start="2025-01-01", test_end="2025-03-31", train_run_id="wf_train_0", test_run_id="wf_test_0", train_metrics=SplitResult(run_id="wf_train_0", trade_count=3, total_return_pct=10.0, profit_factor=1.5), test_metrics=SplitResult(run_id="wf_test_0", trade_count=2, total_return_pct=4.0, profit_factor=1.3), ), WalkForwardFoldResult( fold_index=1, train_start="2024-04-01", train_end="2025-03-31", test_start="2025-04-01", test_end="2025-06-30", train_run_id="wf_train_1", test_run_id="wf_test_1", train_metrics=SplitResult(run_id="wf_train_1", trade_count=4, total_return_pct=11.0, profit_factor=1.4), test_metrics=SplitResult(run_id="wf_test_1", trade_count=3, total_return_pct=5.0, profit_factor=1.2), ), ], train_aggregate=WalkForwardAggregate(mean_return_pct=10.5), test_aggregate=WalkForwardAggregate( mean_return_pct=4.5, median_return_pct=4.5, worst_return_pct=4.0, positive_fold_rate_pct=100.0, mean_profit_factor=1.25, mean_max_drawdown_pct=2.5, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=6.0, worst_train_test_return_gap_pct=6.0, ), ) updated = attach_walk_forward_summary(journal_path, "return_max_long_v999_wfv", summary) assert updated.walk_forward_summary is not None assert updated.wfqs_score is not None assert updated.deployment_score is not None registry = rebuild_registry(journal_path, registry_path, leaderboard_path) assert registry.entries[0].walk_forward_summary is not None assert registry.entries[0].wfqs_score is not None assert registry.entries[0].deployment_score is not None def test_attach_robustness_summary_updates_existing_entry(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-19T10:00:00+00:00", experiment_name="return_max_long_v999_rb", hypothesis="rb", results={ "train": SplitResult(run_id="bt_train", trade_count=5, total_return_pct=12.0), "valid": SplitResult(run_id="bt_valid", trade_count=5, total_return_pct=6.0), "test": SplitResult(run_id="bt_test", trade_count=5, total_return_pct=7.0), }, walk_forward_summary=WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=2, train_aggregate=WalkForwardAggregate(mean_return_pct=10.0), test_aggregate=WalkForwardAggregate( mean_return_pct=6.5, median_return_pct=6.5, worst_return_pct=4.0, positive_fold_rate_pct=100.0, mean_profit_factor=1.5, mean_max_drawdown_pct=3.0, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=8.0, worst_train_test_return_gap_pct=10.0, ), ), ) append_journal_entry(journal_path, entry) summary = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=70.0, overall_worst_return_pct=-15.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=4.0, worst_return_pct=-2.0, positive_window_rate_pct=80.0, mean_max_drawdown_pct=2.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=2, mean_return_pct=10.0, median_return_pct=9.0, worst_return_pct=3.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=4.0, ), ], ) updated = attach_robustness_summary(journal_path, "return_max_long_v999_rb", summary) assert updated.robustness_matrix_summary is not None assert updated.robustness_matrix_summary.overall_window_count == 10 registry = rebuild_registry(journal_path, registry_path, leaderboard_path) assert registry.entries[0].robustness_matrix_summary is not None assert registry.entries[0].deployment_score is not None expected_sqs, _, expected_source = compute_public_sqs( updated.results.get("train"), updated.results.get("valid"), updated.results.get("test"), walk_forward_summary=updated.walk_forward_summary, robustness_matrix_summary=updated.robustness_matrix_summary, rqs_score=registry.entries[0].rqs_score, wfqs_score=registry.entries[0].wfqs_score, deployment_score=registry.entries[0].deployment_score, ) assert expected_source == "pending_validation" assert expected_sqs is None assert registry.entries[0].sqs_score is None def test_attach_out_of_time_robustness_summary_updates_existing_entry(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" entry = JournalEntry( entry_id="IMP-0001", timestamp="2026-03-19T10:00:00+00:00", experiment_name="return_max_long_v999_oot", hypothesis="oot", results={ "train": SplitResult(run_id="bt_train", trade_count=5, total_return_pct=12.0), "valid": SplitResult(run_id="bt_valid", trade_count=5, total_return_pct=6.0), "test": SplitResult(run_id="bt_test", trade_count=5, total_return_pct=7.0), }, walk_forward_summary=WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=2, train_aggregate=WalkForwardAggregate(mean_return_pct=10.0), test_aggregate=WalkForwardAggregate( mean_return_pct=6.5, median_return_pct=6.5, worst_return_pct=4.0, positive_fold_rate_pct=100.0, mean_profit_factor=1.5, mean_max_drawdown_pct=3.0, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=8.0, worst_train_test_return_gap_pct=10.0, ), ), robustness_matrix_summary=RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=80.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=4.0, worst_return_pct=-2.0, positive_window_rate_pct=80.0, mean_max_drawdown_pct=2.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=2, mean_return_pct=10.0, median_return_pct=9.0, worst_return_pct=3.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=4.0, ), ], ), ) append_journal_entry(journal_path, entry) summary = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=60.0, overall_worst_return_pct=-15.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=2.0, median_return_pct=2.0, worst_return_pct=-2.0, positive_window_rate_pct=66.7, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=2, mean_return_pct=7.0, median_return_pct=7.0, worst_return_pct=-1.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) updated = attach_out_of_time_robustness_summary(journal_path, "return_max_long_v999_oot", summary) assert updated.out_of_time_robustness_summary is not None registry = rebuild_registry(journal_path, registry_path, leaderboard_path) assert registry.entries[0].out_of_time_robustness_summary is not None def test_attach_walk_forward_summary_updates_existing_entry(self, tmp_path): journal_path = tmp_path / "journal.jsonl" append_journal_entry( journal_path, JournalEntry( entry_id="IMP-0001", timestamp="2026-03-19T12:00:00+00:00", experiment_name="return_max_long_v334_v326_material_compact", hypothesis="attach wfv", results={ "train": SplitResult(run_id="bt_train", trade_count=10, total_return_pct=55.0), "valid": SplitResult(run_id="bt_valid", trade_count=4, total_return_pct=12.0), "test": SplitResult(run_id="bt_test", trade_count=5, total_return_pct=18.0), }, rqs_score=70.0, ), ) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=3, folds=[ WalkForwardFoldResult( fold_index=0, train_start="2024-01-01", train_end="2024-12-31", test_start="2025-01-01", test_end="2025-03-31", train_run_id="wf_train_0", test_run_id="wf_test_0", train_metrics=SplitResult(run_id="wf_train_0", trade_count=3, total_return_pct=20.0), test_metrics=SplitResult(run_id="wf_test_0", trade_count=2, total_return_pct=8.0), ), ], train_aggregate=WalkForwardAggregate(mean_return_pct=25.0), test_aggregate=WalkForwardAggregate( mean_return_pct=9.0, median_return_pct=8.0, worst_return_pct=2.0, positive_fold_rate_pct=100.0, mean_profit_factor=2.2, mean_max_drawdown_pct=3.0, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=16.0, worst_train_test_return_gap_pct=22.0, ), ) updated = attach_walk_forward_summary(journal_path, "return_max_long_v334_v326_material_compact", summary) assert updated.walk_forward_summary is not None assert updated.walk_forward_summary.fold_count == 3 assert updated.wfqs_score is not None assert updated.deployment_score is not None persisted = load_journal(journal_path) assert len(persisted) == 1 assert persisted[0].walk_forward_summary is not None assert persisted[0].wfqs_score == updated.wfqs_score assert persisted[0].deployment_score == updated.deployment_score def test_retired_short_core_and_legacy_pead_are_hidden_by_default(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" wfv, robustness, oot = self._validated_summary_triplet() test_result = SplitResult( run_id="bt_test", trade_count=60, profit_factor=1.8, total_return_pct=1.2, win_rate=0.55, monthly_win_rate=0.60, equity_curve_r_squared=0.5, avg_gross_exposure_pct=5.0, avg_net_exposure_pct=2.0, days_in_market_pct=60.0, ) valid_result = SplitResult( run_id="bt_valid", trade_count=12, profit_factor=1.4, total_return_pct=0.6, win_rate=0.5, monthly_win_rate=0.5, equity_curve_r_squared=0.2, avg_gross_exposure_pct=4.0, avg_net_exposure_pct=1.5, days_in_market_pct=40.0, ) append_journal_entry( journal_path, JournalEntry( entry_id="IMP-0001", timestamp="2026-03-18T12:00:00", experiment_name="pead_midcap_step56_short_core_macro_block_crashcap_gap10_interleave_longtrend25", hypothesis="legacy", sqs_score=51.3, results={"test": test_result, "valid": valid_result}, walk_forward_summary=wfv, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, tags=["pead", "short_core"], ), ) append_journal_entry( journal_path, JournalEntry( entry_id="IMP-0002", timestamp="2026-03-18T13:00:00", experiment_name="return_max_long_v32_same_day_longer_tail", hypothesis="active", sqs_score=14.9, results={"test": test_result, "valid": valid_result}, walk_forward_summary=wfv, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, tags=["return", "max", "long"], ), ) registry = rebuild_registry(journal_path, registry_path, leaderboard_path) assert len(registry.entries) == 2 assert classify_strategy_family(registry.entries[0].experiment_name) in {"short_core", "return_max_long"} retired_hidden = filter_registry_entries(registry.entries, include_retired=False) assert [entry.experiment_name for entry in retired_hidden] == ["return_max_long_v32_same_day_longer_tail"] assert retired_hidden[0].is_retired is False retired_entry = next(entry for entry in registry.entries if entry.experiment_name.startswith("pead_midcap_step56")) assert retired_entry.is_retired is True assert retired_entry.strategy_family == "short_core" lb_text = leaderboard_path.read_text() assert "return_max_long_v32_same_day_longer_tail" in lb_text assert "pead_midcap_step56_short_core_macro_block_crashcap_gap10_interleave_longtrend25" not in lb_text assert "--include-retired" in lb_text def test_manifest_exact_branches_are_structurally_retired(self, tmp_path, monkeypatch): from libs.backtest import tracker as tracker_module cfg_dir = tmp_path / "configs" / "experiments" cfg_dir.mkdir(parents=True) (cfg_dir / "return_max_long_tmp_exact_probe.json").write_text( json.dumps( { "experiment_name": "return_max_long_tmp_exact_probe", "strategy_engines": [ {"engine_id": "next_open_long_probe_exact"}, ], } ) ) tracker_module._load_manifest_json.cache_clear() monkeypatch.setattr(tracker_module, "_EXPERIMENTS_DIR", cfg_dir) assert classify_strategy_family("return_max_long_tmp_exact_probe") == "exact_pocket_return_max_long" assert classify_strategy_family("return_max_long_clean_restart_v326") == "return_max_long" tracker_module._load_manifest_json.cache_clear() def test_manifest_named_micro_branches_are_structurally_retired_only_when_enabled(self, tmp_path, monkeypatch): from libs.backtest import tracker as tracker_module cfg_dir = tmp_path / "configs" / "experiments" cfg_dir.mkdir(parents=True) (cfg_dir / "return_max_long_tmp_named_micro.json").write_text( json.dumps( { "experiment_name": "return_max_long_tmp_named_micro", "strategy_engines": [ {"engine_id": "next_open_long_unknown_guidance_apld_micro", "enabled": True}, ], } ) ) (cfg_dir / "return_max_long_tmp_named_micro_disabled.json").write_text( json.dumps( { "experiment_name": "return_max_long_tmp_named_micro_disabled", "strategy_engines": [ {"engine_id": "next_open_long_unknown_guidance_apld_micro", "enabled": False}, ], } ) ) tracker_module._load_manifest_json.cache_clear() monkeypatch.setattr(tracker_module, "_EXPERIMENTS_DIR", cfg_dir) assert classify_strategy_family("return_max_long_tmp_named_micro") == "named_micro_return_max_long" assert classify_strategy_family("return_max_long_tmp_named_micro_disabled") == "named_micro_return_max_long" tracker_module._load_manifest_json.cache_clear() def test_scan_runs_for_experiment_recurses_and_prefers_latest_split_run(self, tmp_path): runs_dir = tmp_path / "runs" experiment_name = "return_max_long_v1.10" def _write_run(batch: str, run_name: str, split: str, started_at: str, total_return_pct: float) -> None: run_dir = runs_dir / batch / run_name (run_dir / "metrics").mkdir(parents=True) (run_dir / "manifest.json").write_text(json.dumps({"experiment_name": experiment_name})) (run_dir / "metadata.json").write_text( json.dumps( { "run_id": run_name, "split_name": split, "started_at": started_at, } ) ) metrics = MetricsBundle( trade_count=12, profit_factor=1.8, total_return_pct=total_return_pct, win_rate=0.58, monthly_win_rate=0.6, equity_curve_r_squared=0.4, max_drawdown_pct=3.0, sharpe_ratio=1.2, avg_gross_exposure_pct=11.0, avg_net_exposure_pct=11.0, days_in_market_pct=22.0, ) (run_dir / "metrics" / "metrics_summary.json").write_text(metrics.model_dump_json()) _write_run("batch_old", "bt_old_train", "train", "2026-03-20T01:00:00+00:00", 10.0) _write_run("batch_new", "bt_new_train", "train", "2026-03-21T01:00:00+00:00", 25.0) _write_run("batch_new", "bt_new_valid", "valid", "2026-03-21T01:05:00+00:00", 15.0) _write_run("batch_new", "bt_new_test", "test", "2026-03-21T01:10:00+00:00", 18.0) results = scan_runs_for_experiment(runs_dir, experiment_name) assert set(results) == {"train", "valid", "test"} assert results["train"][0] == "bt_new_train" assert results["train"][1].total_return_pct == 25.0 def test_scan_runs_for_experiment_ignores_walk_forward_pseudo_splits(self, tmp_path): runs_dir = tmp_path / "runs" experiment_name = "return_max_long_v1.128" def _write_run(batch: str, run_name: str, split: str, started_at: str, total_return_pct: float) -> None: run_dir = runs_dir / batch / run_name (run_dir / "metrics").mkdir(parents=True) (run_dir / "manifest.json").write_text(json.dumps({"experiment_name": experiment_name})) (run_dir / "metadata.json").write_text( json.dumps( { "run_id": run_name, "split_name": split, "started_at": started_at, } ) ) metrics = MetricsBundle( trade_count=12, profit_factor=1.8, total_return_pct=total_return_pct, win_rate=0.58, monthly_win_rate=0.6, equity_curve_r_squared=0.4, max_drawdown_pct=3.0, sharpe_ratio=1.2, avg_gross_exposure_pct=11.0, avg_net_exposure_pct=11.0, days_in_market_pct=22.0, ) (run_dir / "metrics" / "metrics_summary.json").write_text(metrics.model_dump_json()) _write_run("fullsplit", "bt_train", "train", "2026-03-21T01:00:00+00:00", 10.0) _write_run("fullsplit", "bt_valid", "valid", "2026-03-21T01:05:00+00:00", 11.0) _write_run("fullsplit", "bt_test", "test", "2026-03-21T01:10:00+00:00", 12.0) _write_run("wfv", "bt_wf_train_00", "wf_train_00", "2026-03-21T02:00:00+00:00", 30.0) _write_run("wfv", "bt_wf_test_00", "wf_test_00", "2026-03-21T02:05:00+00:00", 20.0) results = scan_runs_for_experiment(runs_dir, experiment_name) assert set(results) == {"train", "valid", "test"} assert results["train"][0] == "bt_train" assert results["valid"][0] == "bt_valid" assert results["test"][0] == "bt_test" def test_sync_official_manifests_auto_records_missing_manifest(self, tmp_path): journal_path = tmp_path / "journal.jsonl" configs_dir = tmp_path / "configs" / "experiments" runs_dir = tmp_path / "runs" configs_dir.mkdir(parents=True) experiment_name = "return_max_long_v1.10" (configs_dir / f"{experiment_name}.json").write_text( json.dumps( { "experiment_name": experiment_name, "description": "Clean baseline", "tags": ["return-max", "clean-lineage", "v1.10"], "strategy_engines": [ {"engine_id": "reaction_close_long_core", "enabled": True}, {"engine_id": "next_open_long_unknown_guidance_apld_micro", "enabled": False}, ], } ) ) def _write_run(split: str, total_return_pct: float) -> None: run_dir = runs_dir / "batch" / f"bt_{split}" (run_dir / "metrics").mkdir(parents=True) (run_dir / "manifest.json").write_text(json.dumps({"experiment_name": experiment_name})) (run_dir / "metadata.json").write_text( json.dumps( { "run_id": f"bt_{split}", "split_name": split, "started_at": f"2026-03-21T0{1 if split == 'train' else 2 if split == 'valid' else 3}:00:00+00:00", } ) ) metrics = MetricsBundle( trade_count=15, profit_factor=2.0, total_return_pct=total_return_pct, win_rate=0.6, monthly_win_rate=0.75, equity_curve_r_squared=0.5, max_drawdown_pct=3.0, sharpe_ratio=1.7, avg_gross_exposure_pct=14.0, avg_net_exposure_pct=14.0, days_in_market_pct=28.0, ) (run_dir / "metrics" / "metrics_summary.json").write_text(metrics.model_dump_json()) _write_run("train", 40.0) _write_run("valid", 20.0) _write_run("test", 25.0) wfv_dir = runs_dir / f"{experiment_name}_wfv" / "walk_forward" wfv_dir.mkdir(parents=True) wfv_summary = WalkForwardSummary( train_days=504, test_days=63, step_days=63, fold_count=2, folds=[ WalkForwardFoldResult( fold_index=0, train_start="2024-01-01", train_end="2025-12-31", test_start="2026-01-01", test_end="2026-03-31", train_metrics=SplitResult(run_id="wf_train_0", trade_count=20, total_return_pct=40.0), test_metrics=SplitResult( run_id="wf_test_0", trade_count=12, total_return_pct=8.0, win_rate=0.58, profit_factor=1.8, max_drawdown_pct=3.0, ), train_run_id="wf_train_0", test_run_id="wf_test_0", ), WalkForwardFoldResult( fold_index=1, train_start="2024-03-01", train_end="2026-02-28", test_start="2026-03-01", test_end="2026-05-31", train_metrics=SplitResult(run_id="wf_train_1", trade_count=20, total_return_pct=42.0), test_metrics=SplitResult( run_id="wf_test_1", trade_count=11, total_return_pct=6.0, win_rate=0.55, profit_factor=1.6, max_drawdown_pct=2.5, ), train_run_id="wf_train_1", test_run_id="wf_test_1", ), ], train_aggregate=WalkForwardAggregate(mean_return_pct=41.0), test_aggregate=WalkForwardAggregate( mean_return_pct=7.0, median_return_pct=7.0, worst_return_pct=6.0, positive_fold_rate_pct=100.0, mean_profit_factor=1.7, mean_max_drawdown_pct=2.75, mean_trade_count=11.5, mean_win_rate=0.565, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=34.0, worst_train_test_return_gap_pct=36.0, fold_return_cv=0.4, ), ) (wfv_dir / "walk_forward_summary.json").write_text(wfv_summary.model_dump_json()) rm_dir = runs_dir / f"{experiment_name}_rm" / "robustness_matrix" rm_dir.mkdir(parents=True) rm_summary = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=80.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.5, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.2, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=21.0, median_return_pct=18.0, worst_return_pct=8.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=7.0, ), ], ) (rm_dir / "robustness_matrix_summary.json").write_text(rm_summary.model_dump_json()) synced = sync_official_manifests(journal_path, runs_dir, configs_dir) assert [entry.experiment_name for entry in synced] == [experiment_name] persisted = load_journal(journal_path) assert [entry.experiment_name for entry in persisted] == [experiment_name] assert persisted[0].walk_forward_summary is not None assert persisted[0].robustness_matrix_summary is not None assert persisted[0].verdict_reasoning == "Auto-synced from official manifest." def test_leveraged_return_max_entries_are_hidden_by_default(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" wfv, robustness, oot = self._validated_summary_triplet() valid_result = SplitResult( run_id="bt_valid", trade_count=5, profit_factor=1.6, total_return_pct=1.8, win_rate=0.4, monthly_win_rate=0.33, equity_curve_r_squared=0.02, avg_gross_exposure_pct=4.7, avg_net_exposure_pct=4.7, days_in_market_pct=10.8, ) test_result = SplitResult( run_id="bt_test", trade_count=13, profit_factor=3.8, total_return_pct=5.4, win_rate=0.54, monthly_win_rate=0.6, equity_curve_r_squared=0.63, avg_gross_exposure_pct=7.3, avg_net_exposure_pct=7.3, days_in_market_pct=32.6, ) append_journal_entry( journal_path, JournalEntry( entry_id="IMP-0001", timestamp="2026-03-18T16:35:14+00:00", experiment_name="return_max_long_v47_same_day_longer_tail_size200_bp150", hypothesis="leveraged", sqs_score=13.3, results={"valid": valid_result, "test": test_result}, walk_forward_summary=wfv, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, ), ) append_journal_entry( journal_path, JournalEntry( entry_id="IMP-0002", timestamp="2026-03-18T16:40:14+00:00", experiment_name="return_max_long_v42_same_day_longer_tail_size200", hypothesis="unlevered", sqs_score=13.3, results={"valid": valid_result, "test": test_result}, walk_forward_summary=wfv, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, ), ) registry = rebuild_registry(journal_path, registry_path, leaderboard_path) leveraged_entry = next(entry for entry in registry.entries if "_bp" in entry.experiment_name) assert leveraged_entry.strategy_family == "leveraged_return_max_long" assert leveraged_entry.is_retired is True visible = filter_registry_entries(registry.entries, include_retired=False) assert [entry.experiment_name for entry in visible] == [ "return_max_long_v42_same_day_longer_tail_size200" ] def test_incomplete_train_only_entries_are_hidden_by_default(self, tmp_path): journal_path = tmp_path / "journal.jsonl" registry_path = tmp_path / "registry.json" leaderboard_path = tmp_path / "LEADERBOARD.md" train_result = SplitResult( run_id="bt_train_only", trade_count=44, profit_factor=1.9, total_return_pct=33.5, win_rate=0.61, monthly_win_rate=0.53, equity_curve_r_squared=0.73, avg_gross_exposure_pct=18.7, avg_net_exposure_pct=18.7, days_in_market_pct=16.6, ) valid_result = SplitResult( run_id="bt_valid", trade_count=5, profit_factor=1.8, total_return_pct=1.1, win_rate=0.4, monthly_win_rate=0.33, equity_curve_r_squared=0.02, avg_gross_exposure_pct=4.7, avg_net_exposure_pct=4.7, days_in_market_pct=10.8, ) test_result = SplitResult( run_id="bt_test", trade_count=13, profit_factor=4.0, total_return_pct=5.4, win_rate=0.54, monthly_win_rate=0.6, equity_curve_r_squared=0.63, avg_gross_exposure_pct=7.3, avg_net_exposure_pct=7.3, days_in_market_pct=32.6, ) append_journal_entry( journal_path, JournalEntry( entry_id="IMP-0001", timestamp="2026-03-18T16:35:14+00:00", experiment_name="return_max_long_v48_same_day_longer_tail_size400_bp300", hypothesis="train ceiling", sqs_score=71.2, results={"train": train_result}, ), ) append_journal_entry( journal_path, JournalEntry( entry_id="IMP-0002", timestamp="2026-03-18T16:40:14+00:00", experiment_name="return_max_long_v47_same_day_longer_tail_size200_bp150", hypothesis="validated", sqs_score=13.3, results={"valid": valid_result, "test": test_result}, ), ) registry = rebuild_registry(journal_path, registry_path, leaderboard_path) visible = filter_registry_entries(registry.entries, include_retired=False) assert visible == [] lb_text = leaderboard_path.read_text() assert "return_max_long_v47_same_day_longer_tail_size200_bp150" not in lb_text assert "return_max_long_v48_same_day_longer_tail_size400_bp300" not in lb_text assert "incomplete train-only scans" in lb_text # --------------------------------------------------------------------------- # WFQS v2: multiplicative penalty helpers # --------------------------------------------------------------------------- class TestGapPenalty: def test_low_gap_no_penalty(self): assert _gap_penalty(20.0) == 1.0 assert _gap_penalty(30.0) == 1.0 def test_moderate_gap(self): assert _gap_penalty(65.0) == pytest.approx(0.8, abs=0.01) def test_high_gap(self): assert _gap_penalty(100.0) == pytest.approx(0.6, abs=0.01) def test_very_high_gap(self): assert _gap_penalty(200.0) == pytest.approx(0.45, abs=0.01) def test_extreme_gap(self): assert _gap_penalty(300.0) == pytest.approx(0.3, abs=0.01) assert _gap_penalty(500.0) == 0.2 def test_none_gap(self): assert _gap_penalty(None) == 1.0 class TestFoldVariancePenalty: def test_low_cv_no_penalty(self): assert _fold_variance_penalty(0.3) == 1.0 assert _fold_variance_penalty(0.5) == 1.0 def test_moderate_cv(self): assert _fold_variance_penalty(1.0) == pytest.approx(0.85, abs=0.01) def test_high_cv(self): assert _fold_variance_penalty(1.5) == pytest.approx(0.7, abs=0.01) def test_extreme_cv(self): assert _fold_variance_penalty(2.0) == 0.6 def test_none_cv(self): assert _fold_variance_penalty(None) == 1.0 class TestTradeCredibility: def test_many_trades(self): assert _trade_credibility(25.0, 0.6) == 1.0 def test_few_trades(self): assert _trade_credibility(3.0, 0.6) == 0.5 def test_moderate_trades(self): assert _trade_credibility(7.0, 0.6) == 0.7 def test_suspicious_win_rate(self): assert _trade_credibility(12.0, 0.98) == pytest.approx(0.85 * 0.8, abs=0.01) def test_high_wr_with_enough_trades(self): # win_rate > 0.95 but mean_trades >= 15 → no additional penalty assert _trade_credibility(20.0, 0.98) == 1.0 def test_none_trades(self): assert _trade_credibility(None, 0.6) == 0.5 class TestEngineReliabilityPenalty: def test_high_ratio(self): assert _engine_reliability_penalty(0.8) == 1.0 def test_medium_ratio(self): assert _engine_reliability_penalty(0.5) == pytest.approx(0.75, abs=0.01) def test_low_ratio(self): assert _engine_reliability_penalty(0.2) == 0.4 def test_none_ratio(self): assert _engine_reliability_penalty(None) == 1.0 # --------------------------------------------------------------------------- # WFQS v2: full computation # --------------------------------------------------------------------------- class TestComputeWfqsV2: def test_returns_none_for_no_summary(self): score, breakdown = compute_wfqs_v2(None) assert score is None assert breakdown == {} def test_returns_none_for_zero_folds(self): summary = WalkForwardSummary(train_days=252, test_days=63, step_days=63, fold_count=0) score, breakdown = compute_wfqs_v2(summary) assert score is None def test_base_score_without_penalties(self): """When gap is low and CV is low, penalties are ~1.0 so wfqs_v2 ≈ base_score.""" summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=14.0, median_return_pct=12.0, worst_return_pct=2.0, positive_fold_rate_pct=87.5, mean_profit_factor=2.2, mean_max_drawdown_pct=4.0, mean_trade_count=25.0, mean_win_rate=0.6, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=18.0, worst_train_test_return_gap_pct=30.0, fold_return_cv=0.3, ), ) score, breakdown = compute_wfqs_v2(summary) assert score is not None assert score > 0 # All penalties should be ~1.0 assert breakdown["gap_penalty"] == 1.0 assert breakdown["fold_variance_penalty"] == 1.0 assert breakdown["trade_credibility"] == 1.0 assert breakdown["engine_reliability"] == 1.0 # base_score and final should be close assert abs(score - breakdown["base_score"]) < 1.0 def test_overfitted_profile_scores_lower(self): """v1045-like overfitted profile should score much lower than general-only-like.""" # General-only profile: low gap, low CV, good trades general = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=30.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=37.0, worst_train_test_return_gap_pct=60.0, fold_return_cv=0.4, ), ) # Overfitted profile: extreme gap, high CV, few trades, high win rate overfit = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=20.0, median_return_pct=18.0, worst_return_pct=5.0, positive_fold_rate_pct=100.0, mean_profit_factor=3.0, mean_max_drawdown_pct=3.0, mean_trade_count=8.0, mean_win_rate=0.97, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=523.0, worst_train_test_return_gap_pct=900.0, fold_return_cv=1.8, ), ) general_score, _ = compute_wfqs_v2(general) overfit_score, overfit_breakdown = compute_wfqs_v2(overfit) assert general_score is not None assert overfit_score is not None # The overfit profile should be drastically penalized assert general_score > overfit_score # Gap penalty should be severe assert overfit_breakdown["gap_penalty"] == 0.2 # Fold variance should also penalize assert overfit_breakdown["fold_variance_penalty"] == 0.6 # Trade credibility should penalize (few trades + suspicious WR) assert overfit_breakdown["trade_credibility"] < 1.0 def test_recent_fold_strength_receives_modest_bonus(self): def _fold(i: int, end_date: dt.date, ret: float, pf: float = 1.8, dd: float = 5.0) -> WalkForwardFoldResult: metrics = SplitResult( run_id=f"bt_fold_{i}", trade_count=25, total_return_pct=ret, profit_factor=pf, max_drawdown_pct=dd, win_rate=0.55, ) return WalkForwardFoldResult( fold_index=i, train_start=end_date - dt.timedelta(days=315), train_end=end_date - dt.timedelta(days=64), test_start=end_date - dt.timedelta(days=63), test_end=end_date, train_run_id=f"train_{i}", test_run_id=f"test_{i}", train_metrics=metrics, test_metrics=metrics, ) common_kwargs = dict( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=25.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) strong_recent = WalkForwardSummary( **common_kwargs, folds=[ _fold(0, dt.date(2024, 2, 1), 6.0), _fold(1, dt.date(2024, 4, 1), 7.0), _fold(2, dt.date(2024, 6, 1), 8.0), _fold(3, dt.date(2024, 8, 1), 9.0), _fold(4, dt.date(2024, 10, 1), 14.0), _fold(5, dt.date(2024, 12, 1), 16.0), _fold(6, dt.date(2025, 2, 1), 18.0), _fold(7, dt.date(2025, 4, 1), 20.0), ], ) weak_recent = WalkForwardSummary( **common_kwargs, folds=[ _fold(0, dt.date(2024, 2, 1), 20.0), _fold(1, dt.date(2024, 4, 1), 18.0), _fold(2, dt.date(2024, 6, 1), 16.0), _fold(3, dt.date(2024, 8, 1), 14.0), _fold(4, dt.date(2024, 10, 1), 9.0), _fold(5, dt.date(2024, 12, 1), 8.0), _fold(6, dt.date(2025, 2, 1), 7.0), _fold(7, dt.date(2025, 4, 1), 6.0), ], ) strong_score, strong_breakdown = compute_wfqs_v2(strong_recent) weak_score, weak_breakdown = compute_wfqs_v2(weak_recent) assert strong_score is not None assert weak_score is not None assert strong_breakdown["recent_fold_count"] >= 2 assert strong_breakdown["recent_quality"] > weak_breakdown["recent_quality"] assert strong_score > weak_score def test_engine_reliability_applied(self): """Engine reliability penalty should reduce score when ratio is low.""" summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=30.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), engine_reliability_ratio=0.2, ) score, breakdown = compute_wfqs_v2(summary) assert score is not None assert breakdown["engine_reliability"] == 0.4 # Without engine reliability summary_no_er = summary.model_copy(update={"engine_reliability_ratio": None}) score_no_er, _ = compute_wfqs_v2(summary_no_er) assert score_no_er is not None assert score_no_er > score def test_single_fold(self): summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=1, test_aggregate=WalkForwardAggregate( mean_return_pct=5.0, median_return_pct=5.0, worst_return_pct=5.0, positive_fold_rate_pct=100.0, mean_profit_factor=1.5, mean_max_drawdown_pct=3.0, mean_trade_count=10.0, mean_win_rate=0.5, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=10.0, ), ) score, breakdown = compute_wfqs_v2(summary) assert score is not None # Low fold penalty applied assert breakdown["base_score"] > score / breakdown["gap_penalty"] # base was penalized def test_zero_trades(self): summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=4, test_aggregate=WalkForwardAggregate( mean_return_pct=0.0, median_return_pct=0.0, worst_return_pct=0.0, positive_fold_rate_pct=0.0, mean_trade_count=0.0, mean_win_rate=0.0, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=0.0, ), ) score, breakdown = compute_wfqs_v2(summary) assert score is not None assert breakdown["trade_credibility"] == 0.5 class TestComputePublicSqsV2: def test_returns_none_without_rqs(self): score, breakdown, source = compute_public_sqs_v2(None, None, None) assert score is None assert source is None def test_requires_validations(self): train = SplitResult(run_id="bt_train", trade_count=10, total_return_pct=20.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=10, total_return_pct=10.0, win_rate=0.6) test = SplitResult(run_id="bt_test", trade_count=10, total_return_pct=12.0, win_rate=0.6) score, breakdown, source = compute_public_sqs_v2(train, valid, test) assert score is None assert source == "pending_validation" assert breakdown["requires_walk_forward"] == 1.0 assert breakdown["requires_robustness"] == 1.0 assert breakdown["requires_out_of_time_robustness"] == 1.0 def test_deployment_with_wfqs_v2(self): train = SplitResult(run_id="bt_train", trade_count=10, total_return_pct=20.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=10, total_return_pct=10.0, win_rate=0.6) test = SplitResult(run_id="bt_test", trade_count=10, total_return_pct=12.0, win_rate=0.6) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=25.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=90.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=18.0, median_return_pct=12.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) oot = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=80.0, overall_worst_return_pct=-4.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=3.0, median_return_pct=3.0, worst_return_pct=-1.5, positive_window_rate_pct=66.7, mean_max_drawdown_pct=3.2, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=11.0, median_return_pct=9.0, worst_return_pct=2.0, positive_window_rate_pct=75.0, mean_max_drawdown_pct=5.2, ), ], ) score, breakdown, source = compute_public_sqs_v2( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, ) assert score is not None assert source == "v2_deployment+robustness+oot" assert breakdown["oot_gate_factor"] > 0.0 def test_out_of_time_robustness_applies_additional_gate(self): train = SplitResult(run_id="bt_train", trade_count=10, total_return_pct=20.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=10, total_return_pct=10.0, win_rate=0.6) test = SplitResult(run_id="bt_test", trade_count=10, total_return_pct=12.0, win_rate=0.6) summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=25.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=90.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=18.0, median_return_pct=12.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) oot = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=60.0, overall_worst_return_pct=-15.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=2.0, median_return_pct=2.0, worst_return_pct=-2.0, positive_window_rate_pct=66.7, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=7.0, median_return_pct=7.0, worst_return_pct=-1.0, positive_window_rate_pct=75.0, mean_max_drawdown_pct=5.0, ), ], ) strong_oot = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=85.0, overall_worst_return_pct=-3.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=3.0, median_return_pct=3.0, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=2.5, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=10.0, median_return_pct=8.5, worst_return_pct=1.5, positive_window_rate_pct=100.0, mean_max_drawdown_pct=4.8, ), ], ) strong_score, _, _ = compute_public_sqs_v2( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=strong_oot, ) oot_score, breakdown, source = compute_public_sqs_v2( train, valid, test, walk_forward_summary=summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=oot, ) assert strong_score is not None assert oot_score is not None assert oot_score < strong_score assert source == "v2_deployment+robustness+oot" assert "oot_gate_factor" in breakdown assert "oot_quality_factor" in breakdown assert "oot_quality" in breakdown def test_low_activity_strategy_gets_lower_public_score(self): train = SplitResult(run_id="bt_train", trade_count=20, total_return_pct=20.0, win_rate=0.7) valid = SplitResult(run_id="bt_valid", trade_count=15, total_return_pct=10.0, win_rate=0.6) active_test = SplitResult( run_id="bt_test_active", trade_count=24, total_return_pct=12.0, win_rate=0.6, days_in_market_pct=55.0, ) sparse_test = SplitResult( run_id="bt_test_sparse", trade_count=9, total_return_pct=12.0, win_rate=0.6, days_in_market_pct=18.0, ) active_summary = WalkForwardSummary( train_days=252, test_days=63, step_days=63, fold_count=8, test_aggregate=WalkForwardAggregate( mean_return_pct=10.0, median_return_pct=8.0, worst_return_pct=1.0, positive_fold_rate_pct=87.5, mean_profit_factor=1.8, mean_max_drawdown_pct=5.0, mean_trade_count=12.0, mean_win_rate=0.55, ), gap_stats=WalkForwardGapStats( mean_train_test_return_gap_pct=20.0, fold_return_cv=0.3, ), ) sparse_summary = active_summary.model_copy( update={ "test_aggregate": active_summary.test_aggregate.model_copy( update={"mean_trade_count": 3.5} ) } ) robustness = RobustnessMatrixSummary( horizons_days=[63, 252], step_days=21, overall_window_count=10, overall_positive_window_rate_pct=90.0, overall_worst_return_pct=-2.0, horizon_summaries=[ RobustnessHorizonSummary( horizon_days=63, window_count=6, mean_return_pct=4.0, median_return_pct=3.5, worst_return_pct=-1.0, positive_window_rate_pct=83.3, mean_max_drawdown_pct=3.0, ), RobustnessHorizonSummary( horizon_days=252, window_count=4, mean_return_pct=18.0, median_return_pct=12.0, worst_return_pct=4.0, positive_window_rate_pct=100.0, mean_max_drawdown_pct=5.0, ), ], ) active_score, active_breakdown, _ = compute_public_sqs_v2( train, valid, active_test, walk_forward_summary=active_summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=robustness, ) sparse_score, sparse_breakdown, _ = compute_public_sqs_v2( train, valid, sparse_test, walk_forward_summary=sparse_summary, robustness_matrix_summary=robustness, out_of_time_robustness_summary=robustness, ) assert active_score is not None assert sparse_score is not None assert sparse_score < active_score assert active_breakdown["activity_factor"] > sparse_breakdown["activity_factor"]