"""PaperBookService — application layer for the paper-trading cockpit.

Two jobs, both read-only w.r.t. the broker:
1. `derive_and_persist`: turn book_allocator sleeve weights + each sleeve's symbol weights + the day's
   prices into target positions (pure `domain.paper`), diff them against the prior held book to produce
   trades, and persist both to `PaperRepo`.
2. `view`: load the latest open positions, mark them to live prices, and build the cockpit read model
   (combined equity, per-sleeve PnL, open positions, recent trades).
"""
from __future__ import annotations

import dataclasses
import datetime as dt
import math
import statistics as st
from dataclasses import dataclass, field

import numpy as np

from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.application.book_allocator import _marginal_prune
from fxhnt.application.book_allocator import align_edges as _align_edges
from fxhnt.domain.edge_decay import RetirementGate, WindowedEdgeHealth
from fxhnt.domain.paper import (
    Position,
    Trade,
    diff_trades,
    realized_delta,
    target_positions,
)
from fxhnt.ports.live_price import LivePrice

# Default number of recent trades shown in the cockpit read model.
RECENT_TRADES_LIMIT = 25

# --- Realistic trading-cost model (research-grounded; USDT-M perps) ----------------------
# Per-side trading cost = taker_fee + spread + slippage, tiered by sleeve liquidity. The square-root
# market-impact law is the basis for the tiering: liquid majors pay ~fee+a-few-bp; illiquid small-cap
# alt shorts pay a much wider spread/slippage. Module-level so they are easy to tune.
_COST_RATE_BY_SLEEVE: dict[str, float] = {
    "crypto_tstrend": 0.0008,        # 8 bp  — 5 bp taker fee + ~3 bp spread/slippage; liquid majors
    "stablecoin_rotation": 0.0006,   # 6 bp  — liquid stablecoins
    "unlock": 0.0035,                # 35 bp — 5 bp fee + ~30 bp spread/slippage; illiquid small-cap shorts
}
_DEFAULT_COST_RATE = 0.0015          # any other sleeve

# The A8 assumed per-side cost tier in BPS for a testnet fill (taker fee already folded into the tier — see
# _COST_RATE_BY_SLEEVE above). The unlock-short symbols pay the wide illiquid small-cap tier (35 bp); every
# other (liquid) symbol pays the liquid-majors tier (8 bp). Sources the SAME _COST_RATE_BY_SLEEVE constants the
# sim book charges, so there is one source of truth for the cost tiers (no new hardcoded numbers).
_UNLOCK_TIER_BPS = _COST_RATE_BY_SLEEVE["unlock"] * 1e4            # 35 bp
_LIQUID_TIER_BPS = _COST_RATE_BY_SLEEVE["crypto_tstrend"] * 1e4    # 8 bp


def assumed_tier_bps(symbol: str, is_unlock: bool) -> float:
    """The A8 assumed per-side cost tier (bps) the testnet leg validates execution slippage against: the wide
    illiquid small-cap tier (35 bp) for an unlock-short `symbol`, the liquid-majors tier (8 bp) otherwise.
    Reuses the existing `_COST_RATE_BY_SLEEVE` tiers (the SAME numbers the sim book charges). Pure; no I/O."""
    return _UNLOCK_TIER_BPS if is_unlock else _LIQUID_TIER_BPS


def trading_cost(prev_positions: list[Position], target_positions: list[Position],
                 prices: dict[str, float], *, rate_by_sleeve: dict[str, float] = _COST_RATE_BY_SLEEVE,
                 default_rate: float = _DEFAULT_COST_RATE) -> float:
    """Per-rebalance trading cost (>= 0). For each (sleeve, symbol) in the union of prior + target,
    traded_qty = |target_qty − prev_qty| (0 on a side where the key is absent); notional_traded =
    traded_qty · price (the day's close, fallback to the position's entry_price when the symbol has no
    price); cost = Σ notional_traded · rate_by_sleeve.get(sleeve, default_rate). Pure; no I/O."""
    pri = {(p.sleeve, p.symbol): p for p in prev_positions}
    tgt = {(p.sleeve, p.symbol): p for p in target_positions}
    total = 0.0
    for key in set(pri) | set(tgt):
        sleeve, sym = key
        pq = pri[key].qty if key in pri else 0.0
        tq = tgt[key].qty if key in tgt else 0.0
        traded = abs(tq - pq)
        if traded == 0.0:
            continue
        # Fallback to whichever side carries the entry price (target preferred — it's the new lot).
        entry = (tgt[key].entry_price if key in tgt else pri[key].entry_price)
        px = prices.get(sym, entry)
        total += traded * px * rate_by_sleeve.get(sleeve, default_rate)
    return total


def funding_pnl(positions: list[Position], funding_by_symbol: dict[str, float],
                prices: dict[str, float]) -> float:
    """Signed funding cash flow to the holder of `positions` for one date. For each held position the
    notional is SIGNED (qty carries long/short): notional = qty · price (fallback entry_price). The
    cash flow to the holder = −notional · funding (a LONG on positive funding PAYS → negative; a SHORT
    RECEIVES → positive). Unknown symbol → 0 funding. Summed; signed. Pure; no I/O. `funding_by_symbol`
    is the date's per-symbol daily funding rate (already 8h-summed-to-daily in the warehouse 'funding'
    feature)."""
    total = 0.0
    for p in positions:
        notional = p.qty * prices.get(p.symbol, p.entry_price)   # signed
        total += -notional * funding_by_symbol.get(p.symbol, 0.0)
    return total


# Winsorize bound (fractional daily return) for the per-symbol DAILY-CLOSE carry return. A real delta-neutral
# basis+funding move rarely exceeds ~10-15%/day even in a squeeze; ±25% is a generous ~2x-headroom hard cap
# that leaves every plausible real move untouched while neutralizing a data-glitch print. It exists because a
# single bad spot/perp close (the 2025-02-02 xsfunding artifact: +179.7% sleeve / +44.69% book in ONE day —
# ~15x the next-largest ever +8% day) blows up the daily-close basis term `(spot_ret − perp_ret)` and, at its
# cross-sectional weight, dominates the whole `Σ w·carry_ret/Σ|w|` sleeve return and inflates the backtest
# baseline the forward gate reconciles against. We CAP (winsorize) the day rather than DROP it. This bound
# does NOT touch the deliberate intraday `worst_basis` liquidation charge (its own −100% floor is intentional).
_CARRY_RET_DAILY_CAP = 0.25


