"""READ-ONLY, MEMORY-BOUNDED risk-managed BOOK evaluator on the Bybit liquid sleeves.

The standalone Bybit edges are modest (tstrend ~0.89, unlock ~0.81, xsfunding ~0.64, stablecoin thin).
The DEPLOYABLE number is the COMBINATION through the LIVE risk overlay — the same vol-target + killswitch
(+ ISV-adaptive weights + adaptive vol budget) that took the raw ~0.4-0.9 legacy-venue edges to Sharpe ~2.74.
This module computes that analog on the Bybit `bybit_features` warehouse.

HOW IT REUSES THE LIVE OVERLAY (no reimplementation):
  * `bybit_edges_eval`'s real sleeve runners give each edge's per-day RETURN SERIES
    (`sleeve_returns_from_store` below — the same `TrendRunner` / `UnlockShortRunner` /
    `StableReversionRunner` / `XsFundingReplay` engines the live sleeves wrap; xsfunding via
    `carry_metrics_from_store`'s curve).
  * `paper_sim.simulate_book` runs the book forward. It is a thin composition over
    `paper_book.IncrementalRiskOverlay`, which is golden-tested (test_incremental_overlay_matches_per_date)
    to be BIT-IDENTICAL to `paper_book.paper_risk_overlay` — the SINGLE entry point both the live backfill
    and the nightly snapshot call. So the overlay applied here is exactly the live book's overlay:
    `isv_adaptive_weights` + ISV-adaptive `adaptive_target_vol` + `vol_target_leverage` + `DrawdownKillSwitch`
    (controller="killswitch", the live default) — and `controller="combined"` additionally layers the
    per-sleeve structural-retirement gate, also the live code path. The marginal-gate
    (`marginal_gate_active`) is intentionally OFF in the live small-book overlay, so it is OFF here too —
    using the SAME overlay the live legacy combined book uses rather than a hand-tuned variant.

READ-ONLY: there is no `PaperRepo`, and no `store.write_*` / `paper_*` write is ever issued. The runners
and `simulate_book` are pure. A `_WriteTripwireStore` in the tests proves no write method is touched.

MEMORY-BOUNDED (the user explicitly wants NO more OOMs — the `--all` eval OOM'd the dagster daemon):
  * Heavy multi-symbol compute runs as a Kubernetes JOB with its own memory limit
    (`infra/k8s/jobs/fxhnt-bybit-book-eval-job.yaml`), NEVER via `kubectl exec` in the dagster daemon —
    that shared-memory exec is exactly what OOM'd the full-universe eval.
  * The evaluator operates on the LIQUID SUBSET only. `TimescaleFeatureStore.crypto_close_panel()` (and the
    funding/spot variants) SELECT the WHOLE table for a feature (all ~794 coins) with no symbol filter — a
    full-universe panel materialisation. To avoid that, when a `universe` is supplied we wrap the store in
    `_UniverseRestrictedStore`, which serves those panels via `read_panel(sorted(universe), feature)` — a
    symbol-restricted read (`WHERE symbol IN (...)`), so ONLY the liquid symbols ever load into memory. The
    `universe`-restricted read is asserted by a focused test.
"""
from __future__ import annotations

import math
import statistics as st
from typing import Any, Protocol

import numpy as np

from fxhnt.application.bybit_carry_revalidate import carry_metrics_from_store
from fxhnt.application.paper_sim import simulate_book

_PERIODS_PER_YEAR = 365

# The Bybit-computable directional + carry + positioning sleeves, in canonical book order.
# stablecoin_rotation is omitted by default (the Bybit perp warehouse carries no stablecoin SPOT panel — it
# would always be N/A); it is included only if an explicit stablecoin spot panel is injected with >= a couple
# of symbols. positioning (contrarian long/short-account-ratio) is the validated 4th edge — uncorrelated
# with the other three (Sharpe ~2.6-3.1 net); it is read from the `long_ratio` + `close` features.
_DEFAULT_BYBIT_SLEEVES = ("crypto_tstrend", "unlock", "xsfunding", "positioning")

