from __future__ import annotations from pathlib import Path from apps.intraday_bt.lab import ( _deserialize_finalist_eval_row, _engine_specs, _pre_robustness_rank_key, _promotion_status, _read_json, _sample_representative_days, _select_champions, _serialize_finalist_eval_row, ) from apps.intraday_bt.overfit_check import ( summarize_is_oos_from_results, summarize_walk_forward_test_from_summary, ) from libs.backtest.domain import ( SplitResult, WalkForwardAggregate, WalkForwardFoldResult, WalkForwardGapStats, WalkForwardSummary, ) from libs.intraday.domain import IntradayMetrics, ORBStrategyParams def test_sample_representative_days_preserves_order_and_endpoints() -> None: days = [f"2026-01-{day:02d}" for day in range(2, 22)] sampled = _sample_representative_days(days, 5) assert sampled[0] == days[0] assert sampled[-1] == days[-1] assert sampled == sorted(sampled) assert len(sampled) == 5 def test_pre_robustness_rank_key_prefers_test_then_valid_then_activity() -> None: weak_test = { "train_sharpe": 3.0, "valid_sharpe": 2.0, "test_sharpe": 0.5, "test_trade_count": 200, } strong_test = { "train_sharpe": 1.0, "valid_sharpe": 0.8, "test_sharpe": 0.9, "test_trade_count": 50, } assert _pre_robustness_rank_key(strong_test) > _pre_robustness_rank_key(weak_test) def test_engine_specs_quick_are_curated_and_small() -> None: specs = _engine_specs(True) assert [spec.family for spec in specs] == [ "classic_breakout", "quality_breakout", "gainers_leader", "compression_breakout", ] assert all(spec.thesis for spec in specs) assert [len(spec.hypotheses) for spec in specs] == [4, 4, 4, 4] quality = next(spec for spec in specs if spec.family == "quality_breakout") assert all(h["entry_direction"] == "long_only" for h in quality.hypotheses) assert all(h["min_candidate_breadth"] is not None for h in quality.hypotheses) assert all(h["market_regime_spy_threshold"] is not None for h in quality.hypotheses) gainers = next(spec for spec in specs if spec.family == "gainers_leader") assert all(h["entry_direction"] == "long_only" for h in gainers.hypotheses) assert all(h["max_gap_pct"] is None for h in gainers.hypotheses) assert all(h["min_candidates_to_trade"] == 1 for h in gainers.hypotheses) compression = next(spec for spec in specs if spec.family == "compression_breakout") assert all(h["min_candidate_breadth"] is not None for h in compression.hypotheses) assert all(h["market_regime_spy_threshold"] is not None for h in compression.hypotheses) def test_finalist_eval_row_round_trips() -> None: row = { "candidate_id": "abc", "engine_family": "quality_breakout", "live_readiness": "live_ready", "promotion_status": "eligible", "overrides": {"orb_minutes": 5}, "params": ORBStrategyParams(orb_minutes=5), "train_metrics_obj": IntradayMetrics(run_id="tr", trading_days=10, total_trades=5), "valid_metrics_obj": IntradayMetrics(run_id="va", trading_days=10, total_trades=4), "test_metrics_obj": IntradayMetrics(run_id="te", trading_days=10, total_trades=6), "train_result": None, "valid_result": None, "test_result": None, } payload = _serialize_finalist_eval_row(row) restored = _deserialize_finalist_eval_row(payload) assert restored["candidate_id"] == "abc" assert restored["params"].orb_minutes == 5 assert restored["test_metrics_obj"].run_id == "te" def test_read_json_returns_default_for_missing_file(tmp_path: Path) -> None: missing = tmp_path / "missing.json" assert _read_json(missing, default={"ok": True}) == {"ok": True} def test_summarize_walk_forward_test_from_summary_reuses_existing_folds() -> None: wf_summary = WalkForwardSummary( train_days=84, test_days=21, step_days=21, fold_count=3, folds=[ WalkForwardFoldResult( fold_index=1, train_start="2025-01-02", train_end="2025-03-31", test_start="2025-04-01", test_end="2025-04-30", train_run_id="tr1", test_run_id="te1", train_metrics=SplitResult(run_id="tr1", trade_count=10, sharpe_ratio=1.0), test_metrics=SplitResult(run_id="te1", trade_count=10, sharpe_ratio=0.9), ), WalkForwardFoldResult( fold_index=2, train_start="2025-02-01", train_end="2025-04-30", test_start="2025-05-01", test_end="2025-05-31", train_run_id="tr2", test_run_id="te2", train_metrics=SplitResult(run_id="tr2", trade_count=10, sharpe_ratio=1.0), test_metrics=SplitResult(run_id="te2", trade_count=10, sharpe_ratio=0.7), ), WalkForwardFoldResult( fold_index=3, train_start="2025-03-01", train_end="2025-05-31", test_start="2025-06-01", test_end="2025-06-30", train_run_id="tr3", test_run_id="te3", train_metrics=SplitResult(run_id="tr3", trade_count=10, sharpe_ratio=1.0), test_metrics=SplitResult(run_id="te3", trade_count=10, sharpe_ratio=0.8), ), ], train_aggregate=WalkForwardAggregate(), test_aggregate=WalkForwardAggregate(), gap_stats=WalkForwardGapStats(), ) result = summarize_walk_forward_test_from_summary(wf_summary) assert result["source"] == "walk_forward_summary" assert result["n_windows"] == 3 assert result["window_sharpes"] == [0.9, 0.7, 0.8] assert result["verdict"] == "PASS" def test_summarize_is_oos_from_results_reuses_existing_splits() -> None: result = summarize_is_oos_from_results( SplitResult(run_id="is", trade_count=100, sharpe_ratio=1.0), SplitResult(run_id="oos", trade_count=80, sharpe_ratio=0.7), is_period="2024-01-02 → 2025-12-31", oos_period="2026-01-02 → 2026-03-31", ) assert result["source"] == "split_results" assert result["verdict"] == "PASS" assert result["retention_pct"] == 70.0 def test_promotion_status_blocks_negative_oos_even_with_activity() -> None: valid = SplitResult(run_id="valid", trade_count=120, total_return_pct=-1.0) test = SplitResult(run_id="test", trade_count=120, total_return_pct=2.0) assert _promotion_status(valid, test) == "blocked_negative_oos" def test_select_champions_uses_only_eligible_rows() -> None: ranking = [ { "candidate_id": "blocked", "promotion_status": "blocked_negative_oos", "live_readiness": "live_ready", "orbqs_score": 30.0, }, { "candidate_id": "eligible_live", "promotion_status": "eligible", "live_readiness": "live_ready", "orbqs_score": 20.0, }, { "candidate_id": "eligible_research", "promotion_status": "eligible", "live_readiness": "research_only", "orbqs_score": 10.0, }, ] top_candidate, overall, live_ready = _select_champions(ranking) assert top_candidate["candidate_id"] == "blocked" assert overall["candidate_id"] == "eligible_live" assert live_ready["candidate_id"] == "eligible_live"