def carry_return_by_symbol(funding_by_symbol: dict[str, float] | None,
                           basis_by_symbol: dict[str, float] | None,
                           worst_basis_by_symbol: dict[str, float] | None = None, *,
                           cap: float = _CARRY_RET_DAILY_CAP) -> dict[str, float]:
    """The per-symbol FULL daily carry return `basisᵢ + fundingᵢ` for a delta-neutral carry book, merging
    the funding accrual with the per-symbol realized BASIS (the price P&L of the spot+perp legs). This is
    the SINGLE SOURCE consumed by BOTH the carry sleeve return (`sleeve_daily_carry`) AND its equity
    contribution (`carry_pnl`) — they can't drift because they read the same map.

    The realized basis for a symbol is, in priority order:
      1. the CONSERVATIVE intraday `worst_basis` = `max(worst_basis, −1.0)` (capped at −100% = total loss
         of the leg) when an intraday `worst_basisᵢ` is available — this charges the INTRADAY liquidation
         tail (a LUNA-class mid-day squeeze the daily-close basis is blind to), instead of the daily-close;
      2. otherwise the DAILY-CLOSE basis change `(spot_ret − perp_ret)` from
         `funding_panel.basis_change_by_symbol` (the documented fallback when no worst_basis was ingested).
    `worst_basis_by_symbol` absent (None / empty) → the daily-close basis everywhere.

    WINSORIZATION (`cap`, default ±`_CARRY_RET_DAILY_CAP` = ±25%/day). The DAILY-CLOSE carry return
    `fundingᵢ + basisᵢ` is clipped to [−cap, +cap] so a single absurd basis/funding print (a bad spot/perp
    close — e.g. the 2025-02-02 +179.7% xsfunding artifact) cannot dominate the cross-sectional sleeve. This
    is applied IN the SSOT, so it is BOTH retroactive (cleans any curve recomputed on read) AND prospective
    (a future glitch is capped). The day is CAPPED, never dropped. The intraday `worst_basis` OVERRIDE is a
    SEPARATE, already-bounded (−100%) deliberate liquidation charge and is NOT winsorized — a real −80%/−100%
    mid-day excursion must survive; only the daily-close artifact is clipped.

    Union of all three inputs' symbols; a symbol present in only one contributes that leg alone (funding
    but no basis → funding-only; basis but no funding → basis-only; a worst_basis with no close basis is
    still charged its capped value). Pure; no I/O."""
    f = funding_by_symbol or {}
    b = basis_by_symbol or {}
    wb = worst_basis_by_symbol or {}
    out: dict[str, float] = {}
    for s in set(f) | set(b) | set(wb):
        if s in wb:
            # Intraday worst_basis (deliberate liquidation charge, already floored at −100%) overrides the
            # daily-close basis — NOT winsorized, so a real leg-loss survives at its true magnitude.
            out[s] = f.get(s, 0.0) + max(wb[s], -1.0)
        else:
            # Daily-close carry return: winsorize to ±cap to neutralize a data-glitch print (capped, not dropped).
            out[s] = max(-cap, min(cap, f.get(s, 0.0) + b.get(s, 0.0)))
    return out


def carry_pnl(positions: list[Position], carry_ret_by_symbol: dict[str, float]) -> float:
    """FULL-return P&L for a delta-neutral CARRY book (xsfunding) for one date — the equity contribution a
    return-defined carry sleeve makes.

    = Σ (qtyᵢ·entry_priceᵢ)·carry_retᵢ over `positions` (signed unit weight wᵢ = qtyᵢ·entry_priceᵢ), where
    `carry_retᵢ = (spot_retᵢ − perp_retᵢ) + fundingᵢ` is the symbol's FULL daily carry return (funding +
    daily basis change; see `carry_return_by_symbol`). This is the cross-sectional CARRY-BOOK convention
    (`domain.book_return_xs = Σ w·return`): a LONG on a high-funding coin EARNS, a SHORT on a low/negative-
    funding coin EARNS — and the basis term now charges the delta-neutral position's price P&L (where the
    LUNA-class tail lives), no longer dropped. Note this is the OPPOSITE sign of `funding_pnl` (the perp-
    HOLDER cash-flow `−notional·funding`, used for directional funding DRAG): a carry book is delta-neutral
    spot+perp, so its realized return is the dispersion `Σ w·carry_ret`, not the one-sided perp cash flow.
    `carry_pnl / Σ|qtyᵢ·entry_priceᵢ|` equals `sleeve_daily_carry` exactly (same numerator, same map), so
    the equity contribution is consistent with the sleeve return booked into `paper_sleeve_ret` → live ==
    replay. Unknown symbol → 0 carry return. Pure; no I/O.

    BASIS RESOLUTION CAVEAT: the basis is now charged at DAILY-close resolution (perp+spot close from the
    warehouse). The remaining gap is the INTRADAY basis-squeeze / liquidation tail (1m `worst_basis`),
    still a documented future refinement — but the carry Sharpe is no longer fake."""
    return sum(p.qty * p.entry_price * carry_ret_by_symbol.get(p.symbol, 0.0) for p in positions)


def unit_shadow_positions(symbol_weights_by_sleeve: dict[str, dict[str, float]],
                          prices: dict[str, float], date: str) -> list[Position]:
    """The UNIT (sleeve_weight=1, capital=1) composition for every active sleeve — the kill-proof signal
    book. Reuses target_positions so qty/entry semantics match the throttled book. Pure."""
    out: list[Position] = []
    for sleeve, sw in symbol_weights_by_sleeve.items():
        out.extend(target_positions(sleeve=sleeve, sleeve_weight=1.0, capital=1.0,
                                    symbol_weights=sw, prices=prices, date=date))
    return out


def sleeve_daily_return(prev_positions: list[Position],
                        close_by_symbol: dict[str, float]) -> float | None:
    """One date's return for a sleeve from its prior-date positions, marked to today's closes.

    The prior close is carried by `Position.entry_price` (positions are re-derived each date at that
    date's close). Return = Σ qty·(close − entry_price) / Σ|qty·entry_price|. Unknown symbol → entry
    (0 pnl). None when there is no prior position or the notional is 0. Pure; no I/O.
    """
    notional = sum(abs(p.qty * p.entry_price) for p in prev_positions)
    if notional <= 0.0:
        return None
    pnl = sum(p.qty * (close_by_symbol.get(p.symbol, p.entry_price) - p.entry_price)
              for p in prev_positions)
    return pnl / notional


def sleeve_daily_carry(prev_positions: list[Position],
                       carry_ret_by_symbol: dict[str, float]) -> float | None:
    """One date's CARRY return for a funding sleeve: Σ wᵢ·carry_retᵢ / Σ|wᵢ|, where the signed unit weight
    wᵢ = qtyᵢ·entry_priceᵢ is recovered from the unit shadow book and `carry_retᵢ = (spot_retᵢ − perp_retᵢ)
    + fundingᵢ` is the symbol's FULL daily carry return (funding + daily basis change; see
    `carry_return_by_symbol`). A short (w<0) on a positive-funding coin earns negative here only as the BOOK
    direction dictates (the domain's `book_return_xs = Σ w·return`). The basis term now charges the delta-
    neutral price P&L — the carry return is no longer funding-only. Unknown symbol → 0 carry return. None
    when no prior position or the gross notional is 0. Pure; no I/O."""
    gross = sum(abs(p.qty * p.entry_price) for p in prev_positions)
    if gross <= 0.0:
        return None
    carry = sum((p.qty * p.entry_price) * carry_ret_by_symbol.get(p.symbol, 0.0) for p in prev_positions)
    return carry / gross


_L_MIN = 10
_L_MAX = 60
_PHI = 0.5
_KAPPA = 2.0

_TARGET_VOL = 0.30
_KELLY_FRACTION = 1.0
_MAX_LEVERAGE = 3.0
_VOL_LOOKBACK = 60
_PERIODS_PER_YEAR = 365
_KILL_DD = 0.25
_REENTER_DD = 0.12

_DD_BUDGET = -0.25       # default ex-ante drawdown budget for the edge_health controller's base Kelly

_VOL_MIN = 0.25          # adaptive vol-budget floor (never "massive")
_VOL_MAX = 0.35          # adaptive vol-budget ceiling (user cap)
_VOL_SH_REF = 2.0        # trailing book Sharpe at which the budget reaches vol_max
_VOL_MIN_OBS = 20        # min trailing book returns before adapting (else bootstrap to base)


