research: ricerca dispersion/correlation index (165 agenti) — 2 edge su 60, nessun nuovo motore
Workflow 60 celle (15 famiglie x 4 finestre), verifica avversariale 2-skeptic. Esito: 2 edge confermati causali (index_comp_disp W168, rel_idio_fade W24), ma entrambi fade-BTC del residuo idiosincratico -> sovrapposti alle MR esistenti (P migliora-PORT06 ~20%, lezione FR01). 13/15 famiglie rumore (overfit/regime, non look-ahead). Edge preservati in scripts/analysis/dispersion_edges/; harness riusabile dispersion_lab.py (gia' committato). Diario + TODO follow-up. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,152 @@
|
||||
"""rel_idio_fade (W=24): fade della componente idiosincratica rel_A vs indice.
|
||||
|
||||
Idea: rel_A = ret(A) - ret(indice EW) e' il rendimento idiosincratico (residuo di
|
||||
mercato). Quando l'asset diverge troppo dall'indice (z-score di rel_A su finestra
|
||||
W=24 elevato), si fada il residuo verso l'indice: se A ha sovraperformato troppo
|
||||
(z alto) -> SHORT A; se ha sottoperformato (z basso) -> LONG A. Mean-reversion del
|
||||
residuo.
|
||||
|
||||
ENTRY CAUSALE: la decisione a close[i] usa SOLO rel_A fino a i incluso. Lo z-score
|
||||
e' costruito con media/std rolling su [i-W+1 .. i] (causale). Ingresso eseguibile a
|
||||
close[i]; exit a tempo (max_bars), opzionale TP/SL ad ATR.
|
||||
|
||||
Esegui: cd /opt/docker/PythagorasGoal && PYTHONPATH=. uv run python \
|
||||
scripts/analysis/_disp_scratch/rel_idio_fade_24.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from scripts.analysis.dispersion_lab import features, align_to, UNIVERSE # noqa: E402
|
||||
from scripts.analysis.explore_lab import get_df, evaluate, robust, atr # noqa: E402
|
||||
|
||||
W = 24 # finestra correlazione/dispersione (richiesta dalla famiglia)
|
||||
ASSETS = ["BTC", "ETH", "ADA", "BNB", "DOGE", "LTC", "SOL", "XRP"]
|
||||
Z_GRID = [1.5, 2.0, 2.5, 3.0]
|
||||
MB_GRID = [12, 24, 48]
|
||||
TP_ATR = None # exit a tempo puro per il primo screening
|
||||
SL_ATR = None
|
||||
|
||||
|
||||
def build_entries(asset: str, df: pd.DataFrame, fa: pd.DataFrame,
|
||||
z_thr: float, max_bars: int,
|
||||
tp_atr=None, sl_atr=None) -> list[dict]:
|
||||
"""Entries CAUSALI per il fade del residuo idiosincratico.
|
||||
|
||||
z[i] = (rel[i] - mean(rel[i-W+1..i])) / std(rel[i-W+1..i]) -> usa solo dati <= i.
|
||||
rel[i] e' gia' causale (deriva da log-ret fino a close[i]). Quando |z[i]|>=thr:
|
||||
z>0 (A ha sovraperformato l'indice) -> SHORT (d=-1), fade verso l'indice
|
||||
z<0 (A ha sottoperformato) -> LONG (d=+1)
|
||||
"""
|
||||
rel = fa[f"rel_{asset}"].values.astype(float)
|
||||
s = pd.Series(rel)
|
||||
mu = s.rolling(W).mean().values
|
||||
sd = s.rolling(W).std(ddof=0).values
|
||||
z = (rel - mu) / np.where(sd > 0, sd, np.nan)
|
||||
|
||||
a = atr(df, 14)
|
||||
c = df["close"].values
|
||||
n = len(c)
|
||||
entries: list[dict] = []
|
||||
for i in range(W, n - 1):
|
||||
zi = z[i]
|
||||
if not np.isfinite(zi) or abs(zi) < z_thr:
|
||||
continue
|
||||
d = -1 if zi > 0 else 1 # fade del residuo
|
||||
e = {"i": i, "d": d, "max_bars": max_bars}
|
||||
if tp_atr is not None and np.isfinite(a[i]):
|
||||
e["tp"] = c[i] + d * tp_atr * a[i] # TP nella direzione del fade
|
||||
if sl_atr is not None and np.isfinite(a[i]):
|
||||
e["sl"] = c[i] - d * sl_atr * a[i]
|
||||
entries.append(e)
|
||||
return entries
|
||||
|
||||
|
||||
def check_no_lookahead(asset: str, df: pd.DataFrame, fa: pd.DataFrame,
|
||||
z_thr: float, max_bars: int) -> bool:
|
||||
"""Perturba i prezzi DOPO un indice T e verifica che le entries con i<=T non
|
||||
cambino (la entry-rule usa solo dati <= close[i])."""
|
||||
rel = fa[f"rel_{asset}"].values.astype(float)
|
||||
n = len(rel)
|
||||
T = int(n * 0.6)
|
||||
|
||||
def z_of(relv):
|
||||
s = pd.Series(relv)
|
||||
mu = s.rolling(W).mean().values
|
||||
sd = s.rolling(W).std(ddof=0).values
|
||||
return (relv - mu) / np.where(sd > 0, sd, np.nan)
|
||||
|
||||
z0 = z_of(rel)
|
||||
rel2 = rel.copy()
|
||||
rel2[T + 1:] = rel2[T + 1:] + 0.05 # shock del futuro
|
||||
z2 = z_of(rel2)
|
||||
|
||||
def ents_from(z):
|
||||
out = []
|
||||
for i in range(W, n - 1):
|
||||
if i > T:
|
||||
break
|
||||
zi = z[i]
|
||||
if np.isfinite(zi) and abs(zi) >= z_thr:
|
||||
out.append((i, -1 if zi > 0 else 1))
|
||||
return out
|
||||
|
||||
ok = ents_from(z0) == ents_from(z2)
|
||||
print(f" [no-look-ahead {asset}] entries i<=T={T} invarianti al futuro: "
|
||||
f"{'OK' if ok else 'VIOLATO'}")
|
||||
return ok
|
||||
|
||||
|
||||
def main():
|
||||
F = features()
|
||||
print(f"feature caricate: {F.shape[0]} barre, {F.shape[1]} colonne")
|
||||
|
||||
best = None
|
||||
look_ok_all = True
|
||||
for asset in ASSETS:
|
||||
df = get_df(asset, "1h")
|
||||
fa = align_to(F, df)
|
||||
# un check no-look-ahead per asset (config centrale)
|
||||
look_ok_all &= check_no_lookahead(asset, df, fa, z_thr=2.0, max_bars=24)
|
||||
print(f"--- {asset} ---")
|
||||
for z_thr in Z_GRID:
|
||||
for mb in MB_GRID:
|
||||
ents = build_entries(asset, df, fa, z_thr, mb, TP_ATR, SL_ATR)
|
||||
if len(ents) < 30:
|
||||
continue
|
||||
name = f"{asset} z{z_thr} mb{mb}"
|
||||
res = evaluate(name, ents, df)
|
||||
rb = robust(res)
|
||||
score = res["oos"]["ret"] + res["full"]["ret"]
|
||||
cand = {
|
||||
"asset": asset, "z": z_thr, "mb": mb,
|
||||
"full": res["full"]["ret"], "oos": res["oos"]["ret"],
|
||||
"fee02_oos": res["sweep_oos"][0.002],
|
||||
"dd": res["full"]["dd"], "sharpe": res["full"]["sharpe"],
|
||||
"pos_yrs": res["pos_yrs"], "n_yrs": res["n_yrs"],
|
||||
"robust": rb, "score": score, "trades": res["full"]["trades"],
|
||||
}
|
||||
# preferisci robuste; a parita' di robustezza, score piu' alto
|
||||
if best is None or (cand["robust"], cand["score"]) > (best["robust"], best["score"]):
|
||||
best = cand
|
||||
|
||||
print("\n=== CELLA MIGLIORE ===")
|
||||
if best:
|
||||
print(f" asset={best['asset']} z={best['z']} mb={best['mb']} trades={best['trades']}")
|
||||
print(f" FULL={best['full']:+.0f}% OOS={best['oos']:+.0f}% "
|
||||
f"fee0.2%OOS={best['fee02_oos']:+.0f}% DD={best['dd']:.0f}% "
|
||||
f"Sharpe={best['sharpe']:.2f} anniPos={best['pos_yrs']}/{best['n_yrs']} "
|
||||
f"robust={best['robust']}")
|
||||
print(f" no-look-ahead tutti gli asset: {'OK' if look_ok_all else 'VIOLATO'}")
|
||||
return best
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user