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

The idea (Advances in Financial Machine Learning, López de Prado, ch. 3)
-----------------------------------------------------------------------
The positioning edge (`bybit_positioning_eval.py`: contrarian, SHORT high long-ratio / LONG low, per coin per
day) is the PRIMARY model — it sets the SIDE. We add a META-MODEL that learns WHEN to trust each coin-day
signal (filter false positives) + SIZE by confidence P(profitable). Separating SIDE (primary) from SIZE/FILTER
(meta) can raise precision → higher OOS Sharpe — IF it survives PURGED cross-validation. It often does NOT
(ML overfits finance), and "meta-labeling doesn't add" is a VALID, honest outcome. We adopt only what clears
purged out-of-sample by a CLEAR margin.

What this module builds
-----------------------
1. `build_metalabel_dataset` — one row per liquid coin-day where the primary takes a position:
     features (FEW — anti-overfit, see `META_FEATURES`), a binary forward-return label, and the bookkeeping
     arrays (day / coin / side / primary_ret) the apply + CV need. Causal: features use signal-day inputs
     only; the label uses forward returns only.
2. `PurgedKFold` — de Prado purged + embargoed k-fold: drops training rows whose label-windows overlap the
     test fold (plus an embargo gap). Plain k-fold would LEAK (overlapping labels) and lie.
3. `apply_metalabel` — take the primary position only when P>threshold; size ∝ P (bet sizing).
4. `metalabel_experiment` — the SKEPTICAL verdict: run the meta-model out-of-fold under PurgedKFold, build the
     PURGED-OOS meta-labeled return series, compare vs RAW on OOS Sharpe/CAGR/maxDD + precision/lift/trades-
     kept, and print "META-LABELING ADDS (deploy)" vs "NO LIFT — STAY RAW". A `shuffle_label` control proves
     the pipeline does not leak (shuffled labels must show ~no lift).

The meta-model is a SIMPLE, regularized classifier (sklearn LogisticRegression, L2) — NOT a deep model.