def _regime_lookback(series_by_sleeve: dict[str, dict[int, float]], active: list[str]) -> int:
    """Adaptive L_t from a DISJOINT recent-vs-prior vol ratio on the active sleeves' mean return series.
    rho = std(recent L_min) / std(prior L_min); rho>1 (vol rising) -> shorten. Hold L_max until 2*L_min
    obs exist. Floor L_min, anchor L_max."""
    days = sorted({d for s in active for d in series_by_sleeve.get(s, {})})
    if len(days) < 2 * _L_MIN:
        return _L_MAX
    # Only the LAST 2·L_min days feed the disjoint recent/prior slices below; the rest of `rep` is never
    # read. Build the NaN matrix over just that tail (O(active·L_min) instead of O(active·D)) — `rep` is
    # computed per-column so trimming `days` to the tail yields a bit-identical recent/prior.
    days = days[-2 * _L_MIN:]
    mat = np.full((len(active), len(days)), np.nan)
    for r, s in enumerate(active):
        sr = series_by_sleeve.get(s, {})
        for c, d in enumerate(days):       # look up the ≤20 tail days directly (O(active·L_min))
            if d in sr:
                mat[r, c] = sr[d]
    # NaN-aware mean across active sleeves so a sleeve missing a day does NOT get a spurious 0
    # (which would depress the pooled vol and bias the regime ratio toward L_max).
    with np.errstate(invalid="ignore"):
        rep = np.nanmean(mat, axis=0)
    rep = np.nan_to_num(rep, nan=0.0)   # a day with NO active sleeve -> neutral 0 (rare; union axis)
    recent, prior = rep[-_L_MIN:], rep[-2 * _L_MIN:-_L_MIN]
    sp = float(np.std(prior))
    if sp <= 1e-12:
        return _L_MAX
    rho = float(np.std(recent)) / sp
    return int(round(min(_L_MAX, max(_L_MIN, _L_MAX / max(1.0, rho)))))


def _waterfill(init: dict[str, float], w_floor: float, cap: float) -> dict[str, float]:
    """Two-sided iterative clamp-to-[w_floor, cap] + renormalise the free mass, until stable (<=N passes).

    When all sleeves land on a bound (free is empty) but residual mass remains, the residual is
    redistributed proportional to the slack each sleeve has toward the bound it needs to move: if
    residual > 0, distribute proportional to (cap - w); if residual < 0, proportional to (w - w_floor).
    This guarantees the result sums to 1 whenever the bounds are feasible (w_floor*N <= 1 <= cap*N).

    Mirrors the cap loop in book_allocator._inverse_vol_weights.
    """
    w = dict(init)
    n_passes = len(w) + 2   # a couple of extra passes covers the all-clamped residual dispatch
    for _ in range(n_passes):
        clamped = {k: v for k, v in w.items() if v <= w_floor + 1e-15 or v >= cap - 1e-15}
        if clamped:
            fixed = {k: min(cap, max(w_floor, w[k])) for k in clamped}
            free = {k: w[k] for k in w if k not in clamped}
            remaining = 1.0 - sum(fixed.values())
            nxt: dict[str, float]
            if free:
                ft = sum(free.values())
                nxt = dict(fixed)
                for k in free:
                    nxt[k] = free[k] / ft * remaining if ft > 1e-12 else remaining / len(free)
            else:
                # All sleeves are clamped; distribute residual proportional to available slack.
                if abs(remaining) < 1e-15:
                    nxt = dict(fixed)
                elif remaining > 0:
                    # Need to grow some sleeves — slack is (cap - w).
                    slacks = {k: cap - v for k, v in fixed.items()}
                    total_slack = sum(slacks.values())
                    if total_slack > 1e-15:
                        nxt = {k: min(cap, v + slacks[k] / total_slack * remaining)
                               for k, v in fixed.items()}
                    else:
                        nxt = dict(fixed)
                else:
                    # Need to shrink some sleeves — slack is (w - w_floor).
                    slacks = {k: v - w_floor for k, v in fixed.items()}
                    total_slack = sum(slacks.values())
                    if total_slack > 1e-15:
                        nxt = {k: max(w_floor, v + slacks[k] / total_slack * remaining)
                               for k, v in fixed.items()}
                    else:
                        nxt = dict(fixed)
        else:
            nxt = w
        if all(abs(nxt[k] - w[k]) < 1e-12 for k in w):
            w = nxt
            break
        w = nxt

    # Safety net: if any residual remains after the loop (e.g. extremely tight bounds), do one
    # final proportional-slack pass rather than silently truncating.
    total = sum(w.values())
    residual = 1.0 - total
    if abs(residual) > 1e-9:
        if residual > 0:
            slacks = {k: cap - v for k, v in w.items()}
        else:
            slacks = {k: v - w_floor for k, v in w.items()}
        total_slack = sum(slacks.values())
        if total_slack > 1e-15:
            w = {k: min(cap, max(w_floor, v + slacks[k] / total_slack * residual))
                 for k, v in w.items()}

    return {k: min(cap, max(w_floor, v)) for k, v in w.items()}


def isv_adaptive_weights(returns_by_sleeve: dict[str, dict[int, float]],
                         active_sleeves: list[str], *, _debug: bool = False):
    """ISV-adaptive sleeve weights: adaptive lookback L_t (A), N_eff-derived cap_t (C), and a
    max-with-floor cold-start via two-sided water-fill (B). Cold-start / no-history -> equal-weight.
    Pure. Returns {sleeve: weight} (sum 1), or ({...}, {"L_t":..,"cap_t":..}) when _debug."""
    active = [s for s in active_sleeves if s]
    n = len(active)
    if n == 0:
        return ({}, {"L_t": _L_MAX, "cap_t": 0.0}) if _debug else {}
    has = [s for s in active if len(returns_by_sleeve.get(s, {})) >= 2]
    if not has:
        eq = {s: 1.0 / n for s in active}
        return (eq, {"L_t": _L_MAX, "cap_t": 1.0}) if _debug else eq
    L_t = _regime_lookback(returns_by_sleeve, has)
    inv = {}
    for s in has:
        vals = [returns_by_sleeve[s][d] for d in sorted(returns_by_sleeve[s])][-L_t:]
        v = float(np.std(vals)) if len(vals) > 1 else 0.0
        inv[s] = (1.0 / v) if v > 1e-12 else 0.0
    tot = sum(inv.values())
    raw = {s: inv[s] / tot for s in has} if tot > 0 else {s: 1.0 / len(has) for s in has}
    n_eff = (sum(raw.values()) ** 2) / sum(v * v for v in raw.values()) if raw else 1.0
    cap_t = min(1.0, max(1.0 / n, _KAPPA / n_eff))
    w_floor = _PHI / n
    cold = [s for s in active if s not in has]
    init = {s: w_floor for s in cold}
    has_mass = max(0.0, 1.0 - w_floor * len(cold))
    for s in has:
        init[s] = raw[s] * has_mass
    w = _waterfill(init, w_floor, cap_t)
    return (w, {"L_t": L_t, "cap_t": cap_t}) if _debug else w


def marginal_gate_active(returns_by_sleeve: dict[str, dict[int, float]], active_sleeves: list[str], *,
                         vol_lookback: int = _VOL_LOOKBACK, max_sleeve_weight: float = _PHI) -> list[str]:
    """Leave-one-out marginal-Sharpe edge gate (reuses book_allocator._marginal_prune): bench a has-history
    sleeve only when REMOVING it raises the trailing book Sharpe (keeps anticorrelated hedges, benches dead /
    non-diversifying sleeves). Cold sleeves (<2 returns) pass through untouched; <2 has-history → unchanged.
    Causal over the trailing `vol_lookback`. Returns the kept active sleeves in input order. Pure."""
    has = [s for s in active_sleeves if len(returns_by_sleeve.get(s, {})) >= 2]
    cold = [s for s in active_sleeves if s not in has]
    if len(has) < 2:
        return list(active_sleeves)
    _, arrs = _align_edges({s: returns_by_sleeve[s] for s in has})
    window = {s: a[max(0, a.size - vol_lookback):] for s, a in arrs.items()}
    kept_has = set(_marginal_prune(window, max_sleeve_weight))
    return [s for s in active_sleeves if s in cold or s in kept_has]


