from __future__ import annotations from apps.intraday_bt.orb_research import force_simple_returns from apps.intraday_bt.orb_validate import ( chronological_splits, compute_orb_validation_score, delta_payload, metrics_payload, period_quality_score, promotion_gate, tail_days, walk_forward_verdict, ) from libs.intraday.domain import IntradayConfig, IntradayMetrics, ORBStrategyParams def test_tail_days_returns_recent_window() -> None: days = [f"2026-01-{idx:02d}" for idx in range(1, 8)] assert tail_days(days, 3) == ["2026-01-05", "2026-01-06", "2026-01-07"] assert tail_days(days, 20) == days def test_chronological_splits_use_fixed_two_year_layout() -> None: days = [f"2025-01-{idx:03d}" for idx in range(1, 505)] splits = chronological_splits(days) assert len(splits["train"]) == 252 assert len(splits["valid"]) == 126 assert len(splits["test"]) == 126 assert splits["train"][0] == "2025-01-001" assert splits["test"][-1] == "2025-01-504" def test_metrics_payload_converts_decimal_metrics_to_pct_units() -> None: payload = metrics_payload( IntradayMetrics( run_id="r1", start_date="2026-01-02", end_date="2026-01-31", trading_days=20, days_with_trades=10, total_trades=15, total_return_pct=1.2345, max_drawdown_pct=-0.0123, sharpe_ratio=2.345, profit_factor=3.456, loss_day_rate=0.25, worst_day_return_pct=-0.02, ) ) assert payload["return_pct"] == 123.45 assert payload["max_drawdown_pct"] == -1.23 assert payload["sharpe_ratio"] == 2.35 assert payload["profit_factor"] == 3.46 assert payload["loss_day_rate_pct"] == 25.0 assert payload["worst_day_return_pct"] == -2.0 def test_period_quality_penalizes_large_drawdown_profile() -> None: robust = { "period": {"trading_days": 504}, "days_with_trades": 190, "total_trades": 300, "stop_loss_exits": 20, "return_pct": 140.0, "annualized_return_pct": 70.0, "max_drawdown_pct": -8.0, "sharpe_ratio": 3.0, "profit_factor": 2.8, "loss_day_rate_pct": 25.0, "tail_loss_20_pct": -1.2, "worst_day_return_pct": -2.5, } fragile = { **robust, "return_pct": 155.0, "max_drawdown_pct": -32.0, "sharpe_ratio": 1.2, "profit_factor": 1.3, "tail_loss_20_pct": -5.0, "worst_day_return_pct": -9.0, } robust_score, _ = period_quality_score(robust) fragile_score, _ = period_quality_score(fragile) assert robust_score > fragile_score def test_delta_payload_returns_candidate_minus_baseline() -> None: candidate = {"return_pct": 110.0, "max_drawdown_pct": -1.0, "total_trades": 15} baseline = {"return_pct": 100.0, "max_drawdown_pct": -2.5, "total_trades": 10} assert delta_payload(candidate, baseline) == { "return_pct": 10.0, "max_drawdown_pct": 1.5, "total_trades": 5, } def test_walk_forward_verdict_requires_positive_stable_folds() -> None: report = { "fold_count": 4, "test_aggregate": { "positive_fold_rate_pct": 100.0, "worst_return_pct": 1.2, "mean_return_pct": 3.4, }, "gap_stats": {"fold_return_cv": 0.5}, } assert walk_forward_verdict(report)["verdict"] == "PASS" def test_promotion_gate_fails_primary_damage_even_if_recent_guards_pass() -> None: candidate = { "periods": { "primary": {"return_pct": 90.0, "max_drawdown_pct": -3.0}, "guard_200d": {"return_pct": 60.0, "max_drawdown_pct": -1.0}, "guard_60d": {"return_pct": 20.0, "max_drawdown_pct": -0.5}, }, "deltas": { "primary": {"return_pct": -12.0, "max_drawdown_pct": -2.0}, "guard_200d": {"return_pct": 0.0, "max_drawdown_pct": 0.0}, "guard_60d": {"return_pct": 0.0, "max_drawdown_pct": 0.0}, }, "walk_forward": {"verdict": {"verdict": "PASS"}}, } assert promotion_gate(candidate, baseline={"periods": {}})["overall"] == "FAIL" def test_promotion_gate_ignores_guards_that_were_not_requested() -> None: candidate = { "periods": { "primary": {"return_pct": 100.0, "max_drawdown_pct": -1.0}, "guard_60d": {"return_pct": 20.0, "max_drawdown_pct": -0.5}, }, "deltas": { "primary": {"return_pct": 0.0, "max_drawdown_pct": 0.0}, "guard_60d": {"return_pct": 0.0, "max_drawdown_pct": 0.0}, }, "walk_forward": {"verdict": {"verdict": "SKIP"}}, } assert promotion_gate(candidate, baseline={"periods": {}})["overall"] == "PASS" def _split_result(run_id: str, ret: float, dd: float = 5.0) -> dict: return { "run_id": run_id, "trade_count": 40, "profit_factor": 2.0, "total_return_pct": ret, "annualized_return_pct": ret, "win_rate": 0.65, "max_drawdown_pct": dd, "sharpe_ratio": 2.0, "avg_gross_exposure_pct": 20.0, "avg_net_exposure_pct": 20.0, "days_in_market_pct": 40.0, } def _candidate_for_score(*, primary_return: float, guard_200_return: float) -> dict: period = { "period": {"trading_days": 504}, "days_with_trades": 180, "total_trades": 260, "stop_loss_exits": 25, "return_pct": primary_return, "annualized_return_pct": 65.0, "max_drawdown_pct": -10.0, "sharpe_ratio": 2.5, "profit_factor": 2.4, "loss_day_rate_pct": 25.0, "tail_loss_20_pct": -1.5, "worst_day_return_pct": -3.0, } return { "periods": { "primary": period, "guard_200d": {**period, "period": {"trading_days": 200}, "return_pct": guard_200_return}, "guard_60d": {**period, "period": {"trading_days": 60}, "return_pct": 20.0}, }, "splits": { "train": {"split_result": _split_result("train", 45.0)}, "valid": {"split_result": _split_result("valid", 35.0)}, "test": {"split_result": _split_result("test", 40.0)}, }, "walk_forward": { "train_days": 252, "test_days": 63, "step_days": 63, "fold_count": 4, "folds": [], "train_aggregate": {"mean_win_rate": 0.65}, "test_aggregate": { "mean_return_pct": 8.0, "median_return_pct": 8.0, "worst_return_pct": 2.0, "positive_fold_rate_pct": 100.0, "mean_profit_factor": 2.0, "mean_max_drawdown_pct": 4.0, "mean_trade_count": 25.0, "mean_win_rate": 0.62, }, "gap_stats": { "mean_train_test_return_gap_pct": 20.0, "worst_train_test_return_gap_pct": 35.0, "fold_return_cv": 0.4, }, "engine_reliability_ratio": 1.0, }, } def test_orb_validation_score_penalizes_recent_window_concentration() -> None: balanced = _candidate_for_score(primary_return=150.0, guard_200_return=80.0) concentrated = _candidate_for_score(primary_return=150.0, guard_200_return=145.0) balanced_score = compute_orb_validation_score(balanced) concentrated_score = compute_orb_validation_score(concentrated) assert balanced_score["score"] > concentrated_score["score"] assert concentrated_score["temporal_breakdown"]["pre_200_return_pct"] == 5.0 def test_orb_validation_score_does_not_rank_compound_runs() -> None: candidate = _candidate_for_score(primary_return=150.0, guard_200_return=80.0) candidate["capital_mode"] = { "compound_returns": True, "daily_budget_reset": False, "settlement_days": 1, } score = compute_orb_validation_score(candidate) assert score["score"] is None assert score["source"] == "compound_diagnostic_not_ranked" assert score["policy"]["rank_compound_runs"] is False assert score["requires"] == ["daily_budget_reset_primary"] def test_force_simple_returns_uses_daily_reset_for_orb_sqs() -> None: config = IntradayConfig( strategy_mode="orb", orb_strategy=ORBStrategyParams( compound_returns=True, daily_budget_reset=False, ), ) normalized = force_simple_returns(config) assert normalized.orb_strategy is not None assert normalized.orb_strategy.compound_returns is False assert normalized.orb_strategy.daily_budget_reset is True assert config.orb_strategy is not None assert config.orb_strategy.compound_returns is True assert config.orb_strategy.daily_budget_reset is False