READ-ONLY: no PaperRepo, no `store.write_*`. Memory-bounded: reads restricted to `universe` (the liquid
subset) via `_bounded_store`. Reuses the live signal construction (`positioning_weights`, `_signal_panels`).
"""
from __future__ import annotations

import math
import statistics as st
from dataclasses import dataclass
from typing import Any

import numpy as np

from fxhnt.application.bybit_book_eval import _bounded_store
from fxhnt.application.bybit_positioning_eval import (
    _FeatureStore,
    _panel_symbols,
    _ratio_by_day,
    positioning_weights,
)

# --- the FEW documented meta-features (anti-overfit: 5 features) --------------------------------
#
# Each is computed from inputs KNOWABLE ON THE SIGNAL DAY d (the day the primary forms its weight). The label
# is the only forward-looking quantity. Order is fixed (the X matrix columns follow this order).
META_FEATURES: tuple[str, ...] = (
    "signal_strength",   # |demeaned long_ratio| — the conviction of the primary's bet on this coin-day.
    "trailing_vol",      # the coin's trailing realized vol (stdev of recent daily returns) up to & incl. d.
    "momentum",          # the coin's recent cumulative return over the trailing window (up to & incl. d).
    "market_dispersion", # cross-sectional stdev of long_ratio across coins on day d (a market-regime proxy).
    "primary_side",      # +1 if the primary goes LONG this coin-day, -1 if SHORT (the side being filtered).
)

_PERIODS_PER_YEAR = 365
_VOL_LOOKBACK = 20          # trailing window (days) for trailing_vol + momentum.
_DEFAULT_THRESHOLD = 0.50   # P>threshold to take a position (tuned on TRAIN folds; default 0.5 is neutral).
_ADD_MARGIN = 0.25          # PURGED-OOS Sharpe lift required to declare "META ADDS" (clear, not marginal).


# --- 1. the per-coin-day dataset ---------------------------------------------------------------

@dataclass(frozen=True)
class MetalabelDataset:
    """A meta-labeling dataset: one row per coin-day where the primary takes a position.

    `X` is `(n_rows, len(feature_names))` causal features; `y` the binary forward-return label (1=profitable).
    `days`/`coins`/`sides`/`primary_ret` are aligned per-row bookkeeping the apply + purged CV consume.
    `feature_names` == `META_FEATURES`. All arrays are aligned and sorted by (day, coin)."""

    X: np.ndarray
    y: np.ndarray
    days: list[int]
    coins: list[str]
    sides: list[float]
    primary_ret: list[float]
    feature_names: list[str]

    @property
    def n_rows(self) -> int:
        return int(self.X.shape[0])


def _forward_position_return(close_series: dict[int, float], day: int, side: float,
                             label_horizon: int) -> float | None:
    """The PRIMARY position's realized return over the next `label_horizon` days: side × (close-to-close move
    from day d to d+horizon). Forward-only (causal label). None if either close is missing/non-finite."""
    days = sorted(close_series)
    if day not in close_series:
        return None
    # find the day that is exactly `label_horizon` calendar-days ahead among available closes is too strict
    # for sparse panels; use the close at the index `label_horizon` ahead of d in the sorted day list.
    idx = days.index(day)
    if idx + label_horizon >= len(days):
        return None
    p0 = close_series[day]
    p1 = close_series[days[idx + label_horizon]]
    if not (p0 and math.isfinite(p0) and math.isfinite(p1)):
        return None
    return float(side) * (p1 / p0 - 1.0)


def _trailing_stats(close_series: dict[int, float], day: int, lookback: int) -> tuple[float, float]:
    """`(trailing_vol, momentum)` from the daily returns ending ON day d (inclusive), over up to `lookback`
    prior steps. vol = pstdev of those returns; momentum = their cumulative compounded return. Causal: uses
    only closes up to and including day d. Sparse/short history → (0.0, 0.0)."""
    days = [d for d in sorted(close_series) if d <= day]
    if len(days) < 2:
        return 0.0, 0.0
    rets: list[float] = []
    for i in range(max(1, len(days) - lookback), len(days)):
        p0, p1 = close_series[days[i - 1]], close_series[days[i]]
        if p0 and math.isfinite(p0) and math.isfinite(p1):
            rets.append(p1 / p0 - 1.0)
    if len(rets) < 2:
        return 0.0, 0.0
    vol = st.pstdev(rets)
    mom = 1.0
    for r in rets:
        mom *= (1.0 + r)
    return float(vol), float(mom - 1.0)


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

    For every coin-day on which the PRIMARY (`positioning_weights(direction)`) takes a non-zero position AND a
    forward return is realizable, emit a row:

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

    Rows sorted by (day, coin). Memory-bounded (reads restricted to `universe`); writes NOTHING."""
    bounded = _bounded_store(store, universe)
    long_ratio = bounded.read_panel(sorted(_panel_symbols(bounded, universe)), "long_ratio")
    long_ratio = {s: v for s, v in long_ratio.items() if v}
    close_panel = bounded.crypto_close_panel()
    if not long_ratio or not close_panel:
        return MetalabelDataset(np.empty((0, len(META_FEATURES))), np.empty(0, dtype=int),
                                [], [], [], [], list(META_FEATURES))

    ratio_by_day = _ratio_by_day(long_ratio)
    weights_by_day = positioning_weights(ratio_by_day, direction=direction)

    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 sorted(weights_by_day):
        row_ratio = ratio_by_day.get(day, {})
        if len(row_ratio) < 2:
            continue
        ratio_mean = st.mean(row_ratio.values())
        dispersion = st.pstdev(row_ratio.values())   # market-regime proxy (cross-sectional spread)
        for coin, w in weights_by_day[day].items():
            if w == 0.0 or coin not in close_panel:
                continue
            side = 1.0 if w > 0.0 else -1.0
            pr = _forward_position_return(close_panel[coin], day, side, label_horizon)
            if pr is None:
                continue
            vol, mom = _trailing_stats(close_panel[coin], day, _VOL_LOOKBACK)
            signal_strength = abs(row_ratio[coin] - ratio_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 MetalabelDataset(np.empty((0, len(META_FEATURES))), np.empty(0, dtype=int),
                                [], [], [], [], list(META_FEATURES))

    # sort by (day, coin) so the row order is deterministic and CV folds are time-contiguous-friendly.
    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))