def vol_target_leverage(weights: dict[str, float], returns_by_sleeve: dict[str, dict[int, float]], *,
                        target_vol: float = _TARGET_VOL, kelly_fraction: float = _KELLY_FRACTION,
                        vol_lookback: int = _VOL_LOOKBACK, periods_per_year: int = _PERIODS_PER_YEAR,
                        max_leverage: float = _MAX_LEVERAGE, warmup_obs: int = _VOL_LOOKBACK) -> float:
    """Fractional-Kelly vol-target leverage over the trailing weighted-portfolio return (mirrors
    book_allocator.combine_book step 4). lev = min(max_leverage, kelly·(target_vol/√ppy)/pv) where pv is the
    std of Σ wᵢ·rᵢ over the last `vol_lookback` of the union date axis. WARMUP: returns 1.0 (unlevered) until
    there are ≥ `warmup_obs` (default `vol_lookback`) book-return observations — no maximal leverage on
    cold-start thin history (a thin window has a tiny, unreliable vol estimate that over-levers). Also 1.0 on
    pv≤1e-12 / non-finite. Never flat from lack of data. Pure."""
    days = sorted({d for s in weights for d in returns_by_sleeve.get(s, {})})
    if len(days) < warmup_obs:
        return 1.0
    win = days[max(0, len(days) - vol_lookback):]
    port = [sum(weights.get(s, 0.0) * returns_by_sleeve.get(s, {}).get(d, 0.0) for s in weights) for d in win]
    pv = float(np.std(port)) if len(port) > 1 else 0.0
    if not math.isfinite(pv) or pv <= 1e-12:
        return 1.0
    return min(max_leverage, kelly_fraction * (target_vol / math.sqrt(periods_per_year)) / pv)


def kelly_for_dd_budget(dd_budget: float, *, intercept: float = 21.0, slope: float = 12.0,
                        kmin: float = 0.1, kmax: float = 1.5) -> float:
    """Map an ex-ante drawdown budget to a base Kelly by inverting the empirically-measured leverage→DD
    curve `DD% ≈ −(intercept + slope·kelly)` (Kelly-sweep finding: DD is a SIZING choice, Sharpe is ~leverage-
    invariant). Inverting: `kelly = clip((|dd_budget|·100 − intercept) / slope, kmin, kmax)`. A bigger DD
    budget buys more (higher Kelly); a tight budget clamps to kmin. Pure. e.g. −0.25→~0.33, −0.33→1.0."""
    kelly = (abs(dd_budget) * 100.0 - intercept) / slope
    return max(kmin, min(kmax, kelly))


def adaptive_target_vol(book_returns: list[float], *, vol_min: float = _VOL_MIN, vol_max: float = _VOL_MAX,
                        sh_ref: float = _VOL_SH_REF, base: float = _TARGET_VOL,
                        win: int = _VOL_LOOKBACK, min_obs: int = _VOL_MIN_OBS,
                        periods_per_year: int = _PERIODS_PER_YEAR) -> float:
    """ISV-adaptive vol budget: scale the target vol by the trailing book Sharpe, bounded [vol_min, vol_max].
    Eases toward vol_min in weak patches, rides vol_max when the edge is firing. Bootstrap to `base` until
    `min_obs` trailing book returns exist. Pure; causal (caller passes the strictly-prior book series)."""
    w = book_returns[-win:]
    if len(w) < min_obs:
        return base
    sd = st.pstdev(w)
    ts = (st.mean(w) / sd * math.sqrt(periods_per_year)) if sd > 0 else 0.0
    frac = max(0.0, min(1.0, ts / sh_ref))
    return vol_min + (vol_max - vol_min) * frac


class DrawdownKillSwitch:
    """Drawdown kill-switch with hysteresis (mirrors book_allocator.apply_drawdown_killswitch). Tracks the
    UN-throttled shadow equity; `.killed` (read before `observe`-ing the next return) is the state to apply to
    the next date. kill when running dd > kill_dd; re-enter when dd < reenter_dd. Causal; pure (no I/O)."""

    def __init__(self, kill_dd: float = _KILL_DD, reenter_dd: float = _REENTER_DD) -> None:
        self.kill_dd = kill_dd
        self.reenter_dd = reenter_dd
        self.shadow_eq = 1.0
        self.peak = 1.0
        self.killed = False

    def observe(self, book_return: float) -> None:
        self.shadow_eq *= (1.0 + book_return)
        self.peak = max(self.peak, self.shadow_eq)
        dd = (self.peak - self.shadow_eq) / self.peak if self.peak > 0 else 0.0
        if not self.killed and dd > self.kill_dd:
            self.killed = True
        elif self.killed and dd < self.reenter_dd:
            self.killed = False


def _sleeve_health_stateless(sleeve_returns: dict[int, float]) -> float:
    """Edge-health θ for ONE sleeve: replay a fresh WindowedEdgeHealth over the sleeve's realized return
    series (strictly-prior, ascending) and read `.health()`. Pure. Empty history → 1.0 (bootstrap: healthy
    by default; nothing to de-size on). Mirrors the incremental per-sleeve detector so live==replay.

    WindowedEdgeHealth is a bounded, SCALE-INVARIANT trailing risk-adjusted drift (t-stat) — it replaces
    the broken reuse of PageHinkleyDecay.health (which was unbounded, scale-dependent, and re-armed via a
    consecutive-day streak that almost never fires on daily crypto)."""
    det = WindowedEdgeHealth()
    for d in sorted(sleeve_returns):
        det.update(sleeve_returns[d])
    return det.health()


def _edge_health_weights(returns_by_sleeve: dict[str, dict[int, float]], active_sleeves: list[str], *,
                         theta_by_sleeve: dict[str, float]) -> dict[str, float]:
    """ISV weights health-adjusted (NO renorm): weights[s] = isv_weight[s] · θ_s. A decaying sleeve REDUCES
    total book exposure (that IS the de-sizing) — deliberately un-renormalised. Pure."""
    isv = isv_adaptive_weights(returns_by_sleeve, active_sleeves)
    return {s: w * theta_by_sleeve.get(s, 1.0) for s, w in isv.items()}


def _retirement_active_stateless(returns_by_sleeve: dict[str, dict[int, float]],
                                 active_sleeves: list[str]) -> list[str]:
    """The COMBINED controller's structural-retirement gate, replayed from scratch per call (stateless).
    For each active sleeve, replay a fresh RetirementGate over THAT sleeve's realized return series
    (strictly-prior, ascending) and keep the sleeve iff its status is not "retired". Returns the kept
    active sleeves in input order. Mirrors the incremental per-sleeve gate so live==replay.

    The gate is fed the sleeve's daily return series — the kill-proof signal stays alive even while a
    sleeve is retired, so a structural recovery is detectable and the sleeve can be re-admitted. A young
    sleeve (WEH still bootstrapping) is never retired. Pure."""
    kept: list[str] = []
    for s in active_sleeves:
        if not s:
            continue
        gate = RetirementGate()
        for d in sorted(returns_by_sleeve.get(s, {})):
            gate.update(returns_by_sleeve[s][d])
        if gate.status != "retired":
            kept.append(s)
    return kept


