fix(live): parita' worker multi-asset — TR01 fee x leva, forming-bar su TR01/Pairs, WARN panel corto
Punti 2-4 dell'improvement-sweep 2026-06-06 (fix di parita', nessun cambio di strategia): - TR01 (basket_trend_worker): fee per flip = POS * LEV * fee_rt/2 come il reference honest_improve2._tr_basket_daily:150 (mancava il fattore leva -> fee sotto-caricate 2x, bias ottimistico che gonfiava total_capital al ribilancio). - TR01: crossover EMA e booking del return valutati SOLO su barre 4h COMPLETE (la riga -1 e' la candela in corso finche' non e' trascorsa la sua durata) — lezione EXIT-16; evidenza live: flip SOL 0->1->0 in 59min nella stessa finestra 4h. - PairsWorker: entry ED exit valutati sul close di barra COMPLETA, come il backtest pairs_research (close settled). Stesso pattern bar_ms di strategy_worker.tick. - TSM01/ROT02: il silent return su panel inner-join troncato sotto il lookback ora emette WARN log + Telegram PANEL_SHORT (helper _warn_panel_short condiviso), gated su "era gia' operativo" (no falsi positivi al cold-start), una notifica per episodio. Test: 72/72 verdi (7 nuovi: forming-bar TR01/pairs + controllo positivo, fee con leva, PANEL_SHORT warn/dedup/cold-start). Replay parity: validate_worker_pairs ESATTO (ETH/BTC e BTC/LTC == backtest); validate_honest_workers invariato (TR01 -44% vs ref +42% IDENTICO pre/post fix: e' la divergenza di convenzione capitale-unico vs media-equity gia' documentata, da rivisitare a parte). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -3,6 +3,7 @@ Replica live di honest_improve2._tr_basket_daily."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
@@ -52,21 +53,35 @@ class BasketTrendWorker:
|
||||
"ts": datetime.now(timezone.utc).isoformat()}, indent=2))
|
||||
|
||||
def tick(self, data: dict):
|
||||
now_ms = int(time.time() * 1000)
|
||||
rets = []
|
||||
for a in self.universe:
|
||||
df = data.get(a)
|
||||
if df is None or len(df) < 110:
|
||||
if df is None or len(df) < 111:
|
||||
continue
|
||||
# Scarta la barra 4h IN FORMAZIONE (la riga -1 e' la candela in corso
|
||||
# finche' non e' trascorsa la sua durata): crossover EMA e booking del
|
||||
# return valutati SOLO su barre COMPLETE, come il reference
|
||||
# honest_improve2._tr_basket_daily (lezione EXIT-16; evidenza live: flip
|
||||
# SOL 0->1->0 in 59min nella stessa finestra 4h, -9.3% di glitch).
|
||||
ts_arr = df["timestamp"].values.astype("int64")
|
||||
bar_ms = int(np.median(np.diff(ts_arr[-50:]))) if len(ts_arr) > 1 else 0
|
||||
c = df["close"].values
|
||||
if bar_ms and now_ms < int(ts_arr[-1]) + bar_ms:
|
||||
c, ts_arr = c[:-1], ts_arr[:-1]
|
||||
if len(c) < 110:
|
||||
continue
|
||||
ef, es = _ema(c, 20)[-1], _ema(c, 100)[-1]
|
||||
target = 1.0 if ef > es else 0.0
|
||||
bar_ts = int(df["timestamp"].iloc[-1])
|
||||
bar_ts = int(ts_arr[-1])
|
||||
prev = self.positions[a]
|
||||
if self.last_bar_ts[a] and bar_ts > self.last_bar_ts[a] and prev > 0:
|
||||
r = (c[-1] - c[-2]) / c[-2]
|
||||
rets.append(self.position_size * self.leverage * r * prev)
|
||||
if target != prev:
|
||||
self.capital -= self.capital * self.position_size * (self.fee_rt / 2) * abs(target - prev) / len(self.universe)
|
||||
# fee = FEE_RT/2 * LEV come il reference (honest_improve2.py:150):
|
||||
# il notional e' leveraged, la fee si paga sul notional
|
||||
self.capital -= self.capital * self.position_size * self.leverage * (self.fee_rt / 2) * abs(target - prev) / len(self.universe)
|
||||
self._log(a, prev, target, float(c[-1]))
|
||||
self.positions[a] = target
|
||||
self.last_bar_ts[a] = bar_ts
|
||||
|
||||
@@ -16,6 +16,7 @@ Stato persistente (resume al restart) e log come StrategyWorker.
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
@@ -178,6 +179,14 @@ class PairsWorker:
|
||||
m = df_a[["timestamp", "close"]].rename(columns={"close": "ca"}).merge(
|
||||
df_b[["timestamp", "close"]].rename(columns={"close": "cb"}), on="timestamp", how="inner"
|
||||
).sort_values("timestamp").reset_index(drop=True)
|
||||
# Scarta la barra IN FORMAZIONE (la riga -1 e' la candela in corso finche'
|
||||
# non e' trascorsa la sua durata): entry ED exit valutati SOLO sul close di
|
||||
# barra COMPLETA, come il backtest (pairs_research: close settled) —
|
||||
# lezione EXIT-16 (strategy_worker.tick).
|
||||
ts_arr = m["timestamp"].values.astype("int64")
|
||||
bar_ms = int(np.median(np.diff(ts_arr[-50:]))) if len(ts_arr) > 1 else 0
|
||||
if bar_ms and int(time.time() * 1000) < int(ts_arr[-1]) + bar_ms:
|
||||
m = m.iloc[:-1]
|
||||
if len(m) < self.n + 2:
|
||||
return
|
||||
ca, cb = m["ca"].values, m["cb"].values
|
||||
|
||||
@@ -29,6 +29,25 @@ def _panel(data: dict, universe: list):
|
||||
return panel, cols
|
||||
|
||||
|
||||
def _warn_panel_short(worker_id: str, panel, cols: list, need: int,
|
||||
last_bar_ts: int, already_warned: bool) -> bool:
|
||||
"""WARN (log + Telegram) quando il panel inner-join e' troncato sotto il lookback
|
||||
richiesto e tick() salterebbe in SILENZIO (worker inerte senza segnale di vita).
|
||||
Gated su last_bar_ts != 0 ("era gia' operativo") per evitare falsi positivi al
|
||||
cold-start; una notifica per episodio. Ritorna il nuovo flag warned."""
|
||||
if not last_bar_ts: # mai stato operativo: cold-start, non un guasto
|
||||
return already_warned
|
||||
if already_warned:
|
||||
return True
|
||||
from src.live.telegram_notifier import notify_event
|
||||
got = 0 if panel is None else len(panel)
|
||||
msg = {"worker": worker_id, "panel_rows": got, "need": need,
|
||||
"assets": cols or "nessuno (BTC mancante?)"}
|
||||
print(f" [{worker_id}] WARN panel corto: {got}/{need} righe — tick saltato")
|
||||
notify_event("PANEL_SHORT", msg)
|
||||
return True
|
||||
|
||||
|
||||
class RotationWorker:
|
||||
def __init__(self, universe, lookback=60, top_k=3, gross=0.45, regime_n=100,
|
||||
tf="1d", capital=1000.0, fee_rt=FEE_RT, name="ROT02_rot",
|
||||
@@ -50,6 +69,7 @@ class RotationWorker:
|
||||
self.weights = {a: 0.0 for a in self.universe}
|
||||
self.last_bar_ts = 0
|
||||
self.in_position = False
|
||||
self._panel_warned = False # dedup WARN panel corto (per episodio, non persistito)
|
||||
self._load()
|
||||
|
||||
def _load(self):
|
||||
@@ -67,9 +87,13 @@ class RotationWorker:
|
||||
"ts": datetime.now(timezone.utc).isoformat()}, indent=2))
|
||||
|
||||
def tick(self, data: dict):
|
||||
need = max(self.lookback + 1, self.regime_n + 1)
|
||||
panel, cols = _panel(data, self.universe)
|
||||
if panel is None or len(panel) < max(self.lookback + 1, self.regime_n + 1) or "BTC" not in cols:
|
||||
if panel is None or len(panel) < need or "BTC" not in cols:
|
||||
self._panel_warned = _warn_panel_short(
|
||||
self.worker_id, panel, cols, need, self.last_bar_ts, self._panel_warned)
|
||||
return
|
||||
self._panel_warned = False
|
||||
P = panel[cols].values
|
||||
bar_ts = int(panel["timestamp"].iloc[-1])
|
||||
# 1) realizza il rendimento dei pesi correnti sull'ultima barra chiusa
|
||||
|
||||
@@ -19,6 +19,8 @@ NOTIFY_EVENTS = {
|
||||
"REAL_OPEN_FAIL", # un'apertura reale NON si e' verificata (problema da guardare)
|
||||
"STALE_FEED", # feed flat/fermo da >= N barre 1h (worker ciechi: il prossimo
|
||||
# prezzo reale puo' gappare, come ETH 2026-06-05 1655->1600)
|
||||
"PANEL_SHORT", # TSM01/ROT02: panel inner-join troncato sotto il lookback
|
||||
# richiesto -> tick() salterebbe in SILENZIO (worker inerte)
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -9,7 +9,7 @@ from pathlib import Path
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src.live.rotation_worker import _panel, FEE_RT
|
||||
from src.live.rotation_worker import _panel, _warn_panel_short, FEE_RT
|
||||
|
||||
|
||||
class TsmomWorker:
|
||||
@@ -33,6 +33,7 @@ class TsmomWorker:
|
||||
self.weights = {a: 0.0 for a in self.universe}
|
||||
self.last_bar_ts = 0
|
||||
self.in_position = False
|
||||
self._panel_warned = False # dedup WARN panel corto (per episodio, non persistito)
|
||||
self._load()
|
||||
|
||||
def _load(self):
|
||||
@@ -53,7 +54,10 @@ class TsmomWorker:
|
||||
need = max(max(self.horizons) + 1, self.regime_n + 1)
|
||||
panel, cols = _panel(data, self.universe)
|
||||
if panel is None or len(panel) < need or "BTC" not in cols:
|
||||
self._panel_warned = _warn_panel_short(
|
||||
self.worker_id, panel, cols, need, self.last_bar_ts, self._panel_warned)
|
||||
return
|
||||
self._panel_warned = False
|
||||
P = panel[cols].values
|
||||
bar_ts = int(panel["timestamp"].iloc[-1])
|
||||
if self.last_bar_ts and bar_ts > self.last_bar_ts:
|
||||
|
||||
@@ -28,3 +28,32 @@ def test_basket_persists_and_resumes(tmp_path):
|
||||
w.tick({"AAA": _ramp_df(slope=1.0)})
|
||||
w2 = BasketTrendWorker(universe=["AAA"], tf="4h", capital=1000.0, data_dir=tmp_path)
|
||||
assert w2.positions["AAA"] == 1.0 # stato ripreso da status.json
|
||||
|
||||
|
||||
def _ramp_df_now(n=300, slope=1.0, forming_close=None):
|
||||
"""Serie 4h che termina ADESSO: l'ultima riga e' la candela IN FORMAZIONE."""
|
||||
import time
|
||||
c = np.linspace(100, 100 + slope * n, n)
|
||||
if forming_close is not None:
|
||||
c = c.copy()
|
||||
c[-1] = forming_close
|
||||
bar = 4 * 3_600_000
|
||||
now_ms = int(time.time() * 1000)
|
||||
ts = now_ms - bar * np.arange(n - 1, -1, -1)
|
||||
return pd.DataFrame({"timestamp": ts, "open": c, "high": c, "low": c, "close": c, "volume": 1.0})
|
||||
|
||||
|
||||
def test_basket_ignores_forming_bar(tmp_path):
|
||||
# downtrend su barre COMPLETE (target flat); la candela in formazione e' un
|
||||
# spike estremo che flipperebbe EMA20>EMA100 se contata -> va ignorata.
|
||||
w = BasketTrendWorker(universe=["AAA"], tf="4h", capital=1000.0, data_dir=tmp_path)
|
||||
w.tick({"AAA": _ramp_df_now(slope=-0.3, forming_close=1e6)})
|
||||
assert w.positions["AAA"] == 0.0
|
||||
|
||||
|
||||
def test_basket_fee_includes_leverage(tmp_path):
|
||||
# flip 0->1: fee = cap * POS * LEV * fee_rt/2 (reference honest_improve2:150)
|
||||
w = BasketTrendWorker(universe=["AAA"], tf="4h", capital=1000.0, data_dir=tmp_path)
|
||||
w.tick({"AAA": _ramp_df(slope=1.0)})
|
||||
expected = 1000.0 - 1000.0 * w.position_size * w.leverage * (w.fee_rt / 2)
|
||||
assert np.isclose(w.capital, expected)
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
"""PairsWorker: entry/exit valutati SOLO su barre COMPLETE (lezione EXIT-16)."""
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src.live.pairs_worker import PairsWorker
|
||||
|
||||
|
||||
def _pair_dfs(n=80, jump=0.05, forming=True):
|
||||
"""Log-ratio = rumore alternato ±0.001 (z ~ ±1, nessun segnale) tranne l'ULTIMA
|
||||
barra, dove un salto `jump` del ratio porta z >> z_in (e dr <= jump_max).
|
||||
Con forming=True la serie termina ADESSO -> l'ultima riga e' in formazione."""
|
||||
r = 0.001 * np.array([(-1) ** i for i in range(n)], dtype=float)
|
||||
r[-1] = jump
|
||||
ca = 100.0 * np.exp(r)
|
||||
cb = np.full(n, 100.0)
|
||||
if forming:
|
||||
now_ms = int(time.time() * 1000)
|
||||
ts = now_ms - 3_600_000 * np.arange(n - 1, -1, -1)
|
||||
else:
|
||||
ts = (pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC").astype("int64") // 10**6)
|
||||
dfa = pd.DataFrame({"timestamp": ts, "close": ca})
|
||||
dfb = pd.DataFrame({"timestamp": ts, "close": cb})
|
||||
return dfa, dfb
|
||||
|
||||
|
||||
def test_pairs_opens_on_completed_extreme_bar(tmp_path):
|
||||
# controllo: la stessa barra estrema, se COMPLETA, apre (z >= z_in, short ratio)
|
||||
w = PairsWorker("AAA", "BBB", "1h", data_dir=tmp_path)
|
||||
dfa, dfb = _pair_dfs(forming=False)
|
||||
w.tick(dfa, dfb)
|
||||
assert w.in_position and w.direction == -1
|
||||
|
||||
|
||||
def test_pairs_ignores_forming_bar(tmp_path):
|
||||
# la stessa barra estrema IN FORMAZIONE va ignorata: nessun ingresso
|
||||
w = PairsWorker("AAA", "BBB", "1h", data_dir=tmp_path)
|
||||
dfa, dfb = _pair_dfs(forming=True)
|
||||
w.tick(dfa, dfb)
|
||||
assert not w.in_position
|
||||
@@ -30,3 +30,27 @@ def test_rotation_persists_and_resumes(tmp_path):
|
||||
w.tick({"BTC": _df(slope=1.0), "AAA": _df(slope=3.0)})
|
||||
w2 = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
|
||||
assert w2.weights == w.weights
|
||||
|
||||
|
||||
def test_rotation_warns_once_on_short_panel(tmp_path, monkeypatch):
|
||||
# worker GIA' operativo + panel troncato -> WARN (una sola notifica per episodio)
|
||||
calls = []
|
||||
monkeypatch.setattr("src.live.telegram_notifier.notify_event",
|
||||
lambda e, d=None: calls.append(e))
|
||||
w = RotationWorker(universe=["BTC", "AAA"], data_dir=tmp_path)
|
||||
w.last_bar_ts = 123 # era gia' operativo
|
||||
short = {"BTC": _df(n=20), "AAA": _df(n=20)}
|
||||
w.tick(short)
|
||||
w.tick(short) # dedup: niente seconda notifica
|
||||
assert calls == ["PANEL_SHORT"]
|
||||
w.tick({"BTC": _df(n=200), "AAA": _df(n=200)}) # recovery resetta il dedup
|
||||
assert w._panel_warned is False
|
||||
|
||||
|
||||
def test_rotation_no_warn_on_cold_start(tmp_path, monkeypatch):
|
||||
calls = []
|
||||
monkeypatch.setattr("src.live.telegram_notifier.notify_event",
|
||||
lambda e, d=None: calls.append(e))
|
||||
w = RotationWorker(universe=["BTC", "AAA"], data_dir=tmp_path) # last_bar_ts=0
|
||||
w.tick({"BTC": _df(n=20), "AAA": _df(n=20)})
|
||||
assert calls == []
|
||||
|
||||
@@ -31,3 +31,14 @@ def test_tsmom_persists_and_resumes(tmp_path):
|
||||
w.tick({"BTC": _df(slope=1.0), "AAA": _df(slope=2.0)})
|
||||
w2 = TsmomWorker(universe=["BTC", "AAA"], gross=0.30, data_dir=tmp_path)
|
||||
assert w2.weights == w.weights
|
||||
|
||||
|
||||
def test_tsmom_warns_on_short_panel(tmp_path, monkeypatch):
|
||||
# worker GIA' operativo + panel sotto need=253 -> WARN invece di silent return
|
||||
calls = []
|
||||
monkeypatch.setattr("src.live.telegram_notifier.notify_event",
|
||||
lambda e, d=None: calls.append(e))
|
||||
w = TsmomWorker(universe=["BTC", "AAA"], gross=0.30, data_dir=tmp_path)
|
||||
w.last_bar_ts = 123
|
||||
w.tick({"BTC": _df(n=100), "AAA": _df(n=100)})
|
||||
assert calls == ["PANEL_SHORT"] and w._panel_warned
|
||||
|
||||
Reference in New Issue
Block a user