# Human display names for the book breakdown, INDEPENDENT of STRATEGY_REGISTRY. The 4 book sleeves live in a
# SEPARATE namespace from the forward-track strategy_id registry — a sleeve string id happening to match a
# (possibly retired) standalone registry entry must never blank the book breakdown's own label. This bit the
# `crypto_tstrend` sleeve when its standalone forward-track entry was retired (Phase 0b Task 4): the
# breakdown's display-name lookup fell back to the raw sleeve id. unlock/xsfunding/positioning mirror their
# current registry `display_name` (both namespaces happen to use the same human copy today; they are free to
# diverge without either breaking).
_SLEEVE_DISPLAY_NAMES = {
    "crypto_tstrend": "Crypto trend (long/flat, TS-momentum)",
    "unlock": "Token-unlock dilution shorts (market-neutral)",
    "xsfunding": "Crypto funding harvest (cross-sectional, delta-neutral)",
    "positioning": "Positioning (contrarian long/short ratio)",
}

# Map a sleeve name (paper_sleeves vocabulary) to the edge name the runners use.
_SLEEVE_TO_EDGE = {
    "crypto_tstrend": "tstrend",
    "unlock": "unlock",
    "stablecoin_rotation": "stablecoin_rotation",
    "xsfunding": "xsfunding",
    "positioning": "positioning",
}


class _FeatureStore(Protocol):
    """Just the panel reads the edge engines need (TimescaleFeatureStore satisfies this). READ-ONLY."""

    def crypto_close_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]: ...
    def crypto_funding_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]: ...
    def crypto_spot_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]: ...
    def read_panel(self, symbols: list[str], feature: str) -> dict[str, dict[int, float]]: ...


class _UniverseRestrictedStore:
    """READ-ONLY proxy that makes the full-table panel reads MEMORY-BOUNDED.

    `crypto_close_panel` / `crypto_funding_panel` / `crypto_spot_panel` on the real store SELECT the whole
    table for a feature (no symbol filter) → all ~794 coins materialise. This proxy serves those three
    panels via `read_panel(sorted(universe), feature)` instead — a `WHERE symbol IN (...)` read that loads
    ONLY the liquid `universe` symbols. Every OTHER attribute (incl. `read_panel` itself, used by the carry
    pipeline's `worst_basis`/`turnover` reads) is forwarded UNCHANGED to the inner store. No write method is
    forwarded that the evaluator uses; the inner store is never mutated by this class."""

    # close/funding/spot panel-method name -> the warehouse feature it reads.
    _PANEL_FEATURE = {
        "crypto_close_panel": "close",
        "crypto_funding_panel": "funding",
        "crypto_spot_panel": "spot_close",
    }

    def __init__(self, inner: _FeatureStore, universe: set[str]) -> None:
        object.__setattr__(self, "_inner", inner)
        # Freeze the symbol list ONCE (sorted, deterministic) so every restricted read uses the same set.
        object.__setattr__(self, "_symbols", sorted(universe))

    def _restricted_panel(self, feature: str, *, suffix: str = "USDT") -> dict[str, dict[int, float]]:
        # Only the liquid symbols with the requested suffix are read — bounded by |universe|, not the table.
        syms = [s for s in self._symbols if s.endswith(suffix)]
        if not syms:
            return {}
        panel = self._inner.read_panel(syms, feature)
        # read_panel pre-seeds every requested symbol with an empty dict; drop the ones with no rows so the
        # downstream runners see only symbols that actually have data (same shape as _feature_panel).
        return {s: v for s, v in panel.items() if v}

    def crypto_close_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]:
        return self._restricted_panel("close", suffix=suffix)

    def crypto_funding_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]:
        return self._restricted_panel("funding", suffix=suffix)

    def crypto_spot_panel(self, *, suffix: str = "USDT") -> dict[str, dict[int, float]]:
        return self._restricted_panel("spot_close", suffix=suffix)

    def __getattr__(self, name: str) -> Any:
        # read_panel, close(), and anything else fall through to the inner store unchanged.
        return getattr(self._inner, name)


