# =============================================================================
#  Project   : anomaly-atlas
#  File      : experiments/micro/expG_cost_frontier/benchmark.py
#  Purpose   : Cost frontier: net-of-spread sweep over the double-filtered pool
#  Author    : Simon-Pierre Boucher
#  Contact   : contact@spboucher.ai
#  Data src  : hfmarketdata.io (sole data source)
#  Created   : 2026-08-12
#  Modified  : 2026-08-12
#  Platform  : macOS / Apple Silicon (arm64)
#  License   : All rights reserved (research code)
# =============================================================================
"""Experiment G — the cost frontier (protocol pre-specified in hypothesis.md).

Pool = mechanical intersection of committed expC/expD/expF outputs. For each
rule: gross daily stream + daily turnover, then net = gross − κ·(EDGE/2)·
turnover across the declared κ sweep, with the analytic breakeven κ*.
"""

from __future__ import annotations

import glob
import json
import sys
from collections import defaultdict
from datetime import UTC, datetime
from pathlib import Path

import numpy as np

REPO_ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(REPO_ROOT / "benchmarks"))
sys.path.insert(0, str(REPO_ROOT / "src"))

from hardware_manifest import collect_manifest  # noqa: E402

from anomaly_atlas.data.cleaning import RTH_SLOTS, rth_day_grids  # noqa: E402
from anomaly_atlas.data.hf_client import HFMarketDataClient  # noqa: E402
from anomaly_atlas.data.universe import TRAIN_SUBPERIODS  # noqa: E402
from anomaly_atlas.stats.bootstrap import moving_block_bootstrap, percentile_ci  # noqa: E402
from anomaly_atlas.validation.artifacts import edge_spread  # noqa: E402

ADJ = "adj_split"
KAPPAS = [0.0, 0.1, 0.25, 0.5, 1.0, 2.0]
ONE_MIN_WINDOW = ("2014-01-01", "2016-01-01")
D_WINDOW = ("2014-01-01", "2016-01-01")


def latest(pattern: str) -> dict:
    return json.loads(Path(sorted(glob.glob(pattern))[-1]).read_text())


def pool_from_committed_results() -> tuple[list[dict], list[dict]]:
    """Rebuild the double-filtered pool mechanically (no hand-picking)."""
    rc = latest(str(REPO_ROOT / "results/expC_reversion_scan/*/results.json"))
    rf = latest(str(REPO_ROOT / "results/expF_multiple_testing/*/results.json"))
    rd = latest(str(REPO_ROOT / "results/expD_leadlag_scan/*/results.json"))

    cells = rc["cells"]
    for c in cells:
        c["vr30_excess"] = c["vr30"] - (1 + 2 * c["ac1"] * (1 - 1 / 30))
    triage = {(c["ticker"], c["timeframe"], c["period"]) for c in cells
              if c.get("fdr_vr30") and c["vr30_excess"] < -0.05}
    spa_r = {tuple(s[3:].split(":")) for s in rf["funnel"]["spa_step1_survivors"]
             if s[1] == "R"}  # (ticker, timeframe)
    rev_pool = [{"ticker": t, "timeframe": tf, "period": per,
                 "asset": "etf" if t in ("SPY", "QQQ") else "stock"}
                for (t, tf, per) in sorted(triage) if (t, tf) in spa_r]

    ll_pool = []
    for p in rd["pairs"]:
        if p["window"] != "2014-2015" or p["bucket"] == "index":
            continue
        if not (p.get("fdr_fresh_+1") or p.get("fdr_fresh_-1")):
            continue
        follower = p["pair"].split("->")[1] if p["pair"].startswith("SPY->") else "SPY"
        leader = "SPY" if follower != "SPY" else p["pair"].split("->")[0].split("[")[0]
        sign = np.sign(p["fresh_xcorr"]["1"]) if p.get("fdr_fresh_+1") else np.sign(
            p["fresh_xcorr"]["-1"])
        ll_pool.append({"pair": p["pair"], "leader": leader, "follower": follower,
                        "sign": int(sign), "bucket": p["bucket"]})
    return rev_pool, ll_pool