# --- 2. PurgedKFold + embargo (de Prado leakage-free CV) ---------------------------------------

class PurgedKFold:
    """de Prado PURGED + EMBARGOED k-fold CV (Advances in Financial ML, §7.4).

    Each sample's LABEL spans `label_horizon` forward days, so a training sample whose label-window overlaps
    the test fold's time span leaks. `split` drops every training row whose signal day falls inside the
    forbidden zone `[t_lo - label_horizon - embargo_days, t_hi + label_horizon + embargo_days]` around the
    test block `[t_lo, t_hi]`. Plain k-fold (no purge) would train on overlapping-label rows and over-state OOS
    performance — the whole reason this exists.

    Folds are CONTIGUOUS blocks of the (time-sorted) index, so every sample is in exactly one test fold and
    the test block is a single time span (purging is then a simple window exclusion)."""

    def __init__(self, n_splits: int = 5, *, label_horizon: int = 1, embargo_days: int = 1) -> None:
        if n_splits < 2:
            raise ValueError(f"n_splits must be >= 2, got {n_splits}")
        self.n_splits = n_splits
        self.label_horizon = label_horizon
        self.embargo_days = embargo_days

    def split(self, days: list[int]):
        """Yield `(train_idx, test_idx)` index lists. `days[i]` is the signal (epoch) day of sample i; the
        index order is assumed time-sorted (as `build_metalabel_dataset` emits). Test folds are contiguous
        blocks; training rows in the purge+embargo window around each test block are excluded."""
        n = len(days)
        if n < self.n_splits:
            return
        bounds = np.linspace(0, n, self.n_splits + 1).astype(int)
        forbidden = self.label_horizon + self.embargo_days
        for k in range(self.n_splits):
            lo, hi = int(bounds[k]), int(bounds[k + 1])
            if hi <= lo:
                continue
            test_idx = list(range(lo, hi))
            t_lo, t_hi = days[lo], days[hi - 1]
            ban_lo, ban_hi = t_lo - forbidden, t_hi + forbidden
            train_idx = [i for i in range(n)
                         if i < lo or i >= hi
                         if not (ban_lo <= days[i] <= ban_hi)]
            yield train_idx, test_idx


# --- 3. apply: filter by P>threshold, size ∝ P -------------------------------------------------

def apply_metalabel(primary_ret: list[float] | np.ndarray, p: list[float] | np.ndarray, *,
                    threshold: float = _DEFAULT_THRESHOLD) -> list[float]:
    """Bet-sizing apply: for each coin-day take the primary position ONLY when P>threshold, sized ∝ P.

    Returns the per-row meta-labeled return: `primary_ret * P` when `P > threshold`, else `0.0` (filtered
    out — no position). A kept LOSER still counts at `primary_ret * P` (the outcome is not used to size — that
    would be look-ahead). Pure."""
    out: list[float] = []
    for r, pi in zip(primary_ret, p, strict=True):
        out.append(float(r) * float(pi) if float(pi) > threshold else 0.0)
    return out


# --- 4. metrics on a {day: ret} series ---------------------------------------------------------

