"""READ-ONLY evaluator for a NEW Bybit edge: OI-DELEVERAGING REVERSION (open-interest unwind overshoot).

Economic thesis
---------------
A sharp DROP in open interest alongside a price move is the fingerprint of a FORCED unwind — longs liquidated
(price down) or a short squeeze (price up). Forced flow overshoots; the move reverts next day. Bybit is
retail-heavy → frequent cascades. This is a FLOW event (a forced ΔOI), NOT a positioning LEVEL — so it should
be uncorrelated with the existing positioning (account-ratio LEVEL) edge.

The signal (per coin/day), documented exactly
---------------------------------------------
From the `open_interest` panel and the perp `close` panel:
  * `doi_t = (OI_t - OI_{t-1}) / OI_{t-1}`            — the day's OI change RATE;
  * `ret_t = close_t / close_{t-1} - 1`              — the day's close-to-close return (the move that day).
Both are knowable AT THE END OF day t. The DELEVERAGING-OVERSHOOT signal on day t:

      signal_t = -sign(ret_t) * (|doi_t| * |ret_t|)     when doi_t < 0   (OI unwinding → forced move)
      signal_t = 0                                       when doi_t >= 0   (OI building → not forced)

So on an OI-DROP day we FADE that day's move, with magnitude scaling with the size of the unwind AND the size
of the move (bigger unwind + bigger move ⇒ stronger expected overshoot). On OI-BUILD days the signal is zero
(the move is fresh positioning, not a forced unwind).

`direction="reversion"` (DEFAULT) uses the signal as defined above (fade the forced move). `direction=
"momentum"` is the EXACT SIGN-FLIP (the forced move continues) — the same robustness probe positioning runs.

The bet is on PRICE: the day-t signal forms cross-sectional, market-neutral, unit-gross weights applied to
the NEXT-day (t→t+1) price return, so the signal is strictly causal (no look-ahead).

Reuse (no return-engine reimplemented)
--------------------------------------
The per-coin daily PRICE return is close-to-close off the Bybit perp `close` panel. The signal comes from the
`open_interest` + `close` panels. The standard `_curve_metrics` wraps the series. The adversarial verify
mirrors the positioning verify shape: both directions net @ 5.5/11/22 bp, per-coin top-5 share, per-year
Sharpe, turnover, AND the correlation of THIS edge's returns with EACH of the FOUR existing edges (tstrend,
unlock, xsfunding, positioning) so additivity is visible.

READ-ONLY + memory-bounded
--------------------------
No PaperRepo, no `store.write_*` — the live paper_* tables are never touched. With a `universe` the reads go
through `_UniverseRestrictedStore` (a `WHERE symbol IN (...)` read), so the full all-symbol panels are never
materialised.
"""
from __future__ import annotations

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

from fxhnt.application.bybit_book_eval import _bounded_store, sleeve_returns_from_store
from fxhnt.application.bybit_carry_revalidate import _curve_metrics

_PERIODS_PER_YEAR = 365
# All FOUR existing edges — including positioning, since OI-reversion must be additive to it specifically
# (if it just re-expresses the positioning LEVEL, it is not a new edge).
_EXISTING_EDGES = ("tstrend", "unlock", "xsfunding", "positioning")
_DIRECTIONS = ("reversion", "momentum")


