Warrior_EA/research/altdata/screen_macro.py

"""Macro screen: rates, curve, inflation expectations, policy differentials,
and - to settle the question rather than assert it - monthly country statistics.

THE POWER POINT, made concretely instead of rhetorically. A feature joined to
D1 bars as-of its publication is a STEP FUNCTION: monthly unemployment takes
~200 distinct values across a 16-year sample no matter how many bars it is
pasted onto. The `distinct` column below is the honest sample size, and it is
what decides whether a feature can clear a permutation gate at all. Daily rate
series do not have this problem; monthly country statistics do, badly.

Vintage discipline: every series here comes through fred.py's ALFRED first-print
path where an archive exists (so revised series are joined at what was actually
knowable), and the unrevised path with published = observed + 1 day otherwise.

Usage:
    python -m altdata.screen_macro [--perms 499]
"""
import argparse

import numpy as np
import pandas as pd

from .common import DATA_ROOT
from .screen import _tercile, asof_join, cmi_3x3x3, load_bars, mi_3x3, true_range


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


def build(bars: pd.DataFrame) -> pd.DataFrame:
    """One column per candidate, as-of joined onto the bar grid."""
    out = pd.DataFrame(index=bars.index)

    def col(sid: str, name: str, kind: str, k: int = 5):
        f = series(sid)
        v = f["value"]
        if kind == "level":
            val = v
        elif kind == "diff":
            val = v - v.shift(k)
        elif kind == "yoy":
            val = v / v.shift(k) - 1.0
        frame = pd.DataFrame({"published": f["published"], name: val})
        out[name] = asof_join(bars, frame)[name].to_numpy()

    # ---- daily: rates, curve, inflation expectations
    col("DGS10", "dgs10", "level")
    col("DGS10", "dgs10_chg5", "diff", 5)
    col("DGS10", "dgs10_chg20", "diff", 20)
    col("DGS2", "dgs2_chg5", "diff", 5)
    col("T10Y2Y", "curve", "level")
    col("T10Y2Y", "curve_chg20", "diff", 20)
    col("T5YIE", "bei5_chg20", "diff", 20)
    col("DFF", "dff_chg20", "diff", 20)
    # ---- policy differential (each leg joined independently, then differenced)
    col("DFF", "_dff", "level")
    col("ECBDFR", "_ecb", "level")
    out["polidiff"] = out["_dff"] - out["_ecb"]
    out["polidiff_chg20"] = out["polidiff"] - out["polidiff"].shift(20)
    out.drop(columns=["_dff", "_ecb"], inplace=True)
    # ---- monthly country statistics: the power demonstration
    col("UNRATE", "unrate", "level")
    col("UNRATE", "unrate_chg12", "diff", 12)
    col("CPIAUCSL", "cpi_yoy", "yoy", 12)
    return out


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--sym", nargs="*", default=["SP500", "EURUSD", "USDJPY", "XAUUSD"])
    ap.add_argument("--perms", type=int, default=499)
    ap.add_argument("--seed", type=int, default=17)
    args = ap.parse_args()
    rng = np.random.default_rng(args.seed)

    for sym in args.sym:
        bars = load_bars(sym)
        tr = true_range(bars)
        atr = tr.ewm(alpha=1 / 14, min_periods=14).mean()
        trailing = tr.rolling(5).sum() / atr
        y_raw = tr.shift(-1).rolling(5).sum().shift(-4) / atr

        feats = build(bars)
        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())
        print(f"\n=== {sym}: {int(valid.sum())} D1 bars, forward-5d range, {args.perms} perms ===")

        results, null_max = [], np.zeros(args.perms)
        for c in feats.columns:
            x = feats.loc[valid, c].to_numpy(dtype=float)
            ok = ~np.isnan(x)
            if ok.sum() < 500:
                continue
            xo, yo, zo = x[ok], y_bin[ok], z_bin[ok]
            n = len(xo)
            distinct = len(np.unique(xo))
            obs, obs_c = mi_3x3(xo, yo), cmi_3x3x3(xo, yo, zo)
            null = np.empty(args.perms)
            null_c = np.empty(args.perms)
            for k in range(args.perms):
                xs = np.roll(xo, rng.integers(63, n - 63))
                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((c, obs, p, obs_c, p_c, n, distinct))
            if not c.startswith("CTRL_"):
                null_max = np.maximum(null_max, null)
        fam = np.quantile(null_max, 0.95)
        print(f"  family-wise 5% bar: {fam:.5f}")
        print(f"  {'feature':20s} {'n':>6s} {'distinct':>9s}  MI / p          | MI|vol / p")
        for c, obs, p, obs_c, p_c, n, dis in sorted(results, key=lambda r: -r[3]):
            mark = "FAM" if obs > fam else "   "
            inc = "INCR" if p_c <= 0.05 else "    "
            print(f"  {c:20s} {n:6d} {dis:9d}  {obs:.5f} p={p:.4f} {mark} | "
                  f"{obs_c:.5f} p={p_c:.4f} {inc}")


if __name__ == "__main__":
    main()