"""OOS-disciplined SLEEVE-WEIGHTING A/B for the Bybit 4-edge book — READ-ONLY, memory-bounded.

The 4 sleeves (crypto_tstrend, unlock, xsfunding, positioning) have very different vols, so EQUAL CAPITAL
!= EQUAL RISK. Full-sample (look-ahead): xsfunding is Sharpe ~0.64 but ~78% vol → ~66% of book risk under
naive 1/n, while positioning (Sharpe ~3.14) carries only ~13% of the risk. Naive eq-wt book Sharpe ~1.82;
a LOOK-AHEAD inverse-vol book ~2.18. So IN-SAMPLE risk-weighting helps by reining in xsfunding. The ONLY
question this harness answers is whether a CAUSAL risk-weighting scheme survives that edge OUT-OF-SAMPLE,
NET of the extra turnover it costs. A clean "naive still wins" is a valid, valuable verdict — the 2.18 is
look-ahead-optimistic, and our memory (A3g/A3h, bybit_overlay_ab) warns causal trailing-vol estimates are
noisy/lagged and that the LIVE legacy-venue-tuned leverage overlay already LOST OOS on this book.

This harness isolates ONLY the SLEEVE-WEIGHTING question — there is NO leverage / kelly / vol-target /
killswitch here (those already lost in `bybit_overlay_ab`). The schemes (all weights for day t use ONLY
returns STRICTLY BEFORE t — no look-ahead):
  * naive       — static 1/n equal weight (the current book). Its metrics come from the EXACT
                  `_equal_weight_metrics` `risk_managed_book_from_store` uses, so the baseline is
                  bit-identical to the deployed book's naive number.
  * slow_rp     — inverse-vol from a LONG/SLOW causal trailing vol (lookback picked on TRAIN from
                  `_SLOW_RP_LOOKBACKS` = 90d / 180d). w_s ∝ 1/trailing_vol_s, renormalized over the sleeves
                  present that day. Deliberately slow to avoid the jumpy-trailing-vol noise that killed the
                  adaptive overlays.
  * risk_capped — start from 1/n, but CAP any single sleeve's RISK contribution at a ceiling (picked on
                  TRAIN from `_RISK_CAPS` = 30% / 40%), using causal trailing vol to estimate each sleeve's
                  risk share (w_s·vol_s / Σ w·vol); iteratively shrink an over-cap sleeve's weight and
                  redistribute to the others, renormalize. The minimal, hardest-to-overfit fix — it only
                  reins in the dominant sleeve (xsfunding) without a full optimization.

OOS DISCIPLINE: each scheme's hyperparam (lookback / cap level) is chosen on the TRAIN split ONLY, then the
chosen variant is scored on the held-out TEST. A scheme is recommended over naive ONLY IF it beats naive's
TEST/OOS Sharpe AND is positive in > `min_window_frac` of consecutive non-overlapping rolling windows.
Otherwise STAY NAIVE.

NET OF COST: risk-weighting/capping adds turnover (weights drift with vol). Each scheme is charged the same
per-turnover cost as the rest of the book; per-scheme turnover is reported.

REUSE: the inverse-vol primitive is `book_allocator._inverse_vol_weights` (the live allocator). The naive
baseline is `bybit_book_eval._equal_weight_metrics`. The store assembly / universe-restriction is
`bybit_book_eval.sleeve_returns_from_store` (the same `_UniverseRestrictedStore` seam). The causal forward
loop mirrors `bybit_overlay_ab._overlay_curve` / `paper_book_returns` (weights for day d from returns
strictly before d). READ-ONLY: no PaperRepo, no store.write_* (WriteTripwire test proves it). MEMORY-BOUNDED:
a `universe` restricts the panel reads to the liquid subset (FullPanel tripwire test proves it).
"""
from __future__ import annotations

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

import numpy as np

from fxhnt.application.book_allocator import _inverse_vol_weights
from fxhnt.application.bybit_book_eval import (
    _DEFAULT_BYBIT_SLEEVES,
    _equal_weight_metrics,
    sleeve_returns_from_store,
)