def _bounded_store(store: _FeatureStore, universe: set[str] | None) -> _FeatureStore:
    """Wrap `store` so its close/funding/spot panel reads are restricted to `universe` (the memory-bound
    seam). `universe is None` → the raw store (full universe; the documented OOM-risk path, used only when
    the caller explicitly asks for --all). An EMPTY universe still wraps (→ empty panels), not the raw
    store, so `--liquid` with no qualifying symbol does not silently fall back to the full table."""
    if universe is None:
        return store
    return _UniverseRestrictedStore(store, universe)


def _series_from_run(dates: list[int], returns: np.ndarray) -> dict[int, float]:
    """A {epoch_day: ret} series from a runner's parallel (dates, returns). Last-write-wins on a dup day
    (runners emit one return per realized day, so there are none in practice). Non-finite returns dropped."""
    out: dict[int, float] = {}
    for d, r in zip(dates, returns, strict=True):
        rv = float(r)
        if math.isfinite(rv):
            out[int(d)] = rv
    return out


def _tstrend_series(store: _FeatureStore, *, cost_bps: float) -> dict[int, float]:
    """tstrend per-day return series off the (already universe-bounded) store's close panel — the live
    sleeve's EXACT engine `TrendRunner(panel, long_short=False).run()`."""
    from fxhnt.application.trend_runner import TrendRunner

    panel = store.crypto_close_panel()
    if not panel:
        return {}
    # charge the perp funding a LONG trend book pays (longs pay positive funding) — the honest deploy cost.
    res = TrendRunner(panel, long_short=False, cost_bps=cost_bps,
                      funding=store.crypto_funding_panel()).run()
    return _series_from_run(res.dates, res.returns)


def _unlock_series(store: _FeatureStore, *, cost_bps: float,
                   unlock_events: list[dict[str, Any]] | None) -> dict[int, float]:
    """unlock per-day return series via the live `UnlockShortRunner` over the store's close panel. Needs the
    INJECTED DefiLlama-style calendar; empty/absent calendar → empty series (the edge is uncomputable)."""
    from fxhnt.application.unlock_runner import UnlockShortRunner, warehouse_unlock_panel

    if not unlock_events:
        return {}
    panel = store.crypto_close_panel()
    if not panel:
        return {}
    res = UnlockShortRunner(warehouse_unlock_panel(panel), unlock_events).run()
    return _series_from_run(res.dates, res.returns)


def _stablecoin_series(*, stablecoin_spot_panel: dict[str, dict[int, float]] | None) -> dict[int, float]:
    """stablecoin per-day return series via the live `StableReversionRunner` over an INJECTED stablecoin
    SPOT panel (the Bybit perp warehouse carries none). Absent panel → empty series."""
    from fxhnt.application.stablecoin_runner import StableReversionRunner

    if not stablecoin_spot_panel:
        return {}
    res = StableReversionRunner(stablecoin_spot_panel).run()
    return _series_from_run(res.dates, res.returns)


def _xsfunding_series(store: _FeatureStore, *, cost_bps: float) -> dict[int, float]:
    """xsfunding (carry) per-day return series. Reuses the live carry pipeline by driving
    `carry_metrics_from_store` with a CAPTURING `_curve_metrics` so the per-day (epoch_day, ret) the carry
    curve already builds is recovered WITHOUT reimplementing XsFundingReplay/basis accounting. The store is
    already universe-bounded by the caller, so this is the liquid carry series."""
    import fxhnt.application.bybit_carry_revalidate as carry_mod

    captured: dict[int, float] = {}
    orig = carry_mod._curve_metrics

    def _capture(daily: list[tuple[int, float]], *, days_meta: dict[str, Any]) -> dict[str, Any]:
        captured.clear()
        for d, r in daily:
            rv = float(r)
            if math.isfinite(rv):
                captured[int(d)] = rv
        return orig(daily, days_meta=days_meta)

    carry_mod._curve_metrics = _capture
    try:
        # universe=None: the store passed in is ALREADY the universe-restricted proxy, so its panels only
        # carry the liquid symbols — restricting again here would be a redundant set-intersect over an
        # already-bounded panel. cost_bps is the per-turnover carry haircut (same as the edge evaluator).
        carry_metrics_from_store(store, cost_bps=cost_bps, universe=None)
    finally:
        carry_mod._curve_metrics = orig
    return dict(captured)