def paper_book_returns(returns_by_sleeve: dict[str, dict[int, float]], *,
                       target_vol: float = _TARGET_VOL, kelly_fraction: float = _KELLY_FRACTION,
                       vol_lookback: int = _VOL_LOOKBACK, periods_per_year: int = _PERIODS_PER_YEAR,
                       max_leverage: float = _MAX_LEVERAGE) -> dict[int, float]:
    """UN-throttled paper book daily return series (the killswitch's regime signal). Ascending over the union
    date axis; weights/leverage are recomputed from returns STRICTLY BEFORE each date and held into it, so
    bret(d) = lev_held·Σ w_held·r_d. Causal — no look-ahead. The historical 'active' set is every sleeve with
    prior return history (a faithful proxy; identical in both call paths). The vol budget is ISV-adaptive
    (trailing-Sharpe of the running book series, bounded [25%,35%]); `target_vol` is the bootstrap base. Pure."""
    all_days = sorted({d for r in returns_by_sleeve.values() for d in r})
    out: dict[int, float] = {}
    w_held: dict[str, float] | None = None
    lev_held = 1.0
    # ACCUMULATING strictly-before view: instead of rebuilding `before_d = {s:{k:v if k<d}}` (O(history))
    # every day → O(D²) total, grow `before_d` once and append day d's returns at the END of iteration d.
    # At the TOP of iteration d, `before_d` holds exactly the keys < d (the prior iterations appended only
    # < d), so it is bit-identical to the old per-day rebuild. isv_adaptive_weights / vol_target_leverage /
    # _regime_lookback are READ-ONLY over this dict (verified), so the shared growing dict is safe.
    before_d: dict[str, dict[int, float]] = {s: {} for s in returns_by_sleeve}
    # Chronological mirror of `out`'s values (out is filled in ascending-day order) so adaptive_target_vol
    # is fed the running list rather than rebuilding `list(out.values())` (O(D)) every day. Bit-identical:
    # same values, same order; adaptive_target_vol reads only its `[-win:]` tail.
    out_vals: list[float] = []
    for d in all_days:
        if w_held is not None:
            bret = lev_held * sum(w_held.get(s, 0.0) * returns_by_sleeve.get(s, {}).get(d, 0.0)
                                  for s in w_held)
            out[d] = bret
            out_vals.append(bret)
        active = [s for s, r in before_d.items() if r]
        w_held = isv_adaptive_weights(before_d, active)
        tv = adaptive_target_vol(out_vals, base=target_vol)
        lev_held = vol_target_leverage(w_held, before_d, target_vol=tv, kelly_fraction=kelly_fraction,
                                       vol_lookback=vol_lookback, periods_per_year=periods_per_year,
                                       max_leverage=max_leverage)
        # Append day d AFTER sizing into d, so the next iteration (d' > d) sees d as strictly-before.
        for s, r in returns_by_sleeve.items():
            if d in r:
                before_d[s][d] = r[d]
    return out


def paper_risk_overlay(returns_by_sleeve: dict[str, dict[int, float]], active_sleeves: list[str], *,
                       target_vol: float = _TARGET_VOL, kelly_fraction: float = _KELLY_FRACTION,
                       vol_lookback: int = _VOL_LOOKBACK, periods_per_year: int = _PERIODS_PER_YEAR,
                       max_leverage: float = _MAX_LEVERAGE, kill_dd: float = _KILL_DD,
                       reenter_dd: float = _REENTER_DD,
                       controller: str = "killswitch") -> tuple[dict[str, float], float, bool]:
    """Single entry point for one date's risk overlay given its trailing (strictly-prior) returns. Returns
    (weights, vol-target leverage, killed). Both backfill and snapshot call this on their trailing window so
    live /paper == /paper/replay. Pure.

    `controller="killswitch"` (DEFAULT, backward-compatible): ISV weights + ISV-adaptive vol budget (trailing-
    Sharpe of the prior book series, bounded [25%,35%], bootstrap 30%) + a dormant wide (−25/−12)
    DrawdownKillSwitch replayed over the un-throttled book-return history → `killed` for THIS date. Marginal-
    gate intentionally OFF for the current small book.

    `controller="edge_health"`: replaces the binary killswitch with per-sleeve CONTINUOUS de-sizing. For each
    active sleeve, replay a WindowedEdgeHealth over THAT sleeve's realized return series → θ_s = det.health();
    weights[s] = isv_weight[s]·θ_s (NO renorm — a decaying sleeve cuts total exposure). Base Kelly = the DD-
    budget Kelly `kelly_for_dd_budget(_DD_BUDGET)` (not 1.0). No killswitch → `killed=False` always (all de-
    sizing is via θ_s in the weights).

    `controller="combined"`: K(D*) sizing + per-sleeve WindowedEdgeHealth STRUCTURAL-retirement gate +
    the killswitch backstop. The ONLY behavioural delta vs `"killswitch"` is the retirement gate, which
    replays a per-sleeve `RetirementGate` over each active sleeve's return series and drops any sleeve
    whose edge is structurally dead → `active` survivors feed `isv_adaptive_weights` (natural re-norm;
    retired sleeves get weight 0). Leverage + killswitch are computed exactly as `"killswitch"`, so when
    NO sleeve is retired (all healthy) `combined` == `killswitch` bit-identically. Default
    `kelly_fraction=1.0` (current sizing; `kelly_for_dd_budget` available but not the combined default)."""
    if controller == "edge_health":
        theta = {s: _sleeve_health_stateless(returns_by_sleeve.get(s, {}))
                 for s in active_sleeves if s}
        weights = _edge_health_weights(returns_by_sleeve, active_sleeves, theta_by_sleeve=theta)
        budget_kelly = kelly_for_dd_budget(_DD_BUDGET)
        bret = paper_book_returns(returns_by_sleeve, target_vol=target_vol, kelly_fraction=budget_kelly,
                                  vol_lookback=vol_lookback, periods_per_year=periods_per_year,
                                  max_leverage=max_leverage)
        tv = adaptive_target_vol(list(bret.values()), base=target_vol)
        leverage = vol_target_leverage(weights, returns_by_sleeve, target_vol=tv,
                                       kelly_fraction=budget_kelly, vol_lookback=vol_lookback,
                                       periods_per_year=periods_per_year, max_leverage=max_leverage)
        return weights, leverage, False
    # `combined` = killswitch path, but the structural-retirement gate first filters the active set.
    # When NO sleeve is retired (all healthy) the active set is unchanged ⇒ combined == killswitch
    # bit-identically (same ISV weights, same book series, same leverage, same kill state).
    active = (_retirement_active_stateless(returns_by_sleeve, active_sleeves)
              if controller == "combined" else active_sleeves)
    weights = isv_adaptive_weights(returns_by_sleeve, active)
    bret = paper_book_returns(returns_by_sleeve, target_vol=target_vol, kelly_fraction=kelly_fraction,
                              vol_lookback=vol_lookback, periods_per_year=periods_per_year,
                              max_leverage=max_leverage)
    tv = adaptive_target_vol(list(bret.values()), base=target_vol)
    leverage = vol_target_leverage(weights, returns_by_sleeve, target_vol=tv, kelly_fraction=kelly_fraction,
                                   vol_lookback=vol_lookback, periods_per_year=periods_per_year,
                                   max_leverage=max_leverage)
    ks = DrawdownKillSwitch(kill_dd=kill_dd, reenter_dd=reenter_dd)
    for d in sorted(bret):
        ks.observe(bret[d])
    return weights, leverage, ks.killed