def contrarian_stream(bars: list[dict], timeframe: str) -> dict[str, tuple[float, float]]:
    """day -> (gross rule return, turnover) for the +contrarian rule."""
    by_day: dict[str, list[float]] = defaultdict(list)
    for b in bars:
        dt = b["datetime"]
        if timeframe == "1day" or "09:30" <= dt[11:16] < "16:00":
            by_day[dt[:10]].append(np.log(b["close"]))
    days = sorted(by_day)
    out: dict[str, tuple[float, float]] = {}
    if timeframe == "1day":
        closes = np.array([by_day[d][0] for d in days])
        r = np.diff(closes)
        pos_prev = 0.0
        for i in range(1, len(r)):
            pos = -np.sign(r[i - 1])
            out[days[i + 1]] = (float(pos * r[i]), float(abs(pos - pos_prev)))
            pos_prev = pos
        return out
    for d in days:
        p = np.array(by_day[d])
        if len(p) < 3:
            continue
        r = np.diff(p)
        pos = -np.sign(r[:-1])
        gross = float(np.sum(pos * r[1:]))
        turnover = float(abs(pos[0]) + np.sum(np.abs(np.diff(pos))) + abs(pos[-1]))
        out[d] = (gross, turnover)
    return out


def leadlag_stream(gx: dict, gy: dict, day_list: list[str], sign: int
                   ) -> dict[str, tuple[float, float]]:
    """day -> (gross, turnover) for pos_t = sign * sign(x_{t-1}) on fresh minutes."""
    out: dict[str, tuple[float, float]] = {}
    for day in day_list:
        px, py = gx.get(day), gy.get(day)
        if px is None or py is None:
            continue
        ox, oy = np.isfinite(px), np.isfinite(py)
        fpx, fpy = px.copy(), py.copy()
        for t in range(1, RTH_SLOTS):
            if not ox[t]:
                fpx[t] = fpx[t - 1]
            if not oy[t]:
                fpy[t] = fpy[t - 1]
        rx, ry = np.diff(fpx), np.diff(fpy)
        fx = ox[1:] & ox[:-1]
        fy = oy[1:] & oy[:-1]
        keep = fx[:-1] & fy[1:] & np.isfinite(rx[:-1]) & np.isfinite(ry[1:])
        if keep.sum() < 30:
            continue
        pos = np.zeros(len(ry))
        pos[1:][keep] = sign * np.sign(rx[:-1][keep])
        gross = float(np.sum(pos[1:] * ry[1:]))
        turnover = float(np.sum(np.abs(np.diff(np.concatenate([[0.0], pos, [0.0]])))))
        out[day] = (gross, turnover)
    return out


def half_spread(client: HFMarketDataClient, asset: str, ticker: str,
                adj: str | None, start: str, end: str) -> float:
    bars = client.get_bars(asset, ticker, "1day", adj, start, end)
    if len(bars) < 100:
        return float("nan")
    o = np.array([b["open"] for b in bars])
    h = np.array([b["high"] for b in bars])
    lo = np.array([b["low"] for b in bars])
    c = np.array([b["close"] for b in bars])
    s = edge_spread(o, h, lo, c)
    return s / 2.0 if np.isfinite(s) else float("nan")


