# =============================================================================
#  Project   : anomaly-atlas
#  File      : experiments/micro/expA_data_reality/benchmark.py
#  Purpose   : Benchmark runner: empirical data reality check of hfmarketdata.io
#  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 A — establish empirically what hfmarketdata.io actually returns.

Probes (all through the single hf_client, so every response is cached and
manifest-indexed):
  A1  dataset inventory (/v1/status): assets x timeframes x adjustments
  A2  history bounds per asset class (earliest/latest 1min and 1day)
  A3  intraday session structure and missing-minute patterns (liquid vs sparse)
  A4  daily-bar vs 1min-aggregate semantics (official close vs last bar)
  A5  corporate-action adjustment semantics around the AAPL 2020 4:1 split
  A6  response row cap + pagination correctness on a full year of 1min bars
  A7  request latency profile (small cached-miss requests)
  A8  options coverage (quarters, expirations, chain columns)

Writes results/expA_data_reality/<UTC timestamp>/results.json embedding the
hardware manifest and the client's instrumentation stats.
"""

from __future__ import annotations

import json
import sys
import time
from collections import Counter
from datetime import UTC, datetime
from pathlib import Path

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

REPRESENTATIVE = {
    "stock": ("AAPL", "adj_splitdiv"),
    "etf": ("SPY", "adj_splitdiv"),
    "futures": ("ES", "contin_adj_ratio"),
    "index": ("SPX", None),
    "fx": ("EURUSD", None),
    "crypto": ("BTC", None),
}
PROBE_DAY = "2026-08-06"  # a regular Thursday inside every dataset's coverage


def bounds(
    client: HFMarketDataClient, asset: str, ticker: str, adjustment: str | None, timeframe: str
) -> dict:
    def one(order: str) -> str | None:
        rows = client.get(
            f"/v1/bars/{asset}/{ticker}",
            {
                "timeframe": timeframe,
                "adjustment": adjustment,
                "order": order,
                "limit": 1,
            },
        ).get("data", [])
        return rows[0]["datetime"] if rows else None

    return {"earliest": one("asc"), "latest": one("desc")}


def session_structure(
    client: HFMarketDataClient, asset: str, ticker: str, adjustment: str | None
) -> dict:
    rows = client.get(
        f"/v1/bars/{asset}/{ticker}",
        {
            "timeframe": "1min",
            "adjustment": adjustment,
            "start": PROBE_DAY,
            "end": "2026-08-07",
            "order": "asc",
            "limit": 50_000,
        },
    ).get("data", [])
    rows = [r for r in rows if r["datetime"][:10] == PROBE_DAY]
    if not rows:
        return {"bars": 0}
    per_hour = Counter(r["datetime"][11:13] for r in rows)
    rth = [r for r in rows if "09:30" <= r["datetime"][11:16] < "16:00"]
    return {
        "bars": len(rows),
        "first": rows[0]["datetime"],
        "last": rows[-1]["datetime"],
        "bars_per_hour": dict(sorted(per_hour.items())),
        "zero_volume_bars": sum(1 for r in rows if r.get("volume") == 0),
        "has_volume_field": "volume" in rows[0],
        "rth_bars_of_390": len(rth),
    }


def main() -> None:
    run_utc = datetime.now(UTC)
    client = HFMarketDataClient()
    results: dict = {
        "experiment": "expA_data_reality",
        "run_utc": run_utc.isoformat(),
        "author": "Simon-Pierre Boucher",
        "contact": "contact@spboucher.ai",
        "data_source": "hfmarketdata.io",
    }

    # A1 — inventory
    status = client.status()
    results["A1_inventory"] = status.get("datasets", {})
    results["A1_options_quarters"] = (
        status.get("datasets", {}).get("options", {}).get("quarters", [])
    )

    # A2 — history bounds per class, 1min and 1day
    results["A2_bounds"] = {
        asset: {tf: bounds(client, asset, tk, adj, tf) for tf in ("1min", "1day")}
        for asset, (tk, adj) in REPRESENTATIVE.items()
    }

    # A3 — session structure, liquid + sparse
    results["A3_session"] = {
        f"{asset}:{tk}": session_structure(client, asset, tk, adj)
        for asset, (tk, adj) in REPRESENTATIVE.items()
    }
    results["A3_session"]["stock:AIZN(sparse)"] = session_structure(
        client, "stock", "AIZN", "adj_splitdiv"
    )

    # A4 — daily bar vs 1min RTH aggregate (close semantics)
    day = client.get(
        "/v1/bars/stock/AAPL",
        {
            "timeframe": "1day",
            "adjustment": "adj_splitdiv",
            "start": PROBE_DAY,
            "end": "2026-08-07",
        },
    )["data"][0]
    intraday = client.get(
        "/v1/bars/stock/AAPL",
        {
            "timeframe": "1min",
            "adjustment": "adj_splitdiv",
            "start": PROBE_DAY,
            "end": "2026-08-07",
            "order": "asc",
            "limit": 50_000,
        },
    )["data"]
    rth = [r for r in intraday if "09:30" <= r["datetime"][11:16] < "16:00"]
    results["A4_daily_vs_intraday"] = {
        "daily_bar": day,
        "rth_1min_aggregate": {
            "open": rth[0]["open"],
            "high": max(r["high"] for r in rth),
            "low": min(r["low"] for r in rth),
            "close": rth[-1]["close"],
            "volume": sum(r["volume"] for r in rth),
            "bars": len(rth),
        },
        "extended_1min_volume": sum(r["volume"] for r in intraday),
    }

    # A5 — adjustment semantics around AAPL 2020-08-31 4:1 split
    results["A5_adjustments"] = {}
    for adj in ("UNADJUSTED", "adj_split", "adj_splitdiv"):
        rows = client.get(
            "/v1/bars/stock/AAPL",
            {
                "timeframe": "1day",
                "adjustment": adj,
                "start": "2020-08-28",
                "end": "2020-09-01",
            },
        )["data"]
        results["A5_adjustments"][adj] = [
            {"date": r["datetime"][:10], "close": r["close"]} for r in rows
        ]

    # A6 — row cap + pagination on a full year of SPY 1min
    capped = client.get(
        "/v1/bars/etf/SPY",
        {
            "timeframe": "1min",
            "adjustment": "adj_splitdiv",
            "start": "2020-01-01",
            "end": "2021-01-01",
            "order": "asc",
            "limit": 1_000_000,
        },
    )
    year = client.get_bars("etf", "SPY", "1min", "adj_splitdiv", "2020-01-01", "2021-01-01")
    dts = [r["datetime"] for r in year]
    results["A6_row_cap_and_pagination"] = {
        "requested_rows": 1_000_000,
        "returned_rows_single_request": capped.get("count"),
        "paginated_total_rows": len(year),
        "paginated_first": dts[0],
        "paginated_last": dts[-1],
        "duplicates_after_pagination": len(dts) - len(set(dts)),
        "monotonic_ascending": all(a < b for a, b in zip(dts, dts[1:], strict=False)),
    }

    # A7 — latency profile (cache misses: distinct small requests)
    latencies = []
    for month in range(1, 11):
        t0 = time.perf_counter()
        client.get(
            "/v1/bars/stock/MSFT",
            {
                "timeframe": "1day",
                "adjustment": "adj_splitdiv",
                "start": f"2025-{month:02d}-01",
                "end": f"2025-{month:02d}-05",
            },
        )
        latencies.append(round(time.perf_counter() - t0, 4))
    results["A7_latency_s"] = {
        "samples": latencies,
        "note": "sequential small requests, cold cache; no rate-limit headers observed",
    }

    # A8 — options coverage
    expiries = client.options_expirations("SPY", trade_date="2026-06-15")
    chain = client.options_chain("SPY", "2026-06-15", limit=2)
    results["A8_options"] = {
        "n_quarters": len(results["A1_options_quarters"]),
        "first_quarter": results["A1_options_quarters"][:1],
        "last_quarter": results["A1_options_quarters"][-1:],
        "spy_expirations_on_2026-06-15": len(expiries),
        "chain_columns": sorted(chain[0].keys()) if chain else None,
        "granularity": "daily (one row per contract per trade_date)",
    }

    # instrumentation + hardware
    results["client_stats"] = {
        "network_requests": client.stats.network_requests,
        "cache_hits": client.stats.cache_hits,
        "rows_fetched": client.stats.rows_fetched,
        "seconds_waiting": round(client.stats.seconds_waiting, 2),
        "errors_retried": client.stats.errors_retried,
    }
    results["manifest"] = collect_manifest()

    out_dir = REPO_ROOT / "results" / "expA_data_reality" / run_utc.strftime("%Y%m%dT%H%M%SZ")
    out_dir.mkdir(parents=True)
    out = out_dir / "results.json"
    out.write_text(json.dumps(results, indent=2) + "\n")
    print(f"wrote {out.relative_to(REPO_ROOT)}")
    print(json.dumps(results["client_stats"], indent=2))


if __name__ == "__main__":
    main()