_PERIODS_PER_YEAR = 365

# Deliberately SLOW lookbacks for slow_rp (avoid jumpy-trailing-vol noise). Picked on TRAIN.
_SLOW_RP_LOOKBACKS: tuple[int, ...] = (90, 180)
# Risk-share ceilings for risk_capped (cap the dominant sleeve's risk contribution). Picked on TRAIN.
_RISK_CAPS: tuple[float, ...] = (0.40, 0.30)

# A scheme must be positive in > this fraction of non-overlapping rolling windows to be adopted over naive.
_MIN_WINDOW_FRAC = 0.60
_ROLLING_WINDOW_DAYS = 60

SCHEME_NAMES: tuple[str, ...] = ("naive", "slow_rp", "risk_capped")


class _FeatureStore(Protocol):
    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]]: ...


# --- causal trailing vol + weighting primitives -------------------------------------------------

def _trailing_vol(before: dict[int, float], *, lookback: int) -> float:
    """Std of a sleeve's last `lookback` STRICTLY-PRIOR daily returns (the causal vol estimate). <2 obs → 0."""
    vals = [before[d] for d in sorted(before)][-lookback:]
    return float(np.std(vals)) if len(vals) > 1 else 0.0


def _slow_rp_weights(before: dict[str, dict[int, float]], active: list[str], *,
                     lookback: int, cap: float = 1.0) -> dict[str, float]:
    """Inverse-vol (risk-parity-lite) weights from each sleeve's SLOW causal trailing vol — reuses the live
    `book_allocator._inverse_vol_weights` verbatim (with `max_sleeve_weight=cap`). Weights for day t depend
    only on returns STRICTLY BEFORE t (the `before` window the caller passes). Cold-start (no usable vol
    signal) → equal weight (the primitive's own fallback). The `cap` arg is the inverse-vol WEIGHT cap (1.0 =
    none); risk_capped does the RISK-share capping instead."""
    if not active:
        return {}
    window: dict[str, np.ndarray] = {}
    for s in active:
        vals = [before[s][d] for d in sorted(before.get(s, {}))][-lookback:]
        window[s] = np.asarray(vals, dtype=float)
    return _inverse_vol_weights(window, cap)


def _causal_realized_risk_share(weights: dict[str, float], before: dict[str, dict[int, float]],
                                active: list[str], *, lookback: int) -> dict[str, float]:
    """Each sleeve's RISK contribution share `w_s·vol_s / Σ w·vol` using the SLOW causal trailing vol — the
    quantity risk_capped bounds and the report surfaces. Σ shares == 1 (or all 0 if no vol signal)."""
    risk = {s: weights.get(s, 0.0) * _trailing_vol(before.get(s, {}), lookback=lookback) for s in active}
    total = sum(risk.values())
    if total <= 0.0:
        return {s: 0.0 for s in active}
    return {s: risk[s] / total for s in active}


