Pietro GiacobazziandDevin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
19c2e0ac72
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
84 lines
3.8 KiB
Python
84 lines
3.8 KiB
Python
"""Verifiche di correttezza + tabella completa 36 coppie x 2 direzioni.
|
|
|
|
Verifiche:
|
|
V1 la direzione registrata coincide col segno di (A-B) all'incrocio
|
|
V2 cret_1 coincide con ret_1
|
|
V3 ricalcolo indipendente della media di un pattern
|
|
V4 baseline drift ~0
|
|
|
|
Tabella: per ogni (coppia, dir) extra-rendimento 5/10/20g, quota che batte il baseline,
|
|
q-value a 10g, tasso di ritorno alla Mediana (rev10/rev20), giorni mediani al ritorno,
|
|
ed etichetta verbale "dopo 10g".
|
|
|
|
Uso:
|
|
python full_table.py [SYMBOL] [YEAR_MIN]
|
|
Output:
|
|
../results/tabella_incroci_<SYM>_<YEAR>.csv (+ .md)
|
|
"""
|
|
import os, sys, numpy as np, pandas as pd
|
|
from _common import load_pair, add_regime, regime_edges, block_bootstrap_p, benjamini_hochberg, HZ, HERE
|
|
|
|
SYM = sys.argv[1] if len(sys.argv) > 1 else "EURUSD"
|
|
YEAR = int(sys.argv[2]) if len(sys.argv) > 2 else 1999
|
|
RESULTS = os.path.normpath(os.path.join(HERE, "..", "results"))
|
|
|
|
cr, ba = load_pair(SYM, YEAR)
|
|
|
|
# ---------------- VERIFICHE ----------------
|
|
sub = cr[cr.pair == "PRICExMED"]
|
|
chk = ((sub.price - sub.med) > 0).astype(int).replace(0, -1)
|
|
print(f"[V1] PRICExMED dir vs segno(price-med): {(np.sign(sub.dir)==np.sign(chk)).mean()*100:.1f}% (atteso ~100%)")
|
|
print(f"[V2] max|cret_1 - ret_1| = {(cr.cret_1-cr.ret_1).abs().max():.6f} (atteso 0)")
|
|
g0 = cr[(cr.pair=='MA121xMA7') & (cr.dir==1)]
|
|
print(f"[V3] MA121xMA7+ n={len(g0)} media cret_10 = {g0.cret_10.mean():.4f}%")
|
|
print(f"[V4] baseline cret_10 medio={ba.cret_10.mean():.4f}% sd={ba.cret_10.std():.3f}% (drift ~0)")
|
|
|
|
# ---------------- TABELLA ----------------
|
|
edges = regime_edges(ba)
|
|
cr = add_regime(cr, edges); ba = add_regime(ba, edges)
|
|
bmean = {h: ba.groupby("regime")[f"cret_{h}"].mean() for h in HZ}
|
|
bglob = {h: ba[f"cret_{h}"].mean() for h in HZ}
|
|
bsd = {h: ba[f"cret_{h}"].std() for h in HZ}
|
|
|
|
rows = []
|
|
for (pair, d), g in cr.groupby(["pair", "dir"]):
|
|
rec = {"pair": pair, "dir": int(d), "n": len(g)}
|
|
for h in HZ:
|
|
ev = g[f"cret_{h}"].values
|
|
bexp = g["regime"].map(bmean[h]).fillna(bglob[h]).values
|
|
exc = ev - bexp
|
|
rec[f"raw_{h}"] = np.nanmean(ev)
|
|
rec[f"exc_{h}"] = np.nanmean(exc)
|
|
rec[f"eff_{h}"] = np.nanmean(exc) / bsd[h]
|
|
rec[f"pos_{h}"] = (exc > 0).mean() * 100
|
|
rec[f"p_{h}"] = block_bootstrap_p(exc)
|
|
rec["rev10"] = g["rev_10"].mean() * 100
|
|
rec["rev20"] = g["rev_20"].mean() * 100
|
|
rec["btr_med"] = g["bars_to_revert"].median()
|
|
rows.append(rec)
|
|
res = pd.DataFrame(rows)
|
|
res["q10"] = benjamini_hochberg(res["p_10"].values)
|
|
|
|
def verb(r):
|
|
if r["q10"] < 0.10 and r["exc_10"] > 0: return "SALE (extra +)"
|
|
if r["q10"] < 0.10 and r["exc_10"] < 0: return "SCENDE (extra -)"
|
|
return "neutro"
|
|
res["dopo_10g"] = res.apply(verb, axis=1)
|
|
res = res.sort_values(["q10", "exc_10"]).reset_index(drop=True)
|
|
|
|
os.makedirs(RESULTS, exist_ok=True)
|
|
csvp = os.path.join(RESULTS, f"tabella_incroci_{SYM}_{YEAR}.csv")
|
|
res.to_csv(csvp, index=False)
|
|
|
|
md = [f"# {SYM} (D1, {YEAR}+) — cosa succede dopo ogni incrocio (36 coppie x 2 dir)", "",
|
|
"extra5/10/20 = extra-rendimento % vs baseline stesso regime; pos10 = % che batte il baseline;",
|
|
"q10 = significativita' (BH, <0.10 robusto); rev10/20 = % ritorno alla Mediana; btr = giorni mediani.", "",
|
|
"| coppia | dir | n | extra5 | extra10 | extra20 | pos10 | q10 | rev10 | rev20 | btr | dopo 10g |",
|
|
"|---|---|---|---|---|---|---|---|---|---|---|---|"]
|
|
for _, x in res.iterrows():
|
|
md.append(f"| {x['pair']} | {int(x['dir']):+d} | {int(x['n'])} | {x['exc_5']:+.3f} | "
|
|
f"{x['exc_10']:+.3f} | {x['exc_20']:+.3f} | {x['pos_10']:.0f}% | {x['q10']:.3f} | "
|
|
f"{x['rev10']:.0f}% | {x['rev20']:.0f}% | {x['btr_med']:.0f} | {x['dopo_10g']} |")
|
|
open(os.path.join(RESULTS, f"tabella_incroci_{SYM}_{YEAR}.md"), "w").write("\n".join(md))
|
|
print(f"\nsalvato: {csvp} (+ .md) righe={len(res)}")
|