Warrior_EA/research/altdata/export.py

"""Export EA-facing alt-data feature files.

Writes Common\\Files\\Warrior_EA\\AltData\\{SYM}_D1.csv with the features that
survived BOTH the marginal family-wise bar and the incremental
(conditional-on-trailing-range) test - see DESIGN.md and screen.py results
(commits 4a56d8d, 6d50e63) - PLUS the exploratory EIA petroleum block on every
symbol (user directive 2026-08-16: "the NN might find patterns in it for both
oil and regular symbols"; screened null on WTI's short sample, so it ships as
exploratory input, not certified edge). For the symbols listed here the columns
must stay identical to the EA's own writer (System\\AltDataFetch.mqh
RebuildFeatures/FeatureValue) - change BOTH together, append-only. The EA's
catalog is broader (indices, more FX, energy, crypto); it is authoritative for
production, this exporter only covers the symbols research screens.

Format (semicolon-separated, header row, chronological):
    date;feat1;feat2;...
    2010.07.24;0.1234;-0.0567;...
Every row is as-of its DATE at 00:00: built only from source rows with
published <= that date. The EA joins each D1 bar to the last row <= bar open.
A sidecar {SYM}_D1.meta carries the feature list + build stamp for staleness
checks on the EA side.

Usage:
    python -m altdata.export
"""
import datetime as dt
from pathlib import Path

import numpy as np
import pandas as pd

from .common import DATA_ROOT
from .screen import cot_features, fred_features

EA_DIR = Path(r"C:\Users\admin\AppData\Roaming\MetaQuotes\Terminal\Common\Files\Warrior_EA\AltData")

# symbol -> feature names (order = EA input order, append-only). Mirror of the
# EA's symbol catalog in System\AltDataFetch.mqh BuildCatalog(). Policy 2026-08-16
# (user, twice): every feature the sources can serve is wired - screens are priors,
# not gates. IVOL[sym] names the instrument's own implied-vol FRED series.
EIA_BLOCK = ["eia_stk_idx1y", "eia_stk_chg4", "eia_util"]
RISK_BLOCK = ["vix_chg5", "vix", "usd_chg5"]
COT_BLOCK = ["cot_idx_1y", "cot_idx_3y", "cot_chg_4w"]
MAC_BLOCK = ["mac_y10", "mac_curve", "mac_bei", "mac_gap", "mac_cpi", "mac_unemp"]
IVOL = {"XAUUSD": "GVZCLS", "XTIUSD": "OVXCLS"}
SURVIVORS = {
    "SP500":  RISK_BLOCK + ["cot_spec_net"] + EIA_BLOCK + MAC_BLOCK,
    "USDJPY": COT_BLOCK + RISK_BLOCK + EIA_BLOCK + MAC_BLOCK,
    "XAUUSD": RISK_BLOCK + ["ivol_chg5", "ivol"] + EIA_BLOCK + MAC_BLOCK,
    "EURUSD": COT_BLOCK + RISK_BLOCK + EIA_BLOCK + MAC_BLOCK,
    "XTIUSD": RISK_BLOCK + ["ivol_chg5", "ivol"] + EIA_BLOCK + MAC_BLOCK,
}

# Fixed A-PRIORI scale constants (units-based, never fitted to data - fitting
# them would leak the sample's distribution into every bar). Goal: values
# land roughly in the +/-1 band the EA's other feature blocks occupy, so the
# first BN layer sees nothing exotic. MI is invariant to these transforms.
TRANSFORMS = {
    "vix":         lambda v: v / 100.0,          # 10..80 -> 0.1..0.8
    "vix_chg5":    lambda v: v / 10.0,           # +/-30 spikes -> +/-3
    "usd_chg5":    lambda v: v,                  # broad-index 5d change, ~+/-3
    "cot_spec_net": lambda v: v,                 # net/OI, already +/-0.5
    "cot_comm_net": lambda v: v,
    "cot_idx_1y":  lambda v: v - 0.5,            # percentile 0..1 -> +/-0.5
    "cot_idx_3y":  lambda v: v - 0.5,
    "cot_chg_4w":  lambda v: v,                  # net/OI 4w delta, ~+/-0.2
    "eia_stk_idx1y": lambda v: v - 0.5,          # percentile 0..1 -> +/-0.5
    "eia_stk_chg4": lambda v: v * 10.0,          # 4w fractional change ~+/-0.03 -> +/-0.3
    "eia_util":    lambda v: (v - 90.0) / 10.0,  # utilization % ~80..98 -> +/-1
    "ivol":        lambda v: v / 100.0,          # instrument IV level, same scale as vix
    "ivol_chg5":   lambda v: v / 10.0,
    "mac_y10":     lambda v: v,                  # 20-obs yield change, ~+/-0.8
    "mac_curve":   lambda v: v,                  # slope in pct-points, ~-1..3
    "mac_bei":     lambda v: v,                  # 20-obs breakeven change, ~+/-0.5
    "mac_gap":     lambda v: v / 10.0,           # Fed-ECB differential, -2..5 -> -0.2..0.5
    "mac_cpi":     lambda v: v * 10.0,           # yoy fraction 0..0.09 -> 0..0.9
    "mac_unemp":   lambda v: v / 10.0,           # 12m change in pp; COVID +10 -> +1
}


def ivol_features(sym: str):
    sid = IVOL.get(sym)
    if not sid:
        return []
    f = pd.read_csv(DATA_ROOT / "fred" / f"{sid}.csv",
                    parse_dates=["observed", "published"]).sort_values("observed")
    v = f["value"].reset_index(drop=True)
    return [("ivol", pd.DataFrame({"published": f["published"].reset_index(drop=True),
                                   "ivol": v, "ivol_chg5": v - v.shift(5)}))]


