diff --git a/src/live/basket_trend_worker.py b/src/live/basket_trend_worker.py index 3a37649..4460c7a 100644 --- a/src/live/basket_trend_worker.py +++ b/src/live/basket_trend_worker.py @@ -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 diff --git a/src/live/pairs_worker.py b/src/live/pairs_worker.py index 56a55a6..7ba50df 100644 --- a/src/live/pairs_worker.py +++ b/src/live/pairs_worker.py @@ -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 diff --git a/src/live/rotation_worker.py b/src/live/rotation_worker.py index 60e65bd..b683ca6 100644 --- a/src/live/rotation_worker.py +++ b/src/live/rotation_worker.py @@ -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 diff --git a/src/live/telegram_notifier.py b/src/live/telegram_notifier.py index 5c5ff06..68c8564 100644 --- a/src/live/telegram_notifier.py +++ b/src/live/telegram_notifier.py @@ -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) } diff --git a/src/live/tsmom_worker.py b/src/live/tsmom_worker.py index 9e4091f..b096bd9 100644 --- a/src/live/tsmom_worker.py +++ b/src/live/tsmom_worker.py @@ -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: diff --git a/tests/portfolio/test_basket_worker.py b/tests/portfolio/test_basket_worker.py index b953e8a..dd2561f 100644 --- a/tests/portfolio/test_basket_worker.py +++ b/tests/portfolio/test_basket_worker.py @@ -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) diff --git a/tests/portfolio/test_pairs_forming_bar.py b/tests/portfolio/test_pairs_forming_bar.py new file mode 100644 index 0000000..18e4b49 --- /dev/null +++ b/tests/portfolio/test_pairs_forming_bar.py @@ -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 diff --git a/tests/portfolio/test_rotation_worker.py b/tests/portfolio/test_rotation_worker.py index 46509c8..0a7bfb7 100644 --- a/tests/portfolio/test_rotation_worker.py +++ b/tests/portfolio/test_rotation_worker.py @@ -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 == [] diff --git a/tests/portfolio/test_tsmom_worker.py b/tests/portfolio/test_tsmom_worker.py index ead37fc..17fd2da 100644 --- a/tests/portfolio/test_tsmom_worker.py +++ b/tests/portfolio/test_tsmom_worker.py @@ -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