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.
1271 lines
46 KiB
Python
1271 lines
46 KiB
Python
"""Common ORB research helpers for streaming lab/evaluation workflows."""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import datetime as dt
|
|
import gzip
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import pickle
|
|
import random
|
|
import statistics
|
|
import sys
|
|
from dataclasses import asdict
|
|
from dataclasses import dataclass
|
|
from dataclasses import field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from libs.backtest.domain import (
|
|
SplitResult,
|
|
WalkForwardAggregate,
|
|
WalkForwardFoldResult,
|
|
WalkForwardGapStats,
|
|
WalkForwardSummary,
|
|
)
|
|
from libs.backtest.tracker import compute_rqs, compute_wfqs_v2
|
|
from libs.common.config import get_settings
|
|
from libs.intraday.cache import DailyBarCache, IntradayCache
|
|
from libs.intraday.domain import (
|
|
IntradayConfig,
|
|
IntradayMetrics,
|
|
ORBStrategyParams,
|
|
)
|
|
from libs.intraday.features import enrich_daily_bars
|
|
from libs.intraday.metrics import IntradayMetricsAccumulator
|
|
from libs.intraday.orb_simulator import ORBSimulationState, run_orb_simulation_with_state
|
|
from libs.oracle_client import OracleClient
|
|
|
|
from apps.intraday_bt.run import (
|
|
_chunk_trading_days_by_pairs,
|
|
get_trading_days,
|
|
load_config,
|
|
_make_progress_bar,
|
|
)
|
|
from apps.intraday_bt.sweep import apply_overrides
|
|
from libs.intraday.screener import (
|
|
fetch_daily_bars_bulk,
|
|
fetch_intraday_bulk,
|
|
orb_pre_screen_candidates,
|
|
resolve_universe,
|
|
)
|
|
|
|
|
|
LAB_TRAIN_START = "2024-01-02"
|
|
LAB_TRAIN_END = "2024-12-31"
|
|
LAB_VALID_START = "2025-01-02"
|
|
LAB_VALID_END = "2025-12-31"
|
|
LAB_TEST_START = "2026-01-02"
|
|
LAB_TEST_END = "2026-03-31"
|
|
LAB_ROBUSTNESS_START = "2022-01-03"
|
|
LAB_ROBUSTNESS_END = "2023-12-29"
|
|
_ORB_RESEARCH_SNAPSHOT_VERSION = 1
|
|
_ORB_PERIOD_METRICS_CACHE_VERSION = 1
|
|
_ORB_TAPE_CACHE_VERSION = 1
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ORBResearchPeriods:
|
|
train_start: str = LAB_TRAIN_START
|
|
train_end: str = LAB_TRAIN_END
|
|
valid_start: str = LAB_VALID_START
|
|
valid_end: str = LAB_VALID_END
|
|
test_start: str = LAB_TEST_START
|
|
test_end: str = LAB_TEST_END
|
|
robustness_start: str = LAB_ROBUSTNESS_START
|
|
robustness_end: str = LAB_ROBUSTNESS_END
|
|
|
|
|
|
DEFAULT_ORB_RESEARCH_PERIODS = ORBResearchPeriods()
|
|
|
|
|
|
def _load_orb_ticker_sectors(tickers: list[str]) -> dict[str, str]:
|
|
settings = get_settings()
|
|
path = Path(settings.data_root) / "cache" / "sector_cache.json"
|
|
if not path.exists():
|
|
return {ticker: "UNKNOWN" for ticker in tickers}
|
|
try:
|
|
payload = json.loads(path.read_text())
|
|
except Exception:
|
|
return {ticker: "UNKNOWN" for ticker in tickers}
|
|
return {ticker: str(payload.get(ticker) or "UNKNOWN") for ticker in tickers}
|
|
|
|
|
|
@dataclass
|
|
class ORBResearchContext:
|
|
config: IntradayConfig
|
|
tickers: list[str]
|
|
trading_days: list[str]
|
|
daily_bars: dict[str, list[dict]]
|
|
enrichment: dict[str, dict[str, dict]]
|
|
candidates: dict[str, list[str]]
|
|
cache: IntradayCache | None
|
|
daily_cache: DailyBarCache | None
|
|
eval_cache: ORBPeriodMetricsCache | None
|
|
tape_cache: ORBPreparedTapeStore | None
|
|
oracle_url: str
|
|
ticker_sectors: dict[str, str] = field(default_factory=dict)
|
|
vix_by_day: dict[str, float] | None = None
|
|
research_snapshot_key: str | None = None
|
|
|
|
@property
|
|
def total_pairs(self) -> int:
|
|
return sum(len(v) for v in self.candidates.values())
|
|
|
|
|
|
class ORBResearchSnapshotStore:
|
|
"""Disk snapshot for expensive ORB research context preparation."""
|
|
|
|
def __init__(self, root: str | Path) -> None:
|
|
self.root = Path(root)
|
|
|
|
def _path(self, key: str) -> Path:
|
|
return self.root / key[:2] / f"{key}.pkl.gz"
|
|
|
|
@staticmethod
|
|
def _signature_payload(
|
|
config: IntradayConfig,
|
|
*,
|
|
start_date: str,
|
|
end_date: str,
|
|
tickers: list[str],
|
|
trading_days: list[str],
|
|
) -> dict[str, Any]:
|
|
orb = config.orb_strategy or ORBStrategyParams()
|
|
return {
|
|
"version": _ORB_RESEARCH_SNAPSHOT_VERSION,
|
|
"strategy_mode": config.strategy_mode,
|
|
"start_date": start_date,
|
|
"end_date": end_date,
|
|
"universe": config.universe.model_dump(mode="json"),
|
|
"context_filters": {
|
|
"min_price": orb.min_price,
|
|
"min_atr_14": orb.min_atr_14,
|
|
"min_avg_dollar_volume": orb.min_avg_dollar_volume,
|
|
"market_regime_ticker": getattr(orb, "market_regime_ticker", "SPY") or "SPY",
|
|
"market_regime_spy_threshold": orb.market_regime_spy_threshold,
|
|
},
|
|
"tickers": tickers,
|
|
"trading_days": trading_days,
|
|
}
|
|
|
|
@classmethod
|
|
def build_key(
|
|
cls,
|
|
config: IntradayConfig,
|
|
*,
|
|
start_date: str,
|
|
end_date: str,
|
|
tickers: list[str],
|
|
trading_days: list[str],
|
|
) -> str:
|
|
payload = cls._signature_payload(
|
|
config,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
tickers=tickers,
|
|
trading_days=trading_days,
|
|
)
|
|
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
|
|
return hashlib.sha1(blob.encode("utf-8")).hexdigest()
|
|
|
|
def load(self, key: str) -> dict[str, Any] | None:
|
|
path = self._path(key)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
with gzip.open(path, "rb") as fh:
|
|
payload = pickle.load(fh)
|
|
except Exception:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
|
|
if not isinstance(payload, dict):
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("version") != _ORB_RESEARCH_SNAPSHOT_VERSION:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("key") != key:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
for required in ("tickers", "trading_days", "daily_bars", "enrichment", "candidates"):
|
|
if required not in payload:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
return payload
|
|
|
|
def save(self, key: str, payload: dict[str, Any]) -> Path:
|
|
path = self._path(key)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(".tmp")
|
|
record = dict(payload)
|
|
record["version"] = _ORB_RESEARCH_SNAPSHOT_VERSION
|
|
record["key"] = key
|
|
with gzip.open(tmp, "wb", compresslevel=3) as fh:
|
|
pickle.dump(record, fh, protocol=pickle.HIGHEST_PROTOCOL)
|
|
tmp.replace(path)
|
|
return path
|
|
|
|
|
|
class ORBPeriodMetricsCache:
|
|
"""Disk cache for expensive period-level ORB simulations."""
|
|
|
|
def __init__(self, root: str | Path) -> None:
|
|
self.root = Path(root)
|
|
|
|
def _path(self, key: str) -> Path:
|
|
return self.root / key[:2] / f"{key}.json.gz"
|
|
|
|
def _checkpoint_path(self, key: str) -> Path:
|
|
return self.root / key[:2] / f"{key}.checkpoint.json.gz"
|
|
|
|
@classmethod
|
|
def build_key(
|
|
cls,
|
|
*,
|
|
research_snapshot_key: str,
|
|
orb_params: ORBStrategyParams,
|
|
trading_days: list[str],
|
|
shuffle_candidates_seed: int | None,
|
|
) -> str:
|
|
payload = {
|
|
"version": _ORB_PERIOD_METRICS_CACHE_VERSION,
|
|
"research_snapshot_key": research_snapshot_key,
|
|
"orb_params": orb_params.model_dump(mode="json"),
|
|
"trading_day_count": len(trading_days),
|
|
"trading_day_start": trading_days[0] if trading_days else "",
|
|
"trading_day_end": trading_days[-1] if trading_days else "",
|
|
"trading_days_hash": hashlib.sha1(
|
|
",".join(trading_days).encode("utf-8")
|
|
).hexdigest(),
|
|
"shuffle_candidates_seed": shuffle_candidates_seed,
|
|
}
|
|
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
|
|
return hashlib.sha1(blob.encode("utf-8")).hexdigest()
|
|
|
|
def load(self, key: str) -> dict[str, Any] | None:
|
|
path = self._path(key)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
with gzip.open(path, "rt", encoding="utf-8") as fh:
|
|
payload = json.load(fh)
|
|
except Exception:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if not isinstance(payload, dict):
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("version") != _ORB_PERIOD_METRICS_CACHE_VERSION:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("key") != key:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if "metrics" not in payload:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
return payload
|
|
|
|
def load_checkpoint(self, key: str) -> dict[str, Any] | None:
|
|
path = self._checkpoint_path(key)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
with gzip.open(path, "rt", encoding="utf-8") as fh:
|
|
payload = json.load(fh)
|
|
except Exception:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if not isinstance(payload, dict):
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("version") != _ORB_PERIOD_METRICS_CACHE_VERSION:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("key") != key:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
required = {"completed_chunks", "total_chunks", "chunk_layout", "accumulator", "sim_state"}
|
|
if not required.issubset(set(payload)):
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
return payload
|
|
|
|
def save(self, key: str, payload: dict[str, Any]) -> Path:
|
|
path = self._path(key)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(".tmp")
|
|
record = dict(payload)
|
|
record["version"] = _ORB_PERIOD_METRICS_CACHE_VERSION
|
|
record["key"] = key
|
|
with gzip.open(tmp, "wt", encoding="utf-8", compresslevel=3) as fh:
|
|
json.dump(record, fh, ensure_ascii=True)
|
|
tmp.replace(path)
|
|
return path
|
|
|
|
def save_checkpoint(self, key: str, payload: dict[str, Any]) -> Path:
|
|
path = self._checkpoint_path(key)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(".tmp")
|
|
record = dict(payload)
|
|
record["version"] = _ORB_PERIOD_METRICS_CACHE_VERSION
|
|
record["key"] = key
|
|
with gzip.open(tmp, "wt", encoding="utf-8", compresslevel=3) as fh:
|
|
json.dump(record, fh, ensure_ascii=True)
|
|
tmp.replace(path)
|
|
return path
|
|
|
|
def clear_checkpoint(self, key: str) -> None:
|
|
self._checkpoint_path(key).unlink(missing_ok=True)
|
|
|
|
|
|
class ORBPreparedTapeStore:
|
|
"""Prepared ORB tape cache to avoid rereading raw intraday parquet files."""
|
|
|
|
def __init__(self, root: str | Path) -> None:
|
|
self.root = Path(root)
|
|
|
|
def _path(self, key: str) -> Path:
|
|
return self.root / key[:2] / f"{key}.pkl.gz"
|
|
|
|
@classmethod
|
|
def build_key(
|
|
cls,
|
|
*,
|
|
research_snapshot_key: str,
|
|
trading_days: list[str],
|
|
candidates: dict[str, list[str]],
|
|
) -> str:
|
|
ordered_candidates = {
|
|
day: list(candidates.get(day, []))
|
|
for day in trading_days
|
|
if candidates.get(day)
|
|
}
|
|
payload = {
|
|
"version": _ORB_TAPE_CACHE_VERSION,
|
|
"research_snapshot_key": research_snapshot_key,
|
|
"trading_day_count": len(trading_days),
|
|
"trading_day_start": trading_days[0] if trading_days else "",
|
|
"trading_day_end": trading_days[-1] if trading_days else "",
|
|
"trading_days_hash": hashlib.sha1(
|
|
",".join(trading_days).encode("utf-8")
|
|
).hexdigest(),
|
|
"candidate_hash": hashlib.sha1(
|
|
json.dumps(
|
|
ordered_candidates,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
default=str,
|
|
).encode("utf-8")
|
|
).hexdigest(),
|
|
}
|
|
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
|
|
return hashlib.sha1(blob.encode("utf-8")).hexdigest()
|
|
|
|
def load(self, key: str) -> dict[str, Any] | None:
|
|
path = self._path(key)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
with gzip.open(path, "rb") as fh:
|
|
payload = pickle.load(fh)
|
|
except Exception:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if not isinstance(payload, dict):
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("version") != _ORB_TAPE_CACHE_VERSION:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if payload.get("key") != key:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
if "bars_by_day" not in payload:
|
|
path.unlink(missing_ok=True)
|
|
return None
|
|
return payload
|
|
|
|
def save(self, key: str, payload: dict[str, Any]) -> Path:
|
|
path = self._path(key)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(".tmp")
|
|
record = dict(payload)
|
|
record["version"] = _ORB_TAPE_CACHE_VERSION
|
|
record["key"] = key
|
|
with gzip.open(tmp, "wb", compresslevel=3) as fh:
|
|
pickle.dump(record, fh, protocol=pickle.HIGHEST_PROTOCOL)
|
|
tmp.replace(path)
|
|
return path
|
|
|
|
|
|
def resolve_orb_config(name_or_path: str) -> tuple[Path, IntradayConfig]:
|
|
"""Resolve YAML path or slug to a validated IntradayConfig."""
|
|
p = Path(name_or_path)
|
|
if p.exists() and p.suffix in {".yaml", ".yml"}:
|
|
yaml_path = p
|
|
else:
|
|
strategies_dir = Path("configs/intraday/strategies")
|
|
candidates = sorted(strategies_dir.glob(f"{name_or_path}*.yaml"))
|
|
if not candidates:
|
|
candidates = sorted(strategies_dir.glob(f"orb_{name_or_path}*.yaml"))
|
|
if not candidates:
|
|
raise FileNotFoundError(
|
|
f"Cannot find config for '{name_or_path}'. "
|
|
f"Provide a full YAML path or a slug matching files in {strategies_dir}/"
|
|
)
|
|
yaml_path = candidates[0]
|
|
|
|
config = load_config(str(yaml_path))
|
|
if config.strategy_mode != "orb":
|
|
raise ValueError(f"{yaml_path} is not an ORB intraday config")
|
|
if config.orb_strategy is None:
|
|
config = config.model_copy(update={"orb_strategy": ORBStrategyParams()})
|
|
return yaml_path, config
|
|
|
|
|
|
def force_simple_returns(config: IntradayConfig) -> IntradayConfig:
|
|
"""Research/SQS runs default to daily-reset simple returns.
|
|
|
|
ORB signal research should not optimize against compound path dependence.
|
|
Compound runs are deployment diagnostics for allocator/capital policy.
|
|
"""
|
|
orb = (config.orb_strategy or ORBStrategyParams()).model_copy(
|
|
update={"compound_returns": False, "daily_budget_reset": True}
|
|
)
|
|
return config.model_copy(update={"orb_strategy": orb})
|
|
|
|
|
|
async def build_orb_research_context(
|
|
config: IntradayConfig,
|
|
start_date: str,
|
|
end_date: str,
|
|
client: OracleClient,
|
|
*,
|
|
daily_concurrency: int = 3,
|
|
print_progress: bool = False,
|
|
) -> ORBResearchContext:
|
|
"""Fetch shared ORB daily context once; intraday is fetched later per period chunk."""
|
|
settings = get_settings()
|
|
orb_params = config.orb_strategy or ORBStrategyParams()
|
|
cache = IntradayCache(config.cache.dir) if config.cache.enabled else None
|
|
daily_cache = (
|
|
DailyBarCache(str(Path(config.cache.dir).with_name("daily")))
|
|
if config.cache.enabled else None
|
|
)
|
|
eval_cache = (
|
|
ORBPeriodMetricsCache(Path(config.cache.dir).with_name("orb_eval"))
|
|
if config.cache.enabled else None
|
|
)
|
|
tape_cache = (
|
|
ORBPreparedTapeStore(Path(config.cache.dir).with_name("orb_tape"))
|
|
if config.cache.enabled else None
|
|
)
|
|
research_snapshot = (
|
|
ORBResearchSnapshotStore(Path(config.cache.dir).with_name("orb_research"))
|
|
if config.cache.enabled else None
|
|
)
|
|
|
|
tickers = await resolve_universe(config.universe, client)
|
|
ticker_sectors = _load_orb_ticker_sectors(tickers)
|
|
trading_days = await get_trading_days(client, start_date, end_date, lookback=0)
|
|
if not trading_days:
|
|
raise ValueError(f"No trading days resolved for {start_date} → {end_date}")
|
|
|
|
snapshot_key = None
|
|
if research_snapshot is not None:
|
|
snapshot_key = research_snapshot.build_key(
|
|
config,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
tickers=tickers,
|
|
trading_days=trading_days,
|
|
)
|
|
snapshot = research_snapshot.load(snapshot_key)
|
|
if snapshot is not None:
|
|
if print_progress:
|
|
print(
|
|
" Research snapshot hit: "
|
|
f"{len(snapshot['tickers'])} tickers, {len(snapshot['trading_days'])} days"
|
|
)
|
|
return ORBResearchContext(
|
|
config=config,
|
|
tickers=list(snapshot["tickers"]),
|
|
trading_days=list(snapshot["trading_days"]),
|
|
daily_bars=dict(snapshot["daily_bars"]),
|
|
enrichment=dict(snapshot["enrichment"]),
|
|
candidates=dict(snapshot["candidates"]),
|
|
ticker_sectors=ticker_sectors,
|
|
cache=cache,
|
|
daily_cache=daily_cache,
|
|
eval_cache=eval_cache,
|
|
tape_cache=tape_cache,
|
|
oracle_url=settings.stock_oracle_url,
|
|
research_snapshot_key=snapshot_key,
|
|
)
|
|
|
|
warmup_start = (dt.date.fromisoformat(trading_days[0]) - dt.timedelta(days=90)).isoformat()
|
|
|
|
def _daily_progress(done: int, total: int) -> None:
|
|
if not print_progress:
|
|
return
|
|
sys.stdout.write(f"\r Daily: {_make_progress_bar(done, total)}")
|
|
sys.stdout.flush()
|
|
|
|
daily_bars = await fetch_daily_bars_bulk(
|
|
tickers,
|
|
warmup_start,
|
|
trading_days[-1],
|
|
client,
|
|
cache=daily_cache,
|
|
intraday_cache_fallback=cache,
|
|
concurrency=daily_concurrency,
|
|
progress_callback=_daily_progress if print_progress else None,
|
|
)
|
|
if print_progress:
|
|
print(f"\r Daily: {len(daily_bars)}/{len(tickers)} tickers loaded")
|
|
|
|
regime_ticker = getattr(orb_params, "market_regime_ticker", "SPY") or "SPY"
|
|
if orb_params.market_regime_spy_threshold is not None and regime_ticker not in daily_bars:
|
|
extra = await fetch_daily_bars_bulk(
|
|
[regime_ticker],
|
|
warmup_start,
|
|
trading_days[-1],
|
|
client,
|
|
cache=daily_cache,
|
|
intraday_cache_fallback=cache,
|
|
concurrency=1,
|
|
)
|
|
daily_bars.update(extra)
|
|
|
|
enrichment = enrich_daily_bars(daily_bars, trading_days)
|
|
candidates = orb_pre_screen_candidates(
|
|
daily_bars,
|
|
trading_days,
|
|
enrichment,
|
|
min_price=orb_params.min_price,
|
|
min_atr=orb_params.min_atr_14,
|
|
min_avg_dollar_vol=orb_params.min_avg_dollar_volume,
|
|
max_per_day=None,
|
|
)
|
|
|
|
# Fetch VIX if the strategy uses it
|
|
from apps.intraday_bt.run import _orb_strategy_uses_vix, _fetch_vix_by_day
|
|
orb_vix_by_day: dict[str, float] | None = None
|
|
if _orb_strategy_uses_vix(orb_params):
|
|
if print_progress:
|
|
print(" Fetching VIX regime series for ORB research...")
|
|
orb_vix_by_day = await _fetch_vix_by_day(client, trading_days)
|
|
|
|
context = ORBResearchContext(
|
|
config=config,
|
|
tickers=tickers,
|
|
trading_days=trading_days,
|
|
daily_bars=daily_bars,
|
|
enrichment=enrichment,
|
|
candidates=candidates,
|
|
ticker_sectors=ticker_sectors,
|
|
vix_by_day=orb_vix_by_day,
|
|
cache=cache,
|
|
daily_cache=daily_cache,
|
|
eval_cache=eval_cache,
|
|
tape_cache=tape_cache,
|
|
oracle_url=settings.stock_oracle_url,
|
|
research_snapshot_key=snapshot_key,
|
|
)
|
|
if research_snapshot is not None and snapshot_key is not None:
|
|
snapshot_path = research_snapshot.save(
|
|
snapshot_key,
|
|
{
|
|
"tickers": tickers,
|
|
"trading_days": trading_days,
|
|
"daily_bars": daily_bars,
|
|
"enrichment": enrichment,
|
|
"candidates": candidates,
|
|
},
|
|
)
|
|
if print_progress:
|
|
print(f" Research snapshot saved: {snapshot_path}")
|
|
return context
|
|
|
|
|
|
def filter_days(trading_days: list[str], start_date: str, end_date: str) -> list[str]:
|
|
return [day for day in trading_days if start_date <= day <= end_date]
|
|
|
|
|
|
def split_trading_days(
|
|
trading_days: list[str],
|
|
split_date: str | None = None,
|
|
train_ratio: float | None = None,
|
|
) -> tuple[list[str], list[str]]:
|
|
if split_date:
|
|
train = [d for d in trading_days if d < split_date]
|
|
test = [d for d in trading_days if d >= split_date]
|
|
elif train_ratio is not None:
|
|
split_idx = int(len(trading_days) * train_ratio)
|
|
train = trading_days[:split_idx]
|
|
test = trading_days[split_idx:]
|
|
else:
|
|
mid = len(trading_days) // 2
|
|
train = trading_days[:mid]
|
|
test = trading_days[mid:]
|
|
return train, test
|
|
|
|
|
|
def generate_walk_forward_windows(
|
|
trading_days: list[str],
|
|
train_days: int,
|
|
test_days: int,
|
|
step_days: int | None = None,
|
|
) -> list[tuple[list[str], list[str]]]:
|
|
if step_days is None:
|
|
step_days = test_days
|
|
|
|
windows: list[tuple[list[str], list[str]]] = []
|
|
idx = 0
|
|
while idx + train_days + test_days <= len(trading_days):
|
|
train = trading_days[idx : idx + train_days]
|
|
test = trading_days[idx + train_days : idx + train_days + test_days]
|
|
windows.append((train, test))
|
|
idx += step_days
|
|
return windows
|
|
|
|
|
|
def resolve_lab_splits(
|
|
trading_days: list[str],
|
|
periods: ORBResearchPeriods = DEFAULT_ORB_RESEARCH_PERIODS,
|
|
) -> dict[str, list[str]]:
|
|
return {
|
|
"train": filter_days(trading_days, periods.train_start, periods.train_end),
|
|
"valid": filter_days(trading_days, periods.valid_start, periods.valid_end),
|
|
"test": filter_days(trading_days, periods.test_start, periods.test_end),
|
|
"robustness": filter_days(trading_days, periods.robustness_start, periods.robustness_end),
|
|
}
|
|
|
|
|
|
def build_orb_params(base_config: IntradayConfig, overrides: dict[str, Any] | None = None) -> ORBStrategyParams:
|
|
if not overrides:
|
|
return base_config.orb_strategy or ORBStrategyParams()
|
|
updated = apply_overrides(base_config, overrides)
|
|
return updated.orb_strategy or ORBStrategyParams()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _candidate_shuffle(seed: int | None):
|
|
if seed is None:
|
|
yield
|
|
return
|
|
|
|
import libs.intraday.orb_simulator as _orb_mod
|
|
|
|
rng = random.Random(seed)
|
|
original = _orb_mod.compute_orb_candidates
|
|
|
|
def _shuffled(*args: Any, **kwargs: Any) -> list[dict]:
|
|
candidates = list(original(*args, **kwargs))
|
|
rng.shuffle(candidates)
|
|
return candidates
|
|
|
|
_orb_mod.compute_orb_candidates = _shuffled
|
|
try:
|
|
yield
|
|
finally:
|
|
_orb_mod.compute_orb_candidates = original
|
|
|
|
|
|
def _chunk_layout_signature(chunks: list[list[str]]) -> list[dict[str, Any]]:
|
|
"""Describe period chunk boundaries so checkpoint resumes can verify layout."""
|
|
return [
|
|
{
|
|
"start": chunk[0],
|
|
"end": chunk[-1],
|
|
"days": len(chunk),
|
|
}
|
|
for chunk in chunks
|
|
if chunk
|
|
]
|
|
|
|
|
|
async def simulate_orb_period(
|
|
context: ORBResearchContext,
|
|
client: OracleClient,
|
|
orb_params: ORBStrategyParams,
|
|
trading_days: list[str],
|
|
*,
|
|
run_id: str = "",
|
|
shuffle_candidates_seed: int | None = None,
|
|
intraday_concurrency: int = 3,
|
|
max_pairs_per_chunk: int = 5_000,
|
|
progress_prefix: str = "",
|
|
) -> IntradayMetrics:
|
|
"""Run a streaming ORB backtest over a subset of days using shared daily context."""
|
|
cache_key: str | None = None
|
|
if (
|
|
context.eval_cache is not None
|
|
and context.research_snapshot_key is not None
|
|
and trading_days
|
|
):
|
|
cache_key = context.eval_cache.build_key(
|
|
research_snapshot_key=context.research_snapshot_key,
|
|
orb_params=orb_params,
|
|
trading_days=trading_days,
|
|
shuffle_candidates_seed=shuffle_candidates_seed,
|
|
)
|
|
cached = context.eval_cache.load(cache_key)
|
|
if cached is not None:
|
|
metrics = IntradayMetrics.model_validate(cached["metrics"])
|
|
if run_id and run_id != metrics.run_id:
|
|
metrics = metrics.model_copy(update={"run_id": run_id})
|
|
if progress_prefix:
|
|
print(
|
|
f"{progress_prefix}cached metrics hit: "
|
|
f"{trading_days[0]} → {trading_days[-1]} ({len(trading_days)} days)"
|
|
)
|
|
return metrics
|
|
|
|
config = context.config.model_copy(update={"orb_strategy": orb_params})
|
|
if not trading_days:
|
|
return IntradayMetricsAccumulator(config, run_id=run_id).finalize()
|
|
|
|
chunks = _chunk_trading_days_by_pairs(
|
|
trading_days,
|
|
context.candidates,
|
|
max_pairs_per_chunk=max_pairs_per_chunk,
|
|
)
|
|
chunk_layout = _chunk_layout_signature(chunks)
|
|
checkpoint_enabled = (
|
|
context.eval_cache is not None
|
|
and cache_key is not None
|
|
and shuffle_candidates_seed is None
|
|
)
|
|
accumulator = IntradayMetricsAccumulator(config, run_id=run_id)
|
|
sim_state = None
|
|
start_chunk_idx = 0
|
|
if checkpoint_enabled:
|
|
checkpoint = context.eval_cache.load_checkpoint(cache_key)
|
|
if checkpoint is not None:
|
|
if checkpoint.get("chunk_layout") == chunk_layout:
|
|
accumulator = IntradayMetricsAccumulator.from_snapshot(
|
|
config,
|
|
checkpoint["accumulator"],
|
|
run_id=run_id,
|
|
)
|
|
state_payload = checkpoint.get("sim_state")
|
|
if state_payload:
|
|
sim_state = ORBSimulationState(**state_payload)
|
|
start_chunk_idx = int(checkpoint.get("completed_chunks", 0))
|
|
if progress_prefix:
|
|
print(
|
|
f"{progress_prefix}checkpoint resume hit: "
|
|
f"chunk {start_chunk_idx}/{len(chunks)}"
|
|
)
|
|
else:
|
|
context.eval_cache.clear_checkpoint(cache_key)
|
|
|
|
if start_chunk_idx >= len(chunks):
|
|
metrics = accumulator.finalize()
|
|
if cache_key is not None and context.eval_cache is not None:
|
|
context.eval_cache.save(
|
|
cache_key,
|
|
{
|
|
"metrics": metrics.model_dump(mode="json"),
|
|
},
|
|
)
|
|
context.eval_cache.clear_checkpoint(cache_key)
|
|
return metrics
|
|
|
|
progress_enabled = bool(progress_prefix)
|
|
|
|
with _candidate_shuffle(shuffle_candidates_seed):
|
|
for chunk_idx, day_chunk in enumerate(chunks[start_chunk_idx:], start=start_chunk_idx + 1):
|
|
chunk_candidates = {
|
|
day: context.candidates.get(day, [])
|
|
for day in day_chunk
|
|
if context.candidates.get(day)
|
|
}
|
|
tape_key = None
|
|
chunk_intraday: dict[str, dict[str, list[dict]]] | None = None
|
|
fetched_from_source = False
|
|
if (
|
|
context.tape_cache is not None
|
|
and context.research_snapshot_key is not None
|
|
and chunk_candidates
|
|
):
|
|
tape_key = context.tape_cache.build_key(
|
|
research_snapshot_key=context.research_snapshot_key,
|
|
trading_days=day_chunk,
|
|
candidates=chunk_candidates,
|
|
)
|
|
tape_payload = context.tape_cache.load(tape_key)
|
|
if tape_payload is not None:
|
|
chunk_intraday = tape_payload["bars_by_day"]
|
|
if progress_enabled:
|
|
pair_count = sum(len(v) for v in chunk_candidates.values())
|
|
print(
|
|
f"{progress_prefix}tape hit {chunk_idx}/{len(chunks)}: "
|
|
f"{day_chunk[0]} → {day_chunk[-1]} ({pair_count} pairs)"
|
|
)
|
|
|
|
if progress_enabled:
|
|
pair_count = sum(len(v) for v in chunk_candidates.values())
|
|
print(
|
|
f"{progress_prefix}batch {chunk_idx}/{len(chunks)}: "
|
|
f"{day_chunk[0]} → {day_chunk[-1]} ({pair_count} pairs)"
|
|
)
|
|
|
|
def _intraday_progress(done: int, total: int, hits: int, calls: int) -> None:
|
|
if not progress_enabled:
|
|
return
|
|
sys.stdout.write(
|
|
f"\r{progress_prefix} {_make_progress_bar(done, total)} cache:{hits} api:{calls}"
|
|
)
|
|
sys.stdout.flush()
|
|
|
|
if chunk_intraday is None:
|
|
chunk_intraday = await fetch_intraday_bulk(
|
|
chunk_candidates,
|
|
client,
|
|
context.cache,
|
|
concurrency=intraday_concurrency,
|
|
progress_callback=_intraday_progress if progress_enabled else None,
|
|
)
|
|
fetched_from_source = True
|
|
if tape_key is not None and context.tape_cache is not None:
|
|
context.tape_cache.save(
|
|
tape_key,
|
|
{
|
|
"bars_by_day": chunk_intraday,
|
|
},
|
|
)
|
|
if progress_enabled and chunk_candidates and fetched_from_source:
|
|
print()
|
|
|
|
stderr_buffer = io.StringIO()
|
|
with contextlib.redirect_stderr(stderr_buffer):
|
|
chunk_results, sim_state = run_orb_simulation_with_state(
|
|
chunk_intraday,
|
|
day_chunk,
|
|
orb_params,
|
|
context.enrichment,
|
|
ticker_sectors=context.ticker_sectors,
|
|
state=sim_state,
|
|
vix_by_day=context.vix_by_day,
|
|
)
|
|
accumulator.extend(chunk_results)
|
|
del chunk_intraday
|
|
if checkpoint_enabled and cache_key is not None and context.eval_cache is not None:
|
|
context.eval_cache.save_checkpoint(
|
|
cache_key,
|
|
{
|
|
"completed_chunks": chunk_idx,
|
|
"total_chunks": len(chunks),
|
|
"chunk_layout": chunk_layout,
|
|
"accumulator": accumulator.snapshot(),
|
|
"sim_state": asdict(sim_state) if sim_state is not None else None,
|
|
},
|
|
)
|
|
|
|
metrics = accumulator.finalize()
|
|
if (
|
|
context.eval_cache is not None
|
|
and cache_key is not None
|
|
):
|
|
context.eval_cache.save(
|
|
cache_key,
|
|
{
|
|
"metrics": metrics.model_dump(mode="json"),
|
|
},
|
|
)
|
|
context.eval_cache.clear_checkpoint(cache_key)
|
|
return metrics
|
|
|
|
|
|
async def simulate_orb_overrides(
|
|
context: ORBResearchContext,
|
|
client: OracleClient,
|
|
overrides: dict[str, Any] | None,
|
|
trading_days: list[str],
|
|
*,
|
|
run_id: str = "",
|
|
shuffle_candidates_seed: int | None = None,
|
|
progress_prefix: str = "",
|
|
intraday_concurrency: int = 3,
|
|
max_pairs_per_chunk: int = 5_000,
|
|
) -> tuple[ORBStrategyParams, IntradayMetrics]:
|
|
params = build_orb_params(context.config, overrides)
|
|
metrics = await simulate_orb_period(
|
|
context,
|
|
client,
|
|
params,
|
|
trading_days,
|
|
run_id=run_id,
|
|
shuffle_candidates_seed=shuffle_candidates_seed,
|
|
progress_prefix=progress_prefix,
|
|
intraday_concurrency=intraday_concurrency,
|
|
max_pairs_per_chunk=max_pairs_per_chunk,
|
|
)
|
|
return params, metrics
|
|
|
|
|
|
def intraday_metrics_to_split_result(
|
|
metrics: IntradayMetrics,
|
|
orb_params: ORBStrategyParams,
|
|
) -> SplitResult:
|
|
"""Adapt intraday metrics into the generic split schema used by RQS/WFQS."""
|
|
days_in_market_pct = None
|
|
if metrics.trading_days > 0:
|
|
days_in_market_pct = round(metrics.days_with_trades / metrics.trading_days * 100.0, 1)
|
|
|
|
# Intraday ORB does not maintain exposure aggregates today; use a bounded proxy
|
|
# from strategy constraints so RQS does not zero out the exposure dimensions.
|
|
gross_proxy = min(
|
|
40.0,
|
|
max(
|
|
8.0,
|
|
orb_params.max_position_pct * 100.0 * max(1.0, min(float(orb_params.max_candidates), 4.0)),
|
|
),
|
|
)
|
|
total_return_pct = metrics.total_return_pct * 100.0 if metrics.total_return_pct is not None else None
|
|
annualized_return_pct = (
|
|
metrics.annualized_return_pct * 100.0 if metrics.annualized_return_pct is not None else None
|
|
)
|
|
max_drawdown_pct = (
|
|
abs(metrics.max_drawdown_pct) * 100.0 if metrics.max_drawdown_pct is not None else None
|
|
)
|
|
|
|
return SplitResult(
|
|
run_id=metrics.run_id,
|
|
trade_count=metrics.total_trades,
|
|
profit_factor=metrics.profit_factor,
|
|
total_return_pct=total_return_pct,
|
|
annualized_return_pct=annualized_return_pct,
|
|
win_rate=metrics.win_rate,
|
|
max_drawdown_pct=max_drawdown_pct,
|
|
sharpe_ratio=metrics.sharpe_ratio,
|
|
monthly_win_rate=None,
|
|
equity_curve_r_squared=None,
|
|
avg_gross_exposure_pct=round(gross_proxy, 1),
|
|
avg_net_exposure_pct=round(gross_proxy, 1),
|
|
days_in_market_pct=days_in_market_pct,
|
|
)
|
|
|
|
|
|
def build_walk_forward_summary(
|
|
folds: list[dict[str, Any]],
|
|
*,
|
|
train_days: int,
|
|
test_days: int,
|
|
step_days: int,
|
|
) -> WalkForwardSummary:
|
|
fold_models: list[WalkForwardFoldResult] = []
|
|
train_results: list[SplitResult] = []
|
|
test_results: list[SplitResult] = []
|
|
|
|
for idx, fold in enumerate(folds, start=1):
|
|
train_result = fold["train_result"]
|
|
test_result = fold["test_result"]
|
|
train_results.append(train_result)
|
|
test_results.append(test_result)
|
|
fold_models.append(
|
|
WalkForwardFoldResult(
|
|
fold_index=idx,
|
|
train_start=dt.date.fromisoformat(fold["train_start"]),
|
|
train_end=dt.date.fromisoformat(fold["train_end"]),
|
|
test_start=dt.date.fromisoformat(fold["test_start"]),
|
|
test_end=dt.date.fromisoformat(fold["test_end"]),
|
|
train_run_id=train_result.run_id,
|
|
test_run_id=test_result.run_id,
|
|
train_metrics=train_result,
|
|
test_metrics=test_result,
|
|
)
|
|
)
|
|
|
|
def _aggregate(results: list[SplitResult]) -> WalkForwardAggregate:
|
|
returns = [r.total_return_pct for r in results if r.total_return_pct is not None]
|
|
profit_factors = [r.profit_factor for r in results if r.profit_factor is not None]
|
|
drawdowns = [r.max_drawdown_pct for r in results if r.max_drawdown_pct is not None]
|
|
trade_counts = [float(r.trade_count) for r in results]
|
|
win_rates = [r.win_rate for r in results if r.win_rate is not None]
|
|
positives = [r for r in returns if r > 0]
|
|
return WalkForwardAggregate(
|
|
mean_return_pct=round(statistics.mean(returns), 2) if returns else None,
|
|
median_return_pct=round(statistics.median(returns), 2) if returns else None,
|
|
worst_return_pct=round(min(returns), 2) if returns else None,
|
|
positive_fold_rate_pct=round(len(positives) / len(results) * 100.0, 1) if results else None,
|
|
mean_profit_factor=round(statistics.mean(profit_factors), 2) if profit_factors else None,
|
|
mean_max_drawdown_pct=round(statistics.mean(drawdowns), 2) if drawdowns else None,
|
|
mean_trade_count=round(statistics.mean(trade_counts), 1) if trade_counts else None,
|
|
mean_win_rate=round(statistics.mean(win_rates), 4) if win_rates else None,
|
|
)
|
|
|
|
train_aggregate = _aggregate(train_results)
|
|
test_aggregate = _aggregate(test_results)
|
|
train_test_gaps = [
|
|
abs((train.total_return_pct or 0.0) - (test.total_return_pct or 0.0))
|
|
for train, test in zip(train_results, test_results, strict=False)
|
|
]
|
|
test_returns = [r.total_return_pct for r in test_results if r.total_return_pct is not None]
|
|
fold_cv = None
|
|
if len(test_returns) >= 2:
|
|
mean_return = statistics.mean(test_returns)
|
|
if abs(mean_return) > 1e-9:
|
|
fold_cv = round(statistics.stdev(test_returns) / abs(mean_return), 3)
|
|
|
|
gap_stats = WalkForwardGapStats(
|
|
mean_train_test_return_gap_pct=round(statistics.mean(train_test_gaps), 2) if train_test_gaps else None,
|
|
worst_train_test_return_gap_pct=round(max(train_test_gaps), 2) if train_test_gaps else None,
|
|
fold_return_cv=fold_cv,
|
|
)
|
|
|
|
return WalkForwardSummary(
|
|
train_days=train_days,
|
|
test_days=test_days,
|
|
step_days=step_days,
|
|
fold_count=len(fold_models),
|
|
folds=fold_models,
|
|
train_aggregate=train_aggregate,
|
|
test_aggregate=test_aggregate,
|
|
gap_stats=gap_stats,
|
|
engine_reliability_ratio=1.0,
|
|
)
|
|
|
|
|
|
def compute_orb_overfit_score(
|
|
is_oos_test: dict[str, Any],
|
|
walk_forward_test: dict[str, Any],
|
|
plateau_test: dict[str, Any],
|
|
permutation_test: dict[str, Any],
|
|
) -> tuple[float, dict[str, float]]:
|
|
"""Blend the 4 ORB overfit diagnostics into a 0-100 score."""
|
|
retention_score = max(0.0, min(100.0, float(is_oos_test.get("retention_pct", 0.0))))
|
|
|
|
mean_sharpe = float(walk_forward_test.get("mean_sharpe", 0.0))
|
|
cv = walk_forward_test.get("cv")
|
|
wf_stability = 0.0
|
|
if cv is not None:
|
|
wf_stability = max(0.0, min(100.0, 100.0 * (1.0 - min(float(cv), 2.0) / 2.0)))
|
|
if mean_sharpe <= 0:
|
|
wf_stability *= 0.6
|
|
|
|
plateau_params = plateau_test.get("params", [])
|
|
plateau_values = [float(p.get("plateau", 0.0)) * 100.0 for p in plateau_params]
|
|
plateau_score = statistics.mean(plateau_values) if plateau_values else 0.0
|
|
|
|
p_value = permutation_test.get("p_value")
|
|
permutation_score = 0.0
|
|
if p_value is not None:
|
|
permutation_score = max(0.0, min(100.0, 100.0 * (1.0 - min(float(p_value), 0.50) / 0.50)))
|
|
|
|
score = (
|
|
0.35 * retention_score
|
|
+ 0.25 * wf_stability
|
|
+ 0.20 * plateau_score
|
|
+ 0.20 * permutation_score
|
|
)
|
|
breakdown = {
|
|
"is_oos_retention": round(retention_score, 1),
|
|
"wf_stability": round(wf_stability, 1),
|
|
"parameter_plateau": round(plateau_score, 1),
|
|
"candidate_permutation": round(permutation_score, 1),
|
|
}
|
|
return round(score, 1), breakdown
|
|
|
|
|
|
def compute_orb_rrs(scenario_results: dict[str, dict[str, Any]]) -> tuple[float, dict[str, float]]:
|
|
"""ORB-specific Regime Robustness Score (0-100)."""
|
|
sharpes = {
|
|
key: value.get("sharpe_ratio", 0.0)
|
|
for key, value in scenario_results.items()
|
|
if "sharpe_ratio" in value
|
|
}
|
|
drawdowns = {
|
|
key: abs(value.get("max_drawdown_pct", 0.0))
|
|
for key, value in scenario_results.items()
|
|
if "max_drawdown_pct" in value
|
|
}
|
|
|
|
bear_sr = sharpes.get("bear_2022")
|
|
bear_survival = 50.0 if bear_sr is None else min(100.0, max(0.0, 100.0 + bear_sr * 25.0))
|
|
|
|
n_positive = sum(1 for s in sharpes.values() if s > 0)
|
|
breadth = 100.0 * n_positive / max(len(sharpes), 1)
|
|
|
|
worst_dd = max(drawdowns.values()) if drawdowns else 0.0
|
|
drawdown_resilience = max(0.0, min(100.0, 100.0 * (1.0 - worst_dd / 50.0)))
|
|
|
|
oos_sr = sharpes.get("oos_2026")
|
|
oos_integrity = 50.0 if oos_sr is None else min(100.0, max(0.0, 50.0 + oos_sr * 25.0))
|
|
|
|
stability_keys = [k for k in ["recovery_2023h1", "bull_2023h2", "mixed_2024", "bull_2025"] if k in sharpes]
|
|
if len(stability_keys) >= 2:
|
|
stab_sharpes = [sharpes[k] for k in stability_keys]
|
|
mean_s = statistics.mean(stab_sharpes)
|
|
std_s = statistics.stdev(stab_sharpes)
|
|
if abs(mean_s) > 0.01:
|
|
cv = std_s / abs(mean_s)
|
|
stability = max(0.0, min(100.0, 100.0 * (1.0 - min(cv, 2.0) / 2.0)))
|
|
else:
|
|
stability = max(0.0, 50.0 - std_s * 25.0)
|
|
else:
|
|
stability = 50.0
|
|
|
|
rrs = (
|
|
0.25 * bear_survival
|
|
+ 0.25 * breadth
|
|
+ 0.20 * drawdown_resilience
|
|
+ 0.20 * oos_integrity
|
|
+ 0.10 * stability
|
|
)
|
|
breakdown = {
|
|
"bear_survival": round(bear_survival, 1),
|
|
"breadth": round(breadth, 1),
|
|
"drawdown_resilience": round(drawdown_resilience, 1),
|
|
"oos_integrity": round(oos_integrity, 1),
|
|
"stability": round(stability, 1),
|
|
}
|
|
return round(rrs, 1), breakdown
|
|
|
|
|
|
def compute_orbqs(
|
|
train_result: SplitResult | None,
|
|
valid_result: SplitResult | None,
|
|
test_result: SplitResult | None,
|
|
walk_forward_summary: WalkForwardSummary | None,
|
|
scenario_results: dict[str, dict[str, Any]],
|
|
overfit_tests: dict[str, dict[str, Any]],
|
|
) -> tuple[float | None, dict[str, Any]]:
|
|
"""Compute ORB Quality Score (ORBQS) using split/WF/scenario/overfit components."""
|
|
rqs_score, rqs_breakdown = compute_rqs(train_result, valid_result, test_result)
|
|
wfqs_score, wfqs_breakdown = compute_wfqs_v2(walk_forward_summary)
|
|
rrs_score, rrs_breakdown = compute_orb_rrs(scenario_results) if scenario_results else (None, {})
|
|
overfit_score, overfit_breakdown = compute_orb_overfit_score(
|
|
overfit_tests.get("is_oos", {}),
|
|
overfit_tests.get("walk_forward", {}),
|
|
overfit_tests.get("param_plateau", {}),
|
|
overfit_tests.get("permutation", {}),
|
|
)
|
|
|
|
valid_trades = valid_result.trade_count if valid_result else 0
|
|
test_trades = test_result.trade_count if test_result else 0
|
|
valid_test_trades = valid_trades + test_trades
|
|
if valid_test_trades < 80:
|
|
activity_factor = 0.70
|
|
elif valid_test_trades < 150:
|
|
activity_factor = 0.85
|
|
else:
|
|
activity_factor = 1.00
|
|
|
|
if rqs_score is None or wfqs_score is None or rrs_score is None:
|
|
return None, {
|
|
"rqs": rqs_score,
|
|
"wfqs_v2": wfqs_score,
|
|
"rrs": rrs_score,
|
|
"overfit": overfit_score,
|
|
"activity_factor": activity_factor,
|
|
}
|
|
|
|
orbqs = (
|
|
0.45 * rqs_score
|
|
+ 0.30 * wfqs_score
|
|
+ 0.15 * rrs_score
|
|
+ 0.10 * overfit_score
|
|
) * activity_factor
|
|
|
|
breakdown = {
|
|
"rqs": round(rqs_score, 1),
|
|
"wfqs_v2": round(wfqs_score, 1),
|
|
"rrs": round(rrs_score, 1),
|
|
"overfit": round(overfit_score, 1),
|
|
"activity_factor": round(activity_factor, 2),
|
|
"valid_test_trade_count": valid_test_trades,
|
|
"rqs_breakdown": rqs_breakdown,
|
|
"wfqs_v2_breakdown": wfqs_breakdown,
|
|
"rrs_breakdown": rrs_breakdown,
|
|
"overfit_breakdown": overfit_breakdown,
|
|
}
|
|
return round(orbqs, 1), breakdown
|
|
|
|
|
|
def write_json(path: Path, payload: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(payload, indent=2, ensure_ascii=True, default=str))
|
|
|
|
|
|
def analyze_skip_reasons(json_path: str, focus_date: str | None = None) -> None:
|
|
"""Print a diagnostic summary of why days were skipped in a backtest run.
|
|
|
|
Args:
|
|
json_path: Path to the backtest JSON output file.
|
|
focus_date: Optional date ('YYYY-MM-DD') for detailed per-day drill-down.
|
|
"""
|
|
data = json.loads(Path(json_path).read_text())
|
|
|
|
skip_breakdown = data.get("skip_breakdown", {})
|
|
agg_filter_stats = data.get("aggregate_filter_stats", {})
|
|
daily_summary = data.get("daily_summary", [])
|
|
total_days = len(daily_summary)
|
|
|
|
print(f"\n=== Skip Reason Breakdown ({total_days} trading days) ===")
|
|
for key in ("traded", "traded_no_fill", "market_regime", "breadth", "vix_gate",
|
|
"rolling_loss", "spy_trend", "no_candidates", "below_min_candidates"):
|
|
n = skip_breakdown.get(key, 0)
|
|
if n > 0:
|
|
pct = n / total_days * 100
|
|
print(f" {key:<25} {n:>4} ({pct:.1f}%)")
|
|
|
|
if agg_filter_stats:
|
|
print("\n=== Aggregate Candidate Filter Drops (all days combined) ===")
|
|
for key in ("gap", "rvol", "atr", "dolvol", "dir", "no_bars", "late", "price"):
|
|
n = agg_filter_stats.get(key, 0)
|
|
if n > 0:
|
|
print(f" {key:<10} {n:>6} tickers dropped")
|
|
|
|
# V20: soft-day breakdown
|
|
soft_days = [r for r in daily_summary if r.get("is_soft_day")]
|
|
if soft_days:
|
|
soft_trades = [t for t in data.get("trades", []) if any(
|
|
r["date"] == t.get("date") and r.get("is_soft_day") for r in daily_summary
|
|
)]
|
|
soft_wins = sum(1 for t in soft_trades if t.get("pnl", 0) > 0)
|
|
soft_wr = soft_wins / len(soft_trades) * 100 if soft_trades else 0.0
|
|
soft_pnl = sum(t.get("pnl", 0) for t in soft_trades)
|
|
print(f"\n=== V20 Soft-Day Breakdown ({len(soft_days)} days) ===")
|
|
print(f" soft days : {len(soft_days)}")
|
|
print(f" soft-day trades : {len(soft_trades)}")
|
|
print(f" soft-day WR : {soft_wr:.1f}%")
|
|
print(f" soft-day total PnL : {soft_pnl:+.2f}")
|
|
print(f" (thesis: soft-day WR>=52% and PnL>0 = viable)")
|
|
|
|
if focus_date:
|
|
match = next((r for r in daily_summary if r["date"] == focus_date), None)
|
|
if match is None:
|
|
print(f"\nfocus_date {focus_date}: NOT FOUND in results")
|
|
return
|
|
print(f"\n=== Focus Date: {focus_date} ===")
|
|
print(f" skip_reason : {match.get('skip_reason') or '(traded)'}")
|
|
print(f" candidates_found : {match.get('candidates_found', 0)}")
|
|
print(f" trades : {match.get('trades', 0)}")
|
|
print(f" daily_pnl : {match.get('daily_pnl', 0):+.2f}")
|
|
print(f" regime_scaler : {match.get('regime_scaler')}")
|
|
print(f" breadth_scaler : {match.get('breadth_scaler')}")
|
|
print(f" is_soft_day : {match.get('is_soft_day')}")
|
|
fs = match.get("candidate_filter_stats")
|
|
if fs:
|
|
print(" candidate_filter_stats:")
|
|
for k, v in sorted(fs.items(), key=lambda x: -x[1]):
|
|
if v > 0:
|
|
print(f" {k:<10} {v:>4} dropped")
|
|
trades_detail = [t for t in data.get("trades", []) if t.get("date") == focus_date]
|
|
if trades_detail:
|
|
print(" selected tickers:")
|
|
for t in trades_detail:
|
|
ticker = t.get("ticker", "?")
|
|
pnl = t.get("pnl", 0)
|
|
pnl_r = t.get("pnl_r", 0)
|
|
print(f" {ticker:<8} pnl={pnl:+.2f} R={pnl_r:+.2f}")
|