def sanitize(obj):
    """Replace NaN/inf with None recursively — Python json emits bare NaN,
    which is invalid JSON for every other consumer (incl. the web platform)."""
    if isinstance(obj, dict):
        return {k: sanitize(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [sanitize(v) for v in obj]
    if isinstance(obj, float) and not np.isfinite(obj):
        return None
    return obj


def sweep(stream: dict[str, tuple[float, float]], hs: float) -> dict:
    days = sorted(d for d in stream if np.isfinite(stream[d][0]) and np.isfinite(stream[d][1]))
    gross = np.array([stream[d][0] for d in days])
    turn = np.array([stream[d][1] for d in days])
    res = {
        "n_days": len(days),
        "gross_mean_daily_bp": round(float(gross.mean()) * 1e4, 3),
        "turnover_per_day": round(float(turn.mean()), 2),
        "half_spread_bp": round(hs * 1e4, 3) if np.isfinite(hs) else None,
        "net": {},
    }
    if not np.isfinite(hs) or hs <= 0:
        res["kappa_star"] = None
        return res
    for k in KAPPAS:
        net = gross - k * hs * turn
        boot = moving_block_bootstrap(net, lambda x: float(np.mean(x)),
                                      block=21, n_boot=300, seed=42)
        lo_ci, hi_ci = percentile_ci(boot)
        res["net"][str(k)] = {
            "mean_daily_bp": round(float(net.mean()) * 1e4, 3),
            "ci95_bp": [round(lo_ci * 1e4, 3), round(hi_ci * 1e4, 3)],
            "positive": bool(net.mean() > 0),
        }
    denom = hs * turn.mean()
    res["kappa_star"] = round(float(gross.mean() / denom), 4) if denom > 0 else None
    return res


def main() -> None:
    run_utc = datetime.now(UTC)
    client = HFMarketDataClient()
    rev_pool, ll_pool = pool_from_committed_results()
    print("reversion pool:", [(r["ticker"], r["timeframe"], r["period"]) for r in rev_pool])
    print("leadlag pool:", [(p["pair"], p["sign"]) for p in ll_pool])

    items = []
    for r in rev_pool:
        if r["period"] in TRAIN_SUBPERIODS:
            s, e = TRAIN_SUBPERIODS[r["period"]]
        else:
            s, e = ONE_MIN_WINDOW
        bars = client.get_bars(r["asset"], r["ticker"], r["timeframe"], ADJ, s, e)
        stream = contrarian_stream(bars, r["timeframe"])
        hs = half_spread(client, r["asset"], r["ticker"], ADJ, s, e)
        item = {"rule": f"R:{r['ticker']}:{r['timeframe']}:{r['period']}",
                "family": "reversion"} | sweep(stream, hs)
        items.append(item)
        print(item["rule"], "kappa* =", item["kappa_star"])

    # lead-lag pool (2014-2015 window)
    s, e = D_WINDOW
    needed = {"SPY"} | {p["leader"] for p in ll_pool} | {p["follower"] for p in ll_pool}
    grids = {}
    for name in sorted(needed):
        if name == "ES":
            g = rth_day_grids(client.get_bars("futures", "ES", "1min",
                                              "contin_adj_ratio", s, e))
        else:
            asset = "etf" if name in ("SPY", "QQQ", "XLF", "XLE", "XLK", "XLV", "XLI",
                                      "XLY", "XLP", "XLU", "XLB") else "stock"
            g = rth_day_grids(client.get_bars(asset, name, "1min", ADJ, s, e))
        grids[name] = g
    day_list = sorted(grids["SPY"].keys())
    for p in ll_pool:
        stream = leadlag_stream(grids[p["leader"]], grids[p["follower"]],
                                day_list, p["sign"])
        traded = p["follower"]
        asset = ("futures" if traded == "ES" else
                 "etf" if traded in ("SPY", "QQQ") or traded.startswith("XL") else "stock")
        adj = "contin_adj_ratio" if traded == "ES" else ADJ
        hs = half_spread(client, asset, traded, adj, s, e)
        item = {"rule": f"L:{p['pair']}:{'+' if p['sign'] > 0 else '-'}",
                "family": "leadlag", "traded": traded} | sweep(stream, hs)
        items.append(item)
        print(item["rule"], "kappa* =", item["kappa_star"])

    ks = [i["kappa_star"] for i in items
          if i["kappa_star"] is not None and np.isfinite(i["kappa_star"])]
    summary = {
        "pool_size": len(items),
        "n_kappa_star_defined": len(ks),
        "kappa_star_median": round(float(np.median(ks)), 4) if ks else None,
        "kappa_star_max": round(float(max(ks)), 4) if ks else None,
        "survivors_at": {str(k): [i["rule"] for i in items
                                  if i["net"].get(str(k), {}).get("positive")]
                         for k in (0.1, 0.25, 0.5, 1.0)},
        "intraday_survivor_at_1.0": [
            i["rule"] for i in items
            if ":1day:" not in i["rule"] and i["net"].get("1.0", {}).get("positive")],
    }

    results = {
        "experiment": "expG_cost_frontier",
        "run_utc": run_utc.isoformat(),
        "author": "Simon-Pierre Boucher",
        "contact": "contact@spboucher.ai",
        "data_source": "hfmarketdata.io",
        "confidence_level": 0,
        "protocol": {"kappas": KAPPAS, "cost_model": "net = gross - k*(EDGE/2)*turnover",
                     "pool": "mechanical intersection of committed expC/expD/expF results"},
        "items": items,
        "summary": summary,
        "client_stats": vars(client.stats) | {"refreshes": list(client.stats.refreshes)},
        "manifest": collect_manifest(),
    }
    out_dir = REPO_ROOT / "results" / "expG_cost_frontier" / run_utc.strftime("%Y%m%dT%H%M%SZ")
    out_dir.mkdir(parents=True)
    (out_dir / "results.json").write_text(
        json.dumps(sanitize(results), indent=2, allow_nan=False) + "\n")
    print(f"\nwrote {out_dir.relative_to(REPO_ROOT)}/results.json")
    print(json.dumps(summary, indent=1))


if __name__ == "__main__":
    main()