|
|
"""Parameter grid search engine for intraday backtesting.
|
|
|
|
|
|
Data is fetched once; simulations run repeatedly with different params.
|
|
|
288 combinations × ~200 days ≈ 5 minutes total simulation time.
|
|
|
"""
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import itertools
|
|
|
from typing import Any
|
|
|
|
|
|
import yaml
|
|
|
|
|
|
from libs.intraday.domain import (
|
|
|
IntradayConfig,
|
|
|
ORBStrategyParams,
|
|
|
StrategyParams,
|
|
|
SweepResult,
|
|
|
)
|
|
|
from libs.intraday.metrics import compute_metrics
|
|
|
from libs.intraday.simulator import run_simulation
|
|
|
|
|
|
|
|
|
class SweepConfig:
|
|
|
"""Parsed sweep configuration."""
|
|
|
|
|
|
def __init__(
|
|
|
self,
|
|
|
base_config: IntradayConfig,
|
|
|
sweep_params: dict[str, list[Any]],
|
|
|
objective: dict[str, Any] | None = None,
|
|
|
) -> None:
|
|
|
self.base_config = base_config
|
|
|
self.sweep_params = sweep_params
|
|
|
self.objective = objective or {}
|
|
|
|
|
|
@property
|
|
|
def total_combinations(self) -> int:
|
|
|
total = 1
|
|
|
for vals in self.sweep_params.values():
|
|
|
total *= len(vals)
|
|
|
return total
|
|
|
|
|
|
|
|
|
def _normalize_sweep_value(
|
|
|
field_name: str,
|
|
|
value: Any,
|
|
|
model_fields: dict[str, Any],
|
|
|
) -> Any:
|
|
|
"""Normalize YAML sweep values without breaking literal string enums like 'none'."""
|
|
|
if value is None or value in ("null", "None"):
|
|
|
return None
|
|
|
if value != "none":
|
|
|
return value
|
|
|
field = model_fields.get(field_name)
|
|
|
if field is not None and getattr(field, "default", None) == "none":
|
|
|
return "none"
|
|
|
return None
|
|
|
|
|
|
|
|
|
def load_sweep_config(sweep_path: str, base_config: IntradayConfig) -> SweepConfig:
|
|
|
"""Load a sweep YAML and merge with the base config."""
|
|
|
with open(sweep_path) as f:
|
|
|
raw = yaml.safe_load(f)
|
|
|
|
|
|
model_fields = (
|
|
|
ORBStrategyParams.model_fields
|
|
|
if base_config.strategy_mode == "orb"
|
|
|
else StrategyParams.model_fields
|
|
|
)
|
|
|
sweep_params: dict[str, list[Any]] = {}
|
|
|
for key, vals in raw.get("sweep", {}).items():
|
|
|
if not isinstance(vals, list):
|
|
|
vals = [vals]
|
|
|
# Normalize None strings and null values
|
|
|
normalized = [_normalize_sweep_value(key, v, model_fields) for v in vals]
|
|
|
sweep_params[key] = normalized
|
|
|
|
|
|
objective = raw.get("objective", {}) or {}
|
|
|
return SweepConfig(base_config=base_config, sweep_params=sweep_params, objective=objective)
|
|
|
|
|
|
|
|
|
def _trade_day_objective_score(metrics: Any, objective: dict[str, Any]) -> float:
|
|
|
"""Score sweep rows with an explicit bonus for safe capital usage.
|
|
|
|
|
|
The default remains Sharpe sorting unless a sweep YAML opts into
|
|
|
objective.name: trade_day_adjusted.
|
|
|
"""
|
|
|
sharpe = float(metrics.sharpe_ratio or -999.0)
|
|
|
total_return = float(metrics.total_return_pct or -999.0)
|
|
|
trade_days = float(metrics.days_with_trades or 0)
|
|
|
target_trade_days = float(objective.get("target_days_with_trades") or 80)
|
|
|
trade_day_bonus = float(objective.get("trade_day_bonus") or 0.0)
|
|
|
return_weight = float(objective.get("return_weight") or 0.10)
|
|
|
min_return = objective.get("min_total_return_pct")
|
|
|
max_drawdown_floor = objective.get("max_drawdown_floor")
|
|
|
min_profit_factor = objective.get("min_profit_factor")
|
|
|
|
|
|
score = sharpe + (total_return * return_weight)
|
|
|
if target_trade_days > 0:
|
|
|
score += min(trade_days / target_trade_days, 1.0) * trade_day_bonus
|
|
|
|
|
|
if min_return is not None and total_return < float(min_return):
|
|
|
score -= 100.0
|
|
|
max_drawdown = metrics.max_drawdown_pct
|
|
|
if max_drawdown_floor is not None and max_drawdown is not None:
|
|
|
if float(max_drawdown) < float(max_drawdown_floor):
|
|
|
score -= 100.0
|
|
|
profit_factor = metrics.profit_factor
|
|
|
if min_profit_factor is not None and profit_factor is not None:
|
|
|
if float(profit_factor) < float(min_profit_factor):
|
|
|
score -= 100.0
|
|
|
return score
|
|
|
|
|
|
|
|
|
def _sweep_objective_score(metrics: Any, objective: dict[str, Any]) -> float:
|
|
|
if objective.get("name") == "trade_day_adjusted":
|
|
|
return _trade_day_objective_score(metrics, objective)
|
|
|
return float(metrics.sharpe_ratio or -999.0) + float(metrics.total_return_pct or -999.0) * 0.001
|
|
|
|
|
|
|
|
|
def generate_combinations(sweep: SweepConfig) -> list[dict[str, Any]]:
|
|
|
"""Generate Cartesian product of all sweep parameters."""
|
|
|
keys = sorted(sweep.sweep_params.keys())
|
|
|
values = [sweep.sweep_params[k] for k in keys]
|
|
|
combos = list(itertools.product(*values))
|
|
|
return [dict(zip(keys, combo)) for combo in combos]
|
|
|
|
|
|
|
|
|
def apply_overrides(base_config: IntradayConfig, overrides: dict[str, Any]) -> IntradayConfig:
|
|
|
"""Apply parameter overrides to base config, returning a new config.
|
|
|
|
|
|
Branches on strategy_mode: momentum overrides go to StrategyParams,
|
|
|
ORB overrides go to ORBStrategyParams.
|
|
|
"""
|
|
|
if base_config.strategy_mode == "orb":
|
|
|
orb = base_config.orb_strategy or ORBStrategyParams()
|
|
|
orb_dict = orb.model_dump()
|
|
|
orb_fields = set(ORBStrategyParams.model_fields.keys())
|
|
|
for key, val in overrides.items():
|
|
|
if key in orb_fields:
|
|
|
orb_dict[key] = val
|
|
|
new_orb = ORBStrategyParams(**orb_dict)
|
|
|
return base_config.model_copy(update={"orb_strategy": new_orb})
|
|
|
|
|
|
# Momentum mode (default)
|
|
|
strategy_dict = base_config.strategy.model_dump()
|
|
|
strategy_fields = set(StrategyParams.model_fields.keys())
|
|
|
for key, val in overrides.items():
|
|
|
if key in strategy_fields:
|
|
|
strategy_dict[key] = val
|
|
|
new_strategy = StrategyParams(**strategy_dict)
|
|
|
return base_config.model_copy(update={"strategy": new_strategy})
|
|
|
|
|
|
|
|
|
def _filter_intraday_by_candidate_map(
|
|
|
all_intraday: dict[str, dict[str, list[dict]]],
|
|
|
candidate_map: dict[str, list[str]] | None,
|
|
|
) -> dict[str, dict[str, list[dict]]]:
|
|
|
"""Restrict preloaded intraday data to the combo-specific candidate set."""
|
|
|
if not candidate_map:
|
|
|
return all_intraday
|
|
|
filtered: dict[str, dict[str, list[dict]]] = {}
|
|
|
for day, tickers in candidate_map.items():
|
|
|
day_intraday = all_intraday.get(day, {})
|
|
|
if not day_intraday:
|
|
|
continue
|
|
|
selected = {
|
|
|
ticker: day_intraday[ticker]
|
|
|
for ticker in tickers
|
|
|
if ticker in day_intraday
|
|
|
}
|
|
|
if selected:
|
|
|
filtered[day] = selected
|
|
|
return filtered
|
|
|
|
|
|
|
|
|
def run_sweep(
|
|
|
sweep: SweepConfig,
|
|
|
all_intraday: dict[str, dict[str, list[dict]]],
|
|
|
trading_days: list[str],
|
|
|
progress_callback: Any = None,
|
|
|
enrichment: dict | None = None,
|
|
|
momentum_enrichment: dict | None = None,
|
|
|
vix_by_day: dict[str, float] | None = None,
|
|
|
ticker_sectors: dict[str, str] | None = None,
|
|
|
sector_proxy_intraday_by_day: dict[str, dict[str, list[dict]]] | None = None,
|
|
|
overlay_tickers_per_day: dict[str, set[str]] | None = None,
|
|
|
orb_context_resolver: Any = None,
|
|
|
) -> list[SweepResult]:
|
|
|
"""Run simulation for each parameter combination.
|
|
|
|
|
|
Data is pre-fetched and shared across all runs.
|
|
|
Only the simulation (pure CPU computation) varies per combination.
|
|
|
|
|
|
Args:
|
|
|
sweep: SweepConfig with base config and param grid.
|
|
|
all_intraday: Pre-loaded {date: {ticker: [bars]}} data.
|
|
|
trading_days: List of dates.
|
|
|
progress_callback: Optional callable(completed, total) for progress.
|
|
|
enrichment: Pre-computed daily enrichment (required for ORB strategy).
|
|
|
|
|
|
Returns:
|
|
|
List of SweepResult sorted by Sharpe ratio descending.
|
|
|
"""
|
|
|
combos = generate_combinations(sweep)
|
|
|
results: list[SweepResult] = []
|
|
|
is_orb = sweep.base_config.strategy_mode == "orb"
|
|
|
|
|
|
for i, overrides in enumerate(combos):
|
|
|
config = apply_overrides(sweep.base_config, overrides)
|
|
|
|
|
|
if is_orb:
|
|
|
combo_enrichment = enrichment or {}
|
|
|
combo_vix = vix_by_day
|
|
|
combo_overlay_tickers = overlay_tickers_per_day
|
|
|
combo_candidate_map = None
|
|
|
if orb_context_resolver is not None:
|
|
|
resolved = orb_context_resolver(config)
|
|
|
if len(resolved) == 4:
|
|
|
combo_enrichment, combo_vix, combo_overlay_tickers, combo_candidate_map = resolved
|
|
|
else:
|
|
|
combo_enrichment, combo_vix, combo_overlay_tickers = resolved
|
|
|
from libs.intraday.orb_simulator import run_orb_simulation
|
|
|
combo_intraday = _filter_intraday_by_candidate_map(all_intraday, combo_candidate_map)
|
|
|
day_results = run_orb_simulation(
|
|
|
combo_intraday, trading_days, config.orb_strategy, combo_enrichment,
|
|
|
ticker_sectors=ticker_sectors,
|
|
|
vix_by_day=combo_vix,
|
|
|
overlay_tickers_per_day=combo_overlay_tickers,
|
|
|
)
|
|
|
else:
|
|
|
day_results = run_simulation(
|
|
|
all_intraday,
|
|
|
trading_days,
|
|
|
config.strategy,
|
|
|
daily_enrichment=momentum_enrichment,
|
|
|
vix_by_day=vix_by_day,
|
|
|
ticker_sectors=ticker_sectors,
|
|
|
sector_proxy_intraday_by_day=sector_proxy_intraday_by_day,
|
|
|
)
|
|
|
|
|
|
metrics = compute_metrics(day_results, config, run_id=f"sw{i:04d}")
|
|
|
objective_score = _sweep_objective_score(metrics, sweep.objective)
|
|
|
results.append(
|
|
|
SweepResult(
|
|
|
params=overrides,
|
|
|
metrics=metrics,
|
|
|
objective_score=round(objective_score, 6),
|
|
|
)
|
|
|
)
|
|
|
|
|
|
if progress_callback:
|
|
|
progress_callback(i + 1, len(combos))
|
|
|
|
|
|
# Sort by configured objective. Default behavior is Sharpe-like unless the
|
|
|
# sweep YAML opts into a trade-day-adjusted objective.
|
|
|
results.sort(
|
|
|
key=lambda r: (
|
|
|
r.objective_score if r.objective_score is not None else float("-inf"),
|
|
|
r.metrics.sharpe_ratio or float("-inf"),
|
|
|
r.metrics.total_return_pct or -999,
|
|
|
),
|
|
|
reverse=True,
|
|
|
)
|
|
|
return results
|