def _series_from_rows(days: list[int], rets: list[float]) -> dict[int, float]:
    """Aggregate per-row returns into a `{day: summed-ret}` daily series (a coin-day book aggregates the day's
    coin legs). Non-finite legs dropped."""
    out: dict[int, float] = {}
    for d, r in zip(days, rets, strict=True):
        if math.isfinite(r):
            out[d] = out.get(d, 0.0) + r
    return out


def _sharpe(series: dict[int, float]) -> float:
    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 _cagr(series: dict[int, float]) -> float:
    days = sorted(series)
    if len(days) < 2:
        return 0.0
    equity = 1.0
    for d in days:
        equity *= (1.0 + series[d])
    span = max(1, days[-1] - days[0])
    if equity <= 0.0:
        return -1.0
    return equity ** (_PERIODS_PER_YEAR / span) - 1.0


def _max_dd(series: dict[int, float]) -> float:
    equity, peak, mdd = 1.0, 1.0, 0.0
    for d in sorted(series):
        equity *= (1.0 + series[d])
        peak = max(peak, equity)
        if peak > 0:
            mdd = min(mdd, equity / peak - 1.0)
    return mdd


# --- 5. the experiment: purged-OOS meta vs raw + skeptical verdict -----------------------------

def _fit_predict_oof(ds: MetalabelDataset, *, n_splits: int, embargo_days: int, label_horizon: int,
                     seed: int, shuffle_label: bool) -> tuple[np.ndarray, dict[str, float]]:
    """Out-of-fold P(profitable) for every row under PurgedKFold, plus the averaged feature importances.

    A SIMPLE regularized classifier (LogisticRegression, L2) is fit on each fold's PURGED training rows and
    predicts the held-out test rows — so every prediction is genuinely out-of-sample. Features standardised
    per fold (fit on TRAIN only — no leakage). `shuffle_label` permutes the labels (the leakage control: the
    model then has nothing real to learn → no OOS lift). Feature importance = |coef| × train-feature-std,
    averaged across folds (comparable across standardised features)."""
    from sklearn.linear_model import LogisticRegression
    from sklearn.preprocessing import StandardScaler

    rng = np.random.default_rng(seed)
    y = ds.y.copy()
    if shuffle_label:
        y = y[rng.permutation(len(y))]

    oof_p = np.full(ds.n_rows, np.nan, dtype=float)
    importances = np.zeros(len(ds.feature_names), dtype=float)
    n_folds_used = 0
    pk = PurgedKFold(n_splits=n_splits, label_horizon=label_horizon, embargo_days=embargo_days)
    for train_idx, test_idx in pk.split(ds.days):
        if not train_idx or not test_idx:
            continue
        ytr = y[train_idx]
        if len(set(ytr.tolist())) < 2:
            # degenerate fold (one class only) → fall back to the base rate as P (no model signal).
            oof_p[test_idx] = float(ytr.mean()) if len(ytr) else 0.5
            continue
        scaler = StandardScaler().fit(ds.X[train_idx])
        Xtr = scaler.transform(ds.X[train_idx])
        Xte = scaler.transform(ds.X[test_idx])
        # L2 is LogisticRegression's default regularization (the `penalty=` kw was deprecated in sklearn 1.8);
        # C=1.0 is the regularization strength — kept simple + regularized (anti-overfit), not a deep model.
        clf = LogisticRegression(C=1.0, max_iter=1000, random_state=seed)
        clf.fit(Xtr, ytr)
        # P(class==1) — robust to which column sklearn assigns to class 1.
        cls = list(clf.classes_)
        pos_col = cls.index(1) if 1 in cls else (len(cls) - 1)
        oof_p[test_idx] = clf.predict_proba(Xte)[:, pos_col]
        importances += np.abs(clf.coef_[0])
        n_folds_used += 1

    if n_folds_used:
        importances /= n_folds_used
    imp = {name: float(importances[i]) for i, name in enumerate(ds.feature_names)}
    # rows never assigned (e.g. fully purged out) → neutral 0.5 so they neither help nor hurt.
    oof_p = np.where(np.isnan(oof_p), 0.5, oof_p)
    return oof_p, imp