class IncrementalRiskOverlay:
    """O(1)-amortised drop-in for repeated `paper_risk_overlay` calls over an APPEND-ONLY trailing-return
    dict (the backfill's growing `acc`). Bit-identical to calling `paper_risk_overlay` per date.

    `paper_risk_overlay` recomputes the WHOLE un-throttled book-return series (`paper_book_returns`) AND
    replays the killswitch from scratch every date → O(D) each, O(N²) over a backfill. Because `acc` only
    ever GROWS (new days appended at the end) and `paper_book_returns` is a deterministic, strictly-causal
    forward loop, the book-return series for date D+1 is the series for date D plus the days that newly
    appeared — so we keep the loop state RESUMABLE (`_before`, `_out`, `_w_held`, `_lev_held`) and advance
    it only over the new days, and keep ONE persistent `DrawdownKillSwitch` that observes only the new
    book returns. Same float ops, same order ⇒ exactly equal output.

    `weights` and `leverage` are still recomputed each date (windowed L_t ≤ _L_MAX and vol_lookback, so
    O(window) not O(D)); only the O(D) pieces (book-return series + killswitch replay) are made incremental.

    The SNAPSHOT path keeps calling the stateless `paper_risk_overlay`; this is backfill-only, guarded by
    a golden equivalence test (test_incremental_overlay_matches_per_date)."""

    def __init__(self, *, target_vol: float = _TARGET_VOL, kelly_fraction: float = _KELLY_FRACTION,
                 vol_lookback: int = _VOL_LOOKBACK, periods_per_year: int = _PERIODS_PER_YEAR,
                 max_leverage: float = _MAX_LEVERAGE, kill_dd: float = _KILL_DD,
                 reenter_dd: float = _REENTER_DD, controller: str = "killswitch") -> None:
        self._controller = controller
        self._target_vol = target_vol
        # In edge_health mode the un-throttled book-return series + leverage are sized with the DD-budget
        # Kelly (matches paper_risk_overlay's edge_health branch); else the supplied kelly_fraction.
        self._book_kelly = kelly_for_dd_budget(_DD_BUDGET) if controller == "edge_health" else kelly_fraction
        self._kelly_fraction = self._book_kelly
        self._vol_lookback = vol_lookback
        self._periods_per_year = periods_per_year
        self._max_leverage = max_leverage
        # Resumable paper_book_returns loop state.
        self._before: dict[str, dict[int, float]] = {}   # strictly-before view, grows as days are consumed
        self._out: dict[int, float] = {}                 # the un-throttled book-return series (bret)
        # Chronological mirror of `self._out`'s values (appended in lockstep) so adaptive_target_vol can be
        # fed the running list without an O(D) `list(self._out.values())` rebuild every day. `_out` is filled
        # in ascending-day order, so `_out_vals` is identical to `list(self._out.values())` element-for-
        # element → bit-identical (adaptive_target_vol only reads its `[-win:]` tail).
        self._out_vals: list[float] = []
        self._w_held: dict[str, float] | None = None
        self._lev_held = 1.0
        self._consumed: set[int] = set()                 # days already advanced through the loop
        # Persistent killswitch over the book-return series (observed once per new day, in order).
        self._ks = DrawdownKillSwitch(kill_dd=kill_dd, reenter_dd=reenter_dd)
        self._ks_observed: set[int] = set()
        # edge_health: ONE persistent WindowedEdgeHealth per sleeve, advanced incrementally over that
        # sleeve's realized returns (resumable; reproduces the stateless replay exactly).
        self._sleeve_health_det: dict[str, WindowedEdgeHealth] = {}
        self._sleeve_health_consumed: dict[str, set[int]] = {}
        # combined: ONE persistent RetirementGate per sleeve, advanced incrementally over that sleeve's
        # realized returns (resumable; reproduces the stateless replay exactly).
        self._retire_gate: dict[str, RetirementGate] = {}
        self._retire_gate_consumed: dict[str, set[int]] = {}

    def _advance_book_returns(self, returns_by_sleeve: dict[str, dict[int, float]]) -> None:
        """Advance the resumable paper_book_returns loop over any days in `returns_by_sleeve` not yet
        consumed. Mirrors paper_book_returns's loop body EXACTLY (same order: emit out[d] from held
        weights, then re-size into d from strictly-before, then append d)."""
        all_days = sorted({d for r in returns_by_sleeve.values() for d in r})
        new_days = [d for d in all_days if d not in self._consumed]
        if not new_days:
            return
        # Keep `_before`'s key set aligned with returns_by_sleeve (new sleeves can appear over time).
        for s in returns_by_sleeve:
            self._before.setdefault(s, {})
        for d in new_days:
            if self._w_held is not None:
                bret = self._lev_held * sum(
                    self._w_held.get(s, 0.0) * returns_by_sleeve.get(s, {}).get(d, 0.0)
                    for s in self._w_held)
                self._out[d] = bret
                self._out_vals.append(bret)
            active = [s for s, r in self._before.items() if r]
            self._w_held = isv_adaptive_weights(self._before, active)
            tv = adaptive_target_vol(self._out_vals, base=self._target_vol)
            self._lev_held = vol_target_leverage(
                self._w_held, self._before, target_vol=tv, kelly_fraction=self._kelly_fraction,
                vol_lookback=self._vol_lookback, periods_per_year=self._periods_per_year,
                max_leverage=self._max_leverage)
            for s, r in returns_by_sleeve.items():
                if d in r:
                    self._before[s][d] = r[d]
            self._consumed.add(d)

    def _sleeve_health(self, returns_by_sleeve: dict[str, dict[int, float]],
                       active_sleeves: list[str]) -> dict[str, float]:
        """Advance each active sleeve's persistent WindowedEdgeHealth over its newly-appeared returns (in
        ascending order) and read `.health()`. `returns_by_sleeve` is append-only, so feeding only the new
        days in order reproduces a from-scratch replay over the full series exactly (deterministic EWMA)."""
        theta: dict[str, float] = {}
        for s in active_sleeves:
            if not s:
                continue
            det = self._sleeve_health_det.get(s)
            if det is None:
                det = WindowedEdgeHealth()
                self._sleeve_health_det[s] = det
                self._sleeve_health_consumed[s] = set()
            consumed = self._sleeve_health_consumed[s]
            for d in sorted(returns_by_sleeve.get(s, {})):
                if d not in consumed:
                    det.update(returns_by_sleeve[s][d])
                    consumed.add(d)
            theta[s] = det.health()
        return theta

    def _retirement_active(self, returns_by_sleeve: dict[str, dict[int, float]],
                           active_sleeves: list[str]) -> list[str]:
        """Advance each active sleeve's persistent RetirementGate over its newly-appeared returns (in
        ascending order) and keep the sleeve iff it is not retired. `returns_by_sleeve` is append-only,
        so feeding only the new days in order reproduces a from-scratch replay over the full series
        exactly (deterministic WEH + counters). Returns kept active sleeves in input order."""
        kept: list[str] = []
        for s in active_sleeves:
            if not s:
                continue
            gate = self._retire_gate.get(s)
            if gate is None:
                gate = RetirementGate()
                self._retire_gate[s] = gate
                self._retire_gate_consumed[s] = set()
            consumed = self._retire_gate_consumed[s]
            for d in sorted(returns_by_sleeve.get(s, {})):
                if d not in consumed:
                    gate.update(returns_by_sleeve[s][d])
                    consumed.add(d)
            if gate.status != "retired":
                kept.append(s)
        return kept

    def overlay(self, returns_by_sleeve: dict[str, dict[int, float]],
                active_sleeves: list[str]) -> tuple[dict[str, float], float, bool]:
        """One date's (weights, leverage, killed) — bit-identical to paper_risk_overlay(returns_by_sleeve,
        active_sleeves, controller=...) with the same parameters. `returns_by_sleeve` MUST be append-only
        across calls."""
        self._advance_book_returns(returns_by_sleeve)
        tv = adaptive_target_vol(self._out_vals, base=self._target_vol)
        if self._controller == "edge_health":
            theta = self._sleeve_health(returns_by_sleeve, active_sleeves)
            weights = _edge_health_weights(returns_by_sleeve, active_sleeves, theta_by_sleeve=theta)
            leverage = vol_target_leverage(
                weights, returns_by_sleeve, target_vol=tv, kelly_fraction=self._book_kelly,
                vol_lookback=self._vol_lookback, periods_per_year=self._periods_per_year,
                max_leverage=self._max_leverage)
            return weights, leverage, False
        # `combined`: structural-retirement gate filters the active set, else identical to killswitch.
        active = (self._retirement_active(returns_by_sleeve, active_sleeves)
                  if self._controller == "combined" else active_sleeves)
        weights = isv_adaptive_weights(returns_by_sleeve, active)
        leverage = vol_target_leverage(
            weights, returns_by_sleeve, target_vol=tv, kelly_fraction=self._kelly_fraction,
            vol_lookback=self._vol_lookback, periods_per_year=self._periods_per_year,
            max_leverage=self._max_leverage)
        for d in sorted(self._out):
            if d not in self._ks_observed:
                self._ks.observe(self._out[d])
                self._ks_observed.add(d)
        return weights, leverage, self._ks.killed


@dataclass(frozen=True)
class PositionView:
    sleeve: str
    symbol: str
    qty: float
    entry_price: float
    last: float
    pnl: float