def _positioning_series(store: _FeatureStore, *, universe: set[str] | None, cost_bps: float,
                        direction: str = "contrarian", capacity_capital: float | None = None,
                        tail_cap_k: float | None = None, coin_gross_cap: float | None = None) -> dict[int, float]:
    """positioning (contrarian long/short account-ratio) per-day return series via the live
    `bybit_positioning_eval.positioning_returns_from_store`. It reads the `long_ratio` + `close` features
    through the SAME `_UniverseRestrictedStore` seam (a `WHERE symbol IN (...)` read), so it is memory-bounded
    on the liquid `universe`. Absent `long_ratio` data → empty series (the edge is uncomputable). The import
    is local to avoid a circular import (`bybit_positioning_eval` imports from this module).

    `capacity_capital` / `tail_cap_k` (both None = off) enable the gains-only capacity + tail-concentration
    haircuts that make positioning's booked return capturable + robust (see `_book_breakdown`).
    `coin_gross_cap` (None = off) caps the per-coin weight to improve concentration robustness."""
    from fxhnt.application.bybit_positioning_eval import positioning_returns_from_store

    return positioning_returns_from_store(store, universe=universe, cost_bps=cost_bps, direction=direction,
                                          capacity_capital=capacity_capital, tail_cap_k=tail_cap_k,
                                          coin_gross_cap=coin_gross_cap)


def sleeve_returns_from_store(store: _FeatureStore, edge: str, *, universe: set[str] | None = None,
                              cost_bps: float = 0.0,
                              unlock_events: list[dict[str, Any]] | None = None,
                              stablecoin_spot_panel: dict[str, dict[int, float]] | None = None,
                              capacity_capital: float | None = None,
                              tail_cap_k: float | None = None,
                              coin_gross_cap: float | None = None,
                              ) -> dict[int, float]:
    """READ-ONLY, memory-bounded per-day RETURN SERIES `{epoch_day: ret}` for ONE edge on `store`.

    The book-level analog of `bybit_edges_eval.edge_metrics_from_store`: instead of the SUMMARY metrics it
    returns the raw daily series `simulate_book` needs to run the overlay over. It reuses the SAME real
    sleeve runners (no edge logic reimplemented). `universe` restricts the crypto symbol set AT THE READ
    (via `_UniverseRestrictedStore`) so only liquid symbols load — the memory bound. External inputs (unlock
    calendar, stablecoin spot panel) are INJECTED so this is pure + network-free; a genuinely-uncomputable
    edge (missing a required input / no data) returns an EMPTY series rather than fabricating one.

    `edge` accepts the edge name (tstrend/unlock/stablecoin_rotation/xsfunding) OR the sleeve name
    (crypto_tstrend/...). Writes NOTHING."""
    edge = _SLEEVE_TO_EDGE.get(edge, edge)
    bounded = _bounded_store(store, universe)
    if edge == "tstrend":
        return _tstrend_series(bounded, cost_bps=cost_bps)
    if edge == "unlock":
        return _unlock_series(bounded, cost_bps=cost_bps, unlock_events=unlock_events)
    if edge == "stablecoin_rotation":
        return _stablecoin_series(stablecoin_spot_panel=stablecoin_spot_panel)
    if edge == "xsfunding":
        return _xsfunding_series(bounded, cost_bps=cost_bps)
    if edge == "positioning":
        # positioning_returns_from_store does its OWN _bounded_store(universe) wrap (it also reads the
        # `long_ratio` panel via read_panel), so pass the RAW store + universe — not the already-bounded
        # proxy — to avoid double-wrapping. capacity_capital/tail_cap_k thread the positioning haircuts
        # (ignored by the other sleeves, which have their own constructions). coin_gross_cap threads the
        # per-coin weight cap (also positioning-only).
        return _positioning_series(store, universe=universe, cost_bps=cost_bps,
                                   capacity_capital=capacity_capital, tail_cap_k=tail_cap_k,
                                   coin_gross_cap=coin_gross_cap)
    raise ValueError(
        f"unknown edge {edge!r} (known: tstrend, unlock, stablecoin_rotation, xsfunding, positioning)")