def _risk_capped_weights(before: dict[str, dict[int, float]], active: list[str], *,
                         lookback: int, cap: float) -> dict[str, float]:
    """Start from 1/n equal weight, then bound the over-cap sleeves' CAUSAL RISK share (w·vol / Σ w·vol) at
    `cap`, redistributing the freed risk budget to the under-cap sleeves in proportion to their NATURAL
    (equal-capital) risk, then converting risk targets back to capital weights (w_s ∝ risk_s / vol_s) and
    renormalizing to sum 1.

    Causal: vols come from the strictly-prior `before` window. Cold start (no usable trailing vol) → equal
    weight. Σ weights == 1 over `active`. This is the minimal, hardest-to-overfit fix — it reins in ONLY the
    sleeve(s) whose risk share blows past `cap` (in practice the high-vol xsfunding) and leaves the rest at
    their natural equal-capital proportions.

    FEASIBILITY NOTE: with the full risk budget summing to 1, at most `floor(1/cap)` sleeves can sit AT the
    cap; a sleeve set too small to absorb the redistribution (e.g. a 2-sleeve book with cap 0.40, where the
    non-dominant sleeve must take 0.60) cannot satisfy the cap for every sleeve. We greedily pin only the
    over-cap sleeves whose NATURAL risk share exceeds `cap`, redistributing to the others — so the originally
    DOMINANT sleeve is always reduced toward `cap`, even when the remainder unavoidably leaves a single
    other sleeve above `cap` (no further sleeve is left to absorb it)."""
    if not active:
        return {}
    n = len(active)
    w = {s: 1.0 / n for s in active}
    vol = {s: _trailing_vol(before.get(s, {}), lookback=lookback) for s in active}
    if sum(vol.values()) <= 0.0:   # no causal vol signal yet → equal weight
        return w

    # The NATURAL (equal-capital) risk shares — risk ∝ vol when capital is equal — set the redistribution
    # proportions among the free sleeves (they keep their natural relative risk).
    natural_share = _causal_realized_risk_share({s: 1.0 / n for s in active}, before, active,
                                                lookback=lookback)

    # Iterative water-fill on RISK shares: pin the largest over-cap sleeve at `cap`, give the free sleeves the
    # remaining risk budget pro-rata by natural risk, and repeat until no free sleeve exceeds the cap OR
    # pinning another would leave no free sleeve (infeasible on a small book — accept best effort, which still
    # reduces the originally-dominant sleeve toward `cap`). At most n-1 pins.
    pinned: set[str] = set()
    target_risk: dict[str, float] = {s: natural_share[s] for s in active}
    for _ in range(n):
        free = [s for s in active if s not in pinned]
        if len(free) <= 1:
            break   # cannot redistribute further — the last free sleeve absorbs whatever remains
        over_free = sorted((s for s in free if target_risk[s] > cap + 1e-9 and vol[s] > 0.0),
                           key=lambda s: target_risk[s], reverse=True)
        if not over_free:
            break
        pinned.add(over_free[0])
        target_risk[over_free[0]] = cap
        free = [s for s in active if s not in pinned]
        remaining_risk = max(0.0, 1.0 - cap * len(pinned))
        free_natural_total = sum(natural_share[s] for s in free)
        for s in free:
            target_risk[s] = (remaining_risk * (natural_share[s] / free_natural_total)
                              if free_natural_total > 0 else remaining_risk / len(free))

    if not pinned:
        return w   # nothing breached the cap → stay equal weight (the minimal-fix intent)

    # Convert the per-sleeve target RISK shares back to capital weights (w_s ∝ risk_s / vol_s), renormalize.
    raw = {s: (target_risk[s] / vol[s]) if vol[s] > 0 else 0.0 for s in active}
    tot = sum(raw.values())
    return {s: (raw[s] / tot if tot > 0 else 1.0 / n) for s in active}


# --- causal forward loop: one scheme's net daily return series ----------------------------------

