src/runtime.py

"""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