Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
031b71bf54473ad44050fe61421c062a1abf19ec
PythagorasGoal/scripts/research/r0724_goal50_intraday_mc.py
T
Adriano Dal Pastro 5c48a8322a research(goal50): MC prop INTRADAY wick-aware — leva eval 1.5x refutata, funded ottimo a 0.75x, P(perdere fee) ~56% 
Recon MTM 1h del book Deribit (TP01 pesi daily + SKH01 230m exit-al-livello SL-first,
fee incluse). Validata: leg TP corr 0.9997 / log-total +0.833 vs 0.827; leg SKH +0.548
vs +0.513; Sharpe MTM 1.62 vs 1.77 sleeve (delta = attribuzione a scalino di SKH).
I wick tagliano 6-37pp di P(pass): de-luck 1.0x HYRO 44.7% / FTMO 59.9% / Breakout 26.6%
(declassata). Funded sweep: leva ottima 0.75x (HYRO P(vivo) 58% vs 20.5% a 1x), atteso
~EUR 14.5/g a 100k. EV biglietto resta positivo ma sottile (de-luck +491/+1048/+2197 USD).
Bug tz-naive/aware nel reindex catturato dalla sanity (TP01 spariva, corr 0.43).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-24 21:55:23 +00:00

359 lines
18 KiB
Python
Raw Blame History

