Warrior_EA/research/altdata/screen_tf.py

"""Does the wired alt-data + volume information survive a move from D1 to H4?

The user is choosing a chart timeframe. The D1 screens are the evidence for
D1; this is the SAME test at H4 - forward range over 30 H4 bars (~5 trading
days, comparable horizon), trailing-range positive control, noise negative
control, circular-shift permutation p-values, and the conditional MI|vol
column that decides incrementality.

Also reports the COST side of the timeframe question: ATR in spread units at
each timeframe, because EV per trade = edge x width, and width measured in
spreads is what shrinks when you shorten the timeframe.

Usage:
    python -m altdata.screen_tf [--tf H4] [--perms 199]
"""
import argparse

import numpy as np
import pandas as pd

from .common import DATA_ROOT
from .screen import _tercile, cmi_3x3x3, mi_3x3

HTF = DATA_ROOT.parent / "htf"

# per-symbol candidate map: (name, source csv, transform)
FRED = DATA_ROOT / "fred"


def load_tf(sym: str, tf: str) -> pd.DataFrame:
    z = np.load(HTF / f"{sym}_{tf}_mid.npz")
    return pd.DataFrame({
        "time": pd.to_datetime(z["t"], unit="ms"),
        "open": z["o"], "high": z["h"], "low": z["l"], "close": z["c"],
        "volume": z["v"], "spread": z["spread"],
    })


def fred_series(sid: str) -> pd.DataFrame:
    return pd.read_csv(FRED / f"{sid}.csv", parse_dates=["observed", "published"])


def asof(bars: pd.DataFrame, pub: pd.Series, val: pd.Series, name: str) -> np.ndarray:
    src = pd.DataFrame({"published": pub.astype("datetime64[ns]"), name: val}) \
            .sort_values("published").dropna()
    j = pd.merge_asof(bars[["time"]].astype({"time": "datetime64[ns]"}), src,
                      left_on="time", right_on="published", direction="backward")
    return j[name].to_numpy()


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--sym", nargs="*", default=["SP500", "EURUSD", "USDJPY", "XAUUSD"])
    ap.add_argument("--tf", default="H4")
    ap.add_argument("--fwd", type=int, default=30)       # ~5 trading days of H4 bars
    ap.add_argument("--perms", type=int, default=199)
    ap.add_argument("--seed", type=int, default=23)
    args = ap.parse_args()
    rng = np.random.default_rng(args.seed)

    for sym in args.sym:
        bars = load_tf(sym, args.tf)
        prev_close = bars["close"].shift(1)
        tr = pd.concat([bars["high"] - bars["low"],
                        (bars["high"] - prev_close).abs(),
                        (bars["low"] - prev_close).abs()], axis=1).max(axis=1)
        atr = tr.ewm(alpha=1 / 14, min_periods=14).mean()
        f = args.fwd
        y_raw = tr.shift(-1).rolling(f).sum().shift(-(f - 1)) / atr
        trailing = tr.rolling(f).sum() / atr

        feats = pd.DataFrame(index=bars.index)
        vix = fred_series("VIXCLS")
        feats["vix_chg5"] = asof(bars, vix["published"],
                                 vix["value"] - vix["value"].shift(5), "vix_chg5")
        if sym == "XAUUSD":
            gvz = fred_series("GVZCLS")
            feats["gvz_chg5"] = asof(bars, gvz["published"],
                                     gvz["value"] - gvz["value"].shift(5), "gvz_chg5")
        usd = fred_series("DTWEXBGS")
        feats["usd_chg5"] = asof(bars, usd["published"],
                                 usd["value"] - usd["value"].shift(5), "usd_chg5")
        cotp = DATA_ROOT / "cot" / f"{sym}_cot.csv"
        if cotp.exists():
            c = pd.read_csv(cotp, parse_dates=["observed", "published"])
            oi = c["Open_Interest_All"].replace(0, np.nan)
            # TFF family carries Lev_Money; the disaggregated family (gold, WTI)
            # carries M_Money - same economic role (the speculators)
            if "Lev_Money_Positions_Long_All" in c.columns:
                spec = (c["Lev_Money_Positions_Long_All"]
                        - c["Lev_Money_Positions_Short_All"]) / oi
            else:
                spec = (c["M_Money_Positions_Long_All"]
                        - c["M_Money_Positions_Short_All"]) / oi
            if sym == "USDJPY":
                spec = -spec
            feats["cot_spec_net"] = asof(bars, c["published"], spec, "cot_spec_net")
        #--- the EA's activity feature at this timeframe (tick volume vs 50-bar mean)
        feats["volLevel50"] = bars["volume"] / bars["volume"].rolling(50).mean() \
                                                            .replace(0, np.nan)
        feats["CTRL_trailing_range"] = trailing
        feats["CTRL_noise"] = rng.standard_normal(len(bars))

        valid = y_raw.notna() & trailing.notna()
        y_bin = _tercile(y_raw[valid].to_numpy())
        z_bin = _tercile(trailing[valid].to_numpy())
        spr_atr = (bars["spread"] / atr).median()
        print(f"\n=== {sym} {args.tf}: {int(valid.sum())} bars, fwd {f} bars, "
              f"{args.perms} perms | median spread = {spr_atr:.4f} x ATR({args.tf}) ===")

        results, null_max = [], np.zeros(args.perms)
        for col in feats.columns:
            x = feats.loc[valid, col].to_numpy(dtype=float)
            ok = ~np.isnan(x)
            if ok.sum() < 2000:
                continue
            xo, yo, zo = x[ok], y_bin[ok], z_bin[ok]
            n = len(xo)
            obs, obs_c = mi_3x3(xo, yo), cmi_3x3x3(xo, yo, zo)
            null = np.empty(args.perms)
            null_c = np.empty(args.perms)
            #--- shift by whole ~weeks of H4 bars so the null respects serial dependence
            for k in range(args.perms):
                xs = np.roll(xo, rng.integers(63 * 6, n - 63 * 6))
                null[k] = mi_3x3(xs, yo)
                null_c[k] = cmi_3x3x3(xs, yo, zo)
            p = (np.sum(null >= obs) + 1) / (args.perms + 1)
            p_c = (np.sum(null_c >= obs_c) + 1) / (args.perms + 1)
            results.append((col, obs, p, obs_c, p_c))
            if not col.startswith("CTRL_"):
                null_max = np.maximum(null_max, null)
        fam = np.quantile(null_max, 0.95)
        print(f"  family-wise 5% bar: {fam:.5f}")
        for col, obs, p, obs_c, p_c in sorted(results, key=lambda r: -r[3]):
            mark = "FAM" if obs > fam else "   "
            inc = "INCR" if p_c <= 0.05 else "    "
            print(f"  {col:22s} MI {obs:.5f} p={p:.4f} {mark} | "
                  f"MI|vol {obs_c:.5f} p={p_c:.4f} {inc}")


if __name__ == "__main__":
    main()