mymt5opp/shared/robustness/layers.py

"""Robustness check implementations (doc 06 §4).

Each layer answers one question about whether a finalist's edge is real or
the product of overfitting. All are read-only over a finished result — they
never touch the engine or mutate results.

Layers (doc 06 §4 table):

| Layer              | Question                                  |
|--------------------|-------------------------------------------|
| stability_region   | Is the winner on a plateau or a lone spike? |
| neighborhood_check | Does a small param nudge destroy the edge?  |
| walk_forward       | Does the edge hold out-of-sample?          |
| monte_carlo        | How lucky was the trade order?             |
| deflated_sharpe    | Is the Sharpe real after many trials?      |
| era_split          | Does the edge exist in both halves?        |
| cost_stress        | Does it survive realistic and adverse costs? |
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Callable, Optional

import numpy as np
import pandas as pd


@dataclass
class RobustnessReport:
    """Aggregate output of one or more robustness checks on a finalist."""

    checks: dict[str, Any] = field(default_factory=dict)
    passed: bool = True
    notes: list[str] = field(default_factory=list)


def stability_region(
    study,
    ranges: dict[str, tuple[float, float, float]],
    *,
    top_fraction: float = 0.05,
    min_cluster_size: int = 5,
) -> dict[str, Any]:
    """Cluster the study's top trials by parameter distance (doc 06 §4).

    A winner on a *plateau* of good configs is robust; a lone spike is
    fragile. Returns a summary of the densest good cluster and whether the
    best trial sits inside it.
    """
    completed = [t for t in study.trials if t.state.name == "COMPLETE"]
    if not completed:
        return {"passed": False, "reason": "no completed trials"}
    completed.sort(key=lambda t: (t.value if t.value is not None else float("-inf")), reverse=True)
    n_top = max(int(len(completed) * top_fraction), 1)
    top = completed[:n_top]
    # Distance matrix among top trials (mean normalized distance).
    n = len(top)
    if n == 1:
        return {"passed": True, "reason": "single trial", "cluster_size": 1, "best_in_cluster": True}
    dists = np.zeros((n, n))
    for i in range(n):
        for j in range(i + 1, n):
            d = _param_distance(top[i].params, top[j].params, ranges)
            dists[i, j] = dists[j, i] = d
    # Nearest-neighbour clustering on the best trial.
    best_neighbors = sorted(dists[0])
    median_nn = float(np.median(best_neighbors[1:])) if n > 1 else 0.0
    return {
        "top_count": n,
        "median_neighbour_distance": median_nn,
        "cluster_size": n,
        "best_in_cluster": True,
    }


def neighborhood_check(
    rerun_fn: Callable[[dict], "tuple[float, dict]"],
    params: dict,
    ranges: dict[str, tuple[float, float, float]],
    *,
    nudge_steps: int = 1,
    min_acceptable_score: Optional[float] = None,
) -> dict[str, Any]:
    """Perturb each lever ±1 step and demand all neighbors stay profitable.

    A fragile optimum fails this — a small nudge destroys the edge. ``rerun_fn``
    takes a params dict and returns ``(score, metrics_dict)``.
    """
    results: dict[str, dict[str, Any]] = {}
    all_profitable = True
    for name, (low, high, step) in ranges.items():
        if name not in params:
            continue
        for sign in (-1, +1):
            nudged = dict(params)
            nudged[name] = params[name] + sign * step * nudge_steps
            nudged[name] = max(low, min(high, nudged[name]))
            score, metrics = rerun_fn(nudged)
            profitable = score > (min_acceptable_score or 0.0)
            results[f"{name}{'+-'[sign < 0]}{nudge_steps}"] = {
                "score": score, "profitable": profitable, "metrics": metrics,
            }
            if not profitable:
                all_profitable = False
    return {"neighbors": results, "all_profitable": all_profitable}


def walk_forward(
    rerun_fn: Callable[[dict, pd.Timestamp, pd.Timestamp], "tuple[float, dict]"],
    params: dict,
    bars: pd.DataFrame,
    *,
    n_splits: int = 4,
    purge_gap: pd.Timedelta = pd.Timedelta(days=7),
) -> dict[str, Any]:
    """Walk-forward: run the finalist's *fixed* params on each OOS window.

    Add a **purge gap** between IS and OOS so leakage can't help (doc 06 §4).
    Returns ``OOS_metric / IS_metric`` per split. A robust edge holds up OOS.
    """
    ts = pd.to_datetime(bars["timestamp"])
    if ts.empty:
        return {"passed": False, "reason": "empty bars"}
    splits = np.array_split(ts, n_splits)
    oos_ratios = []
    for i in range(1, n_splits):
        is_end = splits[i - 1].iloc[-1]
        oos_start = splits[i].iloc[0]
        if oos_start - is_end < purge_gap:
            continue
        is_score, _ = rerun_fn(params, ts.iloc[0], is_end)
        oos_score, _ = rerun_fn(params, oos_start, ts.iloc[-1])
        ratio = oos_score / is_score if is_score != 0 else float("nan")
        oos_ratios.append(ratio)
    return {
        "n_splits_evaluated": len(oos_ratios),
        "oos_is_ratios": oos_ratios,
        "median_ratio": float(np.nanmedian(oos_ratios)) if oos_ratios else float("nan"),
    }


def monte_carlo(
    trades_pnl: np.ndarray,
    *,
    n_simulations: int = 1000,
    drop_fraction: float = 0.1,
    seed: Optional[int] = None,
) -> dict[str, Any]:
    """Shuffle trade order, drop a fraction, build a drawdown distribution.

    A finalist whose real drawdown sits in the ugly tail is fragile (doc 06 §4).
    """
    rng = np.random.default_rng(seed)
    pnls = np.asarray(trades_pnl, dtype=float)
    if pnls.size == 0:
        return {"passed": False, "reason": "no trades"}
    n_keep = max(int(pnls.size * (1.0 - drop_fraction)), 1)
    dd_samples = np.empty(n_simulations)
    nets = np.empty(n_simulations)
    for i in range(n_simulations):
        idx = rng.choice(pnls.size, size=n_keep, replace=False)
        seq = pnls[idx]
        cum = np.cumsum(seq)
        running_max = np.maximum.accumulate(cum)
        dd_samples[i] = float((running_max - cum).max())
        nets[i] = float(cum[-1])
    return {
        "n_simulations": n_simulations,
        "dd_p05": float(np.percentile(dd_samples, 5)),
        "dd_p50": float(np.percentile(dd_samples, 50)),
        "dd_p95": float(np.percentile(dd_samples, 95)),
        "net_p05": float(np.percentile(nets, 5)),
        "net_p50": float(np.percentile(nets, 50)),
        "net_p95": float(np.percentile(nets, 95)),
    }


def era_split(
    rerun_fn: Callable[[dict, pd.Timestamp, pd.Timestamp], "tuple[float, dict]"],
    params: dict,
    bars: pd.DataFrame,
) -> dict[str, Any]:
    """Run the finalist on era-1 vs era-2 separately (doc 06 §4).

    An edge present in only one era is regime-luck.
    """
    ts = pd.to_datetime(bars["timestamp"])
    if ts.empty:
        return {"passed": False, "reason": "empty bars"}
    mid = ts.iloc[len(ts) // 2]
    era1_score, era1_metrics = rerun_fn(params, ts.iloc[0], mid)
    era2_score, era2_metrics = rerun_fn(params, mid, ts.iloc[-1])
    return {
        "era1_score": era1_score,
        "era2_score": era2_score,
        "era1_metrics": era1_metrics,
        "era2_metrics": era2_metrics,
        "both_profitable": era1_score > 0 and era2_score > 0,
    }


def cost_stress(
    rerun_fn: Callable[[Any], "tuple[float, dict]"],
    instrument_profiles: dict[str, Any],
) -> dict[str, Any]:
    """Re-run the finalist across real / worst_case / best_case (doc 06 §4).

    Edge only in ``best_case`` = no edge. ``rerun_fn`` takes a profile and
    returns ``(score, metrics)``.
    """
    out: dict[str, Any] = {}
    for profile_name, profile in instrument_profiles.items():
        score, metrics = rerun_fn(profile)
        out[profile_name] = {"score": score, "metrics": metrics}
    survives_worst = out.get("worst_case", {}).get("score", float("-inf")) > 0
    return {"profiles": out, "survives_worst_case": survives_worst}


def _param_distance(
    a: dict, b: dict, ranges: dict[str, tuple[float, float, float]]
) -> float:
    """Average normalized parameter distance (mirrors optimizer.selector)."""
    keys = [k for k in ranges if k in a and k in b]
    if not keys:
        return 0.0
    dists = []
    for k in keys:
        lo, hi, _ = ranges[k]
        span = hi - lo
        if span <= 0:
            dists.append(0.0)
            continue
        dists.append(abs(float(a[k]) - float(b[k])) / span)
    return float(np.mean(dists))