@dataclass(frozen=True)
class SleeveView:
    weight: float
    pnl: float
    n: int


@dataclass(frozen=True)
class PaperView:
    capital: float
    equity: float
    positions: list[PositionView] = field(default_factory=list)
    sleeves: dict[str, SleeveView] = field(default_factory=dict)
    trades: list[Trade] = field(default_factory=list)


class PaperBookService:
    def __init__(self, repo: PaperRepo, capital: float) -> None:
        self._repo = repo
        self._capital = capital

    def derive_and_persist(self, run_date: str, sleeve_weights: dict[str, float],
                           symbol_weights_by_sleeve: dict[str, dict[str, float]],
                           prices: dict[str, float], at: dt.datetime,
                           leverage: float = 1.0, capital_base: float | None = None,
                           max_gross: float = 3.0) -> list[Position]:
        base = self._capital if capital_base is None else capital_base
        target: list[Position] = []
        for sleeve, sw in sleeve_weights.items():
            symbol_weights = symbol_weights_by_sleeve.get(sleeve, {})
            target.extend(target_positions(
                sleeve=sleeve, sleeve_weight=sw, capital=base * leverage,
                symbol_weights=symbol_weights, prices=prices, date=run_date))
        # GROSS-exposure cap (belt-and-suspenders): cap REALIZED gross (Σ|qty·entry|) at max_gross·equity.
        # Scales every position's qty by the same factor when raw gross exceeds the cap (accounts for
        # market-neutral legs automatically, since those net down the gross). Equity base is untouched.
        gross = sum(abs(p.qty * p.entry_price) for p in target)
        cap = max_gross * base
        if gross > cap and gross > 0:
            scale = cap / gross
            target = [dataclasses.replace(p, qty=p.qty * scale) for p in target]
        # Diff against the book as of the PREVIOUS run_date (not the latest), so re-running this run_date
        # reproduces the same trades + realized PnL — making the whole persist idempotent per run_date.
        prior = self._repo.positions_before(run_date)
        ts = at.isoformat()
        trades = diff_trades(prior, target, ts=ts)
        # Realized PnL from positions reduced/exited/flipped this rebalance, marked at the day's prices.
        realized = realized_delta(prior, target, prices)
        self._repo.replace_positions(run_date, target, weight_by_sleeve=sleeve_weights, at=at)
        # Idempotent per run_date: replace this date's trade ledger + realized contribution (re-run rewrites).
        self._repo.replace_trades(run_date, trades, at=at)
        self._repo.set_realized_for_run_date(run_date, realized)
        # Return the persisted target so callers can compute trading cost without re-reading the repo.
        return target

    def view(self, live: LivePrice) -> PaperView:
        positions = self._repo.latest_positions()
        symbols = sorted({p.symbol for p in positions})
        prices = live.get_prices(symbols) if symbols else {}

        # Compute per-lot PnL ONCE in a single loop; unrealized is the sum (no second mark_to_market pass).
        # The unknown-symbol fallback (`prices.get(sym, entry_price)`) lives in exactly one place — here.
        pos_views: list[PositionView] = []
        sleeve_pnl: dict[str, float] = {}
        sleeve_n: dict[str, int] = {}
        unrealized = 0.0
        for p in positions:
            last = prices.get(p.symbol, p.entry_price)   # unknown -> entry (0 pnl)
            pnl = p.qty * (last - p.entry_price)
            unrealized += pnl
            pos_views.append(PositionView(
                sleeve=p.sleeve, symbol=p.symbol, qty=p.qty,
                entry_price=p.entry_price, last=last, pnl=pnl))
            sleeve_pnl[p.sleeve] = sleeve_pnl.get(p.sleeve, 0.0) + pnl
            sleeve_n[p.sleeve] = sleeve_n.get(p.sleeve, 0) + 1

        weights = self._repo.weights_for(list(sleeve_pnl))   # one query, not weight_for per sleeve (N+1)
        sleeves = {
            sleeve: SleeveView(weight=weights.get(sleeve, 0.0), pnl=pnl, n=sleeve_n.get(sleeve, 0))
            for sleeve, pnl in sleeve_pnl.items()
        }
        return PaperView(
            capital=self._capital,
            # equity = nav curve through last close + intraday MTM since that close.
            # No double-count: nav curve is through last close; unrealized is the move since that close.
            equity=(self._repo.latest_nav_equity() or self._capital) + unrealized,
            positions=pos_views,
            sleeves=sleeves,
            trades=self._repo.recent_trades(RECENT_TRADES_LIMIT))