def mac_features():
    """US macro block; identical observation-index arithmetic to the EA's writer."""
    def fred(sid):
        f = pd.read_csv(DATA_ROOT / "fred" / f"{sid}.csv",
                        parse_dates=["observed", "published"]).sort_values("observed")
        return f.reset_index(drop=True)

    out = []
    y10 = fred("DGS10")
    out.append(("mac_y10", pd.DataFrame({"published": y10["published"],
                                         "mac_y10": y10["value"] - y10["value"].shift(20)})))
    cur = fred("T10Y2Y")
    out.append(("mac_curve", pd.DataFrame({"published": cur["published"],
                                           "mac_curve": cur["value"]})))
    bei = fred("T5YIE")
    out.append(("mac_bei", pd.DataFrame({"published": bei["published"],
                                         "mac_bei": bei["value"] - bei["value"].shift(20)})))
    # policy gap: each leg joined on its own publication date, differenced on the day grid
    dff, ecb = fred("DFF"), fred("ECBDFR")
    out.append(("_dff", pd.DataFrame({"published": dff["published"], "_dff": dff["value"]})))
    out.append(("_ecb", pd.DataFrame({"published": ecb["published"], "_ecb": ecb["value"]})))
    cpi = fred("CPIAUCNS")
    out.append(("mac_cpi", pd.DataFrame({"published": cpi["published"],
                                         "mac_cpi": cpi["value"] / cpi["value"].shift(12) - 1.0})))
    un = fred("UNRATE")
    out.append(("mac_unemp", pd.DataFrame({"published": un["published"],
                                           "mac_unemp": un["value"] - un["value"].shift(12)})))
    return out


def eia_features():
    """Weekly WPSR feature frames, same publication stamps as eia.py wrote them.
    Untransformed here (TRANSFORMS applies the fixed constants), matching how
    cot/fred features arrive. Missing collector output -> skip with a warning
    so the price-complex features still export."""
    eia_dir = DATA_ROOT / "eia"
    out = []
    try:
        stk = pd.read_csv(eia_dir / "crude_stocks_ex_spr.csv",
                          parse_dates=["observed", "published"]).sort_values("observed")
        stk = stk.reset_index(drop=True)
        f = pd.DataFrame({"published": stk["published"]})
        # pandas average-rank pct of the LAST window element = EA RollingPctRank
        f["eia_stk_idx1y"] = stk["value"].rolling(52, min_periods=26).apply(
            lambda w: w.rank(pct=True).iloc[-1], raw=False)
        f["eia_stk_chg4"] = stk["value"] / stk["value"].shift(4) - 1.0
        out.append(("eia_stk", f))
        util = pd.read_csv(eia_dir / "refinery_utilization_pct.csv",
                           parse_dates=["observed", "published"]).sort_values("observed")
        out.append(("eia_util", pd.DataFrame({"published": util["published"],
                                              "eia_util": util["value"]})))
    except FileNotFoundError as exc:
        print(f"WARNING: EIA collector output missing ({exc}) - EIA columns will be empty")
    return out


def daily_panel(sym: str) -> pd.DataFrame:
    """One row per calendar day 2010->today; each column as-of that day."""
    days = pd.date_range("2010-01-01", dt.date.today(), freq="D").astype("datetime64[ns]")
    panel = pd.DataFrame(index=days)
    sources = []
    cot = cot_features(sym)
    if cot is not None:
        sources.append(cot)
    sources.extend(f for _, f in fred_features())
    sources.extend(f for _, f in eia_features())
    sources.extend(f for _, f in ivol_features(sym))
    sources.extend(f for _, f in mac_features())
    for src in sources:
        src = src.sort_values("published").dropna()
        src["published"] = src["published"].astype("datetime64[ns]")
        cols = [c for c in src.columns if c != "published"]
        joined = pd.merge_asof(pd.DataFrame({"day": days}), src,
                               left_on="day", right_on="published",
                               direction="backward")
        for c in cols:
            if c not in panel.columns:
                panel[c] = joined[c].to_numpy()
    if "_dff" in panel.columns and "_ecb" in panel.columns:
        panel["mac_gap"] = panel["_dff"] - panel["_ecb"]
        panel.drop(columns=["_dff", "_ecb"], inplace=True)
    return panel


def main() -> None:
    EA_DIR.mkdir(parents=True, exist_ok=True)
    stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    for sym, feats in SURVIVORS.items():
        panel = daily_panel(sym)
        missing = [f for f in feats if f not in panel.columns]
        if missing:
            # Keep the column (empty) rather than narrowing the CSV: the header must
            # stay identical to the EA writer's or the name pin flags a mismatch.
            print(f"WARNING: {sym}: no source data for {missing} - exporting empty columns")
            for f in missing:
                panel[f] = np.nan
        out = panel[feats].dropna(how="all")
        for f in feats:
            out[f] = TRANSFORMS[f](out[f])
        lines = ["date;" + ";".join(feats)]
        for day, row in out.iterrows():
            vals = ";".join("" if np.isnan(v) else f"{v:.6f}" for v in row)
            lines.append(f"{day:%Y.%m.%d};{vals}")
        dest = EA_DIR / f"{sym}_D1.csv"
        dest.write_text("\n".join(lines) + "\n", encoding="ascii")
        meta = EA_DIR / f"{sym}_D1.meta"
        meta.write_text(f"features={len(feats)}\nnames={','.join(feats)}\n"
                        f"built={stamp}\nrows={len(out)}\n", encoding="ascii")
        print(f"{sym}: {len(out)} daily rows x {len(feats)} features -> {dest}")


if __name__ == "__main__":
    main()