"""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()