def _pick_threshold(ds: MetalabelDataset, oof_p: np.ndarray, *, n_splits: int, embargo_days: int,
                    label_horizon: int, seed: int, shuffle_label: bool) -> float:
    """Choose the P-threshold on TRAIN folds only (never on the OOS series): re-run purged OOF on the TRAIN
    portion is heavy, so we approximate de Prado discipline by choosing the threshold that maximises TRAIN
    Sharpe of the sized series — where TRAIN is the first 60% of the time-sorted rows. The chosen threshold is
    then applied to the held-out OOS rows. Candidate grid spans plausible P cutoffs."""
    n = ds.n_rows
    if n < 4:
        return _DEFAULT_THRESHOLD
    cut = max(2, int(n * 0.6))
    tr_days, tr_ret, tr_p = ds.days[:cut], ds.primary_ret[:cut], oof_p[:cut]
    best_thr, best_sharpe = _DEFAULT_THRESHOLD, -math.inf
    for thr in (0.45, 0.50, 0.55, 0.60, 0.65):
        sized = apply_metalabel(tr_ret, tr_p, threshold=thr)
        s = _sharpe(_series_from_rows(tr_days, sized))
        if s > best_sharpe:
            best_thr, best_sharpe = thr, s
    return best_thr


def metalabel_experiment_from_dataset(ds: MetalabelDataset, *, label_horizon: int = 1,
                                      n_splits: int = 5, embargo_days: int = 1, seed: int = 0,
                                      shuffle_label: bool = False,
                                      insufficient_reason: str | None = None) -> dict[str, Any]:
    """EDGE-AGNOSTIC de Prado meta-labeling core — the SKEPTICAL verdict, given a pre-built `MetalabelDataset`.

    This is the ONE place the CV / OOF / threshold / verdict logic lives, so EVERY edge (positioning, carry,
    …) runs the IDENTICAL pipeline: predict P(profitable) OUT-OF-FOLD under PurgedKFold (+ embargo), pick the
    bet-threshold on TRAIN, apply the meta filter+sizing to the PURGED-OOS rows, and compare the META-LABELED
    OOS return series against the RAW (side-only) OOS series on Sharpe / CAGR / maxDD plus OOS precision,
    Sharpe-lift and fraction-of-trades-kept. Reports feature importances.

    The VERDICT is intentionally conservative: META ADDS only if its PURGED-OOS Sharpe beats RAW by a CLEAR
    margin (`_ADD_MARGIN`). `shuffle_label=True` is the leakage control — shuffled labels must show ~no lift,
    so the verdict must be NOT-adds (a leak would show false lift and is caught by the shuffled-label test).

    Pure given the dataset (NO I/O — the caller does the bounded reads). `available=False` (+ reason) when the
    dataset has too few rows for a purged `n_splits`-fold CV. `insufficient_reason` overrides the default
    not-enough-data message so each edge can point at its own ingest commands."""
    if ds.n_rows < n_splits * 2:
        reason = insufficient_reason or (
            f"meta-labeling needs >= {n_splits * 2} coin-day rows for a purged {n_splits}-fold CV")
        return {"available": False, "reason": reason, "n_rows": ds.n_rows}

    oof_p, importances = _fit_predict_oof(
        ds, n_splits=n_splits, embargo_days=embargo_days, label_horizon=label_horizon,
        seed=seed, shuffle_label=shuffle_label)
    threshold = _pick_threshold(ds, oof_p, n_splits=n_splits, embargo_days=embargo_days,
                                label_horizon=label_horizon, seed=seed, shuffle_label=shuffle_label)

    # Score on the held-out OOS portion (the last 40% — never used to pick the threshold) so the comparison is
    # genuinely out-of-sample for BOTH the model (OOF) and the threshold (TRAIN-picked).
    cut = max(2, int(ds.n_rows * 0.6))
    oos_days = ds.days[cut:]
    oos_pret = ds.primary_ret[cut:]
    oos_p = oof_p[cut:]
    oos_y = ds.y[cut:]

    raw_series = _series_from_rows(oos_days, list(oos_pret))                # side-only (no meta filter)
    meta_rows = apply_metalabel(oos_pret, oos_p, threshold=threshold)
    meta_series = _series_from_rows(oos_days, meta_rows)

    raw_sharpe, meta_sharpe = _sharpe(raw_series), _sharpe(meta_series)
    kept_mask = [float(p) > threshold for p in oos_p]
    n_kept = sum(1 for k in kept_mask if k)
    frac_kept = n_kept / len(kept_mask) if kept_mask else 0.0
    # OOS precision: of the coin-days the meta KEEPS, what fraction were actually profitable (y==1).
    kept_y = [int(oos_y[i]) for i in range(len(kept_mask)) if kept_mask[i]]
    meta_precision = (sum(kept_y) / len(kept_y)) if kept_y else 0.0
    base_rate = float(np.mean(oos_y)) if len(oos_y) else 0.0

    sharpe_lift = meta_sharpe - raw_sharpe
    adds = (sharpe_lift > _ADD_MARGIN) and (meta_sharpe > 0.0) and (frac_kept > 0.0)
    if shuffle_label:
        # Belt-and-braces: a shuffled control should never "add"; if the math somehow says it does, the data
        # leaked — keep the label honest (the test asserts lift is small, this keeps the verdict aligned).
        adds = adds and (sharpe_lift > _ADD_MARGIN)
    label = ("META-LABELING ADDS (deploy)" if adds
             else "NO LIFT — STAY RAW (meta overfits / doesn't help)")

    return {
        "available": True,
        "n_rows": ds.n_rows,
        "n_oos_rows": len(oos_days),
        "label_horizon": label_horizon,
        "n_splits": n_splits,
        "embargo_days": embargo_days,
        "threshold": threshold,
        "shuffle_label": shuffle_label,
        "feature_names": list(ds.feature_names),
        "feature_importances": importances,
        "oos": {
            "raw_sharpe": raw_sharpe,
            "meta_sharpe": meta_sharpe,
            "sharpe_lift": sharpe_lift,
            "raw_cagr": _cagr(raw_series),
            "meta_cagr": _cagr(meta_series),
            "raw_max_dd": _max_dd(raw_series),
            "meta_max_dd": _max_dd(meta_series),
            "frac_trades_kept": frac_kept,
            "meta_precision": meta_precision,
            "base_rate": base_rate,
            "precision_lift": meta_precision - base_rate,
            "raw_days": len(raw_series),
            "meta_days": len(meta_series),
        },
        "verdict": {"adds": adds, "label": label, "add_margin": _ADD_MARGIN},
    }


def metalabel_experiment(store: _FeatureStore, *, universe: set[str] | None = None,
                         direction: str = "contrarian", 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 POSITIONING edge — the SKEPTICAL verdict.

    Builds the positioning per-coin-day dataset (`build_metalabel_dataset`), then delegates to the edge-agnostic
    `metalabel_experiment_from_dataset` core (the shared PurgedKFold / OOF / threshold / verdict pipeline). The
    same core is reused by the carry edge — no CV/verdict logic is duplicated. `direction` is echoed back into
    the report (positioning-specific bookkeeping).

    Memory-bounded (universe-restricted reads); writes NOTHING. `available=False` (+ reason) on no/too-little
    data."""
    ds = build_metalabel_dataset(store, universe=universe, direction=direction,
                                 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"meta-labeling needs >= {n_splits * 2} coin-day rows for a purged {n_splits}-fold CV "
            "(run `fxhnt bybit-ingest-long-short` + `fxhnt bybit-ingest-klines` first)"))
    if rep.get("available"):
        rep["direction"] = direction
    return rep
