"""META-LABELING experiment (de Prado) on the Bybit CARRY (xsfunding) edge — READ-ONLY, anti-overfit, skeptical.

This is the SAME experiment we ran on the positioning edge (`positioning_metalabel.py`), pointed at a DIFFERENT
primary edge: the cross-sectional CARRY book (`bybit_carry_revalidate.py`). The carry edge's primary signal is
the per-coin funding/basis carry — a different mechanism than retail positioning — so it MIGHT carry filterable
structure positioning lacked. It also might not: an honest "NO LIFT" is a valid result (positioning's was).

REUSE, not reinvention
----------------------
The whole CV / OOF / threshold / verdict pipeline is the edge-agnostic core in `positioning_metalabel`
(`metalabel_experiment_from_dataset`, `PurgedKFold`, `apply_metalabel`, the LogisticRegression meta-model, the
`_ADD_MARGIN` 0.25 verdict, the shuffled-label leakage control). This module only builds the CARRY
`MetalabelDataset` — the per-coin-day primary signal + forward carry-return + the SAME few features — and hands
it to that core. No CV/verdict logic is duplicated here.

The CARRY primary + per-coin return
-----------------------------------
* PRIMARY (SIDE): `XsFundingReplay.current_weights(d)` — the live xsfunding cross-sectional market-neutral
  weights from the warehouse funding panel (causal: uses only data <= d). A coin-day with `w != 0` is a row;
  `side = sign(w)` (+1 LONG high-carry / -1 SHORT low-carry).
* per-coin CARRY RETURN: `carry_retᵢ = (spot_retᵢ − perp_retᵢ) + fundingᵢ` — funding accrual PLUS realized
  delta-neutral basis, via `carry_return_by_symbol(funding_d, basis_change_by_symbol(spot, perp, d))` — the
  SAME source of truth the paper carry book books. The label is the sign of `side × Σ carry_ret` over the
  next `label_horizon` days (forward-only, causal).

The FEW features (same five, anti-overfit; carry-flavoured):
  signal_strength = |demeaned funding| (the carry conviction on this coin-day),
  trailing_vol    = the coin's trailing realized vol (close-to-close),
  momentum        = the coin's trailing cumulative return,
  market_dispersion = cross-sectional stdev of funding across coins on day d (a funding-regime proxy),
  primary_side    = +1 LONG / -1 SHORT.

READ-ONLY: no writes. Memory-bounded via `_bounded_store(store, universe)`.
"""
from __future__ import annotations

import math
import statistics as st
from typing import Any

import numpy as np

from fxhnt.application.bybit_book_eval import _bounded_store
from fxhnt.application.bybit_carry_revalidate import _epoch_day_to_iso, _FeatureStore
from fxhnt.application.funding_panel import basis_change_by_symbol, warehouse_funding_panel
from fxhnt.application.paper_book import carry_return_by_symbol
from fxhnt.application.positioning_metalabel import (
    META_FEATURES,
    MetalabelDataset,
    _trailing_stats,
    metalabel_experiment_from_dataset,
)
from fxhnt.application.xsfunding_replay import XsFundingReplay

_VOL_LOOKBACK = 20   # trailing window (days) for trailing_vol + momentum — same as positioning.


def _forward_carry_return(carry_by_day: dict[int, dict[str, float]], all_days: list[int], coin: str,
                          day: int, side: float, label_horizon: int) -> float | None:
    """The PRIMARY carry position's realized return over the next `label_horizon` trading days: `side` ×
    Σ over d' in (d, d+horizon] of `carry_retᵢ(d')` (funding + realized basis on each forward day). Forward-only
    (causal label). `all_days` is the sorted union of trading days; the horizon is counted in trading-day steps
    AFTER `day`. None when the coin has no carry value on `day` or there are fewer than `label_horizon` forward
    days available (so the label is fully realized)."""
    if day not in all_days:
        return None
    idx = all_days.index(day)
    if idx + label_horizon >= len(all_days):
        return None
    total = 0.0
    saw = False
    for j in range(idx + 1, idx + 1 + label_horizon):
        cr = carry_by_day.get(all_days[j], {}).get(coin)
        if cr is None or not math.isfinite(cr):
            continue
        total += cr
        saw = True
    if not saw:
        return None
    return float(side) * total