class _FeatureStore(Protocol):
    """Just the panel reads the OI pipeline needs (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]]: ...


def _epoch_day_to_iso(day: int) -> str:
    return (dt.date(1970, 1, 1) + dt.timedelta(days=int(day))).isoformat()


def _year_of_day(day: int) -> str:
    return str((dt.date(1970, 1, 1) + dt.timedelta(days=int(day))).year)


# --- 1. the per-coin/day deleveraging signal ---------------------------------------------------

def _doi_by_day(oi_panel: dict[str, dict[int, float]]) -> dict[int, dict[str, float]]:
    """`{day: {coin: doi}}` where doi_t = (OI_t - OI_{t-1})/OI_{t-1} is the day-t OI change RATE. Keyed by the
    LATER day t (the day on which the change is realized). A non-positive prior OI (can't form a rate) is
    skipped. Pure; no I/O."""
    out: dict[int, dict[str, float]] = {}
    for coin, series in oi_panel.items():
        days = sorted(series)
        for i in range(1, len(days)):
            d0, d1 = days[i - 1], days[i]
            o0 = series[d0]
            if o0 and o0 > 0.0 and math.isfinite(o0):
                out.setdefault(int(d1), {})[coin] = series[d1] / o0 - 1.0
    return out


def _ret_by_day(close_panel: dict[str, dict[int, float]]) -> dict[int, dict[str, float]]:
    """`{day: {coin: ret}}` where ret_t = close_t/close_{t-1} - 1 is the day-t close-to-close return, keyed by
    the LATER day t (the day on which the move happened — knowable AT END of day t, same as doi). Pure."""
    out: dict[int, dict[str, float]] = {}
    for coin, series in close_panel.items():
        days = sorted(series)
        for i in range(1, len(days)):
            d0, d1 = days[i - 1], days[i]
            p0 = series[d0]
            if p0 and math.isfinite(p0):
                out.setdefault(int(d1), {})[coin] = series[d1] / p0 - 1.0
    return out


def deleverage_signal_by_day(doi_by_day: dict[int, dict[str, float]],
                             ret_by_day: dict[int, dict[str, float]], *,
                             direction: str = "reversion") -> dict[int, dict[str, float]]:
    """`{day: {coin: raw_signal}}` the DELEVERAGING-OVERSHOOT signal per coin/day (see module docstring):

        signal = -sign(ret) * (|doi| * |ret|)   on OI-DROP days (doi < 0); 0 on OI-build days (doi >= 0).

    `direction="reversion"` (DEFAULT) uses it as-is (fade the forced move); `direction="momentum"` is the
    EXACT SIGN-FLIP (continuation). Only days where BOTH doi and ret are known for a coin contribute. Pure."""
    if direction not in _DIRECTIONS:
        raise ValueError(f"direction must be 'reversion' or 'momentum', got {direction!r}")
    sign = 1.0 if direction == "reversion" else -1.0
    out: dict[int, dict[str, float]] = {}
    for day, doi_row in doi_by_day.items():
        ret_row = ret_by_day.get(day, {})
        row: dict[str, float] = {}
        for coin, doi in doi_row.items():
            if coin not in ret_row:
                continue
            ret = ret_row[coin]
            if doi >= 0.0:
                row[coin] = 0.0                              # OI building → not a forced unwind → no signal
            else:
                mag = abs(doi) * abs(ret)
                # -sign(ret) fades the move; the reversion/momentum flag flips the whole signal.
                row[coin] = sign * (-_sign(ret)) * mag
        if row:
            out[int(day)] = row
    return out


def _sign(x: float) -> float:
    return 1.0 if x > 0.0 else (-1.0 if x < 0.0 else 0.0)


# --- 2. the cross-sectional, market-neutral weights --------------------------------------------

def deleverage_weights(signal_by_day: dict[int, dict[str, float]]) -> dict[int, dict[str, float]]:
    """`{day: {coin: w}}` market-neutral, unit-gross weights from the raw per-coin deleveraging signal.

    Construction per day:
      1. DEMEAN the per-day signal across coins (so the book is dollar/market-neutral, Σw ≈ 0);
      2. normalise to UNIT GROSS (Σ|w| = 1).
    A day with <2 coins, or zero gross after demeaning (e.g. all signals zero on an all-OI-build day), yields
    no weights (skipped). The reversion/momentum sign is already baked into the raw signal upstream, so the
    momentum weights are the exact per-coin negation of reversion (gross is identical). Pure."""
    out: dict[int, dict[str, float]] = {}
    for day, row in signal_by_day.items():
        if len(row) < 2:
            continue
        mean = st.mean(row.values())
        raw = {c: v - mean for c, v in row.items()}
        gross = sum(abs(w) for w in raw.values())
        if gross <= 0.0:
            continue
        out[int(day)] = {c: w / gross for c, w in raw.items()}
    return out


# --- 3. the read-only, memory-bounded per-day book breakdown -----------------------------------

def _next_day_returns(close_panel: dict[str, dict[int, float]]) -> dict[int, dict[str, float]]:
    """`{day: {coin: ret(day -> day+1)}}` close-to-close per coin. A weight formed on day d is applied to THIS
    return (the realized NEXT-day price move), keeping the signal causal. Pure."""
    out: dict[int, dict[str, float]] = {}
    for coin, series in close_panel.items():
        days = sorted(series)
        for i in range(len(days) - 1):
            d0, d1 = days[i], days[i + 1]
            p0 = series[d0]
            if p0 and math.isfinite(p0):
                out.setdefault(int(d0), {})[coin] = series[d1] / p0 - 1.0
    return out


def _panel_symbols(store: _FeatureStore, universe: set[str] | None) -> set[str]:
    """The candidate symbol set for the open_interest read. With a universe → that set (already the memory
    bound). Without → the close-panel symbols (so we read OI only for coins that have prices)."""
    if universe is not None:
        return set(universe)
    return set(store.crypto_close_panel())


def _signal_panels(store: _FeatureStore, *, universe: set[str] | None,
                   ) -> tuple[dict[int, dict[str, float]], dict[int, dict[str, float]],
                              dict[int, dict[str, float]], dict[int, dict[str, float]]]:
    """ONE read pass -> (doi_by_day, ret_by_day, next_ret causal, same_day leaked).

    `doi_by_day`/`ret_by_day` = day-t OI-change-rate / day-t return (the SIGNAL inputs, both knowable at end
    of day t); `next_ret` = `{day: {coin: ret(day->day+1)}}` (the CAUSAL mapping the live edge uses);
    `same_day` = `{day: {coin: ret(day-1->day)}}` (== ret_by_day; the LEAKED mapping, used ONLY by the
    look-ahead audit as the negative control). Memory-bounded (universe-restricted) + READ-ONLY."""
    bounded = _bounded_store(store, universe)
    oi_panel = bounded.read_panel(sorted(_panel_symbols(bounded, universe)), "open_interest")
    oi_panel = {s: v for s, v in oi_panel.items() if v}
    close_panel = bounded.crypto_close_panel()
    if not oi_panel or not close_panel:
        return {}, {}, {}, {}
    doi_by_day = _doi_by_day(oi_panel)
    ret_by_day = _ret_by_day(close_panel)
    next_ret = _next_day_returns(close_panel)
    return doi_by_day, ret_by_day, next_ret, ret_by_day


def _series_from_weights(weights_by_day: dict[int, dict[str, float]],
                         next_ret: dict[int, dict[str, float]], *, cost_bps: float,
                         day_lo: int | None = None, day_hi: int | None = None,
                         exclude: frozenset[str] = frozenset(),
                         ) -> tuple[dict[int, float], dict[str, float]]:
    """`(net_series {day: ret}, per_coin {coin: Σ contrib})` from precomputed weights + causal returns.

    Optional `[day_lo, day_hi]` (inclusive epoch-day window) restricts the days; `exclude` drops coins BEFORE
    weighting (re-normalised to unit gross over the kept coins). Turnover (cost driver) is recomputed from the
    surviving weights, NOT negated. Pure."""
    net: dict[int, float] = {}
    per_coin: dict[str, float] = {}
    prev_w: dict[str, float] = {}
    for day in sorted(weights_by_day):
        if (day_lo is not None and day < day_lo) or (day_hi is not None and day > day_hi):
            continue
        raw = weights_by_day[day]
        if exclude:
            raw = {c: w for c, w in raw.items() if c not in exclude}
        gross = sum(abs(w) for w in raw.values())
        rets_d = next_ret.get(day, {})
        if gross <= 0.0 or not rets_d:
            prev_w = {}
            continue
        weights = {c: w / gross for c, w in raw.items()}   # re-normalise survivors to unit gross
        book_gross = sum(w * rets_d.get(s, 0.0) for s, w in weights.items())
        turnover = sum(abs(weights.get(s, 0.0) - prev_w.get(s, 0.0))
                       for s in set(weights) | set(prev_w))
        book_net = book_gross - (turnover * (cost_bps / 1e4) if cost_bps else 0.0)
        if math.isfinite(book_net):
            net[int(day)] = float(book_net)
            for s, w in weights.items():
                per_coin[s] = per_coin.get(s, 0.0) + w * rets_d.get(s, 0.0)
        prev_w = weights
    return net, per_coin


def _book_breakdown(store: _FeatureStore, *, universe: set[str] | None, cost_bps: float,
                    direction: str) -> dict[str, Any]:
    """READ-ONLY, memory-bounded per-day book breakdown for the OI-deleveraging edge in ONE pass.

    Returns `net` ({day: net after cost}), `gross` ({day: gross}), `per_coin` ({coin: Σ contrib}),
    `turnover` ({day: Σ|Δw|}). `direction` flips the raw-signal sign (reversion vs momentum); costs are
    re-derived from the actual flipped weights, NOT by negating a number. Writes NOTHING."""
    empty: dict[str, Any] = {"net": {}, "gross": {}, "per_coin": {}, "turnover": {}}
    doi_by_day, ret_by_day, next_ret, _ = _signal_panels(store, universe=universe)
    if not doi_by_day:
        return empty
    signal = deleverage_signal_by_day(doi_by_day, ret_by_day, direction=direction)
    weights_by_day = deleverage_weights(signal)
    if not weights_by_day:
        return empty

    net: dict[int, float] = {}
    gross_s: dict[int, float] = {}
    turnover_s: dict[int, float] = {}
    per_coin: dict[str, float] = {}
    prev_w: dict[str, float] = {}
    for day in sorted(weights_by_day):
        weights = weights_by_day[day]
        gross = sum(abs(w) for w in weights.values())
        rets_d = next_ret.get(day, {})
        if gross <= 0.0 or not rets_d:
            prev_w = {}
            continue
        book_gross = sum(w * rets_d.get(s, 0.0) for s, w in weights.items()) / gross
        turnover = sum(abs(weights.get(s, 0.0) - prev_w.get(s, 0.0))
                       for s in set(weights) | set(prev_w)) / gross
        book_net = book_gross - (turnover * (cost_bps / 1e4) if cost_bps else 0.0)
        if math.isfinite(book_net):
            net[int(day)] = float(book_net)
            gross_s[int(day)] = float(book_gross)
            turnover_s[int(day)] = float(turnover)
            for s, w in weights.items():
                per_coin[s] = per_coin.get(s, 0.0) + (w * rets_d.get(s, 0.0)) / gross
        prev_w = weights
    return {"net": net, "gross": gross_s, "per_coin": per_coin, "turnover": turnover_s}


def oi_deleverage_returns_from_store(store: _FeatureStore, *, universe: set[str] | None = None,
                                     cost_bps: float = 5.5, direction: str = "reversion",
                                     ) -> dict[int, float]:
    """READ-ONLY, memory-bounded per-day RETURN SERIES `{epoch_day: ret}` for the OI-deleveraging edge.
    `direction="reversion"` (default) fades forced moves; `"momentum"` is the sign-flip. Reads restricted to
    `universe` (the memory bound). Writes NOTHING."""
    return _book_breakdown(store, universe=universe, cost_bps=cost_bps, direction=direction)["net"]


def _n_coins(store: _FeatureStore, universe: set[str] | None) -> int:
    """The number of coins with an open_interest panel (∩ universe) — the tradeable breadth."""
    bounded = _bounded_store(store, universe)
    panel = bounded.read_panel(sorted(_panel_symbols(bounded, universe)), "open_interest")
    return sum(1 for v in panel.values() if v)


def oi_deleverage_metrics_from_store(store: _FeatureStore, *, universe: set[str] | None = None,
                                     direction: str = "reversion", cost_bps: float = 5.5,
                                     ) -> dict[str, Any]:
    """READ-ONLY metrics for the OI-deleveraging edge (the standard `_curve_metrics` shape PLUS `n_coins`).

    `available=False` (+ `reason`) when there is no open_interest data / too few days. Memory-bounded (reads
    restricted to `universe` when supplied) + READ-ONLY (no paper_*/warehouse writes)."""
    series = oi_deleverage_returns_from_store(store, universe=universe, cost_bps=cost_bps,
                                              direction=direction)
    n_coins = _n_coins(store, universe)
    if len(series) < 2:
        return {"cagr": 0.0, "sharpe": 0.0, "max_dd": 0.0, "total_return": 0.0, "ann_return": 0.0,
                "days": 0, "symbols": 0, "first_day": None, "last_day": None, "n_coins": n_coins,
                "available": False,
                "reason": "OI-deleverage curve has <2 days (run `fxhnt bybit-ingest-oi` + klines first)"}
    daily = [(d, series[d]) for d in sorted(series)]
    meta = {"symbols": n_coins, "first_day": _epoch_day_to_iso(daily[0][0]),
            "last_day": _epoch_day_to_iso(daily[-1][0])}
    m = _curve_metrics(daily, days_meta=meta)
    m["available"] = True
    m["n_coins"] = n_coins
    return m


# --- 4. adversarial verification ---------------------------------------------------------------

def _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."""
    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 _existing_edge_series(store: _FeatureStore, *, universe: set[str] | None, cost_bps: float,
                          unlock_events: list[dict[str, Any]] | None,
                          ) -> dict[str, dict[int, float]]:
    """The per-day return series for each EXISTING Bybit edge (tstrend, unlock, xsfunding, positioning), via
    the live sleeve runners (`sleeve_returns_from_store`). unlock needs the injected calendar; absent → empty
    series. Memory-bounded (universe-restricted reads). READ-ONLY."""
    out: dict[str, dict[int, float]] = {}
    for edge in _EXISTING_EDGES:
        out[edge] = sleeve_returns_from_store(
            store, edge, universe=universe, cost_bps=cost_bps, unlock_events=unlock_events)
    return out


def _metrics_from_series(series: dict[int, float]) -> dict[str, Any]:
    """`_curve_metrics` over a `{day: ret}` series, plus `sum_gross_ret` (the additive arithmetic Σ)."""
    daily = [(d, series[d]) for d in sorted(series)]
    meta = {"symbols": 0, "first_day": _epoch_day_to_iso(daily[0][0]) if daily else None,
            "last_day": _epoch_day_to_iso(daily[-1][0]) if daily else None}
    m = _curve_metrics(daily, days_meta=meta)
    m["sum_gross_ret"] = sum(series.values())
    return m


def verify_oi_deleverage_edge(store: _FeatureStore, *, universe: set[str] | None = None,
                              cost_bps: float = 5.5, cost_grid: tuple[float, ...] = (5.5, 11.0, 22.0),
                              unlock_events: list[dict[str, Any]] | None = None,
                              ) -> dict[str, Any]:
    """READ-ONLY adversarial verification of the OI-deleveraging edge in BOTH directions (reversion/momentum).

    Returns (and the CLI prints) per direction: `net_cost[dir][bps]` (metrics net of cost at each bps),
    `per_coin[dir]` + `top5_share[dir]` (1-2-coin ⇒ fragile), `per_year[dir]` (all-one-year ⇒ overfit),
    `avg_turnover[dir]`, and `corr_edges[dir]` ({tstrend, unlock, xsfunding, positioning: corr|None} of the
    net @ `cost_bps` series with EACH existing edge — the additive/uncorrelated check, ESPECIALLY positioning).

    Memory-bounded + READ-ONLY. `available=False` (+ reason) when there is no data / too few days. The verdict
    guide: robust only if the winning direction stays Sharpe>0 at 22bps AND is not 1-2-coin AND positive in
    >1 year AND uncorrelated with ALL FOUR existing edges (esp. positioning)."""
    grid = tuple(sorted(set(cost_grid) | {cost_bps}))
    edge_series = _existing_edge_series(store, universe=universe, cost_bps=cost_bps,
                                        unlock_events=unlock_events)

    net_cost: dict[str, dict[float, dict[str, Any]]] = {}
    gross_total: dict[str, float] = {}
    per_coin: dict[str, dict[str, float]] = {}
    top5_share: dict[str, float] = {}
    per_year: dict[str, dict[str, dict[str, Any]]] = {}
    avg_turnover: dict[str, float] = {}
    corr_edges: dict[str, dict[str, float | None]] = {}

    n_days = 0
    for d in _DIRECTIONS:
        bd = _book_breakdown(store, universe=universe, cost_bps=0.0, direction=d)
        per_coin[d] = dict(bd["per_coin"])
        gross_total[d] = sum(bd["gross"].values())
        contribs = sorted((abs(v) for v in per_coin[d].values()), reverse=True)
        tot_abs = sum(contribs)
        top5_share[d] = (sum(contribs[:5]) / tot_abs) if tot_abs > 0 else 0.0
        tov = bd["turnover"]
        avg_turnover[d] = (sum(tov.values()) / len(tov)) if tov else 0.0

        net_cost[d] = {}
        net_at_headline: dict[int, float] = {}
        for bps in grid:
            net_series = {day: bd["gross"][day] - tov.get(day, 0.0) * (bps / 1e4)
                          for day in bd["gross"]}
            net_cost[d][bps] = _metrics_from_series(net_series)
            if bps == cost_bps:
                net_at_headline = net_series
        if not net_at_headline:
            net_at_headline = {day: bd["gross"][day] - tov.get(day, 0.0) * (cost_bps / 1e4)
                               for day in bd["gross"]}
        n_days = max(n_days, len(net_at_headline))

        by_year: dict[str, dict[int, float]] = {}
        for day, r in net_at_headline.items():
            by_year.setdefault(_year_of_day(day), {})[day] = r
        per_year[d] = {}
        for yr, ser in sorted(by_year.items()):
            ym = _metrics_from_series(ser)
            per_year[d][yr] = {"sharpe": ym["sharpe"], "sum_ret": ym["sum_gross_ret"], "days": ym["days"]}

        corr_edges[d] = {edge: _correlation(net_at_headline, edge_series.get(edge, {}))
                         for edge in _EXISTING_EDGES}

    if n_days < 2:
        return {"available": False,
                "reason": "OI-deleverage curve has <2 days (run the OI + kline ingests first)",
                "n_coins": _n_coins(store, universe)}

    return {"available": True, "n_coins": _n_coins(store, universe), "cost_bps": cost_bps,
            "cost_grid": list(grid), "net_cost": net_cost, "gross_total": gross_total,
            "per_coin": per_coin, "top5_share": top5_share, "per_year": per_year,
            "avg_turnover": avg_turnover, "corr_edges": corr_edges}


# --- 5. OOS / WALK-FORWARD validation (the DEPLOY GATE) ----------------------------------------

def _sharpe(series: dict[int, float]) -> float:
    """Annualised Sharpe (mean/pstdev × √365) of a `{day: ret}` series; <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


def _total_return(series: dict[int, float]) -> float:
    """Compounded total return of a `{day: ret}` series."""
    equity = 1.0
    for d in sorted(series):
        equity *= (1.0 + series[d])
    return equity - 1.0


def _rolling_windows(net: dict[int, float], *, window_days: int) -> dict[str, Any]:
    """Split the day range into CONSECUTIVE non-overlapping `window_days`-wide windows; per window report
    Sharpe + total-return + days, and the FRACTION of windows with positive total return. Windows with <2
    days are skipped."""
    if not net:
        return {"n_windows": 0, "fraction_positive": 0.0, "windows": []}
    days = sorted(net)
    lo, hi = days[0], days[-1]
    windows: list[dict[str, Any]] = []
    start = lo
    while start <= hi:
        end = start + window_days - 1
        ser = {d: net[d] for d in days if start <= d <= end}
        if len(ser) >= 2:
            windows.append({"start": _epoch_day_to_iso(start),
                            "end": _epoch_day_to_iso(min(end, hi)),
                            "sharpe": _sharpe(ser),
                            "total_return": _total_return(ser),
                            "days": len(ser)})
        start = end + 1
    n = len(windows)
    frac = (sum(1 for w in windows if w["total_return"] > 0.0) / n) if n else 0.0
    return {"n_windows": n, "fraction_positive": frac, "windows": windows}


def _train_holdout(signal_rev: dict[int, dict[str, float]], signal_mom: dict[int, dict[str, float]],
                   next_ret: dict[int, dict[str, float]], *, cost_bps: float,
                   train_frac: float = 0.6) -> dict[str, Any]:
    """The KEY OOS check. Split time into TRAIN (first `train_frac`) and TEST (the rest, never looked at).
    On TRAIN, choose the winning DIRECTION (reversion vs momentum by TRAIN Sharpe). Apply THAT direction to
    TEST. Report the chosen direction + TRAIN Sharpe + TEST (OOS) Sharpe. Guards the after-the-fact
    'reversion' choice — the edge generalises only if the train-chosen direction is positive OOS."""
    weights_by_dir = {"reversion": deleverage_weights(signal_rev),
                      "momentum": deleverage_weights(signal_mom)}
    days = sorted(set(weights_by_dir["reversion"]))
    if len(days) < 4:
        return {"train_direction": None, "train_sharpe": 0.0, "test_sharpe": 0.0,
                "train_days": 0, "test_days": 0, "available": False}
    split_idx = max(1, min(len(days) - 1, int(len(days) * train_frac)))
    train_hi = days[split_idx - 1]
    test_lo = days[split_idx]

    best_dir, best_sharpe = "reversion", -math.inf
    for d in _DIRECTIONS:
        tr_net, _ = _series_from_weights(weights_by_dir[d], next_ret, cost_bps=cost_bps, day_hi=train_hi)
        s = _sharpe(tr_net)
        if s > best_sharpe:
            best_dir, best_sharpe = d, s

    test_net, _ = _series_from_weights(weights_by_dir[best_dir], next_ret, cost_bps=cost_bps, day_lo=test_lo)
    return {"train_direction": best_dir, "train_sharpe": best_sharpe,
            "test_sharpe": _sharpe(test_net),
            "train_days": split_idx, "test_days": len(days) - split_idx,
            "split_at": _epoch_day_to_iso(test_lo), "available": True}


def _drop_top_n(weights_by_day: dict[int, dict[str, float]], next_ret: dict[int, dict[str, float]], *,
                cost_bps: float, train_frac: float = 0.6) -> dict[str, Any]:
    """Recompute the edge EXCLUDING the top-5 and top-10 gross-contributing coins, identified CAUSALLY on the
    TRAIN slice (never on the full/test sample). If Sharpe collapses without them, the edge is concentration-
    driven (fragile); if it holds, it's broad."""
    full_net, _ = _series_from_weights(weights_by_day, next_ret, cost_bps=cost_bps)
    days = sorted(weights_by_day)
    train_hi = days[max(0, int(len(days) * train_frac)) - 1] if days else None
    _, train_per_coin = _series_from_weights(weights_by_day, next_ret, cost_bps=cost_bps, day_hi=train_hi)
    ranked = [c for c, _ in sorted(train_per_coin.items(), key=lambda kv: abs(kv[1]), reverse=True)]

    out: dict[str, Any] = {"full_sharpe": _sharpe(full_net)}
    for n in (5, 10):
        drop = frozenset(ranked[:n])
        net_d, _ = _series_from_weights(weights_by_day, next_ret, cost_bps=cost_bps, exclude=drop)
        out[f"drop{n}"] = {"sharpe": _sharpe(net_d), "total_return": _total_return(net_d),
                           "n_dropped": len(drop), "dropped": sorted(drop)}
    return out


def _liquidity_sweep(store: _FeatureStore, *, cost_bps: float, direction: str,
                     liquidity_grid: tuple[float, ...]) -> dict[float, dict[str, Any]]:
    """For each `min_dollar_vol` floor, resolve the liquid universe and report the OI-edge Sharpe + n-coins on
    it. Universe robustness. A floor that resolves no coins yields a 0-Sharpe / 0-coin entry (not an error)."""
    from fxhnt.application.bybit_liquidity import liquid_universe

    out: dict[float, dict[str, Any]] = {}
    for thr in liquidity_grid:
        uni = liquid_universe(store, min_dollar_vol=thr)
        if not uni:
            out[thr] = {"sharpe": 0.0, "total_return": 0.0, "n_coins": 0}
            continue
        net = oi_deleverage_returns_from_store(store, universe=uni, cost_bps=cost_bps, direction=direction)
        out[thr] = {"sharpe": _sharpe(net), "total_return": _total_return(net),
                    "n_coins": _n_coins(store, uni)}
    return out


def look_ahead_audit(store: _FeatureStore, *, universe: set[str] | None = None,
                     cost_bps: float = 0.0, direction: str = "reversion",
                     _leak: bool = False) -> dict[str, Any]:
    """Assert the signal->return mapping is STRICTLY CAUSAL: a day-d signal may only earn the realized d->d+1
    return, never the same-day (d-1->d) move it was BUILT from.

    The audit RE-DERIVES the book two ways from the SAME day-d weights: the CAUSAL book (× d->d+1 returns —
    the live mapping) and a LEAKED book (× the SAME-day d-1->d returns the signal itself uses — a deliberate
    look-ahead negative control). `_leak=True` swaps in the leaked book as the candidate (the negative control
    the tests use to prove the audit BITES). `leak_days` counts days where the candidate's book return equals
    the leaked book's. `causal=True` iff `leak_days == 0`. READ-ONLY + memory-bounded."""
    doi_by_day, ret_by_day, next_ret, same_day = _signal_panels(store, universe=universe)
    signal = deleverage_signal_by_day(doi_by_day, ret_by_day, direction=direction)
    weights = deleverage_weights(signal)
    causal_net, _ = _series_from_weights(weights, next_ret, cost_bps=cost_bps)
    leaked_net, _ = _series_from_weights(weights, same_day, cost_bps=cost_bps)

    candidate = leaked_net if _leak else causal_net
    common = sorted(set(candidate) & set(leaked_net))
    leak_days = sum(1 for d in common if abs(candidate[d] - leaked_net[d]) <= 1e-12)
    is_causal = leak_days == 0
    return {"causal": is_causal, "leak_days": leak_days, "n_days_compared": len(common),
            "candidate_sharpe": _sharpe(candidate)}


def walk_forward_oi_deleverage(store: _FeatureStore, *, universe: set[str] | None = None,
                               cost_bps: float = 5.5, window_days: int = 180,
                               train_frac: float = 0.6,
                               liquidity_grid: tuple[float, ...] = (5_000_000.0, 10_000_000.0, 25_000_000.0),
                               ) -> dict[str, Any]:
    """READ-ONLY, memory-bounded OOS / WALK-FORWARD validation of the OI-deleveraging edge — the DEPLOY GATE.

    Computes (off ONE causal panel pass + the same `_curve_metrics`-style accounting as the in-sample verify):
      1. `rolling_windows` — non-overlapping `window_days` windows + `fraction_positive`;
      2. `train_holdout` — TRAIN picks the DIRECTION; applied to never-seen TEST (the OOS number);
      3. `drop_top_n` — Sharpe excluding the top-5 / top-10 gross contributors (ranked CAUSALLY on TRAIN);
      4. `liquidity_sweep` — Sharpe + n-coins at each `min_dollar_vol` floor;
      5. `look_ahead` — the strict-causality audit;
      6. `verdict` — PASS/FAIL: deployable iff (a) >70% rolling windows positive, (b) TEST/OOS Sharpe > 0 with
         the train-chosen direction, (c) survives drop-top-5 (Sharpe still clearly positive), (d) positive
         across the liquidity sweep.

    The construction sweeps (drop-top-N, liquidity) use the TRAIN-CHOSEN direction from (2). Writes NOTHING;
    `available=False` (+ reason) when there is no OI data / too few days."""
    doi_by_day, ret_by_day, next_ret, _ = _signal_panels(store, universe=universe)
    n_coins = _n_coins(store, universe)
    signal_rev = deleverage_signal_by_day(doi_by_day, ret_by_day, direction="reversion")
    signal_mom = deleverage_signal_by_day(doi_by_day, ret_by_day, direction="momentum")
    usable_days = sorted(set(deleverage_weights(signal_rev)))
    if len(usable_days) < 4:
        return {"available": False, "n_coins": n_coins,
                "reason": "OI-deleverage curve has <4 usable days for a walk-forward split "
                          "(run `fxhnt bybit-ingest-oi` + `fxhnt bybit-ingest-klines` first)"}

    train_holdout = _train_holdout(signal_rev, signal_mom, next_ret, cost_bps=cost_bps,
                                   train_frac=train_frac)
    direction = train_holdout["train_direction"] or "reversion"

    signal = signal_rev if direction == "reversion" else signal_mom
    weights = deleverage_weights(signal)
    full_net, _ = _series_from_weights(weights, next_ret, cost_bps=cost_bps)

    rolling = _rolling_windows(full_net, window_days=window_days)
    drop_top = _drop_top_n(weights, next_ret, cost_bps=cost_bps, train_frac=train_frac)
    liquidity = _liquidity_sweep(store, cost_bps=cost_bps, direction=direction,
                                 liquidity_grid=liquidity_grid)
    audit = look_ahead_audit(store, universe=universe, cost_bps=cost_bps, direction=direction)

    # --- PASS/FAIL verdict --------------------------------------------------------------------
    rw_pos = rolling["fraction_positive"] > 0.70
    oos_pos = train_holdout["available"] and train_holdout["test_sharpe"] > 0.0
    drop5_pos = drop_top["drop5"]["sharpe"] > 0.5
    resolved = [m for m in liquidity.values() if m["n_coins"] > 0]
    liq_robust = bool(resolved) and all(m["sharpe"] > 0.0 for m in resolved)
    causal = audit["causal"]
    checks = {
        "rolling_windows_positive": rw_pos,
        "oos_direction_positive": oos_pos,
        "survives_drop_top5": drop5_pos,
        "liquidity_robust": liq_robust,
        "look_ahead_clean": causal,
    }
    deployable = all(checks.values())

    return {"available": True, "n_coins": n_coins, "cost_bps": cost_bps, "window_days": window_days,
            "direction": direction, "full_sharpe": _sharpe(full_net),
            "rolling_windows": rolling, "train_holdout": train_holdout, "drop_top_n": drop_top,
            "liquidity_sweep": liquidity, "look_ahead": audit,
            "verdict": {"deployable": deployable, "checks": checks}}
