Files
Auto-Trader/agent/backtest/validation.py
T
Kenneth Hau aab57cdeb3
Desktop Windows / package (push) Waiting to run
Build and Push Docker Image / validate-build (push) Waiting to run
Build and Push Docker Image / build-and-push (push) Skipped
CI / Hash lock / macOS ARM64 (push) Waiting to run
CI / Hash lock / Linux x64 (push) Waiting to run
CI / Hash lock / Windows x64 (push) Waiting to run
CI / test / Python 3.11 (push) Waiting to run
CI / test / Python 3.14 (push) Waiting to run
CI / frontend (push) Waiting to run
CI / Windows 11 ARM64 / Python 3.14 background lifecycle (push) Waiting to run
CI / Windows Server 2025 x64 / Python 3.14 background lifecycle (push) Waiting to run
CI / Windows desktop source lifecycle (push) Waiting to run
Deploy Wiki / Deploy to Cloudflare Pages (push) Waiting to run
Wiki / validate (push) Waiting to run
feat: Phase 0/1 TradingAgents swarm graft, mandate paper automation, and P0 WhatsApp outbound send
2026-10-16 14:58:47 +08:00

968 lines
37 KiB
Python

"""Statistical validation for backtest results.
Three independent tools:
- Monte Carlo permutation test: is the strategy significantly better than random?
- Bootstrap Sharpe CI: how stable is the risk-adjusted return?
- Walk-Forward analysis: is performance consistent across time windows?
Usage: called automatically by BaseEngine.run_backtest when config[\"validation\"]
is present, or invoked directly on backtest outputs.
"""
from __future__ import annotations
import json
import math
from numbers import Integral, Real
from pathlib import Path
from typing import Any, Dict, List
import numpy as np
import pandas as pd
from backtest.metrics import effective_bars_per_year, period_returns
from backtest.models import TradeRecord
# ─── Monte Carlo Permutation Test ───
def monte_carlo_test(
trades: List[TradeRecord],
initial_capital: float,
n_simulations: int = 1000,
seed: int = 42,
bars_per_year: int = 252,
) -> Dict[str, Any]:
"""Shuffle trade PnL order to test path significance.
Null hypothesis: the observed Sharpe / max-drawdown is no better than
a random ordering of the same trades.
Args:
trades: Completed round-trip trades from backtest.
initial_capital: Starting capital.
n_simulations: Number of random permutations.
seed: Random seed for reproducibility.
bars_per_year: Annualisation factor (must match bootstrap_sharpe_ci
and walk_forward_analysis so the report's Sharpe figures agree).
Returns:
Dict with actual_sharpe, p_value_sharpe, actual_max_dd,
p_value_max_dd, simulated_sharpes (percentiles).
"""
if isinstance(n_simulations, bool) or not isinstance(n_simulations, Integral) or n_simulations < 1:
return {
"error": f"n_simulations must be >= 1, got {n_simulations}",
"p_value_sharpe": 1.0,
}
if isinstance(seed, bool) or not isinstance(seed, Integral) or seed < 0:
return {"error": f"seed must be >= 0, got {seed}", "p_value_sharpe": 1.0}
if len(trades) < 3:
return {"error": "need at least 3 trades", "p_value_sharpe": 1.0}
pnls = np.array([t.pnl for t in trades])
actual = _path_metrics(pnls, initial_capital, bars_per_year)
rng = np.random.default_rng(seed)
sharpe_count = 0
dd_count = 0
sim_sharpes = []
# The full path matrix feeds the fan-chart payload; skip it for
# pathological sizes so a huge run cannot balloon memory or the JSON.
keep_paths = n_simulations * len(pnls) <= 2_000_000
sim_equities = np.empty((n_simulations, len(pnls))) if keep_paths else None
for i in range(n_simulations):
shuffled = rng.permutation(pnls)
if sim_equities is not None:
sim_equities[i] = initial_capital + np.cumsum(shuffled)
sim = _path_metrics(shuffled, initial_capital, bars_per_year)
sim_sharpes.append(sim["sharpe"])
if sim["sharpe"] >= actual["sharpe"]:
sharpe_count += 1
if sim["max_dd"] >= actual["max_dd"]: # less negative = "better"
dd_count += 1
sim_arr = np.array(sim_sharpes)
result = {
"actual_sharpe": round(actual["sharpe"], 4),
"actual_max_dd": round(actual["max_dd"], 4),
"p_value_sharpe": round(sharpe_count / n_simulations, 4),
"p_value_max_dd": round(dd_count / n_simulations, 4),
"simulated_sharpe_mean": round(float(sim_arr.mean()), 4),
"simulated_sharpe_std": round(float(sim_arr.std()), 4),
"simulated_sharpe_p5": round(float(np.percentile(sim_arr, 5)), 4),
"simulated_sharpe_p95": round(float(np.percentile(sim_arr, 95)), 4),
"n_simulations": n_simulations,
"n_trades": len(trades),
"sharpe_samples": [round(float(s), 4) for s in sim_sharpes],
}
if sim_equities is not None:
idx = np.unique(np.linspace(0, len(pnls) - 1, min(len(pnls), 400)).astype(int))
sample_rows = np.unique(
np.linspace(0, n_simulations - 1, min(30, n_simulations)).astype(int)
)
result["equity_paths"] = {
"steps": (idx + 1).tolist(),
"initial_capital": round(float(initial_capital), 2),
"actual": np.round((initial_capital + np.cumsum(pnls))[idx], 2).tolist(),
"band_p5": np.round(np.percentile(sim_equities[:, idx], 5, axis=0), 2).tolist(),
"band_p25": np.round(np.percentile(sim_equities[:, idx], 25, axis=0), 2).tolist(),
"band_p50": np.round(np.percentile(sim_equities[:, idx], 50, axis=0), 2).tolist(),
"band_p75": np.round(np.percentile(sim_equities[:, idx], 75, axis=0), 2).tolist(),
"band_p95": np.round(np.percentile(sim_equities[:, idx], 95, axis=0), 2).tolist(),
"samples": np.round(sim_equities[np.ix_(sample_rows, idx)], 2).tolist(),
}
return result
def _path_metrics(
pnls: np.ndarray, initial_capital: float, bars_per_year: int = 252
) -> Dict[str, float]:
"""Compute Sharpe and max drawdown from a PnL sequence."""
# Include starting cash so the first trade contributes to returns and
# drawdown in both the observed path and every shuffled path.
equity = np.concatenate(([initial_capital], initial_capital + np.cumsum(pnls)))
if len(equity) > 1:
prev = equity[:-1]
diff = np.diff(equity)
returns = np.where(prev != 0, diff / np.where(prev != 0, prev, 1.0), 0.0)
else:
returns = np.array([0.0])
std = returns.std()
sharpe = float(returns.mean() / (std + 1e-10) * np.sqrt(bars_per_year))
peak = np.maximum.accumulate(equity)
dd = (equity - peak) / np.where(peak > 0, peak, 1.0)
max_dd = float(dd.min())
return {"sharpe": sharpe, "max_dd": max_dd}
# ─── Bootstrap Sharpe CI ───
def bootstrap_sharpe_ci(
equity_curve: pd.Series,
n_bootstrap: int = 1000,
confidence: float = 0.95,
bars_per_year: int = 252,
seed: int = 42,
initial_capital: float | None = None,
) -> Dict[str, Any]:
"""Resample daily returns to estimate Sharpe confidence interval.
Args:
equity_curve: Equity time series.
n_bootstrap: Number of bootstrap samples.
confidence: Confidence level (e.g. 0.95 for 95% CI).
bars_per_year: Annualisation factor.
seed: Random seed.
initial_capital: Starting capital. When given, the first return is
measured from this cash mark (the calc_metrics convention), so
``observed_sharpe`` agrees with the report's own Sharpe on paths
whose first recorded bar carries real PnL. None keeps the legacy
adjacent-pair series.
Returns:
Dict with observed_sharpe, ci_lower, ci_upper, median_sharpe,
prob_positive (fraction of samples with Sharpe > 0).
"""
if isinstance(n_bootstrap, bool) or not isinstance(n_bootstrap, Integral) or n_bootstrap < 1:
return {"error": f"n_bootstrap must be >= 1, got {n_bootstrap}"}
if (
isinstance(confidence, bool)
or not isinstance(confidence, Real)
or not math.isfinite(float(confidence))
or not 0.0 < confidence < 1.0
):
return {"error": f"confidence must be in (0, 1), got {confidence}"}
if isinstance(seed, bool) or not isinstance(seed, Integral) or seed < 0:
return {"error": f"seed must be >= 0, got {seed}"}
if initial_capital is not None:
series = period_returns(equity_curve, initial_capital)
else:
series = equity_curve.pct_change(fill_method=None)
# Missing equity is not a flat observation or a one-period recovery return.
returns = (
series
.replace([np.inf, -np.inf], 0.0)
.dropna()
.values
)
if len(returns) < 5:
return {"error": "need at least 5 return observations"}
observed = _sharpe(returns, bars_per_year)
rng = np.random.default_rng(seed)
boot_sharpes = []
for _ in range(n_bootstrap):
sample = rng.choice(returns, size=len(returns), replace=True)
boot_sharpes.append(_sharpe(sample, bars_per_year))
arr = np.array(boot_sharpes)
alpha = (1 - confidence) / 2
lower = float(np.percentile(arr, alpha * 100))
upper = float(np.percentile(arr, (1 - alpha) * 100))
prob_pos = float(np.mean(arr > 0))
result = {
"observed_sharpe": round(observed, 4),
"ci_lower": round(lower, 4),
"ci_upper": round(upper, 4),
"median_sharpe": round(float(np.median(arr)), 4),
"prob_positive": round(prob_pos, 4),
"confidence": confidence,
"n_bootstrap": n_bootstrap,
}
if n_bootstrap <= 20_000:
result["sharpe_samples"] = [round(float(s), 4) for s in boot_sharpes]
return result
def _sharpe(returns: np.ndarray, bars_per_year: int = 252) -> float:
std = returns.std()
return float(returns.mean() / (std + 1e-10) * np.sqrt(bars_per_year))
# ─── Walk-Forward Analysis ───
def walk_forward_analysis(
equity_curve: pd.Series,
trades: List[TradeRecord],
n_windows: int = 5,
bars_per_year: int = 252,
) -> Dict[str, Any]:
"""Split backtest into sequential windows, check consistency.
Each window is evaluated independently (returns normalised to window start).
Args:
equity_curve: Equity time series.
trades: Completed trades.
n_windows: Number of non-overlapping windows.
bars_per_year: Annualisation factor.
Returns:
Dict with per_window stats, consistency metrics.
"""
if isinstance(n_windows, bool) or not isinstance(n_windows, Integral) or n_windows < 1:
return {"error": f"n_windows must be >= 1, got {n_windows}"}
if len(equity_curve) < n_windows * 2:
return {"error": f"need at least {n_windows * 2} bars for {n_windows} windows"}
indices = equity_curve.index
window_size = len(indices) // n_windows
windows = []
for i in range(n_windows):
start_idx = i * window_size
end_idx = (i + 1) * window_size if i < n_windows - 1 else len(indices)
win_eq = equity_curve.iloc[start_idx:end_idx]
win_start = indices[start_idx]
win_end = indices[end_idx - 1]
if not np.isfinite(win_eq.iloc[[0, -1]]).all():
return {"error": f"window {i + 1} requires observed start and end equity"}
# Per-window trades
win_trades = [t for t in trades if win_start <= t.entry_time <= win_end]
# Per-window metrics
ret = float(win_eq.iloc[-1] / win_eq.iloc[0] - 1) if win_eq.iloc[0] > 0 else 0.0
# Use only observed adjacent pairs within this window.
win_returns = (
win_eq.pct_change(fill_method=None)
.replace([np.inf, -np.inf], 0.0)
.dropna()
.values
)
if not len(win_returns):
return {"error": f"window {i + 1} has no observed adjacent equity returns"}
sharpe = _sharpe(win_returns, bars_per_year) if len(win_returns) > 1 else 0.0
peak = win_eq.cummax()
dd = (win_eq - peak) / peak.replace(0, 1)
max_dd = float(dd.min())
win_pnls = [t.pnl for t in win_trades]
win_rate = len([p for p in win_pnls if p > 0]) / len(win_pnls) if win_pnls else 0.0
windows.append(
{
"window": i + 1,
"start": str(win_start.date()) if hasattr(win_start, "date") else str(win_start),
"end": str(win_end.date()) if hasattr(win_end, "date") else str(win_end),
"return": round(ret, 6),
"sharpe": round(sharpe, 4),
"max_dd": round(max_dd, 6),
"trades": len(win_trades),
"win_rate": round(win_rate, 4),
}
)
# Consistency metrics
returns_list = [w["return"] for w in windows]
sharpes_list = [w["sharpe"] for w in windows]
profitable_windows = sum(1 for r in returns_list if r > 0)
return {
"n_windows": n_windows,
"windows": windows,
"profitable_windows": profitable_windows,
"consistency_rate": round(profitable_windows / n_windows, 4),
"return_mean": round(float(np.mean(returns_list)), 6),
"return_std": round(float(np.std(returns_list)), 6),
"sharpe_mean": round(float(np.mean(sharpes_list)), 4),
"sharpe_std": round(float(np.std(sharpes_list)), 4),
}
def purged_walk_forward_analysis(
equity_curve: pd.Series,
trades: List[TradeRecord],
n_windows: int = 5,
purge_bars: int = 2,
embargo_bars: int = 1,
bars_per_year: int = 252,
) -> Dict[str, Any]:
"""Purged walk-forward analysis with purge and embargo gaps between windows.
Prevents holding-period and serial-correlation leakage across sequential
walk-forward folds by:
1. Dropping ``purge_bars`` from the tail of each fold before a boundary.
2. Dropping ``embargo_bars`` from the head of each subsequent fold after a boundary.
3. Excluding any trade whose ``[entry_time, exit_time]`` straddles outside
the purged fold's ``[win_start, win_end]`` interval.
Args:
equity_curve: Equity time series.
trades: Completed trades.
n_windows: Number of sequential evaluation folds.
purge_bars: Bars removed before each inter-fold boundary (>= 0).
embargo_bars: Bars removed after each inter-fold boundary (>= 0).
bars_per_year: Annualisation factor.
Returns:
Dict with per-window stats, purged bar counts, and consistency metrics.
"""
if isinstance(n_windows, bool) or not isinstance(n_windows, Integral) or n_windows < 1:
return {"error": f"n_windows must be >= 1, got {n_windows}"}
if isinstance(purge_bars, bool) or not isinstance(purge_bars, Integral) or purge_bars < 0:
return {"error": f"purge_bars must be >= 0, got {purge_bars}"}
if isinstance(embargo_bars, bool) or not isinstance(embargo_bars, Integral) or embargo_bars < 0:
return {"error": f"embargo_bars must be >= 0, got {embargo_bars}"}
min_bars_per_win = int(purge_bars) + int(embargo_bars) + 3
if len(equity_curve) < n_windows * min_bars_per_win:
return {
"error": (
f"need at least {n_windows * min_bars_per_win} bars for "
f"{n_windows} purged windows (purge={purge_bars}, embargo={embargo_bars})"
)
}
indices = equity_curve.index
raw_window_size = len(indices) // n_windows
windows: List[Dict[str, Any]] = []
purged_bars_total = 0
for i in range(n_windows):
raw_start = i * raw_window_size
raw_end = (i + 1) * raw_window_size if i < n_windows - 1 else len(indices)
eff_start = raw_start + (int(embargo_bars) if i > 0 else 0)
eff_end = raw_end - (int(purge_bars) if i < n_windows - 1 else 0)
purged_in_fold = (eff_start - raw_start) + (raw_end - eff_end)
purged_bars_total += purged_in_fold
if eff_end - eff_start < 2:
return {"error": f"window {i + 1} has fewer than 2 bars after purge/embargo"}
win_eq = equity_curve.iloc[eff_start:eff_end]
win_start = indices[eff_start]
win_end = indices[eff_end - 1]
if not np.isfinite(win_eq.iloc[[0, -1]]).all():
return {"error": f"window {i + 1} requires observed start and end equity"}
# Strict purge on trades: both entry_time and exit_time must fall inside [win_start, win_end]
win_trades = [
t
for t in trades
if win_start <= t.entry_time <= win_end and win_start <= t.exit_time <= win_end
]
ret = float(win_eq.iloc[-1] / win_eq.iloc[0] - 1) if win_eq.iloc[0] > 0 else 0.0
win_returns = (
win_eq.pct_change(fill_method=None)
.replace([np.inf, -np.inf], 0.0)
.dropna()
.values
)
if not len(win_returns):
return {"error": f"window {i + 1} has no observed adjacent equity returns"}
sharpe = _sharpe(win_returns, bars_per_year) if len(win_returns) > 1 else 0.0
peak = win_eq.cummax()
dd = (win_eq - peak) / peak.replace(0, 1)
max_dd = float(dd.min())
win_pnls = [t.pnl for t in win_trades]
win_rate = len([p for p in win_pnls if p > 0]) / len(win_pnls) if win_pnls else 0.0
windows.append(
{
"window": i + 1,
"start": str(win_start.date()) if hasattr(win_start, "date") else str(win_start),
"end": str(win_end.date()) if hasattr(win_end, "date") else str(win_end),
"bars": int(eff_end - eff_start),
"purged_bars": int(purged_in_fold),
"return": round(ret, 6),
"sharpe": round(sharpe, 4),
"max_dd": round(max_dd, 6),
"trades": len(win_trades),
"win_rate": round(win_rate, 4),
}
)
returns_list = [w["return"] for w in windows]
sharpes_list = [w["sharpe"] for w in windows]
profitable_windows = sum(1 for r in returns_list if r > 0)
return {
"n_windows": int(n_windows),
"purge_bars": int(purge_bars),
"embargo_bars": int(embargo_bars),
"purged_bars_total": int(purged_bars_total),
"windows": windows,
"profitable_windows": profitable_windows,
"consistency_rate": round(profitable_windows / n_windows, 4),
"return_mean": round(float(np.mean(returns_list)), 6),
"return_std": round(float(np.std(returns_list)), 6),
"sharpe_mean": round(float(np.mean(sharpes_list)), 4),
"sharpe_std": round(float(np.std(sharpes_list)), 4),
}
# ─── Runner integration ───
def run_validation(
config: Dict[str, Any],
equity_curve: pd.Series,
trades: List[TradeRecord],
initial_capital: float,
bars_per_year: int | None = 252,
) -> Dict[str, Any]:
"""Run configured validation checks.
Reads from config["validation"]:
- monte_carlo: {n_simulations, seed}
- bootstrap: {n_bootstrap, confidence, seed}
- walk_forward: {n_windows}
- purged_walk_forward: {n_windows, purge_bars, embargo_bars}
Args:
config: Backtest config (must contain "validation" key).
equity_curve: Equity time series.
trades: Completed trades.
initial_capital: Starting capital.
bars_per_year: Annualisation factor.
Returns:
Dict keyed by validation type with results.
"""
v_cfg = config.get("validation", {})
results: Dict[str, Any] = {}
# Cross-market convention (runner.py passes bars_per_year=None): resolve
# it through the shared span-derived factor — otherwise _sharpe's
# np.sqrt(bars_per_year) raises TypeError for every validation-enabled
# cross-market run.
if bars_per_year is None:
bars_per_year = effective_bars_per_year(equity_curve.index)
if "monte_carlo" in v_cfg:
mc_cfg = v_cfg["monte_carlo"] if isinstance(v_cfg["monte_carlo"], dict) else {}
results["monte_carlo"] = monte_carlo_test(
trades,
initial_capital,
n_simulations=mc_cfg.get("n_simulations", 1000),
seed=mc_cfg.get("seed", 42),
bars_per_year=bars_per_year,
)
if "bootstrap" in v_cfg:
bs_cfg = v_cfg["bootstrap"] if isinstance(v_cfg["bootstrap"], dict) else {}
results["bootstrap"] = bootstrap_sharpe_ci(
equity_curve,
bars_per_year=bars_per_year,
n_bootstrap=bs_cfg.get("n_bootstrap", 1000),
confidence=bs_cfg.get("confidence", 0.95),
seed=bs_cfg.get("seed", 42),
initial_capital=initial_capital,
)
if "walk_forward" in v_cfg:
wf_cfg = v_cfg["walk_forward"] if isinstance(v_cfg["walk_forward"], dict) else {}
results["walk_forward"] = walk_forward_analysis(
equity_curve,
trades,
n_windows=wf_cfg.get("n_windows", 5),
bars_per_year=bars_per_year,
)
if "purged_walk_forward" in v_cfg:
pwf_cfg = (
v_cfg["purged_walk_forward"]
if isinstance(v_cfg["purged_walk_forward"], dict)
else {}
)
results["purged_walk_forward"] = purged_walk_forward_analysis(
equity_curve,
trades,
n_windows=pwf_cfg.get("n_windows", 5),
purge_bars=pwf_cfg.get("purge_bars", 2),
embargo_bars=pwf_cfg.get("embargo_bars", 1),
bars_per_year=bars_per_year,
)
return results
# ─── Standalone CLI ───
def _load_equity(run_dir: Path) -> pd.Series:
"""Load equity curve from artifacts/equity.csv."""
path = run_dir / "artifacts" / "equity.csv"
df = pd.read_csv(path, index_col=0, parse_dates=True)
for col in ("equity", "nav", "value"):
if col in df.columns:
return df[col]
raise ValueError(
f"equity.csv must contain an equity/nav/value column; got {list(df.columns)}"
)
def _load_trades(run_dir: Path) -> List[TradeRecord]:
"""Load trades from artifacts/trades.csv and convert to TradeRecord list."""
path = run_dir / "artifacts" / "trades.csv"
df = pd.read_csv(path)
if df.empty:
return []
# New artifacts identify exits explicitly, including break-even trades.
# Unmarked historical/other-engine files retain their legacy convention.
trades = []
exit_mask = df["event"].eq("exit") if "event" in df.columns else df["pnl"].ne(0)
exit_rows = df[exit_mask]
for row_idx, row in exit_rows.iterrows():
exit_time = pd.Timestamp(row.get("timestamp", "2000-01-01"))
entry_time = exit_time
# BaseEngine emits adjacent entry/exit pairs. Recover entry dates only
# from an explicit matching pair; legacy/exit-only files stay unchanged.
if "event" in df.columns and row_idx > 0:
entry = df.iloc[row_idx - 1]
expected_side = {"sell": "buy", "buy": "sell"}.get(row.get("side"))
if (
entry["event"] == "entry"
and entry.get("code") == row.get("code")
and expected_side is not None
and entry.get("side") == expected_side
):
candidate_time = pd.to_datetime(entry.get("timestamp"), errors="coerce")
try:
if pd.notna(candidate_time) and candidate_time <= exit_time:
entry_time = candidate_time
except TypeError:
pass # Incompatible timestamp timezones cannot form a pair.
hold = pd.to_numeric(row.get("holding_bars"), errors="coerce")
if pd.isna(hold):
hold = pd.to_numeric(row.get("holding_days", 0), errors="coerce")
holding_bars = 0.0 if pd.isna(hold) else float(hold)
trades.append(
TradeRecord(
symbol=str(row.get("code", "")),
direction=1 if row.get("side") == "sell" else -1,
entry_price=0.0,
exit_price=float(row.get("price", 0)),
entry_time=entry_time,
exit_time=exit_time,
size=float(row.get("qty", 0)),
leverage=1.0,
pnl=float(row.get("pnl", 0)),
pnl_pct=float(row.get("return_pct", 0)),
exit_reason=str(row.get("reason", "signal")),
holding_bars=holding_bars,
commission=0.0,
)
)
return trades
def _parse_run_dir(argv: List[str]) -> Path:
"""Validate CLI input and return a usable run directory path."""
if len(argv) < 2:
raise SystemExit("Usage: python -m backtest.validation <run_dir>")
raw_run_dir = argv[1]
if not raw_run_dir.strip():
raise SystemExit("run_dir must be a non-empty path")
if "\0" in raw_run_dir:
raise SystemExit("Invalid run_dir path: embedded NUL byte")
try:
run_dir = Path(raw_run_dir).expanduser()
exists = run_dir.exists()
is_dir = run_dir.is_dir() if exists else False
except (OSError, RuntimeError, ValueError) as exc:
raise SystemExit(f"Invalid run_dir path: {exc}") from exc
if not exists:
raise SystemExit(f"run_dir does not exist: {run_dir}")
if not is_dir:
raise SystemExit(f"run_dir is not a directory: {run_dir}")
return run_dir
def _json_safe(value: Any) -> Any:
"""Return a JSON-strict copy of validation results."""
if isinstance(value, np.ndarray):
return [_json_safe(item) for item in value.tolist()]
if isinstance(value, np.generic):
return _json_safe(value.item())
if isinstance(value, float):
return value if math.isfinite(value) else None
if isinstance(value, dict):
return {str(key): _json_safe(item) for key, item in value.items()}
if isinstance(value, (list, tuple)):
return [_json_safe(item) for item in value]
return value
def write_validation_json(path: Path, results: Dict[str, Any]) -> Dict[str, Any]:
"""Write validation results to ``path`` as strict, RFC-8259 JSON.
A validation metric can be non-finite (e.g. a Sharpe computed from a path
whose equity touches zero), and ``json.dumps`` emits bare ``NaN`` /
``Infinity`` tokens for those by default (``allow_nan=True``) — tokens that
strict parsers reject. Sanitize with :func:`_json_safe` (non-finite → null)
and serialize with ``allow_nan=False`` so every writer of
``artifacts/validation.json`` produces the same valid JSON. Returns the
sanitized payload that was written.
"""
safe_results = _json_safe(results)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(safe_results, indent=2, ensure_ascii=False, allow_nan=False) + "\n",
encoding="utf-8",
)
return safe_results
def main(run_dir: Path) -> Dict[str, Any]:
"""Run all three validations on existing backtest artifacts.
Reads equity.csv, trades.csv, and config.json from run_dir.
Runs all three tools with saved validation settings, defaulting missing ones.
Args:
run_dir: Directory with artifacts/ subdirectory.
Returns:
Validation results dict.
"""
# Load initial cash and saved validation settings.
config_path = run_dir / "config.json"
if config_path.exists():
config = json.loads(config_path.read_text(encoding="utf-8"))
else:
config = {}
initial_capital = config.get("initial_cash", 1_000_000)
equity = _load_equity(run_dir)
trades = _load_trades(run_dir)
# Standalone validation always runs all tools, even when the original run
# enabled only a subset. Reuse dispatcher defaults and parameter guards.
settings = {name: {} for name in ("monte_carlo", "bootstrap", "walk_forward")}
saved_settings = config.get("validation")
if isinstance(saved_settings, dict):
settings.update(saved_settings)
results = run_validation({"validation": settings}, equity, trades, initial_capital)
out = run_dir / "artifacts" / "validation.json"
safe_results = write_validation_json(out, results)
print(json.dumps(safe_results, indent=2, allow_nan=False))
return safe_results
# ─── Phase 1 Schedule Promotion Gate & Tiny-Universe Multi-Market Proof ───
TINY_UNIVERSE_BY_MARKET: Dict[str, tuple[str, ...]] = {
"US": ("NVDA.US", "AAPL.US"),
"HK": ("0700.HK", "9988.HK"),
"CRYPTO": ("BTC-USD", "ETH-USD"),
}
class _InMemoryFrameLoader:
"""Minimal DataLoader adapter wrapping pre-loaded OHLCV DataFrames."""
name = "in_memory_frame_loader"
def __init__(self, frames: Dict[str, pd.DataFrame]) -> None:
self._frames = dict(frames)
def fetch(
self,
codes: List[str],
start_date: str = "",
end_date: str = "",
fields: List[str] | None = None,
interval: str = "1D",
) -> Dict[str, pd.DataFrame]:
del start_date, end_date, fields, interval
return {code: self._frames[code].copy() for code in codes if code in self._frames}
class _TrendCrossoverSignalEngine:
"""Deterministic fast/slow MA crossover signal engine for tiny-universe proof."""
def __init__(self, fast_window: int = 5, slow_window: int = 15) -> None:
self.fast_window = fast_window
self.slow_window = slow_window
def generate(self, data_map: Dict[str, pd.DataFrame]) -> Dict[str, pd.Series]:
signals: Dict[str, pd.Series] = {}
for code, df in data_map.items():
close = df["close"].astype(float)
fast = close.rolling(self.fast_window, min_periods=1).mean()
slow = close.rolling(self.slow_window, min_periods=1).mean()
sig = pd.Series(np.where(fast > slow, 1.0, 0.0), index=close.index, dtype=float)
signals[code] = sig
return signals
def evaluate_promotion_gate(
market_reports: Dict[str, Dict[str, Any]],
*,
required_markets: tuple[str, ...] = ("US", "HK", "CRYPTO"),
min_bootstrap_ci_lower: float = -0.25,
min_prob_positive: float = 0.55,
min_wf_consistency: float = 0.40,
) -> Dict[str, Any]:
"""Evaluate whether US + HK + Crypto backtest validation gates pass for schedule promotion.
No paper automation schedule may promote or dispatch proposals unless every
required market passes both Purged Walk-Forward consistency and Bootstrap
Sharpe CI thresholds (fail-closed).
Args:
market_reports: Mapping of market key (``"US"``, ``"HK"``, ``"CRYPTO"``)
to backtest metrics/validation dicts.
required_markets: Markets that must all pass before promotion.
min_bootstrap_ci_lower: Minimum lower bound for bootstrap Sharpe CI.
min_prob_positive: Minimum bootstrap probability of positive Sharpe.
min_wf_consistency: Minimum fraction of profitable purged walk-forward folds.
Returns:
Dict with ``passed`` (bool), ``markets_checked``, ``per_market``, and
``failed_reasons``.
"""
from datetime import datetime, timezone
failed_reasons: List[str] = []
per_market: Dict[str, Dict[str, Any]] = {}
normalized_reports = {
str(k).strip().upper(): v for k, v in (market_reports or {}).items() if isinstance(v, dict)
}
for market in required_markets:
mkt = market.strip().upper()
report = normalized_reports.get(mkt)
if report is None:
failed_reasons.append(f"{mkt}: missing backtest validation report")
per_market[mkt] = {"passed": False, "reason": "missing_report"}
continue
val = report.get("validation") if isinstance(report.get("validation"), dict) else report
bs = val.get("bootstrap") if isinstance(val.get("bootstrap"), dict) else None
pwf = (
val.get("purged_walk_forward")
if isinstance(val.get("purged_walk_forward"), dict)
else val.get("walk_forward")
if isinstance(val.get("walk_forward"), dict)
else None
)
mkt_reasons: List[str] = []
if not bs or "error" in bs:
mkt_reasons.append(
f"{mkt}: bootstrap validation missing or failed ({(bs or {}).get('error', 'missing')})"
)
else:
ci_lower = float(bs.get("ci_lower", float("-inf")))
prob_pos = float(bs.get("prob_positive", 0.0))
if ci_lower < min_bootstrap_ci_lower:
mkt_reasons.append(
f"{mkt}: bootstrap ci_lower {ci_lower:.4f} < {min_bootstrap_ci_lower:.4f}"
)
if prob_pos < min_prob_positive:
mkt_reasons.append(
f"{mkt}: bootstrap prob_positive {prob_pos:.4f} < {min_prob_positive:.4f}"
)
if not pwf or "error" in pwf:
mkt_reasons.append(
f"{mkt}: purged_walk_forward missing or failed ({(pwf or {}).get('error', 'missing')})"
)
else:
consistency = float(pwf.get("consistency_rate", 0.0))
if consistency < min_wf_consistency:
mkt_reasons.append(
f"{mkt}: purged_walk_forward consistency_rate {consistency:.4f} < {min_wf_consistency:.4f}"
)
if "purge_bars" in pwf and int(pwf.get("purge_bars", 0)) < 1:
mkt_reasons.append(f"{mkt}: purged_walk_forward requires purge_bars >= 1")
mkt_passed = len(mkt_reasons) == 0
failed_reasons.extend(mkt_reasons)
per_market[mkt] = {
"passed": mkt_passed,
"bootstrap": bs,
"purged_walk_forward": pwf,
"reasons": mkt_reasons,
}
return {
"passed": len(failed_reasons) == 0 and len(required_markets) > 0,
"markets_checked": [m.strip().upper() for m in required_markets],
"per_market": per_market,
"failed_reasons": failed_reasons,
"evaluated_at_utc": datetime.now(timezone.utc).isoformat(),
}
def run_tiny_universe_backtest_proof(
*,
bar_data_by_market: Dict[str, Dict[str, pd.DataFrame]] | None = None,
work_dir: Path | None = None,
n_bootstrap: int = 300,
n_windows: int = 4,
purge_bars: int = 2,
embargo_bars: int = 1,
seed: int = 42,
min_bootstrap_ci_lower: float = -0.25,
min_prob_positive: float = 0.55,
min_wf_consistency: float = 0.40,
) -> Dict[str, Any]:
"""Run US + HK + Crypto backtest engines on a tiny universe with purged walk-forward + bootstrap.
Executes:
- ``GlobalEquityEngine(market="us")`` on ``("NVDA.US", "AAPL.US")``
- ``GlobalEquityEngine(market="hk")`` on ``("0700.HK", "9988.HK")``
- ``CryptoEngine`` on ``("BTC-USD", "ETH-USD")``
and evaluates the schedule promotion gate across all three markets.
"""
import tempfile
from backtest.engines.crypto import CryptoEngine
from backtest.engines.global_equity import GlobalEquityEngine
base_dir_ctx = (
tempfile.TemporaryDirectory(prefix="vibe_bt_proof_")
if work_dir is None
else None
)
try:
root_dir = Path(base_dir_ctx.name) if base_dir_ctx is not None else Path(work_dir) # type: ignore[arg-type]
root_dir.mkdir(parents=True, exist_ok=True)
signal_engine = _TrendCrossoverSignalEngine(fast_window=5, slow_window=15)
market_reports: Dict[str, Dict[str, Any]] = {}
for mkt_key, default_codes in TINY_UNIVERSE_BY_MARKET.items():
frames = (bar_data_by_market or {}).get(mkt_key)
if frames:
codes = sorted(frames.keys())
loader: Any = _InMemoryFrameLoader(frames)
else:
from backtest.loaders.registry import resolve_loader
codes = list(default_codes)
loader_market = {
"US": "us_equity",
"HK": "hk_equity",
"CRYPTO": "crypto",
}[mkt_key]
loader = resolve_loader(loader_market)
bars_per_year = 365 if mkt_key == "CRYPTO" else 252
config: Dict[str, Any] = {
"codes": codes,
"initial_cash": 100_000.0,
"leverage": 1.0,
"interval": "1D",
"validation": {
"bootstrap": {
"n_bootstrap": n_bootstrap,
"confidence": 0.95,
"seed": seed,
},
"purged_walk_forward": {
"n_windows": n_windows,
"purge_bars": purge_bars,
"embargo_bars": embargo_bars,
},
},
}
if mkt_key == "US":
engine: Any = GlobalEquityEngine(config, market="us")
elif mkt_key == "HK":
config["hk_lot_sizes"] = {code: 100 for code in codes}
engine = GlobalEquityEngine(config, market="hk")
else:
engine = CryptoEngine(config)
mkt_run_dir = root_dir / mkt_key.lower()
mkt_run_dir.mkdir(parents=True, exist_ok=True)
metrics = engine.run_backtest(
config,
loader,
signal_engine,
mkt_run_dir,
bars_per_year=bars_per_year,
)
market_reports[mkt_key] = metrics
gate = evaluate_promotion_gate(
market_reports,
min_bootstrap_ci_lower=min_bootstrap_ci_lower,
min_prob_positive=min_prob_positive,
min_wf_consistency=min_wf_consistency,
)
gate["market_metrics"] = {
mkt: {
"codes": list((bar_data_by_market or {}).get(mkt, {}).keys())
or list(TINY_UNIVERSE_BY_MARKET[mkt]),
"total_return": rep.get("total_return"),
"sharpe": rep.get("sharpe"),
"max_drawdown": rep.get("max_drawdown"),
"total_trades": rep.get("total_trades"),
"validation": rep.get("validation"),
}
for mkt, rep in market_reports.items()
}
return gate
finally:
if base_dir_ctx is not None:
base_dir_ctx.cleanup()
if __name__ == "__main__":
import sys
main(_parse_run_dir(sys.argv))