def build_carry_metalabel_dataset(store: _FeatureStore, *, universe: set[str] | None = None,
                                  label_horizon: int = 1) -> MetalabelDataset:
    """READ-ONLY, memory-bounded meta-labeling dataset for the CARRY (xsfunding) edge.

    For every coin-day on which the PRIMARY carry book (`XsFundingReplay.current_weights`) takes a non-zero
    position AND a forward carry-return is realizable, emit a row:

      * features (`META_FEATURES`, causal — signal-day inputs only),
      * label `y` = 1 if the primary carry position is profitable over the next `label_horizon` days else 0
        (forward carry-returns only),
      * bookkeeping: `days`, `coins`, `sides` (+1 long / -1 short), `primary_ret` (the realized side×fwd carry).

    Rows sorted by (day, coin). Memory-bounded (reads restricted to `universe`); writes NOTHING."""
    bounded = _bounded_store(store, universe)
    close = bounded.crypto_close_panel()
    funding = bounded.crypto_funding_panel()
    spot = bounded.crypto_spot_panel()
    if not close:
        return _empty_dataset()

    panel = warehouse_funding_panel(close, funding)
    replay = XsFundingReplay(panel)

    all_days = sorted({d for days in close.values() for d in days})
    if len(all_days) < 2:
        return _empty_dataset()

    # Per-day per-coin FULL carry return (funding + daily-close basis) — the SAME source the paper book books.
    # Built once over the union of days; the forward label sums it over the horizon (causal: only forward days).
    carry_by_day: dict[int, dict[str, float]] = {}
    funding_xs_by_day: dict[int, dict[str, float]] = {}   # cross-sectional funding per day (for the features)
    for d in all_days:
        funding_d = {s: days[d] for s, days in funding.items() if d in days}
        funding_xs_by_day[d] = funding_d
        basis_d = basis_change_by_symbol(spot, close, d)
        carry_by_day[d] = carry_return_by_symbol(funding_d, basis_d)

    rows_X: list[list[float]] = []
    rows_y: list[int] = []
    days_out: list[int] = []
    coins_out: list[str] = []
    sides_out: list[float] = []
    pret_out: list[float] = []

    for day in all_days:
        weights = replay.current_weights(_epoch_day_to_iso(day))
        if not weights:
            continue
        # cross-sectional funding context on the SIGNAL day (causal): the demeaned funding gives conviction,
        # the cross-sectional stdev gives the funding-regime dispersion proxy.
        funding_d = funding_xs_by_day.get(day, {})
        fund_vals = [v for v in funding_d.values() if math.isfinite(v)]
        fund_mean = st.mean(fund_vals) if fund_vals else 0.0
        dispersion = st.pstdev(fund_vals) if len(fund_vals) >= 2 else 0.0
        for coin, w in weights.items():
            if w == 0.0 or coin not in close:
                continue
            side = 1.0 if w > 0.0 else -1.0
            pr = _forward_carry_return(carry_by_day, all_days, coin, day, side, label_horizon)
            if pr is None:
                continue
            vol, mom = _trailing_stats(close[coin], day, _VOL_LOOKBACK)
            # carry conviction: |this coin's funding − cross-sectional mean| on the signal day.
            signal_strength = abs(funding_d.get(coin, fund_mean) - fund_mean)
            rows_X.append([signal_strength, vol, mom, dispersion, side])
            rows_y.append(1 if pr > 0.0 else 0)
            days_out.append(int(day))
            coins_out.append(coin)
            sides_out.append(side)
            pret_out.append(float(pr))

    if not rows_X:
        return _empty_dataset()

    order = sorted(range(len(rows_X)), key=lambda k: (days_out[k], coins_out[k]))
    X = np.array([rows_X[k] for k in order], dtype=float)
    y = np.array([rows_y[k] for k in order], dtype=int)
    return MetalabelDataset(
        X=X, y=y,
        days=[days_out[k] for k in order],
        coins=[coins_out[k] for k in order],
        sides=[sides_out[k] for k in order],
        primary_ret=[pret_out[k] for k in order],
        feature_names=list(META_FEATURES))


def _empty_dataset() -> MetalabelDataset:
    return MetalabelDataset(np.empty((0, len(META_FEATURES))), np.empty(0, dtype=int),
                            [], [], [], [], list(META_FEATURES))


def carry_metalabel_experiment(store: _FeatureStore, *, universe: set[str] | None = None,
                               label_horizon: int = 1, n_splits: int = 5, embargo_days: int = 1,
                               seed: int = 0, shuffle_label: bool = False) -> dict[str, Any]:
    """READ-ONLY de Prado meta-labeling experiment on the CARRY edge — the SKEPTICAL verdict.

    Builds the carry per-coin-day dataset (`build_carry_metalabel_dataset`), then delegates to the edge-agnostic
    `metalabel_experiment_from_dataset` core (the SAME PurgedKFold / OOF / threshold / verdict pipeline the
    positioning edge uses — no CV/verdict logic duplicated). META ADDS only if its PURGED-OOS Sharpe beats RAW
    by a clear margin; `shuffle_label=True` is the leakage tripwire (shuffled labels must show ~no lift).

    Memory-bounded (universe-restricted reads); writes NOTHING. `available=False` (+ reason) on no/too-little
    data. `edge='carry'` is echoed into the report."""
    ds = build_carry_metalabel_dataset(store, universe=universe, label_horizon=label_horizon)
    rep = metalabel_experiment_from_dataset(
        ds, label_horizon=label_horizon, n_splits=n_splits, embargo_days=embargo_days,
        seed=seed, shuffle_label=shuffle_label,
        insufficient_reason=(
            f"carry meta-labeling needs >= {n_splits * 2} coin-day rows for a purged {n_splits}-fold CV "
            "(run the Phase-A/B Bybit ingests: `fxhnt bybit-ingest-funding` + `fxhnt bybit-ingest-klines` "
            "first)"))
    if rep.get("available"):
        rep["edge"] = "carry"
    return rep