def _scheme_curve(returns_by_sleeve: dict[str, dict[int, float]], sleeves: list[str], *,
                  scheme: str, lookback: int, cap: float, cost_bps: float) -> tuple[dict[int, float], float]:
    """One scheme's CAUSAL daily NET return series `{epoch_day: ret}` AND its total turnover, over the union
    date axis. Mirrors `bybit_overlay_ab._overlay_curve` / `paper_book_returns`: for each day d the weights
    are computed from returns STRICTLY BEFORE d (no look-ahead), the gross return is Σ w·r_d, and the
    per-turnover cost (|Δw| vs the prior day) is charged. NO leverage. Pure.

    Returns (series, turnover_sum). `naive` short-circuits to static 1/n (re-normalized over sleeves present
    each day) — bit-identical in spirit to `_equal_weight_metrics` (used directly for the reported naive
    metrics)."""
    selected = [s for s in sleeves if returns_by_sleeve.get(s)]
    if not selected:
        return {}, 0.0

    all_days = sorted({d for s in selected for d in returns_by_sleeve[s]})
    before: dict[str, dict[int, float]] = {s: {} for s in selected}
    out: dict[int, float] = {}
    prev_w: dict[str, float] = {}
    turnover_sum = 0.0
    n = len(selected)

    for d in all_days:
        active = [s for s in selected if before[s]]
        if not active:
            # cold start: no prior history → static equal weight over sleeves PRESENT today.
            active = selected
            weights = {s: 1.0 / n for s in active}
        elif scheme == "naive":
            weights = {s: 1.0 / n for s in selected}   # static 1/n over the FULL present set
        elif scheme == "slow_rp":
            weights = _slow_rp_weights(before, active, lookback=lookback, cap=1.0)
        elif scheme == "risk_capped":
            weights = _risk_capped_weights(before, active, lookback=lookback, cap=cap)
        else:
            raise ValueError(f"unknown scheme {scheme!r} (known: {SCHEME_NAMES})")

        # only sleeves with a return today contribute; absent-today sleeves contribute 0 (their weight is
        # still 'committed' for turnover accounting so a sleeve dropping out books its exit turnover).
        gross = sum(weights.get(s, 0.0) * returns_by_sleeve[s].get(d, 0.0) for s in weights)
        today_w = {s: (weights.get(s, 0.0) if d in returns_by_sleeve[s] else 0.0) for s in selected}
        turnover = sum(abs(today_w.get(s, 0.0) - prev_w.get(s, 0.0))
                       for s in set(today_w) | set(prev_w))
        turnover_sum += turnover
        if cost_bps:
            gross -= turnover * cost_bps / 1e4
        prev_w = today_w

        if math.isfinite(gross):
            out[d] = gross

        for s in selected:
            if d in returns_by_sleeve[s]:
                before[s][d] = returns_by_sleeve[s][d]
    return out, turnover_sum


# --- metrics + split + rolling windows ---------------------------------------------------------

def _metrics(series: dict[int, float]) -> dict[str, Any]:
    days = sorted(series)
    rets = [series[d] for d in days]
    if len(rets) < 2:
        return {"sharpe": 0.0, "cagr": 0.0, "max_dd": 0.0, "vol": 0.0, "total_return": 0.0,
                "days": len(rets)}
    sd = st.pstdev(rets)
    sharpe = (st.mean(rets) / sd * math.sqrt(_PERIODS_PER_YEAR)) if sd > 0 else 0.0
    vol = sd * math.sqrt(_PERIODS_PER_YEAR)
    equity = 1.0
    peak = 1.0
    max_dd = 0.0
    for r in rets:
        equity *= (1.0 + r)
        peak = max(peak, equity)
        if peak > 0:
            max_dd = min(max_dd, equity / peak - 1.0)
    span = days[-1] - days[0]
    years = span / _PERIODS_PER_YEAR if span > 0 else 0.0
    cagr = (equity ** (1.0 / years) - 1.0) if years > 0 and equity > 0 else 0.0
    return {"sharpe": sharpe, "cagr": cagr, "max_dd": max_dd, "vol": vol,
            "total_return": equity - 1.0, "days": len(rets)}


def _split_days(returns_by_sleeve: dict[str, dict[int, float]], train_frac: float) -> tuple[int, int]:
    days = sorted({d for r in returns_by_sleeve.values() for d in r})
    if len(days) < 4:
        return (days[-1] if days else 0), ((days[-1] + 1) if days else 1)
    idx = max(1, min(len(days) - 1, int(len(days) * train_frac)))
    return days[idx - 1], days[idx]


def _window(series: dict[int, float], *, day_hi: int | None = None,
            day_lo: int | None = None) -> dict[int, float]:
    return {d: r for d, r in series.items()
            if (day_hi is None or d <= day_hi) and (day_lo is None or d >= day_lo)}


def _rolling_fraction_positive(series: dict[int, float], *, window_days: int) -> float:
    """Fraction of CONSECUTIVE non-overlapping `window_days` windows with positive total return (stricter
    than per-year — a real edge is positive in MOST windows). Windows with <2 days are skipped. No series → 0."""
    if not series:
        return 0.0
    days = sorted(series)
    lo, hi = days[0], days[-1]
    n_pos = 0
    n_win = 0
    start = lo
    while start <= hi:
        end = start + window_days - 1
        rets = [series[d] for d in days if start <= d <= end]
        if len(rets) >= 2:
            n_win += 1
            tot = 1.0
            for r in rets:
                tot *= (1.0 + r)
            if tot - 1.0 > 0.0:
                n_pos += 1
        start = end + 1
    return (n_pos / n_win) if n_win else 0.0


