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.
493 lines
18 KiB
Python
493 lines
18 KiB
Python
"""ORB strategy evaluation with IS/OOS validation and streaming intraday fetch."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import statistics
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from apps.intraday_bt.oracle import make_intraday_oracle_client
|
|
from libs.common.config import get_settings
|
|
from libs.intraday.domain import (
|
|
BacktestParams,
|
|
CacheParams,
|
|
IntradayConfig,
|
|
IntradayMetrics,
|
|
ORBStrategyParams,
|
|
OutputParams,
|
|
SweepResult,
|
|
UniverseParams,
|
|
)
|
|
from libs.oracle_client import OracleClient
|
|
|
|
from apps.intraday_bt.orb_research import (
|
|
build_orb_research_context,
|
|
force_simple_returns,
|
|
generate_walk_forward_windows,
|
|
simulate_orb_overrides,
|
|
split_trading_days,
|
|
)
|
|
from apps.intraday_bt.sweep import (
|
|
SweepConfig,
|
|
apply_overrides,
|
|
generate_combinations,
|
|
load_sweep_config,
|
|
)
|
|
from apps.intraday_bt.run import get_trading_days
|
|
|
|
|
|
def compute_degradation(is_sharpe: float | None, oos_sharpe: float | None) -> float:
|
|
if is_sharpe is None or is_sharpe <= 0:
|
|
return 0.0
|
|
if oos_sharpe is None:
|
|
return 0.0
|
|
return oos_sharpe / is_sharpe
|
|
|
|
|
|
async def run_sweep_on_period(
|
|
sweep: SweepConfig,
|
|
context,
|
|
client: OracleClient,
|
|
trading_days: list[str],
|
|
progress_prefix: str = "",
|
|
) -> list[SweepResult]:
|
|
combos = generate_combinations(sweep)
|
|
results: list[SweepResult] = []
|
|
|
|
for i, overrides in enumerate(combos):
|
|
_, metrics = await simulate_orb_overrides(
|
|
context,
|
|
client,
|
|
overrides,
|
|
trading_days,
|
|
run_id=f"sw{i:04d}",
|
|
)
|
|
results.append(SweepResult(params=overrides, metrics=metrics))
|
|
sys.stdout.write(
|
|
f"\r {progress_prefix}[{i+1}/{len(combos)}] "
|
|
f"Sharpe={metrics.sharpe_ratio or 0:.2f} "
|
|
f"Ret={(metrics.total_return_pct or 0)*100:.0f}%"
|
|
)
|
|
sys.stdout.flush()
|
|
print()
|
|
|
|
results.sort(
|
|
key=lambda r: (
|
|
r.metrics.sharpe_ratio or float("-inf"),
|
|
r.metrics.total_return_pct or -999,
|
|
),
|
|
reverse=True,
|
|
)
|
|
return results
|
|
|
|
|
|
async def run_single_config_on_period(
|
|
base_config: IntradayConfig,
|
|
overrides: dict[str, Any],
|
|
context,
|
|
client: OracleClient,
|
|
trading_days: list[str],
|
|
*,
|
|
run_id: str = "",
|
|
) -> IntradayMetrics:
|
|
_, metrics = await simulate_orb_overrides(
|
|
context,
|
|
client,
|
|
overrides,
|
|
trading_days,
|
|
run_id=run_id or str(uuid.uuid4())[:8],
|
|
)
|
|
return metrics
|
|
|
|
|
|
async def _run_split_eval(
|
|
sweep: SweepConfig,
|
|
context,
|
|
client: OracleClient,
|
|
train_days: list[str],
|
|
test_days: list[str],
|
|
top_n: int,
|
|
*,
|
|
oos_sort: bool = False,
|
|
) -> dict[str, Any]:
|
|
print(f"\n ── Phase 1: Sweep on Train ({train_days[0]}→{train_days[-1]}, {len(train_days)}d) ──")
|
|
is_results = await run_sweep_on_period(
|
|
sweep, context, client, train_days, progress_prefix="IS "
|
|
)
|
|
|
|
print(f"\n Top {top_n} IS configs (by Sharpe):")
|
|
print(f" {'#':<3} {'Sharpe':>7} {'Return':>8} {'WR':>6} {'PF':>6} {'Trades':>7} {'DD':>7} Params")
|
|
for i, result in enumerate(is_results[:top_n]):
|
|
metrics = result.metrics
|
|
print(
|
|
f" {i+1:<3} {metrics.sharpe_ratio or 0:>7.2f} "
|
|
f"{(metrics.total_return_pct or 0)*100:>7.0f}% "
|
|
f"{(metrics.win_rate or 0)*100:>5.1f}% "
|
|
f"{metrics.profit_factor or 0:>6.2f} "
|
|
f"{metrics.total_trades or 0:>7} "
|
|
f"{abs((metrics.max_drawdown_pct or 0)*100):>6.2f}% "
|
|
f"{_format_params(result.params)}"
|
|
)
|
|
|
|
print(f"\n ── Phase 2: Validate on Test ({test_days[0]}→{test_days[-1]}, {len(test_days)}d) ──")
|
|
oos_results: list[tuple[SweepResult, IntradayMetrics]] = []
|
|
for i, is_result in enumerate(is_results[:top_n]):
|
|
oos_metrics = await run_single_config_on_period(
|
|
sweep.base_config,
|
|
is_result.params,
|
|
context,
|
|
client,
|
|
test_days,
|
|
run_id=f"oos{i:02d}",
|
|
)
|
|
oos_results.append((is_result, oos_metrics))
|
|
sys.stdout.write(
|
|
f"\r OOS [{i+1}/{top_n}] "
|
|
f"Sharpe={oos_metrics.sharpe_ratio or 0:.2f} "
|
|
f"Ret={(oos_metrics.total_return_pct or 0)*100:.0f}%"
|
|
)
|
|
sys.stdout.flush()
|
|
print()
|
|
|
|
print(f"\n {'='*80}")
|
|
print(" IS/OOS Comparison (Target: OOS Sharpe ≥ 60% of IS Sharpe)")
|
|
print(f" {'='*80}")
|
|
print(
|
|
f" {'#':<3} {'IS Sharpe':>10} {'OOS Sharpe':>11} {'Retain':>8} "
|
|
f"{'IS Ret':>8} {'OOS Ret':>8} {'OOS DD':>7} {'Verdict':>8} Params"
|
|
)
|
|
|
|
report_rows: list[dict[str, Any]] = []
|
|
for i, (is_result, oos_metrics) in enumerate(oos_results):
|
|
is_sharpe = is_result.metrics.sharpe_ratio or 0.0
|
|
oos_sharpe = oos_metrics.sharpe_ratio or 0.0
|
|
retain = compute_degradation(is_sharpe, oos_sharpe)
|
|
verdict = "PASS" if retain >= 0.60 else "WARN" if retain >= 0.40 else "FAIL"
|
|
print(
|
|
f" {i+1:<3} {is_sharpe:>10.2f} {oos_sharpe:>11.2f} {retain*100:>7.0f}% "
|
|
f"{(is_result.metrics.total_return_pct or 0)*100:>7.0f}% "
|
|
f"{(oos_metrics.total_return_pct or 0)*100:>7.0f}% "
|
|
f"{abs((oos_metrics.max_drawdown_pct or 0)*100):>6.2f}% "
|
|
f"{verdict:>6} {_format_params(is_result.params)}"
|
|
)
|
|
report_rows.append({
|
|
"rank": i + 1,
|
|
"params": is_result.params,
|
|
"is_sharpe": is_sharpe,
|
|
"oos_sharpe": oos_sharpe,
|
|
"retention_pct": round(retain * 100, 1),
|
|
"is_return_pct": round((is_result.metrics.total_return_pct or 0) * 100, 1),
|
|
"oos_return_pct": round((oos_metrics.total_return_pct or 0) * 100, 1),
|
|
"oos_max_dd_pct": round(abs((oos_metrics.max_drawdown_pct or 0) * 100), 2),
|
|
"oos_trades": oos_metrics.total_trades,
|
|
"oos_win_rate": round((oos_metrics.win_rate or 0) * 100, 1),
|
|
"verdict": verdict,
|
|
})
|
|
|
|
if oos_sort:
|
|
report_rows.sort(key=lambda row: row["oos_sharpe"], reverse=True)
|
|
for i, row in enumerate(report_rows, start=1):
|
|
row["rank"] = i
|
|
print("\n [Re-ranked by OOS Sharpe]")
|
|
|
|
retentions = [row["retention_pct"] for row in report_rows]
|
|
avg_retain = statistics.mean(retentions) if retentions else 0.0
|
|
median_retain = statistics.median(retentions) if retentions else 0.0
|
|
pass_count = sum(1 for row in report_rows if row["verdict"] == "PASS")
|
|
|
|
print("\n Summary:")
|
|
print(f" Avg retention: {avg_retain:.0f}%")
|
|
print(f" Median retention: {median_retain:.0f}%")
|
|
print(f" Passed (≥60%): {pass_count}/{len(report_rows)}")
|
|
|
|
if avg_retain >= 60:
|
|
print("\n Overall: ROBUST — strategy generalizes well to unseen data")
|
|
elif avg_retain >= 40:
|
|
print("\n Overall: FRAGILE — moderate overfitting detected, caution advised")
|
|
else:
|
|
print("\n Overall: OVERFIT — strategy does not generalize, parameter re-tuning needed")
|
|
|
|
return {
|
|
"mode": "split",
|
|
"train_period": f"{train_days[0]} → {train_days[-1]}",
|
|
"train_days": len(train_days),
|
|
"test_period": f"{test_days[0]} → {test_days[-1]}",
|
|
"test_days": len(test_days),
|
|
"total_sweep_combos": sweep.total_combinations,
|
|
"top_n": len(report_rows),
|
|
"results": report_rows,
|
|
"avg_retention_pct": round(avg_retain, 1),
|
|
"median_retention_pct": round(median_retain, 1),
|
|
"pass_count": pass_count,
|
|
}
|
|
|
|
|
|
async def _run_walk_forward_eval(
|
|
sweep: SweepConfig,
|
|
context,
|
|
client: OracleClient,
|
|
windows: list[tuple[list[str], list[str]]],
|
|
) -> dict[str, Any]:
|
|
wf_results: list[dict[str, Any]] = []
|
|
|
|
for window_idx, (train_days, test_days) in enumerate(windows, start=1):
|
|
print(
|
|
f"\n ── Window {window_idx}/{len(windows)}: "
|
|
f"Train {train_days[0]}→{train_days[-1]} | "
|
|
f"Test {test_days[0]}→{test_days[-1]} ──"
|
|
)
|
|
is_results = await run_sweep_on_period(
|
|
sweep, context, client, train_days, progress_prefix=f"W{window_idx} IS "
|
|
)
|
|
if not is_results:
|
|
continue
|
|
|
|
best_is = is_results[0]
|
|
is_sharpe = best_is.metrics.sharpe_ratio or 0.0
|
|
oos_metrics = await run_single_config_on_period(
|
|
sweep.base_config,
|
|
best_is.params,
|
|
context,
|
|
client,
|
|
test_days,
|
|
run_id=f"wf{window_idx:02d}",
|
|
)
|
|
oos_sharpe = oos_metrics.sharpe_ratio or 0.0
|
|
retain = compute_degradation(is_sharpe, oos_sharpe)
|
|
print(
|
|
f" Best: IS Sharpe={is_sharpe:.2f} → OOS Sharpe={oos_sharpe:.2f} "
|
|
f"(retain {retain*100:.0f}%) {_format_params(best_is.params)}"
|
|
)
|
|
wf_results.append({
|
|
"window": window_idx,
|
|
"train_period": f"{train_days[0]} → {train_days[-1]}",
|
|
"test_period": f"{test_days[0]} → {test_days[-1]}",
|
|
"best_params": best_is.params,
|
|
"is_sharpe": is_sharpe,
|
|
"oos_sharpe": oos_sharpe,
|
|
"retention_pct": round(retain * 100, 1),
|
|
"oos_return_pct": round((oos_metrics.total_return_pct or 0) * 100, 1),
|
|
"oos_trades": oos_metrics.total_trades,
|
|
"oos_win_rate": round((oos_metrics.win_rate or 0) * 100, 1),
|
|
})
|
|
|
|
retentions = [row["retention_pct"] for row in wf_results]
|
|
oos_sharpes = [row["oos_sharpe"] for row in wf_results]
|
|
avg_retain = statistics.mean(retentions) if retentions else 0.0
|
|
median_oos = statistics.median(oos_sharpes) if oos_sharpes else 0.0
|
|
positive_oos = sum(1 for sharpe in oos_sharpes if sharpe > 0)
|
|
|
|
print(f"\n {'='*70}")
|
|
print(f" Walk-Forward Summary ({len(wf_results)} windows)")
|
|
print(f" {'='*70}")
|
|
print(f" {'Window':<8} {'Train':>26} {'Test':>26} {'IS Sh':>6} {'OOS Sh':>7} {'Retain':>7}")
|
|
for row in wf_results:
|
|
print(
|
|
f" {row['window']:<8} {row['train_period']:>26} {row['test_period']:>26} "
|
|
f"{row['is_sharpe']:>6.2f} {row['oos_sharpe']:>7.2f} {row['retention_pct']:>6.0f}%"
|
|
)
|
|
|
|
print(f"\n Avg OOS Sharpe: {statistics.mean(oos_sharpes) if oos_sharpes else 0:.2f}")
|
|
print(f" Median OOS Sharpe: {median_oos:.2f}")
|
|
print(f" Avg retention: {avg_retain:.0f}%")
|
|
print(f" Positive OOS: {positive_oos}/{len(wf_results)}")
|
|
|
|
if wf_results:
|
|
serialized = [json.dumps(row["best_params"], sort_keys=True) for row in wf_results]
|
|
unique_params = len(set(serialized))
|
|
print(f" Unique best configs: {unique_params}/{len(wf_results)}")
|
|
if unique_params <= len(wf_results) * 0.5:
|
|
print(" → Parameter stability: GOOD (same params often selected)")
|
|
else:
|
|
print(" → Parameter stability: POOR (different params per window)")
|
|
|
|
return {
|
|
"mode": "walk_forward",
|
|
"n_windows": len(wf_results),
|
|
"windows": wf_results,
|
|
"avg_oos_sharpe": round(statistics.mean(oos_sharpes), 2) if oos_sharpes else 0,
|
|
"median_oos_sharpe": round(median_oos, 2),
|
|
"avg_retention_pct": round(avg_retain, 1),
|
|
"positive_oos_count": positive_oos,
|
|
}
|
|
|
|
|
|
async def evaluate(
|
|
config: IntradayConfig,
|
|
sweep_path: str,
|
|
split_date: str | None = None,
|
|
train_ratio: float | None = None,
|
|
top_n: int = 5,
|
|
walk_forward: bool = False,
|
|
wf_train_days: int = 252,
|
|
wf_test_days: int = 63,
|
|
oos_sort: bool = False,
|
|
) -> dict[str, Any]:
|
|
settings = get_settings()
|
|
|
|
config = force_simple_returns(config)
|
|
sweep = load_sweep_config(sweep_path, config)
|
|
print(f"\n{'='*65}")
|
|
print(" ORB Strategy Evaluation — IS/OOS Validation")
|
|
print(f"{'='*65}")
|
|
print(f"\n Sweep: {sweep.total_combinations} parameter combinations")
|
|
print(f" Sweep params: {list(sweep.sweep_params.keys())}")
|
|
|
|
async with make_intraday_oracle_client(settings) as client:
|
|
print("\n[1/4] Building shared ORB research context...")
|
|
resolved_days = await get_trading_days(
|
|
client,
|
|
config.backtest.start_date,
|
|
config.backtest.end_date,
|
|
config.backtest.lookback_trading_days,
|
|
)
|
|
if not resolved_days:
|
|
raise ValueError("No trading days resolved for evaluation window")
|
|
start_date = resolved_days[0]
|
|
end_date = resolved_days[-1]
|
|
|
|
context = await build_orb_research_context(
|
|
config,
|
|
start_date,
|
|
end_date,
|
|
client,
|
|
print_progress=True,
|
|
)
|
|
print(
|
|
f" Shared context: {len(context.trading_days)} days, "
|
|
f"{context.total_pairs} ticker-day pairs"
|
|
)
|
|
|
|
if walk_forward:
|
|
windows = generate_walk_forward_windows(
|
|
context.trading_days,
|
|
train_days=wf_train_days,
|
|
test_days=wf_test_days,
|
|
)
|
|
print(
|
|
f"\n[2/4] Walk-forward: {len(windows)} windows "
|
|
f"(train={wf_train_days}d, test={wf_test_days}d)"
|
|
)
|
|
print("\n[3/4] Running streaming evaluation...")
|
|
report = await _run_walk_forward_eval(sweep, context, client, windows)
|
|
else:
|
|
train_days, test_days = split_trading_days(
|
|
context.trading_days,
|
|
split_date=split_date,
|
|
train_ratio=train_ratio,
|
|
)
|
|
print("\n[2/4] Train/Test split:")
|
|
print(f" Train (IS): {train_days[0]} → {train_days[-1]} ({len(train_days)} days)")
|
|
print(f" Test (OOS): {test_days[0]} → {test_days[-1]} ({len(test_days)} days)")
|
|
print("\n[3/4] Running streaming evaluation...")
|
|
report = await _run_split_eval(
|
|
sweep,
|
|
context,
|
|
client,
|
|
train_days,
|
|
test_days,
|
|
top_n,
|
|
oos_sort=oos_sort,
|
|
)
|
|
|
|
return report
|
|
|
|
|
|
def _format_params(params: dict[str, Any]) -> str:
|
|
parts = []
|
|
for key, value in sorted(params.items()):
|
|
short_key = key.replace("_multiplier", "").replace("_pct", "%").replace("_at_r", "R")
|
|
if isinstance(value, float):
|
|
parts.append(f"{short_key}={value:.2g}")
|
|
elif value is None:
|
|
parts.append(f"{short_key}=off")
|
|
else:
|
|
parts.append(f"{short_key}={value}")
|
|
return " ".join(parts)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="ORB strategy IS/OOS evaluation",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
parser.add_argument("--config", required=True, help="Base strategy config YAML")
|
|
parser.add_argument("--sweep", required=True, help="Sweep parameter grid YAML")
|
|
parser.add_argument("--start", help="Start date (YYYY-MM-DD)")
|
|
parser.add_argument("--end", help="End date (YYYY-MM-DD)")
|
|
parser.add_argument("--days", type=int, help="Lookback trading days (alternative to --start)")
|
|
parser.add_argument("--split-date", help="Train/test split date (YYYY-MM-DD)")
|
|
parser.add_argument("--train-ratio", type=float, help="Fraction of data for training (0.0-1.0)")
|
|
parser.add_argument("--top-n", type=int, default=5, help="Top N configs to validate OOS")
|
|
parser.add_argument("--walk-forward", action="store_true", help="Enable walk-forward evaluation")
|
|
parser.add_argument("--wf-train-days", type=int, default=252, help="WF train window (trading days)")
|
|
parser.add_argument("--wf-test-days", type=int, default=63, help="WF test window (trading days)")
|
|
parser.add_argument("--output", help="Save report JSON to path")
|
|
parser.add_argument("--oos-sort", action="store_true", help="Re-rank final results by OOS Sharpe")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
with open(args.config) as handle:
|
|
raw = yaml.safe_load(handle) or {}
|
|
|
|
orb_params = ORBStrategyParams(**(raw.get("orb_strategy", {})))
|
|
backtest_params = BacktestParams(**(raw.get("backtest", {})))
|
|
universe_params = UniverseParams(**(raw.get("universe", {})))
|
|
cache_params = CacheParams(**(raw.get("cache", {})))
|
|
output_params = OutputParams(**(raw.get("output", {})))
|
|
|
|
if args.start:
|
|
backtest_params.start_date = args.start
|
|
if args.end:
|
|
backtest_params.end_date = args.end
|
|
if args.days:
|
|
backtest_params.lookback_trading_days = args.days
|
|
|
|
config = IntradayConfig(
|
|
strategy_mode=raw.get("strategy_mode", "orb"),
|
|
orb_strategy=orb_params,
|
|
backtest=backtest_params,
|
|
universe=universe_params,
|
|
cache=cache_params,
|
|
output=output_params,
|
|
)
|
|
|
|
report = asyncio.run(
|
|
evaluate(
|
|
config=config,
|
|
sweep_path=args.sweep,
|
|
split_date=args.split_date,
|
|
train_ratio=args.train_ratio,
|
|
top_n=args.top_n,
|
|
walk_forward=args.walk_forward,
|
|
wf_train_days=args.wf_train_days,
|
|
wf_test_days=args.wf_test_days,
|
|
oos_sort=args.oos_sort,
|
|
)
|
|
)
|
|
|
|
if args.output:
|
|
out_path = Path(args.output)
|
|
else:
|
|
out_dir = Path(config.output.dir)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
out_path = out_dir / f"eval_{ts}.json"
|
|
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
out_path.write_text(json.dumps(report, indent=2))
|
|
print(f"\n Report saved: {out_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|