"""r0724_goal50_intraday_mc — MC prop-firm con regole sui WICK INTRADAY (2026-07-24).
CHIUDE il caveat n.1 di r0724_goal50_math.py parte C: quel MC applicava daily-loss e
max-drawdown alle CHIUSURE giornaliere del book, ma le regole vere delle firm scattano
sull'equity INTRADAY mark-to-market (il wick conta). Qui:
1) ricostruzione MTM a 1h dell'equity del book Deribit (TP01 75% pesi daily + SKH01 25%
posizioni 230m con exit AL LIVELLO SL/TP nell'ora del trigger, SL prioritario — stessa
convenzione di backtest_signals), wick per-barra inclusi (low/high firmati col segno
del peso), fee incluse (TP01 fee_side*|Dheld|, SKH 0.10% RT a fine trade);
2) per ogni giorno UTC: R_d (chiusura MTM) e m_d (minimo intraday cumulato). Il MC gira
su TUPLE (R_d, m_d) della STESSA ricostruzione -> chiusure e wick coerenti tra loro.
NB: il daily MTM e' la lens GIUSTA per un conto prop (l'equity del conto e' marcata
intraday); lo sleeve SKH ufficiale ha attribuzione a scalino per-trade (P&L intero
sulla barra d'ingresso, caveat noto "equity daily-step") -> la corr daily recon-vs-
sleeve su SKH e' strutturalmente bassa; la sanity giusta su SKH e' il TOTALE.
3) block-bootstrap sulle tuple: stesse regole/leve/de-luck del MC daily-close; breach
valutato su m. Confronto diretto close-only vs intraday = "wick haircut" in pp;
4) fase funded intraday-aware + EV del biglietto per taglia ($25k/$50k/$100k HYRO).
APPROSSIMAZIONI DICHIARATE: (a) wick a risoluzione 1h (i 5m sarebbero piu' profondi ->
le P(pass) restano un TETTO, ma molto piu' oneste del close-only); (b) confini barra
230m non allineati all'ora -> piccoli edge sub-orari; (c) leva applicata linearmente al
giorno (ok per lev <= 2); (d) trigger SL/TP ricercato sulle H/L orarie dentro la barra
230m d'uscita, fallback = cap all'ultima ora della barra.
Uso: `uv run python scripts/research/r0724_goal50_intraday_mc.py` (~1-2 min: SKH 5m full)
"""
from __future__ import annotations
import sys
from pathlib import Path
import numpy as np
import pandas as pd
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
from src.data.downloader import load_data # noqa: E402
from src.portfolio.portfolio import StrategyPortfolio, metrics # noqa: E402
from src.portfolio.sleeves import deribit_book_sleeves # noqa: E402
from src.strategies.skyhook import SKH01_V2_DD, build_frames, skyhook_entries # noqa: E402
from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio, resample_1d # noqa: E402
EURUSD = 1.09
RNG = np.random.default_rng(724)
ASSETS = ("BTC", "ETH")
W_TP, W_SKH = 0.75, 0.25 # pesi del book live Deribit (deribit_book_sleeves)
SKH_FEE_RT = 0.001
# ------------------------------------------------------------------ esposizioni
def tp01_daily_frame(asset: str) -> pd.DataFrame:
"""Per giorno UTC: peso TP01 tenuto (deciso a close[d-1]) e fee del ribilanciamento,
gia' scalati al leg (0.5 del book TP; il peso di book W_TP si applica fuori)."""
tp = TrendPortfolio(**CANONICAL)
df = resample_1d(load_data(asset, "1h"))
tgt = tp.target_series(df)
held = np.zeros(len(tgt))
held[1:] = tgt[:-1]
fee = tp.fee_side * np.abs(np.diff(held, prepend=0.0))
# tz-aware UTC come l'indice 1h del chiamante: un mismatch naive/aware nel reindex
# darebbe NaN->0 silenziosi (TP01 sparirebbe dal recon — visto al primo run, corr 0.43)
days = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True).dt.floor("D"))
return pd.DataFrame({"w": held * 0.5, "fee": fee * 0.5}, index=days)
def skh_hourly_contrib(asset: str, idx1h: pd.DatetimeIndex, close: np.ndarray,
low: np.ndarray, high: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""Contributo del leg SKH01 (peso W_SKH*0.5) a ritorno e wick di ogni barra 1h.
Replica trade-per-trade la logica di sleeves._skyhook_positions/backtest_signals:
entry a close 230m [i], exit alla prima barra 230m che tocca SL (prioritario) o TP
— AL LIVELLO — oppure a close dopo max_bars. Qui il trade e' marcato a mercato ora
per ora; nell'ora del trigger il ritorno e' cappato al livello e la posizione muore.
Fee RT sottratta nell'ora d'uscita."""
ltf, htf = build_frames(load_data(asset, "5m"))
ent = skyhook_entries(ltf, htf, SKH01_V2_DD)
H = ltf["high"].values
L = ltf["low"].values
times = pd.DatetimeIndex(pd.to_datetime(ltf["datetime"], utc=True))
bar_td = times[1] - times[0]
n = len(ltf)
w = W_SKH * 0.5
ret = np.zeros(len(idx1h))
wick = np.zeros(len(idx1h))
i = 0
while i < n:
e = ent[i]
if e is None:
i += 1
continue
d, sl, tp_, mb = e["dir"], e["sl"], e["tp"], e["max_bars"]
exit_idx, exit_mode = None, None
for s in range(1, mb + 1):
j = i + s
if j >= n:
break
hit_sl = (L[j] <= sl) if d == 1 else (H[j] >= sl)
hit_tp = (H[j] >= tp_) if d == 1 else (L[j] <= tp_)
if hit_sl or hit_tp or s == mb:
exit_idx = j
exit_mode = "sl" if hit_sl else ("tp" if hit_tp else "time")
break
if exit_idx is None:
exit_idx, exit_mode = n - 1, "open" # trade ancora aperto a fine dati
t0 = times[i] + bar_td # entry a close della barra i
t1 = times[exit_idx] + bar_td # fine della barra d'uscita
h0 = int(idx1h.searchsorted(t0, side="left"))
h1 = int(idx1h.searchsorted(t1, side="left"))
trig_from = int(idx1h.searchsorted(times[exit_idx], side="left"))
done = False
last_h = None
for h in range(max(h0, 1), h1):
prev_c = close[h - 1]
r_h = close[h] / prev_c - 1.0
wk_h = (low[h] / prev_c - 1.0) if d == 1 else (high[h] / prev_c - 1.0)
if exit_mode in ("sl", "tp") and h >= trig_from:
hit_sl_h = (low[h] <= sl) if d == 1 else (high[h] >= sl)
hit_tp_h = (high[h] >= tp_) if d == 1 else (low[h] <= tp_)
if hit_sl_h: # SL prioritario, come backtest_signals
lvl = sl
wk_h = sl / prev_c - 1.0 # il modello esce al livello: wick cappato
elif hit_tp_h:
lvl = tp_ # wick avverso dell'ora resta (adverse-first)
else:
lvl = None
if lvl is not None:
ret[h] += w * d * (lvl / prev_c - 1.0) - w * SKH_FEE_RT
wick[h] += w * d * wk_h
done = True
break
ret[h] += w * d * r_h
wick[h] += w * d * wk_h
last_h = h
if exit_mode in ("sl", "tp") and not done and last_h is not None:
# trigger non trovato sulle H/L orarie (bordi sub-orari): forza il cap
# all'ultima ora -> P&L coerente col prezzo d'uscita modellato
prev_c = close[last_h - 1]
lvl = sl if exit_mode == "sl" else tp_
ret[last_h] += w * d * (lvl / prev_c - 1.0) - w * d * (close[last_h] / prev_c - 1.0) \
- w * SKH_FEE_RT
elif exit_mode == "time" and last_h is not None:
ret[last_h] -= w * SKH_FEE_RT
if exit_mode == "open":
break
i = exit_idx + 1
return ret, wick
def book_intraday_days() -> tuple[pd.DataFrame, dict]:
"""Per giorno UTC: R_d (chiusura MTM netta del book ricostruito) e m_d (min intraday).
Ritorna anche le serie daily per-leg per la sanity."""
book_r = None
book_wk = None
legs: dict[str, pd.Series] = {}
for a in ASSETS:
df = load_data(a, "1h")
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
close = df["close"].values.astype(float)
low = df["low"].values.astype(float)
high = df["high"].values.astype(float)
prev = np.concatenate(([np.nan], close[:-1]))
r_close = close / prev - 1.0
# --- TP01: peso costante nel giorno, fee alla prima ora del giorno
tpf = tp01_daily_frame(a)
day_of = idx.floor("D")
w_tp = tpf["w"].reindex(day_of).values * W_TP
w_tp = np.nan_to_num(w_tp)
fee_tp = tpf["fee"].reindex(day_of).values * W_TP
fee_tp = np.nan_to_num(fee_tp)
first_of_day = np.concatenate(([True], day_of[1:] != day_of[:-1]))
r_wick_tp = np.where(w_tp >= 0, low / prev - 1.0, high / prev - 1.0)
tp_ret = w_tp * r_close - np.where(first_of_day, fee_tp, 0.0)
tp_wick = w_tp * r_wick_tp - np.where(first_of_day, fee_tp, 0.0)
# --- SKH01: MTM per-trade con exit al livello
skh_ret, skh_wick = skh_hourly_contrib(a, idx, close, low, high)
tot_r = pd.Series(tp_ret + skh_ret, index=idx)
tot_wk = pd.Series(tp_wick + skh_wick, index=idx)
legs[f"tp_{a}"] = pd.Series(tp_ret, index=idx)
legs[f"skh_{a}"] = pd.Series(skh_ret, index=idx)
book_r = tot_r if book_r is None else book_r.add(tot_r, fill_value=np.nan)
book_wk = tot_wk if book_wk is None else book_wk.add(tot_wk, fill_value=np.nan)
book_r = book_r.dropna()
book_wk = book_wk.reindex(book_r.index)
rows = []
for day, g in book_r.groupby(book_r.index.floor("D")):
r = g.values
wk = book_wk.loc[g.index].values
cum = np.cumprod(1 + r)
cum_prev = np.concatenate(([1.0], cum[:-1]))
m = min(cum.min(), (cum_prev * (1 + wk)).min()) - 1.0
rows.append((day, cum[-1] - 1.0, min(m, cum[-1] - 1.0)))
out = pd.DataFrame(rows, columns=["day", "R", "m"]).set_index("day")
return out, legs
# ------------------------------------------------------------------ Monte Carlo
def _paths(R: np.ndarray, gap: np.ndarray, n_days: int, n_paths: int, block: int = 20,
drift_scale: float = 1.0) -> tuple[np.ndarray, np.ndarray]:
"""Bootstrap a blocchi sulle TUPLE (R, gap) — il wick resta accoppiato al suo giorno."""
mu = R.mean()
R_adj = (R - mu) + mu * drift_scale
n = len(R_adj)
n_blocks = int(np.ceil(n_days / block))
starts = RNG.integers(0, n - block, size=(n_paths, n_blocks))
idx = (starts[:, :, None] + np.arange(block)[None, None, :]).reshape(n_paths, -1)[:, :n_days]
return R_adj[idx], gap[idx]
def _sim_eval(R: np.ndarray, m: np.ndarray, target: float, max_dd: float,
daily_loss: float, max_days: int, intraday: bool) -> dict:
"""Eval: pass se la chiusura tocca 1+target prima di un breach. Breach su minimo
intraday (m) se intraday=True, altrimenti su chiusura (R) come nel MC vecchio.
Pareggio stesso giorno -> vince il breach (il minimo precede quasi sempre la chiusura)."""
worst = m if intraday else R
eq = np.cumprod(1 + R, axis=1)
eq_start = eq / (1 + R)
dd_breach = eq_start * (1 + worst) < (1 - max_dd)
dl_breach = worst < -daily_loss
fail = dd_breach | dl_breach
passed = eq >= 1 + target
first_fail = np.where(fail.any(axis=1), fail.argmax(axis=1), max_days + 1)
first_pass = np.where(passed.any(axis=1), passed.argmax(axis=1), max_days + 1)
ok = first_pass < first_fail
return {"p_pass": ok.mean(),
"median_days": float(np.median(first_pass[ok])) if ok.any() else np.nan}
RULES = [
("HYRO 1-step: tgt 10% / DD 6% st / daily 4%", 0.10, 0.06, 0.04),
("BREAKOUT Classic: tgt 10% / DD 6% st / dl 3%", 0.10, 0.06, 0.03),
("FTMO Swing: tgt 10% / DD 10% st / daily 5%", 0.10, 0.10, 0.05),
]
FUNDED_RULES = [("HYRO (max loss 6% st, daily 4%)", 0.06, 0.04),
("FTMO (max loss 10% st, daily 5%)", 0.10, 0.05)]
# biglietti HyroTrader 1-step (ricerca R2): fee rimborsata al primo payout se funded
TICKETS = [(25_000, 249.0), (50_000, 379.0), (100_000, 579.0)]
def main() -> None:
print("costruzione MTM intraday del book (TP01 daily + SKH01 230m exit-al-livello, 1h)...")
intr, legs = book_intraday_days()
intr = intr[intr.index >= "2019-03-01"]
R = intr["R"].to_numpy()
m = intr["m"].to_numpy()
gap = m - R
# ---- sanity vs sleeve ufficiali
port = StrategyPortfolio(deribit_book_sleeves())
book = port.combined_daily().dropna()
book.index = pd.to_datetime(book.index, utc=True).floor("D")
book = book[book.index >= "2019-03-01"]
both = intr.index.intersection(book.index)
corr = float(np.corrcoef(R[intr.index.isin(both)], book.loc[both].to_numpy())[0, 1])
mm_rec = metrics(intr.loc[both, "R"])
mm_book = metrics(book.loc[both])
tp_daily = ((legs["tp_BTC"] + legs["tp_ETH"]).groupby(
(legs["tp_BTC"] + legs["tp_ETH"]).index.floor("D")).apply(lambda g: float(np.prod(1 + g) - 1)))
skh_daily = ((legs["skh_BTC"] + legs["skh_ETH"]).groupby(
(legs["skh_BTC"] + legs["skh_ETH"]).index.floor("D")).apply(lambda g: float(np.prod(1 + g) - 1)))
print("=" * 100)
print(" RICOSTRUZIONE MTM — sanity e geometria dei wick")
print("=" * 100)
print(f" giorni: {len(both)} ({both[0].date()} -> {both[-1].date()})")
print(f" corr daily recon vs book certificato: {corr:.3f} "
"(SKH sleeve = equity a scalino per-trade -> <1 atteso)")
print(f" Sharpe/maxDD recon MTM {mm_rec['sharpe']:.2f}/{mm_rec['maxdd']:.1%} "
f"vs book certificato {mm_book['sharpe']:.2f}/{mm_book['maxdd']:.1%}")
sl_tot = {}
for s in port.sleeves:
sd = s.daily()
sd.index = pd.to_datetime(sd.index, utc=True).floor("D")
sl_tot[s.name] = float(np.log1p(sd[sd.index >= "2019-03-01"]).sum())
print(f" log-total leg TP: recon {np.log1p(tp_daily).sum():+.3f} vs sleeve x{W_TP} "
f"{sl_tot['TP01_trend_1d'] * W_TP:+.3f} | leg SKH: recon {np.log1p(skh_daily).sum():+.3f} "
f"vs sleeve x{W_SKH} {sl_tot['SKH01_skyhook'] * W_SKH:+.3f}")
q = np.percentile(gap, [50, 10, 1])
print(f" gap wick (min intraday - chiusura): p50 {q[0]*100:.2f}pp p10 {q[1]*100:.2f}pp "
f"p1 {q[2]*100:.2f}pp worst {gap.min()*100:.2f}pp")
print(f" giorni con wick oltre -2pp sotto la chiusura: {(gap < -0.02).mean():.1%}")
if corr < 0.80 or abs(mm_rec["sharpe"] - mm_book["sharpe"]) > 0.5:
print(" !! recon troppo lontana dal book certificato: NON credere ai numeri sotto")
n_paths, max_days = 20_000, 365
print("\n" + "=" * 100)
print(" EVAL MC — regole sui WICK INTRADAY vs sole chiusure (stesse path, stesso seed)")
print("=" * 100)
for drift_scale, lab in ((1.0, "book modellato"), (0.6, "de-luck x0.6 (onesto)")):
Rp, Gp = _paths(R, gap, max_days, n_paths, drift_scale=drift_scale)
for lev in (1.0, 1.5, 2.0):
R_l, m_l = Rp * lev, (Rp + Gp) * lev
print(f"\n --- {lab}, leva x{lev:.1f} ---")
for name, tgt, dd, dl in RULES:
a = _sim_eval(R_l, m_l, tgt, dd, dl, max_days, intraday=False)
b = _sim_eval(R_l, m_l, tgt, dd, dl, max_days, intraday=True)
d = f"{b['median_days']:.0f}g" if np.isfinite(b["median_days"]) else "n/a"
print(f" {name:<44} close {a['p_pass']:>5.1%} -> intraday {b['p_pass']:>5.1%} "
f"(wick {100*(b['p_pass']-a['p_pass']):+.1f}pp) mediana {d}")
print("\n" + "=" * 100)
print(" FASE FUNDED intraday-aware — $100k, split 80%, 1 anno, SWEEP DI LEVA")
print(" (bust=0 payout: conservativo; con DD 6% la domanda giusta e' se girare SOTTO 1x)")
print("=" * 100)
ev_inputs = {}
for drift_scale, lab in ((1.0, "book modellato"), (0.6, "de-luck x0.6")):
Rp, Gp = _paths(R, gap, 365, n_paths, drift_scale=drift_scale)
for fname, dd_lim, dl_lim in FUNDED_RULES:
best = (None, -1.0)
for lev in (0.5, 0.75, 1.0, 1.25):
Rl, ml = Rp * lev, (Rp + Gp) * lev
eq = np.cumprod(1 + Rl, axis=1)
eq_start = eq / (1 + Rl)
blown = (eq_start * (1 + ml) < 1 - dd_lim) | (ml < -dl_lim)
alive = ~blown.any(axis=1)
pnl = (eq[:, -1] - 1) * 100_000 * 0.80
qq = np.percentile(pnl[alive], [10, 50, 90]) if alive.any() else [np.nan] * 3
e_payout = float(np.where(alive, np.maximum(pnl, 0.0), 0.0).mean())
if e_payout > best[1]:
best = (lev, e_payout)
print(f" {lab:<16} {fname:<32} lev {lev:.2f} P(vivo 1a) {alive.mean():>5.1%} "
f"p10/p50/p90 EUR {qq[0]/EURUSD/365:>5.1f} / {qq[1]/EURUSD/365:>5.1f} / "
f"{qq[2]/EURUSD/365:>5.1f} /g E[payout] ${e_payout:,.0f}")
print(f" {'':<16} {fname:<32} -> lev ottima {best[0]:.2f} (E[payout] ${best[1]:,.0f})")
if fname.startswith("HYRO"):
ev_inputs[lab] = best[1]
print("\n" + "=" * 100)
print(" EV DEL BIGLIETTO (HYRO: eval a leva x1.0, funded a leva ottima; regole intraday;")
print(" fee rimborsata al pass)")
print("=" * 100)
for drift_scale, lab in ((1.0, "book modellato"), (0.6, "de-luck x0.6")):
Rp, Gp = _paths(R, gap, max_days, n_paths, drift_scale=drift_scale)
R_l, m_l = Rp * 1.0, (Rp + Gp) * 1.0
res = _sim_eval(R_l, m_l, 0.10, 0.06, 0.04, max_days, intraday=True)
p = res["p_pass"]
e100 = ev_inputs[lab]
print(f"\n {lab}: P(pass) {p:.1%}")
for size, fee in TICKETS:
e_pay = e100 * size / 100_000
ev = p * (e_pay + fee) - fee
cost_funded = fee / p if p > 0 else np.nan
print(f" ${size//1000}k (fee ${fee:.0f}): EV 1a = ${ev:>8,.0f} "
f"costo atteso per arrivare funded ${cost_funded:,.0f} "
f"P(perdere la fee) {1-p:.0%}")
print("\n NB: EV lordo di tasse (33%) e attriti fuori-MC (consistency, recycling bust,")
print(" controparte non regolata). Bust funded = payout 0 (prelievi settimanali reali")
print(" lo migliorano). I wick sono a risoluzione 1h: i 5m sarebbero ~piu' severi.")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink