spb/crypto-trade-bot

1"""Backtest orchestration: provenance, benchmarks, uncertainty, stress tests, walk-forward; stores results in Postgres."""23from __future__ import annotations45import hashlib6import math7import subprocess8import time9import uuid10from pathlib import Path11from typing import Any1213import asyncpg14import numpy as np15import pandas as pd1617from ktrade import __version__18from ktrade.backtest.costs import CostModel19from ktrade.backtest.engine import run_backtest20from ktrade.backtest.metrics import (21    block_bootstrap_ci,22    compute_metrics,23    deflated_sharpe,24    drawdown_series,25    rolling_sharpe,26    shuffled_trades_test,27    stress_scenarios,28)29from ktrade.backtest.validation import sensitivity_map, walk_forward30from ktrade.market.service import candles_frame31from ktrade.risk.gateway import RiskLimits32from ktrade.strategy.registry import get_strategy3334ROOT = Path(__file__).resolve().parents[3]353637def _git_commit() -> str | None:38    try:39        return subprocess.check_output(["git", "-C", str(ROOT), "rev-parse", "HEAD"], text=True, stderr=subprocess.DEVNULL).strip()40    except Exception:  # noqa: BLE00141        return None424344def dataset_hash(df: pd.DataFrame) -> str:45    h = hashlib.sha256()46    h.update(pd.util.hash_pandas_object(df[["ts", "open", "high", "low", "close", "volume"]], index=False).to_numpy().tobytes())47    return h.hexdigest()[:16]484950def _clean(o: Any) -> Any:51    if isinstance(o, dict):52        return {k: _clean(v) for k, v in o.items()}53    if isinstance(o, list | tuple):54        return [_clean(v) for v in o]55    if isinstance(o, float) and not math.isfinite(o):56        return None57    if isinstance(o, np.bool_):58        return bool(o)59    if isinstance(o, np.floating | np.integer):60        return _clean(o.item())61    return o626364def _downsample(eq: pd.DataFrame, max_points: int = 3000) -> list[dict[str, Any]]:65    if len(eq) > max_points:66        step = int(math.ceil(len(eq) / max_points))67        eq = pd.concat([eq.iloc[::step], eq.iloc[[-1]]]).drop_duplicates("ts")68    dd = drawdown_series(eq["equity"])69    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)]707172async def execute_backtest(pool: asyncpg.Pool, run_id: uuid.UUID, spec: dict[str, Any]) -> None:73    await pool.execute("UPDATE backtest_runs SET status='running' WHERE id=$1", run_id)74    t0 = time.time()75    try:76        strategy_cls = get_strategy(spec["strategy"])77        interval = int(spec["interval_min"])78        df = await candles_frame(pool, spec["symbol"], interval, spec.get("start_ts"), spec.get("end_ts"))79        if len(df) < strategy_cls.meta.min_bars:80            raise ValueError(f"only {len(df)} closed bars available; strategy needs {strategy_cls.meta.min_bars}")81        pair = await pool.fetchrow("SELECT lot_decimals, ordermin, costmin FROM pairs WHERE wsname=$1", spec["symbol"])82        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}83        cm = CostModel(**spec.get("cost_model", {}))84        initial_cash = float(spec.get("initial_cash", 10_000.0))85        limits = RiskLimits(**spec["risk_limits"]) if spec.get("risk_limits") else None86        holdout_frac = float(spec.get("holdout_frac", 0.0))87        df_main = df.iloc[: int(len(df) * (1 - holdout_frac))] if holdout_frac > 0 else df88        df_hold = df.iloc[int(len(df) * (1 - holdout_frac)) - strategy_cls.meta.min_bars * 2 :] if holdout_frac > 0 else None8990        strat = strategy_cls(spec.get("params", {}))91        res = run_backtest(strat, df_main, interval_min=interval, initial_cash=initial_cash, cost_model=cm, risk_limits=limits, **bt_kwargs)92        metrics = compute_metrics(res.equity, res.trades, interval, initial_cash, res.fees_paid, res.slippage_paid, res.turnover)93        bars_per_year = 365 * 24 * 60 / interval94        r = np.diff(np.log(np.maximum(res.equity["equity"].to_numpy(), 1e-12)))95        metrics["bootstrap"] = block_bootstrap_ci(r, bars_per_year)96        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)9798        # benchmarks on the same window & costs99        benchmarks = {}100        for b in ("buy_and_hold", "cash", "simple_momentum", "equal_risk"):101            br = run_backtest(get_strategy(b)(), df_main, interval_min=interval, initial_cash=initial_cash, cost_model=cm, **bt_kwargs)102            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)}103104        def rerun(cost: CostModel) -> dict[str, Any]:105            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)106            mm = compute_metrics(rr.equity, rr.trades, interval, initial_cash, rr.fees_paid, rr.slippage_paid, rr.turnover)107            return {k: mm.get(k) for k in ("sharpe", "total_return", "max_drawdown", "cagr", "fee_drag")}108109        stress = stress_scenarios(rerun, cm, res.equity) if spec.get("stress", True) else {}110        # gap / outage scenario: drop a random 5% contiguous block of bars (exchange outage) and re-run111        if spec.get("stress", True) and len(df_main) > 400:112            cut = len(df_main) // 20113            start = len(df_main) // 2114            df_gap = pd.concat([df_main.iloc[:start], df_main.iloc[start + cut :]]).reset_index(drop=True)115            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)116            gm = compute_metrics(gr.equity, gr.trades, interval, initial_cash, gr.fees_paid, gr.slippage_paid, gr.turnover)117            stress["exchange_outage_gap"] = {k: gm.get(k) for k in ("sharpe", "total_return", "max_drawdown")}118        from ktrade.backtest.metrics import _round_trips119120        shuffled = shuffled_trades_test(_round_trips(res.trades)) if spec.get("stress", True) else {}121122        wf = None123        if spec.get("walk_forward"):124            wfc = spec["walk_forward"]125            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)126        sens = None127        if spec.get("sensitivity"):128            sc = spec["sensitivity"]129            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)130        holdout = None131        if df_hold is not None and len(df_hold) > strategy_cls.meta.min_bars * 2:132            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)133            start_ts = int(df.iloc[int(len(df) * (1 - holdout_frac))]["ts"])134            heq = hr.equity[hr.equity["ts"] >= start_ts].reset_index(drop=True)135            if len(heq) > 2:136                heq = heq.assign(equity=heq["equity"] / heq["equity"].iloc[0] * initial_cash)137                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)}138139        provenance = {140            "code_version": __version__, "git_commit": _git_commit(), "dataset_hash": dataset_hash(df_main), "n_bars": int(len(df_main)),141            "window": [int(df_main["ts"].iloc[0]), int(df_main["ts"].iloc[-1])], "params": strat.params, "cost_model": cm.model_dump(), "seed": 42,142            "initial_cash": initial_cash, "risk_limits": limits.model_dump() if limits else None, "holdout_frac": holdout_frac, "runtime_s": round(time.time() - t0, 2),143            "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.",144        }145        artifacts = {146            "equity": _downsample(res.equity), "rolling_sharpe": rolling_sharpe(res.equity, interval, max(30, min(int(bars_per_year / 12), len(res.equity) // 4))),147            "trades": [t.__dict__ for t in res.trades[-2000:]], "rejected": res.rejected[-200:], "benchmarks": benchmarks, "stress": stress, "shuffled_trades": shuffled,148            "walk_forward": wf, "sensitivity": sens, "holdout": holdout, "features_tail": _features_tail(strat, df_main),149        }150        await pool.execute(151            "UPDATE backtest_runs SET status='done', finished_at=now(), metrics=$2, provenance=$3, artifacts=$4, start_ts=$5, end_ts=$6 WHERE id=$1",152            run_id, _clean(metrics), _clean(provenance), _clean(artifacts), int(df_main["ts"].iloc[0]), int(df_main["ts"].iloc[-1]),153        )154    except Exception as e:  # noqa: BLE001155        await pool.execute("UPDATE backtest_runs SET status='failed', finished_at=now(), error=$2 WHERE id=$1", run_id, f"{type(e).__name__}: {e}")156        raise157158159def _features_tail(strat, df: pd.DataFrame, n: int = 500) -> list[dict[str, Any]]:160    try:161        f = strat.features(df.assign().copy()).tail(n)162        cols = [c for c in f.columns if f[c].dtype.kind in "fi"][:8]163        return _clean(f[cols].round(6).to_dict(orient="records"))164    except Exception:  # noqa: BLE001165        return []166