def _naive_metrics(returns_by_sleeve: dict[str, dict[int, float]], active: list[str], *,
                   cost_bps: float, day_hi: int | None = None,
                   day_lo: int | None = None) -> dict[str, Any]:
    """Naive equal-weight metrics over a date window, via the EXACT `_equal_weight_metrics`
    `risk_managed_book_from_store` uses — so the harness's `naive` number is bit-identical to the deployed
    book's naive baseline. Windows the per-sleeve returns first, then adds `vol` for a uniform report."""
    windowed = {s: _window(returns_by_sleeve[s], day_hi=day_hi, day_lo=day_lo) for s in active}
    present = [s for s in active if windowed.get(s)]
    m = _equal_weight_metrics(windowed, present, capital=100_000.0, cost_bps=cost_bps)
    # _equal_weight_metrics doesn't return vol; recover it from the naive curve for a uniform report.
    curve, _ = _scheme_curve(windowed, present, scheme="naive", lookback=0, cap=1.0, cost_bps=cost_bps)
    vol = _metrics(curve)["vol"]
    return {"sharpe": m["sharpe"], "cagr": m["cagr"], "max_dd": m["max_dd"], "vol": vol,
            "total_return": m["total_return"], "days": m["days"]}


# --- per-scheme TRAIN-chosen-hyperparam evaluation ----------------------------------------------

def _eval_scheme(returns_by_sleeve: dict[str, dict[int, float]], active: list[str], *, scheme: str,
                 cost_bps: float, train_hi: int, test_lo: int) -> dict[str, Any]:
    """Score ONE scheme with OOS discipline: pick its hyperparam (lookback for slow_rp, cap for risk_capped)
    by best TRAIN Sharpe, then report the chosen variant's TRAIN / TEST / FULL metrics + turnover + the
    realized risk composition (final-day causal risk shares on the full sample). `naive` has no hyperparam —
    its metrics come straight from `_naive_metrics` (== the deployed book's naive number)."""
    if scheme == "naive":
        full_curve, turnover = _scheme_curve(returns_by_sleeve, active, scheme="naive", lookback=0, cap=1.0,
                                             cost_bps=cost_bps)
        full_m = _naive_metrics(returns_by_sleeve, active, cost_bps=cost_bps)
        before_full = {s: dict(returns_by_sleeve[s]) for s in active}
        comp = _causal_realized_risk_share({s: 1.0 / len(active) for s in active}, before_full, active,
                                           lookback=_SLOW_RP_LOOKBACKS[-1])
        return {
            "param": None,
            "train": _naive_metrics(returns_by_sleeve, active, cost_bps=cost_bps, day_hi=train_hi),
            "test": _naive_metrics(returns_by_sleeve, active, cost_bps=cost_bps, day_lo=test_lo),
            "full": full_m,
            "turnover": turnover,
            "rolling_frac_positive": _rolling_fraction_positive(full_curve, window_days=_ROLLING_WINDOW_DAYS),
            "risk_composition": comp,
        }

    params = _SLOW_RP_LOOKBACKS if scheme == "slow_rp" else _RISK_CAPS

    # pick the hyperparam on TRAIN ONLY.
    best_param = params[0]
    best_train_sharpe = -math.inf
    train_metrics: dict[Any, dict[str, Any]] = {}
    for p in params:
        lb = int(p) if scheme == "slow_rp" else _SLOW_RP_LOOKBACKS[-1]
        cp = float(p) if scheme == "risk_capped" else 1.0
        curve, _ = _scheme_curve(returns_by_sleeve, active, scheme=scheme, lookback=lb, cap=cp,
                                 cost_bps=cost_bps)
        tm = _metrics(_window(curve, day_hi=train_hi))
        train_metrics[p] = tm
        if tm["sharpe"] > best_train_sharpe:
            best_train_sharpe = tm["sharpe"]
            best_param = p

    lb = int(best_param) if scheme == "slow_rp" else _SLOW_RP_LOOKBACKS[-1]
    cp = float(best_param) if scheme == "risk_capped" else 1.0
    full_curve, turnover = _scheme_curve(returns_by_sleeve, active, scheme=scheme, lookback=lb, cap=cp,
                                         cost_bps=cost_bps)
    # realized risk composition (final-day causal shares over the full sample) — does it cut xsfunding?
    before_full = {s: dict(returns_by_sleeve[s]) for s in active}
    if scheme == "slow_rp":
        final_w = _slow_rp_weights(before_full, active, lookback=lb, cap=1.0)
    else:
        final_w = _risk_capped_weights(before_full, active, lookback=lb, cap=cp)
    comp = _causal_realized_risk_share(final_w, before_full, active, lookback=lb)

    return {
        "param": best_param,
        "train": train_metrics[best_param],
        "test": _metrics(_window(full_curve, day_lo=test_lo)),
        "full": _metrics(full_curve),
        "turnover": turnover,
        "rolling_frac_positive": _rolling_fraction_positive(full_curve, window_days=_ROLLING_WINDOW_DAYS),
        "risk_composition": comp,
    }