def _pairwise_correlation(a_series: dict[int, float], b_series: dict[int, float]) -> float | None:
    """Pearson correlation of two daily-return series over their COMMON days; None if <2 common days or
    either has zero variance. Pure — the diversification visibility for the book report."""
    common = sorted(set(a_series) & set(b_series))
    if len(common) < 2:
        return None
    a = [a_series[d] for d in common]
    b = [b_series[d] for d in common]
    sa, sb = st.pstdev(a), st.pstdev(b)
    if sa <= 0.0 or sb <= 0.0:
        return None
    ma, mb = st.mean(a), st.mean(b)
    cov = sum((x - ma) * (y - mb) for x, y in zip(a, b, strict=True)) / len(common)
    return cov / (sa * sb)


def _correlation_matrix(returns_by_sleeve: dict[str, dict[int, float]],
                        sleeves: list[str]) -> dict[str, dict[str, float | None]]:
    """`{sleeve_i: {sleeve_j: corr|None}}` over each pair's common days (diagonal == 1.0). Makes the
    diversification across the 4 sleeves visible in the book report."""
    out: dict[str, dict[str, float | None]] = {}
    for si in sleeves:
        row: dict[str, float | None] = {}
        for sj in sleeves:
            if si == sj:
                row[sj] = 1.0
            else:
                row[sj] = _pairwise_correlation(returns_by_sleeve[si], returns_by_sleeve[sj])
        out[si] = row
    return out


def _standalone_sharpe(series: dict[int, float]) -> float:
    """Annualised standalone Sharpe of a daily return series (mean/pstdev × √365). <2 obs / zero vol → 0."""
    rets = [series[d] for d in sorted(series)]
    if len(rets) < 2:
        return 0.0
    sd = st.pstdev(rets)
    return (st.mean(rets) / sd * math.sqrt(_PERIODS_PER_YEAR)) if sd > 0 else 0.0


# The single documented annual vol target used to make standalone sleeves COMPARABLE in the cockpit. Raw
# cumulative returns are misleading (unsized, and different sleeves have different history lengths); scaling
# each sleeve's daily series to the SAME annual vol then annualising gives an apples-to-apples number
# (≈ Sharpe × target) and a maxDD on a consistent risk basis. The live book vol-targets the blend, not the
# individual sleeves — this is a standalone reference only.
_SLEEVE_VOL_TARGET = 0.20


def vol_targeted_metrics(series: dict[int, float], *,
                         vol_target: float = _SLEEVE_VOL_TARGET) -> dict[str, float]:
    """Scale a daily return series to a common ANNUAL vol target, then report comparable standalone metrics:
    `{ann_return, max_dd, sharpe, vol_target}`. The scale factor is `target_daily_vol / realized_daily_vol`
    (target_daily_vol = vol_target / √365); applied to every daily return so the scaled series has annual vol
    ≈ vol_target. `ann_return` is the annualised (geometric, per-trading-day-span) return of the scaled
    series, which ≈ Sharpe × vol_target; `max_dd` is the running peak-to-trough of the COMPOUNDED scaled
    series (consistent basis — NOT the raw maxDD). <2 obs / zero vol → all zeros. Pure, deterministic."""
    rets = [series[d] for d in sorted(series)]
    n = len(rets)
    zero = {"ann_return": 0.0, "max_dd": 0.0, "sharpe": 0.0, "vol_target": vol_target}
    if n < 2:
        return zero
    sd = st.pstdev(rets)
    if sd <= 0.0:
        return zero
    scale = (vol_target / math.sqrt(_PERIODS_PER_YEAR)) / sd
    scaled = [r * scale for r in rets]
    equity = 1.0
    peak = 1.0
    max_dd = 0.0
    for r in scaled:
        equity *= (1.0 + r)
        peak = max(peak, equity)
        if peak > 0:
            max_dd = min(max_dd, equity / peak - 1.0)
    days = sorted(series)
    span = days[-1] - days[0]
    years = span / _PERIODS_PER_YEAR if span > 0 else 0.0
    ann_return = ((equity ** (1.0 / years) - 1.0) if years > 0 and equity > 0 else 0.0)
    sharpe = _standalone_sharpe(series)   # scale-invariant — same as the raw series
    return {"ann_return": ann_return, "max_dd": max_dd, "sharpe": sharpe, "vol_target": vol_target}


