You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1187 lines
43 KiB
Python
1187 lines
43 KiB
Python
"""Promotion-grade ORB strategy validation.
|
|
|
|
This command is intentionally narrower than the research lab. It evaluates one
|
|
candidate, optionally against a baseline, using the validation matrix that should
|
|
gate new ORB strategy promotion:
|
|
|
|
- primary two-year-ish daily reset window (default: 504 trading days)
|
|
- recent 200d and 60d guard windows
|
|
- rolling walk-forward validation over the full requested range
|
|
|
|
ORB SQS is intentionally reset-only. Compound runs are useful deployment
|
|
diagnostics, but they must not become the optimization target because late
|
|
equity dominates the objective.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from apps.intraday_bt.oracle import make_intraday_oracle_client
|
|
from apps.intraday_bt.orb_research import (
|
|
build_orb_research_context,
|
|
build_walk_forward_summary,
|
|
force_simple_returns,
|
|
generate_walk_forward_windows,
|
|
intraday_metrics_to_split_result,
|
|
resolve_orb_config,
|
|
simulate_orb_period,
|
|
)
|
|
from apps.intraday_bt.run import _latest_backtest_date, get_trading_days, run as run_backtest
|
|
from libs.backtest.domain import SplitResult, WalkForwardSummary
|
|
from libs.backtest.tracker import compute_rqs, compute_wfqs_v2
|
|
from libs.common.config import get_settings
|
|
from libs.intraday.domain import DayResult, IntradayConfig, IntradayMetrics, ORBStrategyParams
|
|
from libs.intraday.metrics import compute_metrics
|
|
from libs.oracle_client import OracleClient
|
|
|
|
|
|
DEFAULT_START_DATE = "2024-01-01"
|
|
DEFAULT_PRIMARY_DAYS = 504
|
|
DEFAULT_GUARD_DAYS = (200, 60)
|
|
DEFAULT_WF_TRAIN_DAYS = 252
|
|
DEFAULT_WF_TEST_DAYS = 63
|
|
DEFAULT_SPLIT_TRAIN_DAYS = 252
|
|
DEFAULT_SPLIT_VALID_DAYS = 126
|
|
|
|
ORB_SQS_POLICY = {
|
|
"primary_objective": "daily_budget_reset",
|
|
"compound_role": "deployment_diagnostic_only",
|
|
"rank_compound_runs": False,
|
|
"notes": (
|
|
"Develop signals on daily reset ORB SQS. Use compound runs only to "
|
|
"validate allocator/capital-policy settings such as position caps, "
|
|
"loss-cap basis, drawdown governor, settlement, and GFV behavior."
|
|
),
|
|
}
|
|
|
|
|
|
def _round(value: float | None, digits: int = 2) -> float | None:
|
|
if value is None:
|
|
return None
|
|
return round(value, digits)
|
|
|
|
|
|
def _pct(value: float | None, digits: int = 2) -> float | None:
|
|
if value is None:
|
|
return None
|
|
return round(value * 100.0, digits)
|
|
|
|
|
|
def _clip(value: float, low: float = 0.0, high: float = 100.0) -> float:
|
|
return max(low, min(high, value))
|
|
|
|
|
|
def _norm(value: float | int | None, low: float, high: float) -> float:
|
|
if value is None:
|
|
return 0.0
|
|
if high <= low:
|
|
raise ValueError("high must be greater than low")
|
|
return _clip((float(value) - low) / (high - low) * 100.0)
|
|
|
|
|
|
def _norm_inverse(value: float | int | None, low: float, high: float) -> float:
|
|
"""Score high when value is at or below `low`, low when at or above `high`."""
|
|
if value is None:
|
|
return 0.0
|
|
if high <= low:
|
|
raise ValueError("high must be greater than low")
|
|
return _clip((high - float(value)) / (high - low) * 100.0)
|
|
|
|
|
|
def _period(days: list[str]) -> dict[str, Any]:
|
|
return {
|
|
"start": days[0] if days else "",
|
|
"end": days[-1] if days else "",
|
|
"trading_days": len(days),
|
|
}
|
|
|
|
|
|
def tail_days(trading_days: list[str], count: int) -> list[str]:
|
|
"""Return the last `count` trading days, or all days when shorter."""
|
|
if count <= 0:
|
|
raise ValueError("count must be positive")
|
|
return trading_days[-count:]
|
|
|
|
|
|
def chronological_splits(
|
|
trading_days: list[str],
|
|
*,
|
|
train_days: int = DEFAULT_SPLIT_TRAIN_DAYS,
|
|
valid_days: int = DEFAULT_SPLIT_VALID_DAYS,
|
|
) -> dict[str, list[str]]:
|
|
"""Build fixed chronological train/valid/test splits inside the primary period."""
|
|
if len(trading_days) < 60:
|
|
raise ValueError("at least 60 trading days are required for chronological splits")
|
|
|
|
if len(trading_days) >= train_days + valid_days + 20:
|
|
train_end = train_days
|
|
valid_end = train_end + valid_days
|
|
else:
|
|
train_end = max(20, int(len(trading_days) * 0.50))
|
|
valid_end = max(train_end + 20, int(len(trading_days) * 0.75))
|
|
valid_end = min(valid_end, len(trading_days) - 20)
|
|
|
|
return {
|
|
"train": trading_days[:train_end],
|
|
"valid": trading_days[train_end:valid_end],
|
|
"test": trading_days[valid_end:],
|
|
}
|
|
|
|
|
|
def metrics_payload(metrics: IntradayMetrics) -> dict[str, Any]:
|
|
"""Compact JSON-friendly metrics summary with percent fields in pct units."""
|
|
return {
|
|
"run_id": metrics.run_id,
|
|
"period": {
|
|
"start": metrics.start_date,
|
|
"end": metrics.end_date,
|
|
"trading_days": metrics.trading_days,
|
|
},
|
|
"days_with_trades": metrics.days_with_trades,
|
|
"days_with_activity": metrics.days_with_activity,
|
|
"total_trades": metrics.total_trades,
|
|
"stop_loss_exits": metrics.stop_loss_exits,
|
|
"return_pct": _pct(metrics.total_return_pct),
|
|
"annualized_return_pct": _pct(metrics.annualized_return_pct),
|
|
"avg_daily_return_pct": _pct(metrics.avg_daily_return_pct, digits=4),
|
|
"max_drawdown_pct": _pct(metrics.max_drawdown_pct),
|
|
"sharpe_ratio": _round(metrics.sharpe_ratio, 2),
|
|
"sortino_ratio": _round(metrics.sortino_ratio, 2),
|
|
"profit_factor": _round(metrics.profit_factor, 2),
|
|
"win_rate_pct": _pct(metrics.win_rate),
|
|
"loss_day_rate_pct": _pct(metrics.loss_day_rate),
|
|
"tail_loss_20_pct": _pct(metrics.tail_loss_20_pct),
|
|
"worst_day_return_pct": _pct(metrics.worst_day_return_pct),
|
|
"loss_containment_score": _round(metrics.loss_containment_score, 2),
|
|
}
|
|
|
|
|
|
def metrics_with_split_payload(
|
|
metrics: IntradayMetrics,
|
|
params: ORBStrategyParams,
|
|
) -> dict[str, Any]:
|
|
payload = metrics_payload(metrics)
|
|
payload["split_result"] = intraday_metrics_to_split_result(metrics, params).model_dump(mode="json")
|
|
return payload
|
|
|
|
|
|
def delta_payload(candidate: dict[str, Any], baseline: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if baseline is None:
|
|
return None
|
|
keys = [
|
|
"return_pct",
|
|
"max_drawdown_pct",
|
|
"sharpe_ratio",
|
|
"profit_factor",
|
|
"days_with_trades",
|
|
"total_trades",
|
|
"stop_loss_exits",
|
|
"worst_day_return_pct",
|
|
]
|
|
delta: dict[str, Any] = {}
|
|
for key in keys:
|
|
cand_value = candidate.get(key)
|
|
base_value = baseline.get(key)
|
|
if isinstance(cand_value, (int, float)) and isinstance(base_value, (int, float)):
|
|
delta[key] = round(cand_value - base_value, 4)
|
|
return delta
|
|
|
|
|
|
def walk_forward_verdict(wf_report: dict[str, Any]) -> dict[str, Any]:
|
|
if wf_report.get("skipped"):
|
|
return {"verdict": "SKIP", "notes": wf_report.get("notes", "walk-forward skipped")}
|
|
aggregate = wf_report.get("test_aggregate", {})
|
|
positive_rate = aggregate.get("positive_fold_rate_pct")
|
|
worst_return = aggregate.get("worst_return_pct")
|
|
mean_return = aggregate.get("mean_return_pct")
|
|
fold_cv = wf_report.get("gap_stats", {}).get("fold_return_cv")
|
|
|
|
if wf_report.get("fold_count", 0) < 2:
|
|
verdict = "SKIP"
|
|
elif (
|
|
positive_rate is not None
|
|
and worst_return is not None
|
|
and mean_return is not None
|
|
and positive_rate >= 75.0
|
|
and worst_return >= 0.0
|
|
and mean_return > 0.0
|
|
and (fold_cv is None or fold_cv <= 1.25)
|
|
):
|
|
verdict = "PASS"
|
|
elif (
|
|
positive_rate is not None
|
|
and worst_return is not None
|
|
and mean_return is not None
|
|
and positive_rate >= 60.0
|
|
and worst_return > -3.0
|
|
and mean_return > 0.0
|
|
):
|
|
verdict = "WARN"
|
|
else:
|
|
verdict = "FAIL"
|
|
|
|
return {
|
|
"verdict": verdict,
|
|
"positive_fold_rate_pct": positive_rate,
|
|
"worst_return_pct": worst_return,
|
|
"mean_return_pct": mean_return,
|
|
"fold_return_cv": fold_cv,
|
|
}
|
|
|
|
|
|
def period_quality_score(payload: dict[str, Any]) -> tuple[float, dict[str, float]]:
|
|
"""Score one ORB period on return quality, drawdown, tail risk, and activity."""
|
|
dd_abs = abs(float(payload.get("max_drawdown_pct") or 0.0))
|
|
worst_day_abs = abs(float(payload.get("worst_day_return_pct") or 0.0))
|
|
tail_loss_abs = abs(float(payload.get("tail_loss_20_pct") or 0.0))
|
|
loss_day_rate = float(payload.get("loss_day_rate_pct") or 0.0)
|
|
trades = float(payload.get("total_trades") or 0.0)
|
|
stop_rate = 0.0
|
|
if trades > 0:
|
|
stop_rate = float(payload.get("stop_loss_exits") or 0.0) / trades * 100.0
|
|
trading_days = float(payload.get("period", {}).get("trading_days") or 0.0)
|
|
activity_rate = 0.0
|
|
if trading_days > 0:
|
|
activity_rate = float(payload.get("days_with_trades") or 0.0) / trading_days * 100.0
|
|
|
|
return_pct = float(payload.get("return_pct") or 0.0)
|
|
ret_over_dd = return_pct / dd_abs if dd_abs > 1e-9 else max(return_pct, 0.0)
|
|
|
|
components = {
|
|
"return": _norm(return_pct, 0.0, 200.0),
|
|
"annualized_return": _norm(payload.get("annualized_return_pct"), 0.0, 110.0),
|
|
"sharpe": _norm(payload.get("sharpe_ratio"), 0.5, 4.0),
|
|
"profit_factor": _norm(payload.get("profit_factor"), 1.0, 4.0),
|
|
"drawdown": _norm_inverse(dd_abs, 5.0, 30.0),
|
|
"return_over_dd": _norm(ret_over_dd, 0.0, 12.0),
|
|
"worst_day": _norm_inverse(worst_day_abs, 1.0, 8.0),
|
|
"tail_loss": _norm_inverse(tail_loss_abs, 0.5, 5.0),
|
|
"loss_day_rate": _norm_inverse(loss_day_rate, 18.0, 45.0),
|
|
"activity": _norm(activity_rate, 20.0, 55.0),
|
|
"stop_rate": _norm_inverse(stop_rate, 5.0, 35.0),
|
|
}
|
|
score = (
|
|
components["return"] * 0.18
|
|
+ components["annualized_return"] * 0.08
|
|
+ components["sharpe"] * 0.13
|
|
+ components["profit_factor"] * 0.10
|
|
+ components["drawdown"] * 0.14
|
|
+ components["return_over_dd"] * 0.12
|
|
+ components["worst_day"] * 0.08
|
|
+ components["tail_loss"] * 0.05
|
|
+ components["loss_day_rate"] * 0.04
|
|
+ components["activity"] * 0.04
|
|
+ components["stop_rate"] * 0.04
|
|
)
|
|
return round(score, 1), {key: round(value, 1) for key, value in components.items()}
|
|
|
|
|
|
def split_quality_score(splits: dict[str, Any]) -> tuple[float | None, dict[str, Any]]:
|
|
split_results: dict[str, SplitResult] = {}
|
|
for name in ("train", "valid", "test"):
|
|
payload = splits.get(name)
|
|
if payload and payload.get("split_result"):
|
|
split_results[name] = SplitResult.model_validate(payload["split_result"])
|
|
|
|
rqs_score, rqs_breakdown = compute_rqs(
|
|
split_results.get("train"),
|
|
split_results.get("valid"),
|
|
split_results.get("test"),
|
|
)
|
|
return rqs_score, rqs_breakdown
|
|
|
|
|
|
def walk_forward_quality_score(wf_report: dict[str, Any]) -> tuple[float | None, dict[str, Any]]:
|
|
if wf_report.get("skipped"):
|
|
return None, {"requires_walk_forward": 1.0}
|
|
summary = WalkForwardSummary.model_validate(wf_report)
|
|
wfqs_score, wfqs_breakdown = compute_wfqs_v2(summary)
|
|
return wfqs_score, wfqs_breakdown
|
|
|
|
|
|
def temporal_robustness_score(candidate: dict[str, Any]) -> tuple[float, dict[str, float]]:
|
|
"""Penalize strategies whose score is mostly a recent-window artifact."""
|
|
periods = candidate.get("periods", {})
|
|
primary = periods.get("primary", {})
|
|
guard_200 = periods.get("guard_200d", {})
|
|
guard_60 = periods.get("guard_60d", {})
|
|
|
|
primary_return = float(primary.get("return_pct") or 0.0)
|
|
guard_200_return = float(guard_200.get("return_pct") or 0.0)
|
|
guard_60_return = float(guard_60.get("return_pct") or 0.0)
|
|
pre_200_return = primary_return - guard_200_return if guard_200 else primary_return
|
|
|
|
early_score = _norm(pre_200_return, 0.0, 80.0)
|
|
recent_200_score = _norm(guard_200_return, 0.0, 120.0) if guard_200 else 50.0
|
|
recent_60_score = _norm(guard_60_return, -5.0, 35.0) if guard_60 else 50.0
|
|
primary_dd_score = _norm_inverse(abs(float(primary.get("max_drawdown_pct") or 0.0)), 8.0, 30.0)
|
|
|
|
concentration_ratio = 0.0
|
|
if primary_return > 1e-9 and guard_200:
|
|
concentration_ratio = guard_200_return / primary_return
|
|
concentration_score = _norm_inverse(concentration_ratio, 0.45, 0.95)
|
|
|
|
score = (
|
|
early_score * 0.35
|
|
+ recent_200_score * 0.20
|
|
+ recent_60_score * 0.10
|
|
+ primary_dd_score * 0.20
|
|
+ concentration_score * 0.15
|
|
)
|
|
breakdown = {
|
|
"pre_200_return_pct": round(pre_200_return, 2),
|
|
"early_score": round(early_score, 1),
|
|
"recent_200_score": round(recent_200_score, 1),
|
|
"recent_60_score": round(recent_60_score, 1),
|
|
"primary_dd_score": round(primary_dd_score, 1),
|
|
"recent_concentration_ratio": round(concentration_ratio, 3),
|
|
"concentration_score": round(concentration_score, 1),
|
|
}
|
|
return round(score, 1), breakdown
|
|
|
|
|
|
def deployment_gate_factor(
|
|
*,
|
|
primary: dict[str, Any],
|
|
split_score: float | None,
|
|
wf_score: float | None,
|
|
wf_report: dict[str, Any],
|
|
temporal_score: float,
|
|
) -> tuple[float, dict[str, Any]]:
|
|
wf_test = wf_report.get("test_aggregate", {})
|
|
checks = {
|
|
"primary_dd_ok": abs(float(primary.get("max_drawdown_pct") or 0.0)) <= 25.0,
|
|
"primary_sharpe_ok": float(primary.get("sharpe_ratio") or 0.0) >= 1.0,
|
|
"split_rqs_ok": (split_score or 0.0) >= 45.0,
|
|
"wfqs_ok": (wf_score or 0.0) >= 45.0,
|
|
"wf_positive_rate_ok": (wf_test.get("positive_fold_rate_pct") or 0.0) >= 60.0,
|
|
"wf_worst_ok": (wf_test.get("worst_return_pct") or -999.0) >= -8.0,
|
|
"temporal_ok": temporal_score >= 35.0,
|
|
}
|
|
pass_count = sum(1 for passed in checks.values() if passed)
|
|
if pass_count >= 6:
|
|
factor = 1.00
|
|
elif pass_count == 5:
|
|
factor = 0.85
|
|
elif pass_count == 4:
|
|
factor = 0.70
|
|
elif pass_count == 3:
|
|
factor = 0.55
|
|
else:
|
|
factor = 0.35
|
|
return factor, {"pass_count": pass_count, **checks}
|
|
|
|
|
|
def compute_orb_validation_score(candidate: dict[str, Any]) -> dict[str, Any]:
|
|
"""Compute a single anti-overfit ORB score, analogous in spirit to PEAD SQS."""
|
|
capital_mode = candidate.get("capital_mode")
|
|
if isinstance(capital_mode, dict) and (
|
|
capital_mode.get("compound_returns") is True
|
|
or capital_mode.get("daily_budget_reset") is not True
|
|
):
|
|
return {
|
|
"score": None,
|
|
"source": "compound_diagnostic_not_ranked",
|
|
"policy": ORB_SQS_POLICY,
|
|
"capital_mode": capital_mode,
|
|
"requires": ["daily_budget_reset_primary"],
|
|
"notes": (
|
|
"ORB SQS ranks daily-reset signal quality only. Compound "
|
|
"results are deployment diagnostics and should not tune "
|
|
"entry, ranking, or exit signals."
|
|
),
|
|
}
|
|
|
|
primary = candidate.get("periods", {}).get("primary")
|
|
if not primary:
|
|
return {"score": None, "source": "missing_primary"}
|
|
|
|
primary_score, primary_breakdown = period_quality_score(primary)
|
|
split_score, split_breakdown = split_quality_score(candidate.get("splits", {}))
|
|
wf_score, wf_breakdown = walk_forward_quality_score(candidate.get("walk_forward", {}))
|
|
temporal_score, temporal_breakdown = temporal_robustness_score(candidate)
|
|
|
|
if split_score is None or wf_score is None:
|
|
return {
|
|
"score": None,
|
|
"source": "pending_validation",
|
|
"primary_quality": primary_score,
|
|
"split_rqs": split_score,
|
|
"wfqs_v2": wf_score,
|
|
"temporal_robustness": temporal_score,
|
|
"requires": [
|
|
name
|
|
for name, value in (("splits", split_score), ("walk_forward", wf_score))
|
|
if value is None
|
|
],
|
|
}
|
|
|
|
raw_score = (
|
|
primary_score * 0.25
|
|
+ split_score * 0.30
|
|
+ wf_score * 0.30
|
|
+ temporal_score * 0.15
|
|
)
|
|
gate_factor, gate_breakdown = deployment_gate_factor(
|
|
primary=primary,
|
|
split_score=split_score,
|
|
wf_score=wf_score,
|
|
wf_report=candidate.get("walk_forward", {}),
|
|
temporal_score=temporal_score,
|
|
)
|
|
final_score = raw_score * gate_factor
|
|
score = {
|
|
"score": round(final_score, 1),
|
|
"source": "orb_sqs_v1",
|
|
"policy": ORB_SQS_POLICY,
|
|
"raw_score": round(raw_score, 1),
|
|
"deployment_gate_factor": round(gate_factor, 2),
|
|
"primary_quality": primary_score,
|
|
"split_rqs": round(split_score, 1),
|
|
"wfqs_v2": round(wf_score, 1),
|
|
"temporal_robustness": temporal_score,
|
|
"weights": {
|
|
"primary_quality": 0.25,
|
|
"split_rqs": 0.30,
|
|
"wfqs_v2": 0.30,
|
|
"temporal_robustness": 0.15,
|
|
},
|
|
"primary_breakdown": primary_breakdown,
|
|
"split_rqs_breakdown": split_breakdown,
|
|
"wfqs_v2_breakdown": wf_breakdown,
|
|
"temporal_breakdown": temporal_breakdown,
|
|
"deployment_gate": gate_breakdown,
|
|
}
|
|
return score
|
|
|
|
|
|
def promotion_gate(
|
|
candidate: dict[str, Any],
|
|
baseline: dict[str, Any] | None,
|
|
) -> dict[str, Any]:
|
|
checks: list[dict[str, Any]] = []
|
|
|
|
primary = candidate["periods"].get("primary")
|
|
if primary is not None:
|
|
primary_delta = candidate.get("deltas", {}).get("primary") or {}
|
|
if baseline is None:
|
|
checks.append({"name": "primary_504d", "verdict": "INFO", "notes": "no baseline supplied"})
|
|
elif (primary_delta.get("return_pct") or 0.0) >= 0.0:
|
|
checks.append({"name": "primary_504d", "verdict": "PASS", "delta_return_pct": primary_delta.get("return_pct")})
|
|
elif (primary_delta.get("return_pct") or 0.0) >= -5.0 and (
|
|
primary_delta.get("max_drawdown_pct") or 0.0
|
|
) >= -0.5:
|
|
checks.append({
|
|
"name": "primary_504d",
|
|
"verdict": "WARN",
|
|
"delta_return_pct": primary_delta.get("return_pct"),
|
|
"delta_max_drawdown_pct": primary_delta.get("max_drawdown_pct"),
|
|
})
|
|
else:
|
|
checks.append({
|
|
"name": "primary_504d",
|
|
"verdict": "FAIL",
|
|
"delta_return_pct": primary_delta.get("return_pct"),
|
|
"delta_max_drawdown_pct": primary_delta.get("max_drawdown_pct"),
|
|
})
|
|
|
|
for name in ("guard_200d", "guard_60d"):
|
|
if name not in candidate.get("periods", {}):
|
|
continue
|
|
period_delta = candidate.get("deltas", {}).get(name) or {}
|
|
if baseline is None:
|
|
checks.append({"name": name, "verdict": "INFO", "notes": "no baseline supplied"})
|
|
continue
|
|
return_delta = period_delta.get("return_pct")
|
|
dd_delta = period_delta.get("max_drawdown_pct")
|
|
if return_delta is not None and return_delta >= -3.0 and (dd_delta is None or dd_delta >= -1.0):
|
|
checks.append({"name": name, "verdict": "PASS", "delta_return_pct": return_delta, "delta_max_drawdown_pct": dd_delta})
|
|
else:
|
|
checks.append({"name": name, "verdict": "WARN", "delta_return_pct": return_delta, "delta_max_drawdown_pct": dd_delta})
|
|
|
|
wf_check = candidate.get("walk_forward", {}).get("verdict", {})
|
|
if wf_check:
|
|
checks.append({"name": "walk_forward", **wf_check})
|
|
|
|
hard_verdicts = [check["verdict"] for check in checks if check["verdict"] not in {"INFO", "SKIP"}]
|
|
if "FAIL" in hard_verdicts:
|
|
overall = "FAIL"
|
|
elif "WARN" in hard_verdicts:
|
|
overall = "WARN"
|
|
elif hard_verdicts:
|
|
overall = "PASS"
|
|
else:
|
|
overall = "INFO"
|
|
return {"overall": overall, "checks": checks}
|
|
|
|
|
|
async def build_walk_forward_report(
|
|
context,
|
|
client: OracleClient,
|
|
params: ORBStrategyParams,
|
|
*,
|
|
train_days: int,
|
|
test_days: int,
|
|
step_days: int,
|
|
intraday_concurrency: int,
|
|
max_pairs_per_chunk: int,
|
|
label: str,
|
|
progress: bool,
|
|
) -> dict[str, Any]:
|
|
windows = generate_walk_forward_windows(
|
|
context.trading_days,
|
|
train_days=train_days,
|
|
test_days=test_days,
|
|
step_days=step_days,
|
|
)
|
|
folds: list[dict[str, Any]] = []
|
|
for idx, (train, test) in enumerate(windows, start=1):
|
|
prefix = f" {label} WF {idx}/{len(windows)} " if progress else ""
|
|
train_metrics = await simulate_orb_period(
|
|
context,
|
|
client,
|
|
params,
|
|
train,
|
|
run_id=f"{label}_wf{idx:02d}_train",
|
|
progress_prefix=prefix,
|
|
intraday_concurrency=intraday_concurrency,
|
|
max_pairs_per_chunk=max_pairs_per_chunk,
|
|
)
|
|
test_metrics = await simulate_orb_period(
|
|
context,
|
|
client,
|
|
params,
|
|
test,
|
|
run_id=f"{label}_wf{idx:02d}_test",
|
|
progress_prefix=prefix,
|
|
intraday_concurrency=intraday_concurrency,
|
|
max_pairs_per_chunk=max_pairs_per_chunk,
|
|
)
|
|
folds.append(
|
|
{
|
|
"train_start": train[0],
|
|
"train_end": train[-1],
|
|
"test_start": test[0],
|
|
"test_end": test[-1],
|
|
"train_result": intraday_metrics_to_split_result(train_metrics, params),
|
|
"test_result": intraday_metrics_to_split_result(test_metrics, params),
|
|
"train_metrics": metrics_payload(train_metrics),
|
|
"test_metrics": metrics_payload(test_metrics),
|
|
}
|
|
)
|
|
|
|
summary = build_walk_forward_summary(
|
|
folds,
|
|
train_days=train_days,
|
|
test_days=test_days,
|
|
step_days=step_days,
|
|
)
|
|
report = summary.model_dump(mode="json")
|
|
report["fold_metrics"] = [
|
|
{
|
|
"fold_index": idx,
|
|
"train": fold["train_metrics"],
|
|
"test": fold["test_metrics"],
|
|
}
|
|
for idx, fold in enumerate(folds, start=1)
|
|
]
|
|
report["verdict"] = walk_forward_verdict(report)
|
|
return report
|
|
|
|
|
|
async def simulate_official_period(
|
|
config: IntradayConfig,
|
|
trading_days: list[str],
|
|
*,
|
|
run_id: str,
|
|
quiet: bool,
|
|
) -> IntradayMetrics:
|
|
"""Run the official backtest pipeline for an exact trading-day period."""
|
|
period_config = config.model_copy(
|
|
update={
|
|
"backtest": config.backtest.model_copy(
|
|
update={
|
|
"start_date": trading_days[0],
|
|
"end_date": trading_days[-1],
|
|
"lookback_trading_days": len(trading_days),
|
|
}
|
|
)
|
|
}
|
|
)
|
|
if quiet:
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
_, metrics, _, _ = await run_backtest(period_config, refresh_cache=False)
|
|
else:
|
|
_, metrics, _, _ = await run_backtest(period_config, refresh_cache=False)
|
|
return metrics.model_copy(update={"run_id": run_id})
|
|
|
|
|
|
async def run_official_period(
|
|
config: IntradayConfig,
|
|
trading_days: list[str],
|
|
*,
|
|
run_id: str,
|
|
quiet: bool,
|
|
) -> tuple[list[DayResult], IntradayMetrics]:
|
|
"""Run the official backtest pipeline once and keep day-level results."""
|
|
period_config = config.model_copy(
|
|
update={
|
|
"backtest": config.backtest.model_copy(
|
|
update={
|
|
"start_date": trading_days[0],
|
|
"end_date": trading_days[-1],
|
|
"lookback_trading_days": len(trading_days),
|
|
}
|
|
)
|
|
}
|
|
)
|
|
if quiet:
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
day_results, metrics, _, _ = await run_backtest(period_config, refresh_cache=False)
|
|
else:
|
|
day_results, metrics, _, _ = await run_backtest(period_config, refresh_cache=False)
|
|
return day_results, metrics.model_copy(update={"run_id": run_id})
|
|
|
|
|
|
def derive_official_metrics(
|
|
config: IntradayConfig,
|
|
day_results: list[DayResult],
|
|
trading_days: list[str],
|
|
*,
|
|
run_id: str,
|
|
) -> IntradayMetrics:
|
|
"""Derive subperiod metrics from one continuous official 504d run."""
|
|
wanted = set(trading_days)
|
|
subset = [result for result in day_results if result.date in wanted]
|
|
if len(subset) != len(trading_days):
|
|
found = {result.date for result in subset}
|
|
missing = [day for day in trading_days if day not in found]
|
|
raise ValueError(f"Cannot derive official metrics; missing {len(missing)} days, first={missing[:3]}")
|
|
return compute_metrics(subset, config, run_id=run_id)
|
|
|
|
|
|
def build_derived_walk_forward_report(
|
|
config: IntradayConfig,
|
|
day_results: list[DayResult],
|
|
trading_days: list[str],
|
|
*,
|
|
train_days: int,
|
|
test_days: int,
|
|
step_days: int,
|
|
label: str,
|
|
) -> dict[str, Any]:
|
|
"""Build WFV from the already executed continuous official run."""
|
|
windows = generate_walk_forward_windows(
|
|
trading_days,
|
|
train_days=train_days,
|
|
test_days=test_days,
|
|
step_days=step_days,
|
|
)
|
|
params = config.orb_strategy or ORBStrategyParams()
|
|
folds: list[dict[str, Any]] = []
|
|
for idx, (train, test) in enumerate(windows, start=1):
|
|
train_metrics = derive_official_metrics(
|
|
config,
|
|
day_results,
|
|
train,
|
|
run_id=f"{label}_wf{idx:02d}_train",
|
|
)
|
|
test_metrics = derive_official_metrics(
|
|
config,
|
|
day_results,
|
|
test,
|
|
run_id=f"{label}_wf{idx:02d}_test",
|
|
)
|
|
folds.append(
|
|
{
|
|
"train_start": train[0],
|
|
"train_end": train[-1],
|
|
"test_start": test[0],
|
|
"test_end": test[-1],
|
|
"train_result": intraday_metrics_to_split_result(train_metrics, params),
|
|
"test_result": intraday_metrics_to_split_result(test_metrics, params),
|
|
"train_metrics": metrics_payload(train_metrics),
|
|
"test_metrics": metrics_payload(test_metrics),
|
|
}
|
|
)
|
|
|
|
summary = build_walk_forward_summary(
|
|
folds,
|
|
train_days=train_days,
|
|
test_days=test_days,
|
|
step_days=step_days,
|
|
)
|
|
report = summary.model_dump(mode="json")
|
|
report["fold_metrics"] = [
|
|
{
|
|
"fold_index": idx,
|
|
"train": fold["train_metrics"],
|
|
"test": fold["test_metrics"],
|
|
}
|
|
for idx, fold in enumerate(folds, start=1)
|
|
]
|
|
report["verdict"] = walk_forward_verdict(report)
|
|
report["derived_from_continuous_official_run"] = True
|
|
return report
|
|
|
|
|
|
async def build_official_walk_forward_report(
|
|
config: IntradayConfig,
|
|
trading_days: list[str],
|
|
*,
|
|
train_days: int,
|
|
test_days: int,
|
|
step_days: int,
|
|
label: str,
|
|
quiet: bool,
|
|
) -> dict[str, Any]:
|
|
windows = generate_walk_forward_windows(
|
|
trading_days,
|
|
train_days=train_days,
|
|
test_days=test_days,
|
|
step_days=step_days,
|
|
)
|
|
folds: list[dict[str, Any]] = []
|
|
for idx, (train, test) in enumerate(windows, start=1):
|
|
if not quiet:
|
|
print(
|
|
f"\n[{label}] official WF {idx}/{len(windows)} "
|
|
f"train {train[0]} → {train[-1]} test {test[0]} → {test[-1]}"
|
|
)
|
|
train_metrics = await simulate_official_period(
|
|
config,
|
|
train,
|
|
run_id=f"{label}_wf{idx:02d}_train",
|
|
quiet=quiet,
|
|
)
|
|
test_metrics = await simulate_official_period(
|
|
config,
|
|
test,
|
|
run_id=f"{label}_wf{idx:02d}_test",
|
|
quiet=quiet,
|
|
)
|
|
params = config.orb_strategy or ORBStrategyParams()
|
|
folds.append(
|
|
{
|
|
"train_start": train[0],
|
|
"train_end": train[-1],
|
|
"test_start": test[0],
|
|
"test_end": test[-1],
|
|
"train_result": intraday_metrics_to_split_result(train_metrics, params),
|
|
"test_result": intraday_metrics_to_split_result(test_metrics, params),
|
|
"train_metrics": metrics_payload(train_metrics),
|
|
"test_metrics": metrics_payload(test_metrics),
|
|
}
|
|
)
|
|
|
|
summary = build_walk_forward_summary(
|
|
folds,
|
|
train_days=train_days,
|
|
test_days=test_days,
|
|
step_days=step_days,
|
|
)
|
|
report = summary.model_dump(mode="json")
|
|
report["fold_metrics"] = [
|
|
{
|
|
"fold_index": idx,
|
|
"train": fold["train_metrics"],
|
|
"test": fold["test_metrics"],
|
|
}
|
|
for idx, fold in enumerate(folds, start=1)
|
|
]
|
|
report["verdict"] = walk_forward_verdict(report)
|
|
return report
|
|
|
|
|
|
async def validate_strategy(
|
|
*,
|
|
label: str,
|
|
config_path: Path,
|
|
config: IntradayConfig,
|
|
client: OracleClient,
|
|
start: str,
|
|
end: str,
|
|
primary_days: int,
|
|
guard_days: list[int],
|
|
wf_train_days: int,
|
|
wf_test_days: int,
|
|
wf_step_days: int,
|
|
intraday_concurrency: int,
|
|
max_pairs_per_chunk: int,
|
|
run_wfv: bool,
|
|
backend: str,
|
|
progress: bool,
|
|
) -> dict[str, Any]:
|
|
params = config.orb_strategy or ORBStrategyParams()
|
|
context = None
|
|
if backend == "research":
|
|
context = await build_orb_research_context(
|
|
config,
|
|
start,
|
|
end,
|
|
client,
|
|
daily_concurrency=10,
|
|
print_progress=progress,
|
|
)
|
|
trading_days = context.trading_days
|
|
else:
|
|
trading_days = await get_trading_days(client, start, end, lookback=0)
|
|
if not trading_days:
|
|
raise ValueError(f"No trading days resolved for {start} → {end}")
|
|
|
|
period_days: dict[str, list[str]] = {
|
|
"primary": tail_days(trading_days, primary_days),
|
|
}
|
|
for days in guard_days:
|
|
period_days[f"guard_{days}d"] = tail_days(period_days["primary"], days)
|
|
|
|
periods: dict[str, Any] = {}
|
|
official_day_results: list[DayResult] | None = None
|
|
if backend == "official":
|
|
primary_days_list = period_days["primary"]
|
|
if progress:
|
|
print(
|
|
f"\n[{label}] primary official run: "
|
|
f"{primary_days_list[0]} → {primary_days_list[-1]} ({len(primary_days_list)} trading days)"
|
|
)
|
|
official_day_results, primary_metrics = await run_official_period(
|
|
config,
|
|
primary_days_list,
|
|
run_id=f"{label}_primary",
|
|
quiet=not progress,
|
|
)
|
|
periods["primary"] = metrics_payload(primary_metrics)
|
|
for name, days in period_days.items():
|
|
if name == "primary":
|
|
continue
|
|
periods[name] = metrics_payload(
|
|
derive_official_metrics(
|
|
config,
|
|
official_day_results,
|
|
days,
|
|
run_id=f"{label}_{name}",
|
|
)
|
|
)
|
|
else:
|
|
for name, days in period_days.items():
|
|
if progress:
|
|
print(f"\n[{label}] {name}: {days[0]} → {days[-1]} ({len(days)} trading days)")
|
|
if context is None:
|
|
raise RuntimeError("research backend requires context")
|
|
metrics = await simulate_orb_period(
|
|
context,
|
|
client,
|
|
params,
|
|
days,
|
|
run_id=f"{label}_{name}",
|
|
progress_prefix=f" {label} {name} " if progress else "",
|
|
intraday_concurrency=intraday_concurrency,
|
|
max_pairs_per_chunk=max_pairs_per_chunk,
|
|
)
|
|
periods[name] = metrics_payload(metrics)
|
|
|
|
split_days = chronological_splits(period_days["primary"])
|
|
splits: dict[str, Any] = {}
|
|
for name, days in split_days.items():
|
|
if progress:
|
|
print(f"\n[{label}] split_{name}: {days[0]} → {days[-1]} ({len(days)} trading days)")
|
|
if backend == "research":
|
|
if context is None:
|
|
raise RuntimeError("research backend requires context")
|
|
metrics = await simulate_orb_period(
|
|
context,
|
|
client,
|
|
params,
|
|
days,
|
|
run_id=f"{label}_split_{name}",
|
|
progress_prefix=f" {label} split_{name} " if progress else "",
|
|
intraday_concurrency=intraday_concurrency,
|
|
max_pairs_per_chunk=max_pairs_per_chunk,
|
|
)
|
|
else:
|
|
if official_day_results is None:
|
|
raise RuntimeError("official backend requires day results")
|
|
metrics = derive_official_metrics(
|
|
config,
|
|
official_day_results,
|
|
days,
|
|
run_id=f"{label}_split_{name}",
|
|
)
|
|
splits[name] = metrics_with_split_payload(metrics, params)
|
|
|
|
if run_wfv:
|
|
if progress:
|
|
print(f"\n[{label}] walk-forward: train={wf_train_days}, test={wf_test_days}, step={wf_step_days}")
|
|
if backend == "research":
|
|
if context is None:
|
|
raise RuntimeError("research backend requires context")
|
|
wf = await build_walk_forward_report(
|
|
context,
|
|
client,
|
|
params,
|
|
train_days=wf_train_days,
|
|
test_days=wf_test_days,
|
|
step_days=wf_step_days,
|
|
intraday_concurrency=intraday_concurrency,
|
|
max_pairs_per_chunk=max_pairs_per_chunk,
|
|
label=label,
|
|
progress=progress,
|
|
)
|
|
else:
|
|
if official_day_results is None:
|
|
raise RuntimeError("official backend requires day results")
|
|
wf = build_derived_walk_forward_report(
|
|
config,
|
|
official_day_results,
|
|
period_days["primary"],
|
|
train_days=wf_train_days,
|
|
test_days=wf_test_days,
|
|
step_days=wf_step_days,
|
|
label=label,
|
|
)
|
|
else:
|
|
wf = {
|
|
"skipped": True,
|
|
"notes": f"{label} walk-forward skipped by CLI",
|
|
"fold_count": 0,
|
|
"verdict": {"verdict": "SKIP", "notes": f"{label} walk-forward skipped by CLI"},
|
|
}
|
|
|
|
return {
|
|
"label": label,
|
|
"config": str(config_path),
|
|
"strategy_name": (config.orb_strategy or ORBStrategyParams()).engine_family,
|
|
"universe": config.universe.source,
|
|
"backend": backend,
|
|
"capital_mode": {
|
|
"compound_returns": params.compound_returns,
|
|
"daily_budget_reset": params.daily_budget_reset,
|
|
"settlement_days": params.settlement_days,
|
|
"single_trade_loss_cap_pct": params.single_trade_loss_cap_pct,
|
|
"single_trade_loss_cap_basis": params.single_trade_loss_cap_basis,
|
|
},
|
|
"context_period": _period(trading_days),
|
|
"periods": periods,
|
|
"splits": splits,
|
|
"walk_forward": wf,
|
|
}
|
|
|
|
|
|
def attach_deltas(candidate: dict[str, Any], baseline: dict[str, Any] | None) -> None:
|
|
deltas: dict[str, Any] = {}
|
|
if baseline is not None:
|
|
for name, payload in candidate["periods"].items():
|
|
deltas[name] = delta_payload(payload, baseline["periods"].get(name))
|
|
candidate["deltas"] = deltas
|
|
candidate["promotion_gate"] = promotion_gate(candidate, baseline)
|
|
candidate["score"] = compute_orb_validation_score(candidate)
|
|
|
|
|
|
def print_console_report(report: dict[str, Any]) -> None:
|
|
candidate = report["candidate"]
|
|
baseline = report.get("baseline")
|
|
print("\nORB validation summary")
|
|
print(f" candidate: {candidate['config']}")
|
|
if baseline:
|
|
print(f" baseline : {baseline['config']}")
|
|
print(f" period : {candidate['context_period']['start']} → {candidate['context_period']['end']}")
|
|
score = candidate.get("score") or {}
|
|
if score.get("score") is not None:
|
|
print(
|
|
f" ORB SQS : {score['score']:.1f} "
|
|
f"(raw {score['raw_score']:.1f}, gate {score['deployment_gate_factor']:.2f})"
|
|
)
|
|
print(
|
|
f" primary {score['primary_quality']:.1f} "
|
|
f"split {score['split_rqs']:.1f} "
|
|
f"WFQS {score['wfqs_v2']:.1f} "
|
|
f"temporal {score['temporal_robustness']:.1f}"
|
|
)
|
|
else:
|
|
print(f" ORB SQS : pending ({', '.join(score.get('requires', [])) or score.get('source', 'unknown')})")
|
|
print(f" gate : {candidate['promotion_gate']['overall']}")
|
|
|
|
def fmt(value: Any, width: int, digits: int = 2) -> str:
|
|
if isinstance(value, (int, float)):
|
|
return f"{value:>{width}.{digits}f}"
|
|
return f"{str(value):>{width}}"
|
|
|
|
for name in ["primary", "guard_200d", "guard_60d"]:
|
|
payload = candidate["periods"].get(name)
|
|
if not payload:
|
|
continue
|
|
delta = candidate.get("deltas", {}).get(name) or {}
|
|
delta_text = ""
|
|
if delta:
|
|
delta_text = (
|
|
f" Δret {delta.get('return_pct', 0):+.2f}pp"
|
|
f" ΔDD {delta.get('max_drawdown_pct', 0):+.2f}pp"
|
|
)
|
|
print(
|
|
f" {name:<10} {fmt(payload.get('return_pct'), 8)}%"
|
|
f" DD {fmt(payload.get('max_drawdown_pct'), 7)}%"
|
|
f" Sharpe {fmt(payload.get('sharpe_ratio'), 5)}"
|
|
f" PF {fmt(payload.get('profit_factor'), 6)}"
|
|
f" trades {payload['total_trades']:>4}"
|
|
f"{delta_text}"
|
|
)
|
|
|
|
wf = candidate["walk_forward"]
|
|
verdict = wf["verdict"]
|
|
print(
|
|
" WFV "
|
|
f"{verdict['verdict']} folds {wf['fold_count']} "
|
|
f"mean {verdict.get('mean_return_pct')}% "
|
|
f"worst {verdict.get('worst_return_pct')}% "
|
|
f"positive {verdict.get('positive_fold_rate_pct')}% "
|
|
f"cv {verdict.get('fold_return_cv')}"
|
|
)
|
|
|
|
|
|
async def _async_main(args: argparse.Namespace) -> int:
|
|
t0 = time.time()
|
|
end = args.end or _latest_backtest_date().isoformat()
|
|
candidate_path, candidate_config = resolve_orb_config(args.config)
|
|
if not args.allow_compound:
|
|
candidate_config = force_simple_returns(candidate_config)
|
|
|
|
baseline_path: Path | None = None
|
|
baseline_config: IntradayConfig | None = None
|
|
if args.baseline_config:
|
|
baseline_path, baseline_config = resolve_orb_config(args.baseline_config)
|
|
if not args.allow_compound:
|
|
baseline_config = force_simple_returns(baseline_config)
|
|
|
|
settings = get_settings()
|
|
async with make_intraday_oracle_client(settings) as client:
|
|
baseline_report = None
|
|
if baseline_path is not None and baseline_config is not None:
|
|
baseline_report = await validate_strategy(
|
|
label="baseline",
|
|
config_path=baseline_path,
|
|
config=baseline_config,
|
|
client=client,
|
|
start=args.start,
|
|
end=end,
|
|
primary_days=args.primary_days,
|
|
guard_days=args.guard_days,
|
|
wf_train_days=args.wf_train_days,
|
|
wf_test_days=args.wf_test_days,
|
|
wf_step_days=args.wf_step_days,
|
|
intraday_concurrency=args.intraday_concurrency,
|
|
max_pairs_per_chunk=args.max_pairs_per_chunk,
|
|
run_wfv=not args.skip_baseline_wfv and not args.skip_wfv,
|
|
backend=args.backend,
|
|
progress=not args.quiet,
|
|
)
|
|
|
|
candidate_report = await validate_strategy(
|
|
label="candidate",
|
|
config_path=candidate_path,
|
|
config=candidate_config,
|
|
client=client,
|
|
start=args.start,
|
|
end=end,
|
|
primary_days=args.primary_days,
|
|
guard_days=args.guard_days,
|
|
wf_train_days=args.wf_train_days,
|
|
wf_test_days=args.wf_test_days,
|
|
wf_step_days=args.wf_step_days,
|
|
intraday_concurrency=args.intraday_concurrency,
|
|
max_pairs_per_chunk=args.max_pairs_per_chunk,
|
|
run_wfv=not args.skip_wfv,
|
|
backend=args.backend,
|
|
progress=not args.quiet,
|
|
)
|
|
|
|
attach_deltas(candidate_report, baseline_report)
|
|
if baseline_report is not None:
|
|
baseline_report["promotion_gate"] = {"overall": "REFERENCE", "checks": []}
|
|
baseline_report["score"] = compute_orb_validation_score(baseline_report)
|
|
|
|
report = {
|
|
"schema_version": 1,
|
|
"elapsed_seconds": round(time.time() - t0, 1),
|
|
"validation": {
|
|
"start": args.start,
|
|
"end": end,
|
|
"primary_days": args.primary_days,
|
|
"guard_days": args.guard_days,
|
|
"wf_train_days": args.wf_train_days,
|
|
"wf_test_days": args.wf_test_days,
|
|
"wf_step_days": args.wf_step_days,
|
|
"intraday_concurrency": args.intraday_concurrency,
|
|
"max_pairs_per_chunk": args.max_pairs_per_chunk,
|
|
"compound_allowed": args.allow_compound,
|
|
"skip_wfv": args.skip_wfv,
|
|
"skip_baseline_wfv": args.skip_baseline_wfv,
|
|
"backend": args.backend,
|
|
"sqs_policy": ORB_SQS_POLICY,
|
|
},
|
|
"baseline": baseline_report,
|
|
"candidate": candidate_report,
|
|
}
|
|
|
|
output_json = args.output_json
|
|
if output_json is None:
|
|
stamp = time.strftime("%Y%m%d_%H%M%S")
|
|
output_json = f"tmp/orb_validation/orb_validation_{stamp}.json"
|
|
output_path = Path(output_json)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_text(json.dumps(report, indent=2), encoding="utf-8")
|
|
|
|
print_console_report(report)
|
|
print(f" report : {output_path}")
|
|
return 0
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
prog="fithia2 intraday-orb-validate",
|
|
description="Run promotion-grade ORB validation: 504d primary, recent guards, and WFV.",
|
|
)
|
|
parser.add_argument("--config", "-c", required=True, help="Candidate YAML path or strategy slug")
|
|
parser.add_argument("--baseline-config", "-b", default=None, help="Optional baseline YAML path or slug")
|
|
parser.add_argument("--start", default=DEFAULT_START_DATE, help=f"Context start date (default: {DEFAULT_START_DATE})")
|
|
parser.add_argument("--end", default=None, help="Context end date (default: latest completed backtest date)")
|
|
parser.add_argument("--primary-days", type=int, default=DEFAULT_PRIMARY_DAYS)
|
|
parser.add_argument("--guard-days", type=int, nargs="*", default=list(DEFAULT_GUARD_DAYS))
|
|
parser.add_argument("--wf-train-days", type=int, default=DEFAULT_WF_TRAIN_DAYS)
|
|
parser.add_argument("--wf-test-days", type=int, default=DEFAULT_WF_TEST_DAYS)
|
|
parser.add_argument("--wf-step-days", type=int, default=DEFAULT_WF_TEST_DAYS)
|
|
parser.add_argument("--intraday-concurrency", type=int, default=12)
|
|
parser.add_argument("--max-pairs-per-chunk", type=int, default=10_000)
|
|
parser.add_argument(
|
|
"--backend",
|
|
choices=["official", "research"],
|
|
default="official",
|
|
help="official uses apps.intraday_bt.run and is the promotion source of truth; research is faster but approximate",
|
|
)
|
|
parser.add_argument("--skip-wfv", action="store_true", help="Only run primary/guard windows")
|
|
parser.add_argument(
|
|
"--skip-baseline-wfv",
|
|
action="store_true",
|
|
help="Run WFV for the candidate only; baseline still gets primary/guard windows",
|
|
)
|
|
parser.add_argument("--allow-compound", action="store_true", help="Do not force candidate/baseline to simple returns")
|
|
parser.add_argument("--quiet", action="store_true", help="Suppress progress output")
|
|
parser.add_argument("--output-json", default=None, help="Report JSON path")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
raise SystemExit(asyncio.run(_async_main(parse_args())))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|