# --- the harness -------------------------------------------------------------------------------

def sleeve_weighting_ab(returns_by_sleeve_or_store: Any, *, universe: set[str] | None = None,
                        cost_bps: float = 11.0, train_frac: float = 0.6,
                        sleeves: list[str] | None = None,
                        min_window_frac: float = _MIN_WINDOW_FRAC,
                        unlock_events: list[dict[str, Any]] | None = None,
                        stablecoin_spot_panel: dict[str, dict[int, float]] | None = None,
                        ) -> dict[str, Any]:
    """OOS-disciplined SLEEVE-WEIGHTING A/B over the Bybit 4-edge book. READ-ONLY, memory-bounded.

    `returns_by_sleeve_or_store` may be a `{sleeve: {epoch_day: ret}}` dict (the pure path) OR a feature
    store (then the 4-sleeve returns are assembled via `sleeve_returns_from_store`, universe-restricted).

    For EACH scheme (naive / slow_rp / risk_capped): pick its hyperparam on TRAIN, then report TRAIN / TEST
    (OOS) / FULL Sharpe-CAGR-vol-maxDD, turnover, rolling fraction-positive, and the realized risk
    composition (does it cut xsfunding's risk share?). VERDICT: switch away from naive ONLY IF a risk scheme
    beats naive TEST/OOS Sharpe (net of cost) AND is positive in > `min_window_frac` of rolling windows;
    otherwise STAY NAIVE.

    Returns `{available, sleeves, days, cost_bps, train_frac, split_*, table {scheme: {...}}, verdict}`.
    Writes NOTHING."""
    returns_by_sleeve = _assemble_returns(
        returns_by_sleeve_or_store, universe=universe, cost_bps=cost_bps, sleeves=sleeves,
        unlock_events=unlock_events, stablecoin_spot_panel=stablecoin_spot_panel)
    active = [s for s in (sleeves or _DEFAULT_BYBIT_SLEEVES) if returns_by_sleeve.get(s)]
    for s in returns_by_sleeve:   # the pure path may pass arbitrary sleeve names; keep any with data.
        if returns_by_sleeve[s] and s not in active:
            active.append(s)

    n_days = len({d for r in returns_by_sleeve.values() for d in r})
    if not active or n_days < 4:
        return {"available": False, "sleeves": active, "days": n_days,
                "reason": "need >= 2 sleeves-with-data and >= 4 union days for a train/test split "
                          "(run the Phase-A/B Bybit ingests first)"}

    train_hi, test_lo = _split_days(returns_by_sleeve, train_frac)

    table: dict[str, dict[str, Any]] = {}
    for scheme in SCHEME_NAMES:
        table[scheme] = _eval_scheme(returns_by_sleeve, active, scheme=scheme, cost_bps=cost_bps,
                                     train_hi=train_hi, test_lo=test_lo)

    verdict = _verdict(table, min_window_frac=min_window_frac)

    return {"available": True, "sleeves": active, "days": n_days,
            "table": table, "verdict": verdict, "cost_bps": cost_bps, "train_frac": train_frac,
            "min_window_frac": min_window_frac,
            "split_train_hi": train_hi, "split_test_lo": test_lo}