def persist_paper_book(repo: PaperRepo, *, capital: float, run_date: str,
                       sleeve_weights: dict[str, float],
                       symbol_weights_by_sleeve: dict[str, dict[str, float]],
                       prices: dict[str, float], at: dt.datetime,
                       enabled: bool = True, leverage: float = 1.0, max_gross: float = 3.0,
                       cost_rates: dict[str, float] | None = None,
                       funding_by_symbol: dict[str, float] | None = None,
                       carry_sleeves: frozenset[str] | None = None,
                       carry_ret_by_symbol: dict[str, float] | None = None,
                       return_by_sleeve_symbol: dict[str, dict[str, float]] | None = None,
                       cost_frac: float | None = None) -> int:
    """The daily book-rebalance hook (T7). Given the three things the book rebalance produces —
    per-sleeve allocation weights, each sleeve's per-symbol target weights, and the day's prices —
    derive + persist the paper positions/trades and return the number of open positions.

    This is the exact, single call site the daily forward/book advance will invoke. It is split out
    of `PaperBookService` so the advance path can call ONE pure function guarded by `paper_enabled`
    without constructing the service inline. `enabled=False` (from `settings.paper_enabled`) makes it
    a no-op.

    Equity accounting is COMPOUNDING mark-to-market: today's equity = the prior equity plus the PRIOR
    held book marked to today's close (Σ positions_before.qty·(close − entry)). The throttled book is
    then sized off that running equity, so gains reinvest (compound). positions_before /
    nav_equity_before are strictly-prior reads, so this is causal and idempotent: re-running an
    ascending backfill reproduces identical nav rows.

    NOTE: the current forward/book advance carries only daily *returns* and does not yet expose the
    three inputs above, so the automatic wiring from the live advance is not yet active — see the
    project report. This function is the ready-to-wire honest seam, fully covered by tests.

    COST MODEL (toggleable, default OFF). When `cost_rates is None AND funding_by_symbol is None` this
    behaves EXACTLY as the legacy gross path (no funding DRAG, no trading cost) — bit-identical, backward-
    compatible. When `funding_by_symbol` is given, the PRIOR book's signed funding cash flow DRAG is added
    to equity BEFORE sizing for DIRECTIONAL (non-carry) sleeves (longs pay positive funding, shorts
    receive). When `cost_rates` is given, the turnover from the prior book to the persisted target is
    charged as `trading_cost` and SUBTRACTED from equity after sizing (the book is sized off equity_pre;
    the bp-of-turnover cost is a small adjustment).

    CARRY (xsfunding) — FULL-return accounting. Sleeves named in `carry_sleeves` are delta-neutral: their
    P&L is the FULL daily carry return `carry_pnl = Σ (qty·entry)·carry_ret` over the PRIOR carry positions,
    where `carry_ret = (spot_ret − perp_ret) + funding` is the per-symbol full carry return from
    `carry_ret_by_symbol` (funding accrual PLUS the daily basis change — the spot+perp legs' price P&L,
    where the LUNA-class basis tail lives). This is the cross-sectional `Σ w·carry_ret` convention (OPPOSITE
    sign to the perp-holder `funding_pnl` drag) and it is applied WHENEVER `carry_ret_by_symbol` is supplied
    — it is P&L, not a cost, so it does NOT depend on the `cost_rates` toggle (the GROSS curve still carries
    the edge; the toggle only governs the directional funding DRAG + trading cost). Carry sleeves are
    EXCLUDED from the `funding_pnl` drag (else the same legs would be double-counted with the opposite sign).
    `carry_pnl/Σ|qty·entry|` equals the `sleeve_daily_carry` return booked into `paper_sleeve_ret` (same map),
    so the equity contribution and the sizing return are consistent → live == replay for carry. BASIS
    RESOLUTION CAVEAT: the basis is now charged at DAILY-close resolution; the remaining gap is the INTRADAY
    basis-squeeze / liquidation tail (1m `worst_basis`), a documented future refinement — the Sharpe is no
    longer fake.

    RETURN-DEFINED BOOK (`return_by_sleeve_symbol`) — the honest multi-edge mode used by the Bybit book.
    `{sleeve: {symbol: realized_return}}` carries each edge's TRUE per-symbol daily return (carry =
    basis + FUNDING; unlock = short_leg_pnl incl. BTC hedge + funding + squeeze; trend/positioning =
    price). When supplied, today's book P&L is booked CONTEMPORANEOUSLY on TODAY's target weights:
        gain = e_prev · Σ_sleeve sleeve_weight · Σ_symbol symbol_weight · return[sleeve][symbol]
    which is EXACTLY the reconstruction's `(1/n)·Σ_sleeve sleeve_true_gross` (the reconciliation
    invariant) — so the live book == the cockpit Backtest, with NO 1-day establishment lag and NO
    naked-perp mis-mark of the delta-neutral / hedged sleeves. This REPLACES the prior-positions
    price-MTM `gain` (and the funding/carry args, which the per-symbol return already subsumes); the
    sized positions are kept (for the cockpit position view). Gated: when `return_by_sleeve_symbol is
    None` this whole branch is skipped and the legacy price-MTM path below runs bit-identically (the
    legacy-venue book + every existing caller is untouched).

    COST (`cost_frac`) — the per-COIN MEASURED cost as a fraction of start-of-day equity (`Σ_coin
    |Δw_coin|·cost_bps(coin)/1e4`, where `cost_bps = taker_fee + 0.5·real_spread(coin)` — the SAME
    per-coin turnover cost `per_coin_book_returns` / `_bybit_measured_net` charge). When supplied,
    `tcost = e_prev·cost_frac`, so `equity = e_prev·(1 + day_ret − cost_frac) = e_prev·(1 + NET)` and the
    live `/paper` NET == the cockpit Backtest NET BY CONSTRUCTION. When `cost_frac is None` the cost falls
    back to the legacy flat per-sleeve `cost_rates` turnover haircut (the pre-real-spread behaviour).
    """
    if not enabled:
        return 0
    if return_by_sleeve_symbol is not None:
        # Contemporaneous return-defined book: book TODAY's weights × each edge's TRUE per-symbol return
        # (== the reconstruction's (1/n)·Σ sleeve_true_gross). Fixes the live Bybit book that previously
        # marked EVERY sleeve to raw perp price — dropping the carry sleeve's FUNDING (the literal edge)
        # and the unlock sleeve's hedge, and lagging every sleeve a day.
        prior = repo.positions_before(run_date)
        e_prev = repo.nav_equity_before(run_date)
        if e_prev is None:
            e_prev = capital                                   # day 0 seed
        day_ret = 0.0
        for sleeve, sw in sleeve_weights.items():
            if sw == 0.0:
                continue
            rmap = return_by_sleeve_symbol.get(sleeve, {})
            sym_w = symbol_weights_by_sleeve.get(sleeve, {})
            day_ret += sw * sum(w * rmap.get(sym, 0.0) for sym, w in sym_w.items())
        gain = e_prev * leverage * day_ret
        # Size the day's book off the START-of-day equity (e_prev), NOT e_prev+gain: the weight is HELD
        # DURING the day and earns `day_ret`, so it is sized before that return is realized. (NAV is
        # weight-based and independent of the sizing base; this only sets the position qty + turnover.)
        target = PaperBookService(repo, capital).derive_and_persist(
            run_date=run_date, sleeve_weights=sleeve_weights,
            symbol_weights_by_sleeve=symbol_weights_by_sleeve, prices=prices, at=at,
            leverage=leverage, capital_base=e_prev, max_gross=max_gross)
        # Trading cost. `cost_frac` (the per-COIN MEASURED cost as a fraction of equity — `Σ_coin |Δw_coin|·
        # cost_bps(coin)/1e4`, the SAME per-coin turnover cost `per_coin_book_returns`/`_bybit_measured_net`
        # charge) takes precedence: tcost = e_prev·cost_frac, so the live NET == the cockpit Backtest NET by
        # construction (same gross, same per-coin cost). When `cost_frac is None` it falls back to the legacy
        # flat per-sleeve `cost_rates` turnover haircut.
        if cost_frac is not None:
            tcost = e_prev * cost_frac
        else:
            tcost = trading_cost(prior, target, prices, rate_by_sleeve=cost_rates) if cost_rates else 0.0
        equity = e_prev + gain - tcost
        repo.upsert_nav(run_date, capital=capital, realized=equity - capital - gain,
                        unrealized=gain, equity=equity, at=at)
        return len(repo.positions_at(run_date))
    carry = carry_sleeves or frozenset()
    # Compounding mark-to-market: today's equity = prior equity + the PRIOR book marked to today's close.
    # (positions_before / nav_equity_before are strictly-prior, so this is causal + idempotent.)
    prior = repo.positions_before(run_date)
    gain = sum(p.qty * (prices.get(p.symbol, p.entry_price) - p.entry_price) for p in prior)
    e_prev = repo.nav_equity_before(run_date)
    if e_prev is None:
        e_prev = capital                                   # day 0 seed
    # Funding cash-flow DRAG on the PRIOR held DIRECTIONAL book (longs pay positive funding, shorts
    # receive). Carry sleeves are excluded here — their funding is booked as carry-ACCRUAL P&L below
    # (opposite sign), so applying the drag too would double-count the same legs. 0 when funding is off.
    directional_prior = [p for p in prior if p.sleeve not in carry] if carry else prior
    fund = funding_pnl(directional_prior, funding_by_symbol, prices) if funding_by_symbol else 0.0
    # Carry FULL-return P&L on the PRIOR carry book: Σ (qty·entry)·carry_ret where carry_ret =
    # (spot_ret − perp_ret) + funding (funding accrual PLUS the daily basis change). Applied whenever
    # `carry_ret_by_symbol` is supplied (P&L, not a cost; independent of cost_rates / the directional
    # drag). carry_pnl/gross == sleeve_daily_carry (same map) → consistent with the persisted sleeve
    # return → live == replay for carry.
    carry_prior = [p for p in prior if p.sleeve in carry] if carry else []
    carry_gain = (carry_pnl(carry_prior, carry_ret_by_symbol)
                  if (carry_prior and carry_ret_by_symbol) else 0.0)
    equity_pre = e_prev + gain + fund + carry_gain
    # Size today's throttled book off the COMPOUNDED (+funding) equity (gains reinvest).
    target = PaperBookService(repo, capital).derive_and_persist(
        run_date=run_date, sleeve_weights=sleeve_weights,
        symbol_weights_by_sleeve=symbol_weights_by_sleeve, prices=prices, at=at,
        leverage=leverage, capital_base=equity_pre, max_gross=max_gross)
    # Trading cost from the prior book to the persisted target (turnover * bp by sleeve). 0 when off.
    tcost = trading_cost(prior, target, prices, rate_by_sleeve=cost_rates) if cost_rates else 0.0
    equity = equity_pre - tcost
    # realized = cumulative through prior day (incl. funding + costs booked into equity); unrealized =
    # today's MTM gain; equity = capital + realized + unrealized.
    repo.upsert_nav(run_date, capital=capital, realized=equity - capital - gain,
                    unrealized=gain, equity=equity, at=at)
    return len(repo.positions_at(run_date))
