|
|
|
@@ -0,0 +1,116 @@
|
|
|
|
|
"""Validazione esterna: addestra SOLO su EURUSD, testa su altri simboli mai visti.
|
|
|
|
|
|
|
|
|
|
Due livelli:
|
|
|
|
|
A) Pattern: gli incroci robusti di EURUSD (extra-rendimento a 10g) hanno lo
|
|
|
|
|
stesso segno/forza sugli altri simboli? (replica del pattern)
|
|
|
|
|
B) Modello: il classificatore addestrato su EURUSD generalizza? AUC e
|
|
|
|
|
precisione sul decile piu' forte su ogni simbolo esterno.
|
|
|
|
|
|
|
|
|
|
Nessun dato dei simboli di test entra nell'addestramento (zero inquinamento).
|
|
|
|
|
La baseline di ciascun simbolo (definizione del target) e' calcolata sui dati
|
|
|
|
|
di quel simbolo: serve solo a definire la ground-truth, non i pesi del modello.
|
|
|
|
|
|
|
|
|
|
Uso:
|
|
|
|
|
python external_validation.py [config.yaml] SYM_TRAIN SYM_TEST1 SYM_TEST2 ...
|
|
|
|
|
(default: train=EURUSD, test=USDJPY USDCHF GBPUSD)
|
|
|
|
|
"""
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
import os, sys
|
|
|
|
|
import numpy as np, pandas as pd
|
|
|
|
|
|
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
sys.path.insert(0, HERE)
|
|
|
|
|
import train as T
|
|
|
|
|
from data import load
|
|
|
|
|
from labeling import RegimeBaseline
|
|
|
|
|
from features import select_xy
|
|
|
|
|
from model import make_model
|
|
|
|
|
from evaluate import metrics
|
|
|
|
|
|
|
|
|
|
RESULTS = os.path.normpath(os.path.join(HERE, "..", "results"))
|
|
|
|
|
ROBUST = ["MA365xMA7", "MA365xMA182", "MA365xMA121", "MA121xMA7",
|
|
|
|
|
"MA121xMA3", "PRICExMA121", "MA182xMA30", "MA182xMA121"]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def load_sym(sym, cfg):
|
|
|
|
|
cr, ba = load(sym, cfg["year_min"])
|
|
|
|
|
cr = cr.dropna(subset=[f"cret_{cfg['horizon']}"]).reset_index(drop=True)
|
|
|
|
|
return cr, ba
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def main():
|
|
|
|
|
cfg_path = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "..", "config.yaml")
|
|
|
|
|
cfg = T.load_cfg(cfg_path)
|
|
|
|
|
args = sys.argv[2:]
|
|
|
|
|
sym_tr = args[0] if args else "EURUSD"
|
|
|
|
|
sym_te = args[1:] if len(args) > 1 else ["USDJPY", "USDCHF", "GBPUSD"]
|
|
|
|
|
num, cat, h = cfg["features_numeric"], cfg["features_categorical"], cfg["horizon"]
|
|
|
|
|
|
|
|
|
|
cr_tr, ba_tr = load_sym(sym_tr, cfg)
|
|
|
|
|
|
|
|
|
|
# ---------- A) replica dei pattern ----------
|
|
|
|
|
def pair_excess(cr, ba):
|
|
|
|
|
rb = RegimeBaseline(cfg["regime"], h).fit(ba)
|
|
|
|
|
ex = rb.excess(cr)
|
|
|
|
|
out = {}
|
|
|
|
|
for (p, d), idx in cr.groupby(["pair", "dir"]).groups.items():
|
|
|
|
|
out[(p, int(d))] = ex[cr.index.get_indexer(idx)].mean()
|
|
|
|
|
return out
|
|
|
|
|
|
|
|
|
|
base_ex = pair_excess(cr_tr, ba_tr)
|
|
|
|
|
print(f"=== A) Replica pattern robusti: extra-rendimento {h}g (%), segno vs {sym_tr} ===")
|
|
|
|
|
print(f"{'pattern':16s} {sym_tr:>9s}", end="")
|
|
|
|
|
sym_ex = {}
|
|
|
|
|
for s in sym_te:
|
|
|
|
|
c, b = load_sym(s, cfg)
|
|
|
|
|
sym_ex[s] = pair_excess(c, b)
|
|
|
|
|
print(f" {s:>9s}", end="")
|
|
|
|
|
print(" concordi")
|
|
|
|
|
agree_counts = []
|
|
|
|
|
for p in ROBUST:
|
|
|
|
|
for d in (1, -1):
|
|
|
|
|
key = (p, d)
|
|
|
|
|
if key not in base_ex:
|
|
|
|
|
continue
|
|
|
|
|
row = f"{p+('+' if d>0 else '-'):16s} {base_ex[key]*1:+8.3f}"
|
|
|
|
|
signs = []
|
|
|
|
|
for s in sym_te:
|
|
|
|
|
v = sym_ex[s].get(key, np.nan)
|
|
|
|
|
row += f" {v:+8.3f}" if not np.isnan(v) else f" {'n/a':>8s}"
|
|
|
|
|
if not np.isnan(v):
|
|
|
|
|
signs.append(np.sign(v) == np.sign(base_ex[key]))
|
|
|
|
|
conc = f"{sum(signs)}/{len(signs)}" if signs else "-"
|
|
|
|
|
agree_counts.append((sum(signs), len(signs)))
|
|
|
|
|
print(row + f" {conc}")
|
|
|
|
|
tot_a = sum(a for a, _ in agree_counts); tot_n = sum(n for _, n in agree_counts)
|
|
|
|
|
print(f"\nConcordanza di segno totale: {tot_a}/{tot_n} ({tot_a/max(tot_n,1)*100:.0f}%)")
|
|
|
|
|
|
|
|
|
|
# ---------- B) modello EURUSD -> simboli esterni ----------
|
|
|
|
|
print(f"\n=== B) Modello addestrato su {sym_tr}, testato esternamente ===")
|
|
|
|
|
rb_tr = RegimeBaseline(cfg["regime"], h).fit(ba_tr)
|
|
|
|
|
y_tr = rb_tr.label(cr_tr)
|
|
|
|
|
mdl = make_model(cfg, num, cat)
|
|
|
|
|
mdl.fit(select_xy(cr_tr, num, cat), y_tr)
|
|
|
|
|
|
|
|
|
|
rows = []
|
|
|
|
|
# riferimento in-sample (stesso simbolo, solo per confronto)
|
|
|
|
|
p_in = mdl.predict_proba(select_xy(cr_tr, num, cat))[:, 1]
|
|
|
|
|
rows.append({"symbol": sym_tr + " (in-sample)", **metrics(y_tr, p_in)})
|
|
|
|
|
for s in sym_te:
|
|
|
|
|
c, b = load_sym(s, cfg)
|
|
|
|
|
rb = RegimeBaseline(cfg["regime"], h).fit(b)
|
|
|
|
|
y = rb.label(c)
|
|
|
|
|
p = mdl.predict_proba(select_xy(c, num, cat))[:, 1]
|
|
|
|
|
rows.append({"symbol": s, **metrics(y, p)})
|
|
|
|
|
res = pd.DataFrame(rows)
|
|
|
|
|
for _, r in res.iterrows():
|
|
|
|
|
print(f" {r['symbol']:22s} n={int(r['n']):5d} AUC={r['auc']:.3f} "
|
|
|
|
|
f"acc={r['acc']:.3f} prec@10%={r['prec_top_decile']:.3f} "
|
|
|
|
|
f"(base {r['base_rate']:.3f}, lift {r['lift_top_decile']:.2f}x)")
|
|
|
|
|
os.makedirs(RESULTS, exist_ok=True)
|
|
|
|
|
res.to_csv(os.path.join(RESULTS, "external_validation.csv"), index=False)
|
|
|
|
|
print(f"\nsalvato: {os.path.join(RESULTS,'external_validation.csv')}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
|
main()
|
Pietro Giacobazzi