Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research(wave-0822): GROWTH-POLICY — il libro gira al 7% di Kelly, e ogni gate del progetto e' cieco alla scala

Browse Source
This commit is contained in:
Adriano Dal Pastro
2026-08-22 17:10:03 +00:00
parent 3b206c8e6d
commit 14c1a75a7e
8 changed files with 275 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/r0822_oi_pin.py
+53 -28
View File
@@ -60,7 +60,8 @@ FEE_SIDE_CONS = A.FEE_SIDE # 5 bps/lato = convenzione conservativa dei bac
MAX_H = 50 # ore prima della scadenza da tenere in memoria
HTE_GRID = (1, 2, 3, 6, 12, 24, 36, 48) # ore-alla-scadenza all'INGRESSO
RULES = ("oi", "gex", "mp") # max OI · max gamma*OI · max pain
PLACEBOS = ("rand", "adj", "jit", "mid", "mr7")
PLACEBOS_GLOB = ("rand", "mid", "mr7") # non dipendono dalla regola
PLACEBOS_RULE = ("adj", "jit") # costruiti sulla distanza DELLA regola
SEED = 20260822
RNG = np.random.default_rng(SEED)
@@ -144,29 +145,39 @@ def levels_table(chain: pd.DataFrame, spot: dict[str, pd.Series]) -> pd.DataFram
"gex": float(K[int(np.argmax(gx))]) if gx.max() > 0 else np.nan,
"mp": _max_pain(K, oi_c, oi_p),
}
# ---- placebo location-matched -------------------------------------------------
# rand: permutare l'OI fra gli strike e riprendere l'argmax equivale ESATTAMENTE a
# estrarre uno strike uniforme dalla griglia (l'argmax di un vettore permutato
# e' uniforme sulle posizioni) -> lo si implementa cosi', ed e' esatto.
p_rand = float(K[RNG.integers(len(K))])
# adj: lo strike ADIACENTE al max-OI (stesso quartiere, OI diverso)
j = i_oi + (1 if (i_oi == 0 or (i_oi < len(K) - 1 and RNG.random() < 0.5)) else -1)
p_adj = float(K[int(np.clip(j, 0, len(K) - 1))])
# jit: strike arbitrario alla STESSA distanza dallo spot ("0.618 vs 0.58?")
dist = abs(K - S)
d0 = abs(lv["oi"] - S)
cand = np.where((dist >= 0.5 * d0) & (dist <= 1.5 * d0) & (K != lv["oi"]))[0]
p_jit = float(K[cand[RNG.integers(len(cand))]]) if len(cand) else np.nan
# mid: punto medio della griglia quotata = puro artefatto di copertura, zero info OI
p_mid = float(0.5 * (K[0] + K[-1]))
# ---- placebo LOCATION-MATCHED, uno per OGNI regola --------------------------------
# ⚠ difetto trovato e corretto in sessione: la prima stesura costruiva `adj`/`jit`
# attorno al SOLO max-OI e poi li usava come null anche per max-pain e gamma. Ma
# |K_oi - S| mediano e' ~2,4% e |K_mp - S| ~1,1%: confrontare mp con un placebo
# tarato sulla distanza di oi NON e' location-matched, e' proprio l'errore che il
# null doveva prevenire. Ogni regola ha ora i placebo alla PROPRIA distanza.
dist = np.abs(K - S)
for nm, kv in list(lv.items()):
if not np.isfinite(kv):
out_adj = out_jit = np.nan
else:
i_r = int(np.argmin(np.abs(K - kv)))
j = i_r + (1 if (i_r == 0 or (i_r < len(K) - 1 and RNG.random() < 0.5)) else -1)
out_adj = float(K[int(np.clip(j, 0, len(K) - 1))])
d0 = abs(kv - S)
cand = np.where((dist >= 0.5 * d0) & (dist <= 1.5 * d0) & (K != kv))[0]
out_jit = float(K[cand[RNG.integers(len(cand))]]) if len(cand) else np.nan
lv[f"adj_{nm}"] = out_adj
lv[f"jit_{nm}"] = out_jit
# placebo GLOBALI (non dipendono dalla regola)
# rand: permutare l'OI fra gli strike e riprenderne l'argmax equivale ESATTAMENTE a
# estrarre uno strike uniforme dalla griglia (l'argmax di un vettore permutato e'
# uniforme sulle posizioni) -> lo si implementa cosi', ed e' esatto, non simulato.
lv["rand"] = float(K[RNG.integers(len(K))])
# mid: centro della griglia quotata = puro artefatto di copertura, zero informazione OI
lv["mid"] = float(0.5 * (K[0] + K[-1]))
rows.append(dict(
asset=asset, exp=exp, hh=hh, dte_h=float((exp - hh).total_seconds() / 3600.0),
S=float(S), n_k=len(K), tot_oi=float(tot),
k_oi=lv["oi"], k_gex=lv["gex"], k_mp=lv["mp"],
p_rand=p_rand, p_adj=p_adj, p_jit=p_jit, p_mid=p_mid,
top_share=float(oi[i_oi] / tot),
hhi=float(((oi / tot) ** 2).sum()),
k_step=float(np.median(np.diff(K))) if len(K) > 1 else np.nan,
**{f"L_{k}": v for k, v in lv.items()},
))
return pd.DataFrame(rows)
@@ -198,9 +209,8 @@ def build_events(lev: pd.DataFrame, spot: dict[str, pd.Series]) -> pd.DataFrame:
top_share=float(r["top_share"]), hhi=float(r["hhi"]),
k_step=float(r["k_step"]),
dow=int(pd.Timestamp(exp).dayofweek))
lvls = {"oi": r["k_oi"], "gex": r["k_gex"], "mp": r["k_mp"],
"rand": r["p_rand"], "adj": r["p_adj"], "jit": r["p_jit"],
"mid": r["p_mid"], "mr7": float(mr7[a].asof(hh))}
lvls = {c[2:]: r[c] for c in r.index if str(c).startswith("L_")}
lvls["mr7"] = float(mr7[a].asof(hh))
for nm, K in lvls.items():
K = float(K) if K is not None else np.nan
if not np.isfinite(K) or K <= 0:
@@ -421,12 +431,12 @@ def main() -> None:
P("\n LA STESSA STATISTICA SUI LIVELLI PLACEBO — se sale anche li', NON e' pinning:")
P(f" {'hte':>4} " + " ".join(f"{('tw_' + n):>10}" for n in
("oi", "mp", "rand", "jit", "mid", "mr7", "maj", "short")))
("oi", "mp", "rand", "jit_mp", "mid", "mr7", "maj", "short")))
for hte in HTE_GRID:
e = ev[ev["hte"] == hte]
P(f" {hte:>4} " + " ".join(
f"{np.nanmean(e[f'tw_{n}']):>10.5f}" for n in
("oi", "mp", "rand", "jit", "mid", "mr7", "maj", "short")))
("oi", "mp", "rand", "jit_mp", "mid", "mr7", "maj", "short")))
P("\n QUANTO SONO LO STESSO SEGNALE? accordo del SEGNO di (K-S) fra max-pain e i placebo:")
for n in ("mid", "mr7", "oi", "maj"):
agr = [float(np.nanmean(np.sign(ev.loc[ev.hte == h, "d_mp"])
@@ -446,22 +456,25 @@ def main() -> None:
P(" placebo: rand=strike uniforme (== permutazione dell'OI, equivalenza esatta) · adj=strike")
P(" adiacente al max-OI · jit=strike arbitrario a distanza 0.5-1.5x (analogo Fib±jitter) ·")
P(" mid=centro della griglia quotata (zero info OI) · mr7=media 7g dello spot (mean-reversion).")
P(f" {'hte':>4} {'rule':>5} " + " ".join(f"{('Δ vs ' + p):>12}" for p in PLACEBOS))
P(f" {'hte':>4} {'rule':>5} {'Δ vs adj_r':>12} {'Δ vs jit_r':>12}"
f" {'Δ vs rand':>12} {'Δ vs mid':>12} {'Δ vs mr7':>12}")
npos = ntot = 0
for hte in HTE_GRID:
e = ev[ev["hte"] == hte]
for rl in RULES:
cells = []
for p in PLACEBOS:
for p in (f"adj_{rl}", f"jit_{rl}") + PLACEBOS_GLOB:
dd = e[f"tw_{rl}"].to_numpy(float) - e[f"tw_{p}"].to_numpy(float)
m, lo, hi, _ = cluster_boot(dd, e["exp"].to_numpy())
mark = "*" if (np.isfinite(lo) and lo > 0) else (" ")
cells.append(f"{m:>11.5f}{mark}")
cells.append(f"{m:>11.5f}{'*' if (np.isfinite(lo) and lo > 0) else ' '}")
ntot += 1
npos += int(np.isfinite(lo) and lo > 0)
P(f" {hte:>4} {rl:>5} " + " ".join(cells))
P(f" celle (rule x hte x placebo) in cui lo SPECIALE batte il placebo con IC95>0:"
f" {npos}/{ntot} (atteso per caso ~{0.025 * ntot:.0f})")
P(" LETTURA: `adj_r`/`jit_r` chiedono 'e' proprio QUELLO strike?'; `mid`/`mr7` chiedono")
P(" 'serve l'open interest, o basta il centro recente del prezzo?'. La seconda e' la")
P(" domanda che uccide o salva il filone.")
# ---------------------------------------------------------------- 5. CONDIZIONAMENTO
P("\n[5] CONDIZIONAMENTO — la concentrazione di OI e le scadenze grandi cambiano qualcosa?")
@@ -581,7 +594,7 @@ def main() -> None:
# DSR: conta TUTTI i trial al rialzo
all_sr = [float(v) for v in C["sharpe"].to_numpy(float) if np.isfinite(v)]
all_sr_wide = all_sr + [float(v) for v in N["sharpe"].to_numpy(float) if np.isfinite(v)]
n_fact = len(HTE_GRID) * len(RULES) * (1 + len(PLACEBOS)) + 3 * 4 # celle del fatto + condiz.
n_fact = len(HTE_GRID) * len(RULES) * (1 + len(PLACEBOS_GLOB) + len(PLACEBOS_RULE)) + 3 * 4 # celle del fatto + condiz.
trials_tot = len(all_sr_wide) + n_fact
dsr, sr0 = A.deflated_sharpe(mb["sharpe"], all_sr, dbest, dpy=365.25)
dsr_w, _ = A.deflated_sharpe(mb["sharpe"], all_sr_wide, dbest, dpy=365.25)
@@ -632,6 +645,18 @@ def main() -> None:
f" · MEDIANA ONESTA {ab.get('median'):.2f} · banda [{ab.get('lo'):.2f},{ab.get('hi'):.2f}]"
f" · positive {ab.get('frac_positive'):.0%} · gate_pass={ab.get('gate_pass')}")
# LA DOMANDA DECISIVA, sulla banda d'ancora: max-pain aggiunge qualcosa a un livello
# che NON usa opzioni? Mediana delle DIFFERENZE APPAIATE sui 8 hte (mai differenza
# delle mediane: lezione 26/07).
def _fn(rule):
return lambda h: strat_daily(ev, rule, int(h), (0.002, 0.03), "sign", "all",
FEE_SIDE_REAL)[0]
for other in ("mr7", "mid", "oi", "rand"):
dl = A.anchor_luck_delta(_fn("mp"), _fn(other), offsets=list(HTE_GRID))
P(f" Δ appaiato mp-{other:<4}: mediana {dl.get('median_paired'):+.2f}"
f" · positiva in {dl.get('n_positive')}/{dl.get('n_anchors')} hte"
f" · banda [{dl.get('lo'):+.2f},{dl.get('hi'):+.2f}] · gate_pass={dl.get('gate_pass')}")
# eseguibilita' a $600
P(" eseguibilita' $600 :", end=" ")
hc = []