"""Backtest orchestration: provenance, benchmarks, uncertainty, stress tests, walk-forward; stores results in Postgres.""" from __future__ import annotations import hashlib import math import subprocess import time import uuid from pathlib import Path from typing import Any import asyncpg import numpy as np import pandas as pd from ktrade import __version__ from ktrade.backtest.costs import CostModel from ktrade.backtest.engine import run_backtest from ktrade.backtest.metrics import ( block_bootstrap_ci, compute_metrics, deflated_sharpe, drawdown_series, rolling_sharpe, shuffled_trades_test, stress_scenarios, ) from ktrade.backtest.validation import sensitivity_map, walk_forward from ktrade.market.service import candles_frame from ktrade.risk.gateway import RiskLimits from ktrade.strategy.registry import get_strategy ROOT = Path(__file__).resolve().parents[3] def _git_commit() -> str | None: try: return subprocess.check_output(["git", "-C", str(ROOT), "rev-parse", "HEAD"], text=True, stderr=subprocess.DEVNULL).strip() except Exception: # noqa: BLE001 return None def dataset_hash(df: pd.DataFrame) -> str: h = hashlib.sha256() h.update(pd.util.hash_pandas_object(df[["ts", "open", "high", "low", "close", "volume"]], index=False).to_numpy().tobytes()) return h.hexdigest()[:16] def _clean(o: Any) -> Any: if isinstance(o, dict): return {k: _clean(v) for k, v in o.items()} if isinstance(o, list | tuple): return [_clean(v) for v in o] if isinstance(o, float) and not math.isfinite(o): return None if isinstance(o, np.bool_): return bool(o) if isinstance(o, np.floating | np.integer): return _clean(o.item()) return o def _downsample(eq: pd.DataFrame, max_points: int = 3000) -> list[dict[str, Any]]: if len(eq) > max_points: step = int(math.ceil(len(eq) / max_points)) eq = pd.concat([eq.iloc[::step], eq.iloc[[-1]]]).drop_duplicates("ts") dd = drawdown_series(eq["equity"]) return [{"ts": int(t), "equity": float(e), "drawdown": float(d), "exposure": float(x), "price": float(p)} for t, e, d, x, p in zip(eq["ts"], eq["equity"], dd, eq["exposure"], eq["price"], strict=False)] async def execute_backtest(pool: asyncpg.Pool, run_id: uuid.UUID, spec: dict[str, Any]) -> None: await pool.execute("UPDATE backtest_runs SET status='running' WHERE id=$1", run_id) t0 = time.time() try: strategy_cls = get_strategy(spec["strategy"]) interval = int(spec["interval_min"]) df = await candles_frame(pool, spec["symbol"], interval, spec.get("start_ts"), spec.get("end_ts")) if len(df) < strategy_cls.meta.min_bars: raise ValueError(f"only {len(df)} closed bars available; strategy needs {strategy_cls.meta.min_bars}") pair = await pool.fetchrow("SELECT lot_decimals, ordermin, costmin FROM pairs WHERE wsname=$1", spec["symbol"]) bt_kwargs: dict[str, Any] = {"lot_decimals": pair["lot_decimals"] if pair else 8, "ordermin": float(pair["ordermin"]) if pair else 0.0, "costmin": float(pair["costmin"]) if pair else 0.0} cm = CostModel(**spec.get("cost_model", {})) initial_cash = float(spec.get("initial_cash", 10_000.0)) limits = RiskLimits(**spec["risk_limits"]) if spec.get("risk_limits") else None holdout_frac = float(spec.get("holdout_frac", 0.0)) df_main = df.iloc[: int(len(df) * (1 - holdout_frac))] if holdout_frac > 0 else df df_hold = df.iloc[int(len(df) * (1 - holdout_frac)) - strategy_cls.meta.min_bars * 2 :] if holdout_frac > 0 else None strat = strategy_cls(spec.get("params", {})) res = run_backtest(strat, df_main, interval_min=interval, initial_cash=initial_cash, cost_model=cm, risk_limits=limits, **bt_kwargs) metrics = compute_metrics(res.equity, res.trades, interval, initial_cash, res.fees_paid, res.slippage_paid, res.turnover) bars_per_year = 365 * 24 * 60 / interval r = np.diff(np.log(np.maximum(res.equity["equity"].to_numpy(), 1e-12))) metrics["bootstrap"] = block_bootstrap_ci(r, bars_per_year) metrics["deflated_sharpe"] = deflated_sharpe(metrics.get("sharpe", float("nan")), len(r), bars_per_year, int(spec.get("n_trials", 1)), metrics.get("skew", 0.0) or 0.0, metrics.get("kurtosis", 0.0) or 0.0) # benchmarks on the same window & costs benchmarks = {} for b in ("buy_and_hold", "cash", "simple_momentum", "equal_risk"): br = run_backtest(get_strategy(b)(), df_main, interval_min=interval, initial_cash=initial_cash, cost_model=cm, **bt_kwargs) benchmarks[b] = {"metrics": compute_metrics(br.equity, br.trades, interval, initial_cash, br.fees_paid, br.slippage_paid, br.turnover), "equity": _downsample(br.equity, 800)} def rerun(cost: CostModel) -> dict[str, Any]: rr = run_backtest(strategy_cls(spec.get("params", {})), df_main, interval_min=interval, initial_cash=initial_cash, cost_model=cost, risk_limits=limits, **bt_kwargs) mm = compute_metrics(rr.equity, rr.trades, interval, initial_cash, rr.fees_paid, rr.slippage_paid, rr.turnover) return {k: mm.get(k) for k in ("sharpe", "total_return", "max_drawdown", "cagr", "fee_drag")} stress = stress_scenarios(rerun, cm, res.equity) if spec.get("stress", True) else {} # gap / outage scenario: drop a random 5% contiguous block of bars (exchange outage) and re-run if spec.get("stress", True) and len(df_main) > 400: cut = len(df_main) // 20 start = len(df_main) // 2 df_gap = pd.concat([df_main.iloc[:start], df_main.iloc[start + cut :]]).reset_index(drop=True) gr = run_backtest(strategy_cls(spec.get("params", {})), df_gap, interval_min=interval, initial_cash=initial_cash, cost_model=cm, risk_limits=limits, **bt_kwargs) gm = compute_metrics(gr.equity, gr.trades, interval, initial_cash, gr.fees_paid, gr.slippage_paid, gr.turnover) stress["exchange_outage_gap"] = {k: gm.get(k) for k in ("sharpe", "total_return", "max_drawdown")} from ktrade.backtest.metrics import _round_trips shuffled = shuffled_trades_test(_round_trips(res.trades)) if spec.get("stress", True) else {} wf = None if spec.get("walk_forward"): wfc = spec["walk_forward"] wf = walk_forward(strategy_cls, df_main, interval_min=interval, grid=wfc.get("grid", {}), n_folds=int(wfc.get("n_folds", 4)), train_frac=float(wfc.get("train_frac", 0.6)), anchored=bool(wfc.get("anchored", True)), cost_model=cm, initial_cash=initial_cash, embargo_bars=int(wfc.get("embargo_bars", 0)), **bt_kwargs) sens = None if spec.get("sensitivity"): sc = spec["sensitivity"] sens = sensitivity_map(strategy_cls, df_main, interval_min=interval, x=(sc["x"], sc["x_values"]), y=(sc["y"], sc["y_values"]), base_params=spec.get("params", {}), cost_model=cm, initial_cash=initial_cash, metric=sc.get("metric", "sharpe"), **bt_kwargs) holdout = None if df_hold is not None and len(df_hold) > strategy_cls.meta.min_bars * 2: hr = run_backtest(strategy_cls(spec.get("params", {})), df_hold, interval_min=interval, initial_cash=initial_cash, cost_model=cm, risk_limits=limits, **bt_kwargs) start_ts = int(df.iloc[int(len(df) * (1 - holdout_frac))]["ts"]) heq = hr.equity[hr.equity["ts"] >= start_ts].reset_index(drop=True) if len(heq) > 2: heq = heq.assign(equity=heq["equity"] / heq["equity"].iloc[0] * initial_cash) holdout = {"start_ts": start_ts, "metrics": compute_metrics(heq, [t for t in hr.trades if t.ts >= start_ts], interval, initial_cash, 0.0, 0.0, hr.turnover), "equity": _downsample(heq, 800)} provenance = { "code_version": __version__, "git_commit": _git_commit(), "dataset_hash": dataset_hash(df_main), "n_bars": int(len(df_main)), "window": [int(df_main["ts"].iloc[0]), int(df_main["ts"].iloc[-1])], "params": strat.params, "cost_model": cm.model_dump(), "seed": 42, "initial_cash": initial_cash, "risk_limits": limits.model_dump() if limits else None, "holdout_frac": holdout_frac, "runtime_s": round(time.time() - t0, 2), "data_source": "Kraken REST OHLC (≤720 bars/poll, accumulated) + WS ohlc", "evaluation_note": "Targets decided at bar close, executed at next open; fees/spread/slippage/impact/partial fills modelled.", } artifacts = { "equity": _downsample(res.equity), "rolling_sharpe": rolling_sharpe(res.equity, interval, max(30, min(int(bars_per_year / 12), len(res.equity) // 4))), "trades": [t.__dict__ for t in res.trades[-2000:]], "rejected": res.rejected[-200:], "benchmarks": benchmarks, "stress": stress, "shuffled_trades": shuffled, "walk_forward": wf, "sensitivity": sens, "holdout": holdout, "features_tail": _features_tail(strat, df_main), } await pool.execute( "UPDATE backtest_runs SET status='done', finished_at=now(), metrics=$2, provenance=$3, artifacts=$4, start_ts=$5, end_ts=$6 WHERE id=$1", run_id, _clean(metrics), _clean(provenance), _clean(artifacts), int(df_main["ts"].iloc[0]), int(df_main["ts"].iloc[-1]), ) except Exception as e: # noqa: BLE001 await pool.execute("UPDATE backtest_runs SET status='failed', finished_at=now(), error=$2 WHERE id=$1", run_id, f"{type(e).__name__}: {e}") raise def _features_tail(strat, df: pd.DataFrame, n: int = 500) -> list[dict[str, Any]]: try: f = strat.features(df.assign().copy()).tail(n) cols = [c for c in f.columns if f[c].dtype.kind in "fi"][:8] return _clean(f[cols].round(6).to_dict(orient="records")) except Exception: # noqa: BLE001 return []