def _equal_weight_metrics(returns_by_sleeve: dict[str, dict[int, float]], sleeves: list[str], *,
                          capital: float, cost_bps: float) -> dict[str, Any]:
    """Naive EQUAL-WEIGHT (no overlay) book metrics — the un-risk-managed baseline the overlay is compared
    against. Same `simulate_book` engine but with the overlay neutralised: equal static weights, unit
    leverage, killswitch never arms. Implemented by feeding `simulate_book` the SAME returns under
    `kelly_fraction` so small that vol-target leverage clamps to its unlevered floor is NOT how we do it —
    instead we compute the simple Σ(1/n)·r curve directly so the baseline is a true equal-weight, not an
    overlay artifact."""
    present = [s for s in sleeves if returns_by_sleeve.get(s)]
    if not present:
        return {"cagr": 0.0, "sharpe": 0.0, "max_dd": 0.0, "total_return": 0.0, "days": 0}
    n = len(present)
    all_days = sorted({d for s in present for d in returns_by_sleeve[s]})
    daily: list[tuple[int, float]] = []
    prev_w = {s: 0.0 for s in present}
    for d in all_days:
        # Equal weight across the sleeves that HAVE a return today (re-normalised over present-today).
        today = [s for s in present if d in returns_by_sleeve[s]]
        if not today:
            continue
        w = 1.0 / n   # static equal weight over the full present set (absent-today sleeves contribute 0)
        gross = sum(w * returns_by_sleeve[s][d] for s in today)
        if cost_bps:
            cur_w = {s: (w if d in returns_by_sleeve[s] else 0.0) for s in present}
            turnover = sum(abs(cur_w[s] - prev_w[s]) for s in present)
            gross -= turnover * cost_bps / 1e4
            prev_w = cur_w
        daily.append((d, gross))

    equity = capital
    peak = capital
    max_dd = 0.0
    rets = [r for _, r in daily]
    for r in rets:
        equity *= (1.0 + r)
        peak = max(peak, equity)
        if peak > 0:
            max_dd = min(max_dd, equity / peak - 1.0)
    total_return = equity / capital - 1.0 if capital else 0.0
    span = (daily[-1][0] - daily[0][0]) if len(daily) >= 2 else 0
    years = span / _PERIODS_PER_YEAR if span > 0 else 0.0
    cagr = ((equity / capital) ** (1.0 / years) - 1.0
            if years > 0 and capital > 0 and equity > 0 else 0.0)
    sharpe = 0.0
    if len(rets) >= 2:
        sd = st.pstdev(rets)
        sharpe = (st.mean(rets) / sd * math.sqrt(_PERIODS_PER_YEAR)) if sd > 0 else 0.0
    return {"cagr": cagr, "sharpe": sharpe, "max_dd": max_dd, "total_return": total_return,
            "days": len(rets)}


