# =============================================================================
#  Project   : anomaly-atlas
#  File      : experiments/micro/expC_reversion_scan/benchmark.py
#  Purpose   : Mean-reversion scan net of the EDGE bounce null (TRAIN only)
#  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 C — reversion scan (protocol pre-specified in hypothesis.md).

Every output cell is Level 0. The scan's job is triage: which cells show
excess reversion beyond the bounce null after FDR — those go to the expF
correction battery, nothing goes to the atlas from here.
"""

from __future__ import annotations

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.hf_client import HFMarketDataClient  # noqa: E402
from anomaly_atlas.data.universe import (  # noqa: E402
    LIQUID_ETF,
    LIQUID_STOCK,
    TRAIN_SUBPERIODS,
    core_universe,
)
from anomaly_atlas.stats.bootstrap import moving_block_bootstrap, percentile_ci  # noqa: E402
from anomaly_atlas.stats.multiple_testing import (  # noqa: E402
    benjamini_hochberg,
    bootstrap_pvalue,
)
from anomaly_atlas.stats.reversion import ac1, half_life, variance_ratio  # noqa: E402
from anomaly_atlas.validation.artifacts import edge_spread  # noqa: E402

ADJ = "adj_split"
BOOT_N, BOOT_SEED = 200, 42
TIMEFRAMES = ["1day", "30min", "5min"]
BARS_PER_DAY = {"5min": 78, "30min": 13, "1day": 1}
ONE_MIN_WINDOW = ("2014-01-01", "2016-01-01")  # liquid 12 only (declared)
MIN_RETURNS = {"1day": 350, "30min": 2_000, "5min": 5_000, "1min": 5_000}


def rth_returns_by_day(bars: list[dict], timeframe: str) -> np.ndarray:
    """Within-day log returns on RTH bars only (no overnight, no LOCF)."""
    days: dict[str, list[float]] = defaultdict(list)
    for b in bars:
        dt = b["datetime"]
        if timeframe == "1day":
            days[dt[:10]].append(np.log(b["close"]))
            continue
        if "09:30" <= dt[11:16] < "16:00":
            days[dt[:10]].append(np.log(b["close"]))
    if timeframe == "1day":
        allp = [v[0] for _, v in sorted(days.items())]
        return np.diff(allp) if len(allp) > 2 else np.array([])
    out = [np.diff(v) for _, v in sorted(days.items()) if len(v) >= 2]
    return np.concatenate(out) if out else np.array([])


def daily_ohlc(bars: list[dict]) -> tuple[np.ndarray, ...]:
    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])
    return o, h, lo, c


def analyze_cell(r: np.ndarray, timeframe: str, spread: float) -> dict | None:
    if len(r) < MIN_RETURNS[timeframe]:
        return None
    var = r.var()
    if var == 0:
        return None
    bounce_ac1 = -(spread**2) / 4.0 / var if np.isfinite(spread) else 0.0
    block = max(20, BARS_PER_DAY.get(timeframe, 390) * 5)
    boot = moving_block_bootstrap(r, ac1, block=block, n_boot=BOOT_N, seed=BOOT_SEED)
    excess_boot = boot - bounce_ac1
    a = ac1(r)
    lo, hi = percentile_ci(excess_boot)
    boot_vr5 = moving_block_bootstrap(
        r, lambda x: variance_ratio(x, 5), block=block, n_boot=BOOT_N, seed=BOOT_SEED
    )
    boot_vr30 = moving_block_bootstrap(
        r, lambda x: variance_ratio(x, 30), block=block, n_boot=BOOT_N, seed=BOOT_SEED
    )
    return {
        "n": int(len(r)),
        "ac1": round(a, 5),
        "edge_spread": round(spread, 6) if np.isfinite(spread) else None,
        "bounce_ac1": round(bounce_ac1, 5),
        "excess_ac1": round(a - bounce_ac1, 5),
        "excess_ci95": [round(lo, 5), round(hi, 5)],
        "p_excess": bootstrap_pvalue(excess_boot, 0.0),
        "vr5": round(variance_ratio(r, 5), 4),
        "p_vr5": bootstrap_pvalue(boot_vr5, 1.0),
        "vr30": round(variance_ratio(r, 30), 4),
        "p_vr30": bootstrap_pvalue(boot_vr30, 1.0),
    }


def main() -> None:
    run_utc = datetime.now(UTC)
    client = HFMarketDataClient()
    universe = core_universe(client.tickers("stock", timeframe="1min", adjustment=ADJ))

    cells: list[dict] = []
    for asset, ticker, bucket in universe:
        # daily bars per sub-period: returns for 1day cells + EDGE spread input
        for sub, (s, e) in TRAIN_SUBPERIODS.items():
            day_bars = client.get_bars(asset, ticker, "1day", ADJ, s, e)
            if len(day_bars) < 200:
                continue
            o, h, lo, c = daily_ohlc(day_bars)
            spread = edge_spread(o, h, lo, c)
            hl = half_life(np.log(c))
            for tf in TIMEFRAMES:
                bars = day_bars if tf == "1day" else client.get_bars(asset, ticker, tf, ADJ, s, e)
                r = rth_returns_by_day(bars, tf)
                m = analyze_cell(r, tf, spread)
                if m is None:
                    continue
                m |= {"ticker": ticker, "bucket": bucket, "timeframe": tf, "period": sub}
                if tf == "1day":
                    m["half_life_days"] = round(hl, 1) if np.isfinite(hl) else None
                cells.append(m)
            print(f"{ticker} {sub}: done ({len(cells)} cells)")

    # 1min cells: liquid 12, declared window
    for asset, ticker in [("stock", t) for t in LIQUID_STOCK] + [("etf", t) for t in LIQUID_ETF]:
        s, e = ONE_MIN_WINDOW
        day_bars = client.get_bars(asset, ticker, "1day", ADJ, s, e)
        if len(day_bars) < 200:
            continue
        spread = edge_spread(*daily_ohlc(day_bars))
        bars = client.get_bars(asset, ticker, "1min", ADJ, s, e)
        r = rth_returns_by_day(bars, "1min")
        m = analyze_cell(r, "1min", spread)
        if m is not None:
            m |= {"ticker": ticker, "bucket": "liquid", "timeframe": "1min",
                  "period": "2014-2015"}
            cells.append(m)
        print(f"{ticker} 1min: done")

    # FDR within each statistic family, all cells jointly
    for key, pkey in [("excess_ac1", "p_excess"), ("vr5", "p_vr5"), ("vr30", "p_vr30")]:
        mask = benjamini_hochberg(np.array([c[pkey] for c in cells]), alpha=0.05)
        for c, rej in zip(cells, mask, strict=True):
            c[f"fdr_{key}"] = bool(rej)

    def survivors(key: str, sign_key: str, negative: bool) -> list[dict]:
        out = []
        for c in cells:
            if not c[f"fdr_{key}"]:
                continue
            v = c[sign_key] - (1.0 if sign_key.startswith("vr") else 0.0)
            if (v < 0) == negative:
                out.append({k: c[k] for k in ("ticker", "bucket", "timeframe", "period",
                                              sign_key, "edge_spread")})
        return out

    summary = {
        "cells_total": len(cells),
        "families_tested": 3,
        "fdr_alpha": 0.05,
        "excess_ac1_negative_survivors": survivors("excess_ac1", "excess_ac1", True),
        "excess_ac1_positive_survivors": survivors("excess_ac1", "excess_ac1", False),
        "vr30_below_1_survivors": len(survivors("vr30", "vr30", True)),
        "vr30_above_1_survivors": len(survivors("vr30", "vr30", False)),
        "liquid_2008_2015_intraday_negative": [
            c["ticker"] for c in cells
            if c["bucket"] == "liquid" and c["period"] == "2008-2015"
            and c["timeframe"] in ("5min", "30min")
            and c["fdr_excess_ac1"] and c["excess_ac1"] < 0
        ],
    }

    results = {
        "experiment": "expC_reversion_scan",
        "run_utc": run_utc.isoformat(),
        "author": "Simon-Pierre Boucher",
        "contact": "contact@spboucher.ai",
        "data_source": "hfmarketdata.io",
        "confidence_level": 0,
        "protocol": {
            "train_subperiods": TRAIN_SUBPERIODS, "adjustment": ADJ,
            "one_min_window": ONE_MIN_WINDOW, "boot": [BOOT_N, BOOT_SEED],
            "min_returns": MIN_RETURNS,
        },
        "summary": summary,
        "cells": cells,
        "client_stats": vars(client.stats) | {"refreshes": list(client.stats.refreshes)},
        "manifest": collect_manifest(),
    }
    out_dir = REPO_ROOT / "results" / "expC_reversion_scan" / run_utc.strftime("%Y%m%dT%H%M%SZ")
    out_dir.mkdir(parents=True)
    (out_dir / "results.json").write_text(json.dumps(results, indent=2) + "\n")
    print(f"\nwrote {out_dir.relative_to(REPO_ROOT)}/results.json")
    print(json.dumps({k: (v if not isinstance(v, list) else len(v))
                      for k, v in summary.items()}, indent=1))


if __name__ == "__main__":
    main()