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

research(premium): ondata premi cross-venue auto-calcolati — CBPREM morto, KIMCHI d30 ucciso dallo scettico (spike+lag)

Browse Source 
CBPREM (Coinbase vs feed cert, 2015->): HEDGE, hold -0.46. KIMCHI (Upbit/ECB, 2017->,
ancora 00:00 UTC verificata): EARNS_SLOT=True al marginal scorer (ADDS, robust_oos,
uplift + ogni anno) ma DSR 0.891 -> scettico obbligatorio: niente plateau
(d15/30/45/60 = 0.44/1.07/0.51/0.24) e lag +1g azzera (hold 0.74->0.14) = parameter
luck. Lezione codificata: EARNS_SLOT con DSR<0.95 -> sempre plateau+lag skeptic.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-07-24 22:23:16 +00:00
parent 2320fa9cec
commit b65f067f69
2 changed files with 191 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/r0724_kimchi_skeptic.py
+50
View File
@@ -0,0 +1,50 @@
"""r0724_kimchi_skeptic — scettico sul candidato KIMCHI d30-folLF (2026-07-24).
Il candidato usciva da r0724_premium_wave con EARNS_SLOT=True al marginal scorer
(ADDS, robust_oos, persistente ogni anno) ma DSR 0.891<0.95. Due check obbligatori:
(1) PLATEAU: d15/d30/d45/d60 — un edge vero degrada dolcemente, uno spike no;
(2) LAG ESECUZIONE +1g: un segnale di flusso a 30g deve sopravvivere a 24h di ritardo.
ESITO (run 2026-07-24): d30 e' uno SPIKE isolato (0.44/1.07/0.51/0.24) e il lag +1g
lo azzera (hold 0.74->0.14, uplift blend negativo) -> SCARTATO, parameter-luck.
"""
import sys
import pathlib
ROOT = str(pathlib.Path(__file__).resolve().parents[2])
sys.path.insert(0, ROOT)
sys.path.insert(0, ROOT + "/scripts/research/alt")
import numpy as np, pandas as pd
import altlib
from altlib import candidate_daily, tp01_baseline_daily, HOLDOUT
sys.path.insert(0, ROOT + "/scripts/research")
from r0724_premium_wave import _PREM
def _sh(s):
s = s.dropna()
return float(s.mean()/s.std()*np.sqrt(365.25)) if s.std() > 0 else 0.0
def factory(D, extra_lag=0):
def fn(df, asset):
p = _PREM[asset]["kimchi"]
z = (p - p.shift(D)).shift(extra_lag)
pos = (np.sign(z) > 0).astype(float)
days = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)).floor("D")
return np.nan_to_num(pos.reindex(days).values.astype(float))
return fn
B = tp01_baseline_daily()
def blend_uplift(C, w=0.25):
J = pd.concat({"b": B, "c": C}, axis=1, join="inner").fillna(0.0)
bl = (1-w)*J["b"] + w*J["c"]
hb, hbl = J["b"][J.index >= HOLDOUT], bl[bl.index >= HOLDOUT]
return _sh(bl)-_sh(J["b"]), _sh(hbl)-_sh(hb)
print(f"{'cella':<18}{'Sh full':>9}{'Sh hold':>9}{'upl full':>10}{'upl hold':>10}")
for D in (15, 30, 45, 60):
C = candidate_daily(factory(D), tf="1d")
uf, uh = blend_uplift(C)
hold = C[C.index >= HOLDOUT]
print(f"d{D:<3} lag0 {_sh(C):>9.2f}{_sh(hold):>9.2f}{uf:>+10.3f}{uh:>+10.3f}")
C = candidate_daily(factory(30, extra_lag=1), tf="1d")
uf, uh = blend_uplift(C)
hold = C[C.index >= HOLDOUT]
print(f"d30 lag+1g {_sh(C):>9.2f}{_sh(hold):>9.2f}{uf:>+10.3f}{uh:>+10.3f}")
scripts/research/r0724_premium_wave.py
+141
View File
@@ -0,0 +1,141 @@
"""r0724_premium_wave — premi cross-venue AUTO-CALCOLATI: Coinbase premium + Kimchi (2026-07-24).
Seconda parte dell'ondata "trova altre strategie" (dopo r0724_onchain_wave): la ricerca
web (agente on-chain/dati) indica i premi regionali come UNICA famiglia flow con dati
100% auto-calcolabili da candele raw -> zero rischio-revisione/vintage del vendor
(coerente con la dottrina dati del progetto). Evidenza accademica: kimchi = anomalia
documentata (violazione persistente della legge del prezzo unico, capital controls);
lead-lag ASIMMETRICO e TEMPO-VARIANTE (MDPI 2026) -> nessuna regola pubblicata onesta,
qui si meccanizza da zero.
SEGNALI (mai il prezzo): per asset a e giorno d
CBPREM_a(d) = close Coinbase USD (00:00 UTC) / close feed certificato - 1
KIMCHI_a(d) = close Upbit KRW (00:00 UTC, ancora 09:00 KST) / (USDKRW_ECB x close cert) - 1
Le candele Upbit daily sono ancorate a mezzanotte UTC (=09:00 KST) -> stesso istante di
chiusura del feed certificato, nessun premio finto da mismatch orario. FX = fixing ECB
del giorno (ffill weekend; il KRW si muove ~nulla vs la vol crypto — caveat dichiarato).
NB: il feed certificato Deribit e' esso stesso un indice multi-exchange che include
Coinbase -> il CBPREM misurato e' SMORZATO (caveat strutturale).
Griglie (piccole, tutte contate nel deflated-Sharpe): z-score rolling 180g del livello
(follow / contrarian) e segno della variazione 30g (follow), LF e LS -> 6 celle/famiglia.
Gate: study_family_honest (cella in-sample, DSR, marginal scorer vs TP01).
Dati: data/external/premium/ (fetch: scratchpad/fetch_premium.py — Coinbase Exchange
public candles, Upbit public candles, frankfurter.app ECB; tutti tokenless).
Uso: `uv run python scripts/research/r0724_premium_wave.py`
"""
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))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
import altlib # noqa: E402
from altlib import study_family_honest, fmt_marginal, get # noqa: E402
EXT = ROOT / "data" / "external" / "premium"
def _cert_close_by_day(asset: str) -> pd.Series:
df = get(asset, "1d")
days = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)).floor("D")
return pd.Series(df["close"].values.astype(float), index=days)
def _cb_close(asset: str) -> pd.Series:
d = pd.read_csv(EXT / f"cb_{asset.lower()}.csv")
idx = pd.DatetimeIndex(pd.to_datetime(d["ts"], unit="s", utc=True)).floor("D")
return pd.Series(d["close"].values.astype(float), index=idx)
def _upbit_close(asset: str) -> pd.Series:
d = pd.read_csv(EXT / f"upbit_{asset.lower()}.csv")
idx = pd.DatetimeIndex(pd.to_datetime(d["utc"], utc=True)).floor("D")
return pd.Series(d["close_krw"].values.astype(float), index=idx)
def _fx() -> pd.Series:
d = pd.read_csv(EXT / "usdkrw.csv")
idx = pd.DatetimeIndex(pd.to_datetime(d["date"], utc=True))
s = pd.Series(d["usdkrw"].values.astype(float), index=idx)
full = pd.date_range(s.index[0], s.index[-1] + pd.Timedelta(days=3), freq="D", tz="UTC")
return s.reindex(full).ffill()
def _premia() -> dict:
fx = _fx()
out = {}
for a in ("BTC", "ETH"):
cert = _cert_close_by_day(a)
cb = _cb_close(a).reindex(cert.index)
up = _upbit_close(a).reindex(cert.index)
out[a] = pd.DataFrame({
"cbprem": cb / cert - 1.0,
"kimchi": up / (fx.reindex(cert.index) * cert) - 1.0,
}, index=cert.index)
return out
_PREM = _premia()
def prem_factory_maker(col: str):
def factory(tf: str, sig: str = "z180", mode: str = "folLF"):
def fn(df, asset):
p = _PREM[asset][col]
if sig == "z180":
mu = p.rolling(180, min_periods=90).mean()
sd = p.rolling(180, min_periods=90).std()
z = (p - mu) / sd
else: # d30: variazione 30g del premio
z = p - p.shift(30)
s = np.sign(z) if mode.startswith("fol") else -np.sign(z)
pos = s if mode.endswith("LS") else (s > 0).astype(float)
days = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)).floor("D")
return np.nan_to_num(pos.reindex(days).values.astype(float))
return fn
return factory
GRID = [
dict(sig="z180", mode="folLF"), dict(sig="z180", mode="conLF"),
dict(sig="z180", mode="folLS"), dict(sig="z180", mode="conLS"),
dict(sig="d30", mode="folLF"), dict(sig="d30", mode="folLS"),
]
def main() -> None:
print("=" * 100)
print(" ONDATA PREMI CROSS-VENUE — CBPREM + KIMCHI via study_family_honest")
for a in ("BTC", "ETH"):
P = _PREM[a].dropna()
print(f" {a}: {len(P)} giorni | cbprem medio {P['cbprem'].mean()*1e4:+.1f}bps "
f"(p1/p99 {P['cbprem'].quantile(0.01)*1e4:+.0f}/{P['cbprem'].quantile(0.99)*1e4:+.0f}) | "
f"kimchi medio {P['kimchi'].mean()*1e4:+.1f}bps "
f"(p1/p99 {P['kimchi'].quantile(0.01)*1e4:+.0f}/{P['kimchi'].quantile(0.99)*1e4:+.0f})")
print("=" * 100)
for fam, col in (("CBPREM-coinbase", "cbprem"), ("KIMCHI-korea", "kimchi")):
print("-" * 100)
rep = study_family_honest(fam, prem_factory_maker(col), GRID, tfs=("1d",))
if rep.get("chosen") is None:
print(f"=== {fam}: nessuna cella valida in-sample")
continue
ch = rep["chosen"]
print(f"=== {fam}: cella IS {ch['params']} (IS Sh {ch['insample_sharpe']}, "
f"full {ch['full_sharpe']}) su {rep['n_cells']} celle")
print(f" deflated-Sharpe {rep['deflated_sharpe']} (null-max {rep['expected_null_max']})"
f" dsr_pass={rep['dsr_pass']}")
print(fmt_marginal(rep["marginal"]))
print(f" >>> EARNS_SLOT_HONEST = {rep['earns_slot_honest']}")
if __name__ == "__main__":
main()