def risk_managed_book_from_store(store: _FeatureStore, *, universe: set[str] | None = None,
                                 sleeves: list[str] | None = None, capital: float = 100_000.0,
                                 cost_bps: float = 5.5, controller: str = "killswitch",
                                 kelly_fraction: float = 1.0,
                                 unlock_events: list[dict[str, Any]] | None = None,
                                 stablecoin_spot_panel: dict[str, dict[int, float]] | None = None,
                                 ) -> dict[str, Any]:
    """READ-ONLY, memory-bounded RISK-MANAGED book on the Bybit liquid sleeves — the deployable analog of
    the Sharpe ~2.74 legacy-venue number.

    Pipeline:
      1. For each available Bybit sleeve, `sleeve_returns_from_store` gives its per-day return series
         (universe-restricted READS → memory-bounded; same live runners as the edge evaluator).
      2. `simulate_book(returns_by_sleeve, ...)` runs the LIVE risk overlay forward — `IncrementalRiskOverlay`
         == `paper_risk_overlay` (ISV-adaptive weights + adaptive vol-target + DrawdownKillSwitch;
         controller="combined" adds the structural-retirement gate) — the exact overlay the live book uses.
      3. The result is compared to a naive EQUAL-WEIGHT (no overlay) baseline so the overlay's contribution
         (leverage + kill + ISV) is visible.

    The default sleeve set is the 4 Bybit-computable edges (tstrend, unlock, xsfunding, positioning); any
    whose data is absent (no required input / <2 days) is skipped gracefully, so the book is whatever subset
    actually has data.

    Returns a dict: `cagr, sharpe, max_dd, total_return, days, leverage_stats (avg/min/max), killed_days,
    per_sleeve_weights (final-day post-overlay), per_sleeve_standalone_sharpe, sleeves (the active set),
    correlation_matrix ({sleeve_i: {sleeve_j: corr|None}}, diagonal 1.0 — the diversification view),
    naive_equal_weight (the baseline metrics), capital, controller, kelly_fraction`. Writes NOTHING — no
    PaperRepo, no store.write_*; the live paper_* tables are untouched."""
    candidate = list(sleeves) if sleeves is not None else list(_DEFAULT_BYBIT_SLEEVES)

    returns_by_sleeve: dict[str, dict[int, float]] = {}
    standalone: dict[str, float] = {}
    for sleeve in candidate:
        series = sleeve_returns_from_store(
            store, sleeve, universe=universe, cost_bps=cost_bps,
            unlock_events=unlock_events, stablecoin_spot_panel=stablecoin_spot_panel)
        if len(series) >= 2:   # need >= 2 days to be a real sleeve in the book
            returns_by_sleeve[sleeve] = series
            standalone[sleeve] = _standalone_sharpe(series)

    active = [s for s in candidate if s in returns_by_sleeve]
    if not active:
        return {"cagr": 0.0, "sharpe": 0.0, "max_dd": 0.0, "total_return": 0.0, "days": 0,
                "leverage_stats": {"avg": 0.0, "min": 0.0, "max": 0.0}, "killed_days": 0,
                "per_sleeve_weights": {}, "per_sleeve_standalone_sharpe": {}, "sleeves": [],
                "correlation_matrix": {},
                "naive_equal_weight": {"cagr": 0.0, "sharpe": 0.0, "max_dd": 0.0,
                                       "total_return": 0.0, "days": 0},
                "capital": capital, "controller": controller, "kelly_fraction": kelly_fraction,
                "reason": "no Bybit sleeve has >= 2 days of returns (run the Phase-A/B ingests first)"}

    # THE LIVE OVERLAY — simulate_book == IncrementalRiskOverlay == paper_risk_overlay (golden-tested).
    res = simulate_book(returns_by_sleeve, capital=capital, sleeves=active,
                        controller=controller, kelly_fraction=kelly_fraction, cost_bps=cost_bps)

    lev = res.leverage
    leverage_stats = {
        "avg": (sum(lev) / len(lev)) if lev else 0.0,
        "min": min(lev) if lev else 0.0,
        "max": max(lev) if lev else 0.0,
    }
    naive = _equal_weight_metrics(returns_by_sleeve, active, capital=capital, cost_bps=cost_bps)
    corr_matrix = _correlation_matrix(returns_by_sleeve, active)

    return {
        "cagr": res.metrics["cagr"],
        "sharpe": res.metrics["sharpe"],
        "max_dd": res.metrics["max_dd"],
        "total_return": res.metrics["total_return"],
        "days": len(res.dates),
        "leverage_stats": leverage_stats,
        "killed_days": sum(1 for k in res.killed if k),
        "per_sleeve_weights": dict(res.weights[-1]) if res.weights else {},
        "per_sleeve_standalone_sharpe": standalone,
        "sleeves": active,
        "correlation_matrix": corr_matrix,
        "naive_equal_weight": naive,
        "capital": capital,
        "controller": controller,
        "kelly_fraction": kelly_fraction,
    }
