#!/usr/bin/env python3
"""
Phase 1 — historical candle divergence study (data source change test)
=======================================================================
Fetches the RAW closed-bar series for XAU/USD from two providers over a
multi-month window via the normal provider stack (`fetch_range`, which
paginates under the per-request bar cap and drops the forming candle),
then quantifies how the feeds diverge:

  * coverage: % shared timestamps, feed-only bars, where they cluster
  * OHLC |Δ| distribution (mean/median/p90/p95/p99/max, $ and bias)
  * |ΔClose| vs mean ATR14 (the scale the signal engine cares about)
  * threshold exceedance (how many bars differ by more than X $)
  * bar-direction agreement (sign of close−open)
  * close-return correlation on consecutive bars
  * per-hour and per-month drift

Both feeds are mid-price (Oanda mid of MBA candles; Tiingo resampled mid;
Twelve Data its own mid). Frames are cached to parquet so re-runs are
instant; pass --no-cache to force a fresh fetch.

Usage:
    venv/bin/python scripts/feed_divergence_study.py --a oanda --b tiingo \
        --tfs M15,H1,H4 --days 365
"""

import argparse
import sys
import time
from pathlib import Path

import numpy as np
import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from config import PlatformConfig, Timeframe, XAUUSD
from providers import make_data_provider

REPO = Path(__file__).resolve().parent.parent
CACHE_DIR = REPO / "backtest" / "cache"
RESEARCH_DIR = REPO / "backtest" / "research"

_TF_LOOKUP = {t.value: t for t in Timeframe}
_TF_LOOKUP.update({
    "M1": Timeframe.M1, "M5": Timeframe.M5, "M15": Timeframe.M15,
    "M30": Timeframe.M30, "H1": Timeframe.H1, "H4": Timeframe.H4,
    "D1": Timeframe.D1, "W1": Timeframe.W1,
})


def wilder_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
    hl = df["High"] - df["Low"]
    hc = (df["High"] - df["Close"].shift(1)).abs()
    lc = (df["Low"] - df["Close"].shift(1)).abs()
    tr = pd.concat([hl, hc, lc], axis=1).max(axis=1)
    return tr.ewm(alpha=1.0 / period, adjust=False).mean()


def fmt_stats(name: str, s: pd.Series) -> str:
    a = s.abs()
    return (f"  {name:<6} mean ${a.mean():.3f}  med ${a.median():.3f}  "
            f"p90 ${a.quantile(0.90):.3f}  p95 ${a.quantile(0.95):.3f}  "
            f"p99 ${a.quantile(0.99):.3f}  max ${a.max():.2f}  "
            f"bias {s.mean():+.4f}")


def cache_path(name: str, tf: Timeframe, start: pd.Timestamp,
               end: pd.Timestamp) -> Path:
    return (CACHE_DIR /
            f"{name}_XAU_USD_{tf.value}_{start:%Y%m%d}_{end:%Y%m%d}.parquet")


def load_or_fetch(provider, name: str, tf: Timeframe, start, end,
                  no_cache: bool):
    cp = cache_path(name, tf, start, end)
    if not no_cache and cp.exists():
        df = pd.read_parquet(cp)
        print(f"  [{name}] loaded {len(df):,} bars from cache "
              f"{cp.name}")
        return df
    t0 = time.time()
    print(f"  [{name}] fetching {tf.value} {start.date()} → {end.date()} ...")
    df = provider.fetch_range(XAUUSD, tf, start, end)
    print(f"  [{name}] fetched {len(df):,} bars in {time.time() - t0:.1f}s")
    df.to_parquet(cp)
    print(f"  [{name}] cached → {cp.name}")
    return df


