"""App runtime: cached store access, run execution, run history, share links. Hub I/O is cached in-process behind an LRU so a repeated backtest never re-downloads a parquet slice. The cache key includes the store's manifest `updated_at`, so a coverage extension invalidates exactly the slices that changed instead of serving stale data forever. """ from __future__ import annotations import base64 import json import logging import os import threading import time import uuid from dataclasses import dataclass, field, asdict from datetime import datetime, timezone import pandas as pd from . import comparisons, config, strategies from .engine import ( BacktestConfig, Costs, Sizing, Stops, Validation, BacktestResult, run_backtest, ) from .store import SignalStore log = logging.getLogger("bit.runtime") _store: SignalStore | None = None _store_lock = threading.Lock() def get_store() -> SignalStore: """Process-wide store handle. Read-only for anonymous user traffic.""" global _store with _store_lock: if _store is None: token = os.environ.get("HF_WRITE_TOKEN") or os.environ.get("HF_TOKEN") _store = SignalStore( repo_id=config.STORE_REPO, local_root=os.environ.get("BIT_STORE_CACHE", ".cache/store"), token=token, ) return _store # -------------------------------------------------------------------------- # Cached slice access # -------------------------------------------------------------------------- _slice_cache: dict[tuple, pd.DataFrame] = {} _cache_order: list[tuple] = [] _cache_lock = threading.Lock() def _cache_get(key): with _cache_lock: hit = _slice_cache.get(key) if hit is not None: _cache_order.remove(key) _cache_order.append(key) return hit def _cache_put(key, value): with _cache_lock: _slice_cache[key] = value _cache_order.append(key) while len(_cache_order) > config.PARQUET_CACHE_SIZE: old = _cache_order.pop(0) _slice_cache.pop(old, None) def cache_clear() -> None: with _cache_lock: _slice_cache.clear() _cache_order.clear() def _manifest_stamp() -> str: try: return get_store().load_manifest().updated_at except Exception: return "unknown" def load_prices(asset: str, timeframe: str, start=None, end=None) -> pd.DataFrame: key = ("px", asset, timeframe, str(start), str(end), _manifest_stamp()) hit = _cache_get(key) if hit is not None: return hit df = get_store().get_prices(asset, timeframe, start, end) _cache_put(key, df) return df def load_signals(model_slug: str, asset: str, timeframe: str, start=None, end=None) -> pd.DataFrame: if not model_slug: return pd.DataFrame() key = ("sig", model_slug, asset, timeframe, str(start), str(end), _manifest_stamp()) hit = _cache_get(key) if hit is not None: return hit df = get_store().get_signals(model_slug, asset, timeframe, start, end) _cache_put(key, df) return df # -------------------------------------------------------------------------- # Coverage map # -------------------------------------------------------------------------- @dataclass class CoverageCell: model_slug: str asset: str timeframe: str start: str end: str rows: int is_placeholder: bool contributed_by: str def coverage_map() -> list[CoverageCell]: m = get_store().load_manifest() return [ CoverageCell( model_slug=e.model_slug, asset=e.asset, timeframe=e.timeframe, start=e.start_ts[:10], end=e.end_ts[:10], rows=e.rows, is_placeholder=e.is_placeholder, contributed_by=e.contributed_by, ) for e in sorted(m.signals.values(), key=lambda x: (x.model_slug, x.asset, x.timeframe)) ] def coverage_frame() -> pd.DataFrame: cells = coverage_map() if not cells: return pd.DataFrame(columns=["Model", "Asset", "TF", "Coverage", "Rows", "Source", "Real?"]) return pd.DataFrame([{ "Model": c.model_slug, "Asset": c.asset, "TF": c.timeframe, "Coverage": f"{c.start} → {c.end}", "Rows": f"{c.rows:,}", "Source": c.contributed_by, "Real?": "PLACEHOLDER" if c.is_placeholder else "real", } for c in cells]) def available_models(asset: str | None = None, timeframe: str | None = None) -> list[str]: m = get_store().load_manifest() return sorted({e.model_slug for e in m.find_signals(asset=asset, timeframe=timeframe)}) def available_assets() -> list[str]: m = get_store().load_manifest() found = sorted({p.asset for p in m.prices.values()}) return found or list(config.ASSETS) def price_coverage_for(asset: str, timeframe: str) -> tuple[str, str] | None: cov = get_store().load_manifest().prices.get(f"{asset}|{timeframe}") return (cov.start_ts[:10], cov.end_ts[:10]) if cov else None # -------------------------------------------------------------------------- # Run configuration # -------------------------------------------------------------------------- RANGE_YEARS = {"1Y": 1.0, "3Y": 3.0, "5Y": 5.0, "Max": 99.0} @dataclass class RunRequest: """Everything the Strategy Builder collects, in one serialisable object.""" strategy: str = "SMA Crossover" asset: str = "BTC-USD" timeframe: str = "1d" date_range: str = "3Y" model_slug: str = "" params: dict = field(default_factory=dict) costs_on: bool = True commission_bps: float = config.DEFAULT_COMMISSION_BPS slippage_bps: float = config.DEFAULT_SLIPPAGE_BPS slippage_model: str = "fixed" sizing_mode: str = "fixed_pct" size_pct: float = 1.0 leverage: float = 1.0 max_position: float = 1.0 sl_pct: float | None = None tp_pct: float | None = None trail_pct: float | None = None validation_mode: str = "walk_forward" train_months: int = 12 test_months: int = 3 roll_months: int = 3 holdout_months: int = 6 def to_config(self) -> BacktestConfig: return BacktestConfig( asset=self.asset, timeframe=self.timeframe, strategy=self.strategy, params=dict(self.params), costs=Costs(enabled=self.costs_on, commission_bps=self.commission_bps, slippage_bps=self.slippage_bps, slippage_model=self.slippage_model), sizing=Sizing(mode=self.sizing_mode, pct=self.size_pct, leverage=self.leverage, max_position=self.max_position), stops=Stops(sl_pct=self.sl_pct, tp_pct=self.tp_pct, trail_pct=self.trail_pct), validation=Validation(mode=self.validation_mode, train_months=self.train_months, test_months=self.test_months, roll_months=self.roll_months, holdout_months=self.holdout_months), ) # -- share links ------------------------------------------------------ def encode(self) -> str: raw = json.dumps(asdict(self), sort_keys=True, separators=(",", ":")) return base64.urlsafe_b64encode(raw.encode()).decode().rstrip("=") @classmethod def decode(cls, token: str) -> "RunRequest": pad = "=" * (-len(token) % 4) raw = base64.urlsafe_b64decode(token + pad).decode() data = json.loads(raw) known = {k: v for k, v in data.items() if k in cls.__dataclass_fields__} req = cls(**known) req.validate() return req def validate(self) -> None: """Reject anything a share link could smuggle in.""" if self.strategy not in strategies.PRESETS: raise ValueError(f"unknown strategy {self.strategy!r}") if self.asset not in config.ASSETS: raise ValueError(f"unknown asset {self.asset!r}") if self.timeframe not in config.TIMEFRAMES: raise ValueError(f"unknown timeframe {self.timeframe!r}") if self.validation_mode not in ("none", "split", "walk_forward", "holdout"): raise ValueError(f"unknown validation mode {self.validation_mode!r}") if self.slippage_model not in ("fixed", "volume_scaled"): raise ValueError(f"unknown slippage model {self.slippage_model!r}") if self.sizing_mode not in ("fixed_pct", "vol_target", "fixed_units"): raise ValueError(f"unknown sizing mode {self.sizing_mode!r}") if not isinstance(self.params, dict): raise ValueError("params must be an object") for k in self.params: if not isinstance(k, str) or not k.replace("_", "").isalnum(): raise ValueError(f"bad parameter name {k!r}") # -------------------------------------------------------------------------- # Execution # -------------------------------------------------------------------------- @dataclass class RunRecord: run_id: str label: str request: RunRequest result: BacktestResult created_at: str elapsed_s: float @property def sharpe(self) -> float: return self.result.metrics_oos.sharpe or self.result.metrics_all.sharpe @property def meta(self) -> str: r = self.request mode = {"walk_forward": "WF", "holdout": "HOLDOUT", "split": "SPLIT", "none": "—"}.get(r.validation_mode, r.validation_mode) return f"{r.timeframe} · {r.date_range} · {mode}" class RunError(RuntimeError): pass def window_for(asset: str, timeframe: str, date_range: str): cov = price_coverage_for(asset, timeframe) if cov is None: raise RunError( f"No cached price coverage for {asset} {timeframe}. " "Pick another pair, or extend coverage." ) start_cov, end_cov = pd.Timestamp(cov[0], tz="UTC"), pd.Timestamp(cov[1], tz="UTC") years = RANGE_YEARS.get(date_range, 3.0) start = max(start_cov, end_cov - pd.Timedelta(days=int(365 * years))) return start, end_cov def execute(req: RunRequest) -> RunRecord: """Run one backtest against cached data only. Never touches a provider.""" t0 = time.perf_counter() req.validate() preset = strategies.PRESETS[req.strategy] if not preset.available: raise RunError(f"{req.strategy}: {preset.unavailable_reason}") start, end = window_for(req.asset, req.timeframe, req.date_range) prices = load_prices(req.asset, req.timeframe, start, end) if prices.empty or len(prices) < 60: raise RunError( f"Only {len(prices)} cached bars for {req.asset} {req.timeframe} — " "not enough to backtest. Try a longer range or another timeframe." ) signals = pd.DataFrame() if preset.needs_signals: model = req.model_slug or (available_models(req.asset, req.timeframe) or [""])[0] if not model: raise RunError( f"{req.strategy} needs stored model signals, and none are cached " f"for {req.asset} {req.timeframe}. Use Extend coverage to add them." ) signals = load_signals(model, req.asset, req.timeframe, start, end) if signals.empty: raise RunError(f"No signal coverage for {model} on {req.asset} {req.timeframe}.") params = {**strategies.defaults_for(req.strategy), **(req.params or {})} out = strategies.build(req.strategy, prices, params, signals) cfg = req.to_config() result = run_backtest(prices, out, cfg, bars_per_year=config.bars_per_year(req.asset, req.timeframe)) return RunRecord( run_id=uuid.uuid4().hex[:8], label=f"{req.strategy} · {req.asset} {req.timeframe}", request=req, result=result, created_at=datetime.now(timezone.utc).isoformat(timespec="seconds"), elapsed_s=time.perf_counter() - t0, ) # -------------------------------------------------------------------------- # Robustness helpers # -------------------------------------------------------------------------- def parameter_sweep(req: RunRequest, x_key: str, x_vals: list, y_key: str, y_vals: list) -> pd.DataFrame: """Two-parameter OOS-Sharpe sweep for the sensitivity heatmap. Runs against the selection window only; the holdout is never touched, because `build_validation_plan` excludes it from every window it emits. """ rows = [] for xv in x_vals: for yv in y_vals: trial = RunRequest(**{**asdict(req), "params": {**req.params, x_key: xv, y_key: yv}}) try: rec = execute(trial) rows.append({x_key: xv, y_key: yv, "oos_sharpe": rec.result.metrics_oos.sharpe}) except Exception: rows.append({x_key: xv, y_key: yv, "oos_sharpe": float("nan")}) return pd.DataFrame(rows) def slippage_stress(req: RunRequest, bps_points=(0, 5, 10, 20)) -> list[tuple[float, float]]: out = [] for bps in bps_points: trial = RunRequest(**{**asdict(req), "costs_on": True, "slippage_bps": float(bps)}) try: rec = execute(trial) out.append((float(bps), rec.result.metrics_oos.sharpe)) except Exception: out.append((float(bps), float("nan"))) return out def regime_breakdown(rec: RunRecord) -> pd.DataFrame: """Strategy return within each market regime.""" from .charts import classify_regime res = rec.result if res.prices is None or res.prices.empty: return pd.DataFrame() reg = classify_regime(res.prices) eq = res.equity.reindex(reg.index).ffill() rets = eq.pct_change().fillna(0.0) rows = [] for name in ("bull", "bear", "chop"): mask = reg == name if not mask.any(): continue rows.append({"regime": name.upper(), rec.label[:24]: float((1 + rets[mask]).prod() - 1.0)}) return pd.DataFrame(rows) def overfit_verdict(rec: RunRecord) -> tuple[str, list[tuple[str, str]]]: """A blunt grade plus the checks behind it.""" res = rec.result checks: list[tuple[str, str]] = [] score = 0 is_s, oos_s = res.metrics_is.sharpe, res.metrics_oos.sharpe if res.metrics_oos.bars == 0: checks.append(("✗", "no out-of-sample period was produced — this result is " "entirely in-sample and cannot be trusted")) else: ratio = (oos_s / is_s) if is_s else float("nan") if pd.notna(ratio) and ratio >= 0.5: checks.append(("✓", f"OOS Sharpe holds at {ratio:.2f} of in-sample")) score += 1 else: checks.append(("✗", f"OOS Sharpe collapses to {ratio:.2f} of in-sample")) n = res.metrics_all.trade_count if n >= 30: checks.append(("✓", f"{n} trades is enough to mean something")) score += 1 else: checks.append(("✗", f"only {n} trades — the result is mostly noise")) if res.windows: pos = sum(1 for w in res.windows if w.metrics.total_return > 0) if pos >= len(res.windows) * 0.6: checks.append(("✓", f"{pos}/{len(res.windows)} walk-forward windows positive")) score += 1 else: checks.append(("✗", f"only {pos}/{len(res.windows)} windows positive")) else: checks.append(("·", "no walk-forward windows in this configuration")) if res.costs_paid > 0: checks.append(("✓", f"costs modelled: ${res.costs_paid:,.0f} paid")) score += 1 else: checks.append(("✗", "costs are off — this number is not real")) if res.metrics_holdout is not None: hs = res.metrics_holdout.sharpe if hs > 0: checks.append(("✓", f"locked holdout Sharpe {hs:.2f}")) score += 1 else: checks.append(("✗", f"locked holdout Sharpe {hs:.2f} — it fails on unseen data")) else: checks.append(("·", "no locked holdout reserved")) grade = ["FAILS", "FRAGILE", "FRAGILE", "PLAUSIBLE", "PLAUSIBLE", "SOLID"][min(score, 5)] return grade, checks def save_run_summary(rec: RunRecord, *, push: bool = False) -> str: """Write a shareable run summary into the store's runs/ folder. With `push`, the summary is committed immediately so it shows up in the global Run history for everyone rather than waiting for the next batch. """ payload = { "run_id": rec.run_id, "label": rec.label, "created_at": rec.created_at, "config": asdict(rec.request), "share_token": rec.request.encode(), "summary": rec.result.summary(), "metrics": { "all": rec.result.metrics_all.to_dict(), "is": rec.result.metrics_is.to_dict(), "oos": rec.result.metrics_oos.to_dict(), "holdout": rec.result.metrics_holdout.to_dict() if rec.result.metrics_holdout else None, }, } store = get_store() store.write_json(f"runs/{rec.run_id}.json", payload) if push: try: store.flush(f"Save run {rec.run_id}: {rec.label}") except Exception as e: log.warning("run summary staged but not pushed: %s", e) return rec.run_id