def _verdict(table: dict[str, dict[str, Any]], *, min_window_frac: float) -> dict[str, Any]:
    """STAY-NAIVE-by-default verdict. A risk scheme is ADOPTED over naive ONLY IF it beats naive's TEST/OOS
    Sharpe AND is positive in > `min_window_frac` of rolling windows. If several qualify, take the best
    TEST Sharpe. Otherwise STAY NAIVE. The full-sample numbers are reference-only (look-ahead-optimistic)."""
    naive_test = table["naive"]["test"]["sharpe"]
    candidates: list[tuple[str, float]] = []
    for name in SCHEME_NAMES:
        if name == "naive":
            continue
        te = table[name]["test"]["sharpe"]
        rwf = table[name]["rolling_frac_positive"]
        if te > naive_test + 1e-9 and rwf > min_window_frac:
            candidates.append((name, te))

    if candidates:
        candidates.sort(key=lambda x: x[1], reverse=True)
        rec = candidates[0][0]
        switch = True
        note = (f"SWITCH to '{rec}': it beats naive eq-wt OUT-OF-SAMPLE (TEST Sharpe "
                f"{table[rec]['test']['sharpe']:.2f} > naive {naive_test:.2f}) and is positive in "
                f"{table[rec]['rolling_frac_positive']*100:.0f}% of rolling windows "
                f"(> {min_window_frac*100:.0f}%).")
    else:
        rec = "naive"
        switch = False
        # explain why each risk scheme failed (OOS Sharpe and/or window-fraction).
        reasons = []
        for name in SCHEME_NAMES:
            if name == "naive":
                continue
            te = table[name]["test"]["sharpe"]
            rwf = table[name]["rolling_frac_positive"]
            why = ("OOS Sharpe" if te <= naive_test + 1e-9 else "") + (
                " & " if te <= naive_test + 1e-9 and rwf <= min_window_frac else "") + (
                "windows" if rwf <= min_window_frac else "")
            reasons.append(f"{name} (TEST {te:.2f} vs naive {naive_test:.2f}, "
                           f"windows {rwf*100:.0f}%; failed: {why or 'tie'})")
        note = ("STAY NAIVE: no risk-weighting scheme beats naive eq-wt OUT-OF-SAMPLE net of cost "
                f"[{'; '.join(reasons)}]. The full-sample inverse-vol Sharpe (~2.18) is look-ahead-optimistic.")

    return {"recommendation": rec, "switch": switch, "naive_test_sharpe": naive_test,
            "min_window_frac": min_window_frac, "note": note}


def _assemble_returns(returns_by_sleeve_or_store: Any, *, universe: set[str] | None, cost_bps: float,
                      sleeves: list[str] | None, unlock_events: list[dict[str, Any]] | None,
                      stablecoin_spot_panel: dict[str, dict[int, float]] | None,
                      ) -> dict[str, dict[int, float]]:
    """Pass through a prebuilt `{sleeve: {day: ret}}` dict, or assemble the 4-sleeve returns from a feature
    store via `sleeve_returns_from_store` (universe-restricted READS → memory-bounded). A dict is detected by
    its values being dicts; anything else is treated as a store."""
    obj = returns_by_sleeve_or_store
    if isinstance(obj, dict) and (not obj or all(isinstance(v, dict) for v in obj.values())):
        return {s: dict(v) for s, v in obj.items()}
    candidate = list(sleeves) if sleeves is not None else list(_DEFAULT_BYBIT_SLEEVES)
    out: dict[str, dict[int, float]] = {}
    for sleeve in candidate:
        series = sleeve_returns_from_store(
            obj, sleeve, universe=universe, cost_bps=cost_bps,
            unlock_events=unlock_events, stablecoin_spot_panel=stablecoin_spot_panel)
        if len(series) >= 2:
            out[sleeve] = series
    return out