def compare(tf: Timeframe, a_name: str, b_name: str, a: pd.DataFrame,
            b: pd.DataFrame):
    print("\n" + "=" * 80)
    print(f"{tf.value} — {a_name.upper()} vs {b_name.upper()}, XAU/USD")
    print("=" * 80)

    common = a.index.intersection(b.index)
    a_only = a.index.difference(b.index)
    b_only = b.index.difference(a.index)

    print(f"\nCOVERAGE")
    print(f"  {a_name} bars : {len(a):>8,}   head {a.index[0]}  tail {a.index[-1]}")
    print(f"  {b_name} bars : {len(b):>8,}   head {b.index[0]}  tail {b.index[-1]}")
    print(f"  shared        : {len(common):>8,}  "
          f"({100 * len(common) / max(len(a.index.union(b.index)), 1):.2f}% of union)")
    print(f"  {a_name}-only : {len(a_only):>8,}  "
          f"({100 * len(a_only) / max(len(a), 1):.2f}% of {a_name})")
    print(f"  {b_name}-only : {len(b_only):>8,}  "
          f"({100 * len(b_only) / max(len(b), 1):.2f}% of {b_name})")

    if len(common) == 0:
        print("  !! no shared timestamps — feeds on different bar grids")
        return

    for label, idx in ((f"{a_name}-only", a_only),
                       (f"{b_name}-only", b_only)):
        if not len(idx):
            continue
        by_date = pd.Series(1, index=idx).groupby(idx.date).sum()
        top = by_date.sort_values(ascending=False).head(6)
        print(f"  {label} clusters on: "
              + ", ".join(f"{d} ({n})" for d, n in top.items()))

    x = a.loc[common]
    y = b.loc[common]

    atr = wilder_atr(x).dropna()
    mean_price = x["Close"].mean()
    print(f"\nSCALE REFERENCES ({a_name} feed)")
    print(f"  mean close    : ${mean_price:,.2f}")
    print(f"  mean bar range: ${(x['High'] - x['Low']).mean():.3f}")
    print(f"  mean ATR14    : ${atr.mean():.3f}   median ${atr.median():.3f}")

    print(f"\nPER-BAR DIFFS  ({a_name} − {b_name}, $)")
    diffs = {}
    for col in ("Open", "High", "Low", "Close"):
        d = x[col] - y[col]
        diffs[col] = d
        print(fmt_stats(col, d))

    dclose = diffs["Close"].abs()
    print(f"\n  |ΔClose| as % of mean ATR14 : "
          f"mean {100 * dclose.mean() / atr.mean():.1f}%   "
          f"p95 {100 * dclose.quantile(0.95) / atr.mean():.1f}%")

    print(f"\nTHRESHOLD EXCEEDANCE  (|ΔClose|)")
    for thr in (0.10, 0.30, 0.65, 1.00, 2.00, 5.00):
        pct = 100 * (dclose > thr).mean()
        print(f"  > ${thr:<5.2f}: {pct:6.2f}% of bars  ({(dclose > thr).sum():,})")

    # direction agreement
    a_dir = np.sign(x["Close"] - x["Open"])
    b_dir = np.sign(y["Close"] - y["Open"])
    both_nonflat = (a_dir != 0) & (b_dir != 0)
    dir_agree = 100 * (a_dir[both_nonflat] == b_dir[both_nonflat]).mean()
    print(f"\nAGREEMENT")
    print(f"  bar direction (sign of close−open) agree: {dir_agree:.2f}%"
          f"  (n={both_nonflat.sum():,})")

    step = pd.Series(common, index=common).diff().median()
    a_ret = x["Close"].pct_change()
    b_ret = y["Close"].pct_change()
    step_ok = pd.Series(common, index=common).diff().le(step + pd.Timedelta(seconds=1))
    corr = a_ret[step_ok].corr(b_ret[step_ok])
    print(f"  close-return correlation (consecutive bars): {corr:.5f}")

    # by-hour
    hours = common.hour
    rows = []
    for h in range(24):
        m = hours == h
        if not m.any():
            continue
        rows.append({"hour_utc": h, "bars": int(m.sum()),
                     "mean_abs_dclose": dclose[m].mean(),
                     "p95_abs_dclose": dclose[m].quantile(0.95)})
    by_hour = pd.DataFrame(rows)
    print(f"\nBY UTC HOUR  (mean |ΔClose|, $)")
    print(by_hour.to_string(index=False,
                            float_format=lambda v: f"{v:.3f}"))

    print(f"\nBY MONTH  (mean |ΔClose|, $)")
    months = dclose.groupby(common.tz_localize(None).to_period("M")).mean()
    for p, v in months.items():
        print(f"  {p}: {v:.3f}")

    return {
        "tf": tf.value, "a": a_name, "b": b_name,
        "a_bars": len(a), "b_bars": len(b), "common": len(common),
        "a_only": len(a_only), "b_only": len(b_only),
        "dir_agree_pct": round(dir_agree, 3),
        "ret_corr": round(corr, 5),
        "mean_abs_dclose": dclose.mean(),
        "p95_abs_dclose": dclose.quantile(0.95),
        "max_abs_dclose": dclose.max(),
        "dclose_pct_atr_mean": 100 * dclose.mean() / atr.mean(),
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--a", required=True)
    ap.add_argument("--b", required=True)
    ap.add_argument("--tfs", default="M15,H1,H4",
                    help="comma-separated timeframe values (default M15,H1,H4)")
    ap.add_argument("--days", type=int, default=365)
    ap.add_argument("--end", default=None, help="UTC end date (default now)")
    ap.add_argument("--no-cache", action="store_true")
    args = ap.parse_args()

    tfs = []
    for s in args.tfs.split(","):
        s = s.strip()
        tf = _TF_LOOKUP.get(s)
        if tf is None:
            sys.exit(f"unknown timeframe '{s}'; use one of "
                     f"{sorted(_TF_LOOKUP)}")
        tfs.append(tf)

    end = pd.Timestamp(args.end, tz="UTC") if args.end \
        else pd.Timestamp.now(tz="UTC")
    start = end - pd.Timedelta(days=args.days)

    cfg = PlatformConfig()
    pa = make_data_provider(cfg, args.a)
    pb = make_data_provider(cfg, args.b)

    summary = []
    for tf in tfs:
        print(f"\n{'─' * 80}\nWINDOW {start.date()} → {end.date()}  "
              f"({args.days} days), {tf.value}")
        try:
            a = load_or_fetch(pa, args.a, tf, start, end, args.no_cache)
        except Exception as e:
            print(f"  !! {args.a} {tf.value} fetch failed: {e}")
            continue
        try:
            b = load_or_fetch(pb, args.b, tf, start, end, args.no_cache)
        except Exception as e:
            print(f"  !! {args.b} {tf.value} fetch failed: {e}")
            continue
        r = compare(tf, args.a, args.b, a, b)
        if r:
            summary.append(r)

    print("\n" + "=" * 80)
    print("SUMMARY TABLE")
    print("=" * 80)
    sdf = pd.DataFrame(summary)
    print(sdf.to_string(index=False,
                        float_format=lambda v: f"{v:.4f}"))
    out = RESEARCH_DIR / f"feed_divergence_{args.a}_vs_{args.b}.csv"
    if len(sdf):
        sdf.to_csv(out, index=False, float_format="%.6f")
        print(f"\nSummary written to {out}")


if __name__ == "__main__":
    main()