Adriano Dal Pastro
14522262e6
Reset del progetto su fondamenta verificate dopo la scoperta che l'intera libreria "validata OOS" era artefatto di feed contaminato (print fantasma del feed Cerbero TESTNET + storico Binance/USDT). - Storico ricostruito da Deribit MAINNET (ccxt pubblico, tokenless) e CERTIFICATO (certify_feed.py): BTC/ETH puliti su TUTTA la storia (mediana 2-6 bps vs Coinbase USD), integrita' OHLC + coerenza resample (maxΔ 0.00) + cross-venue OK. Alt esclusi (illiquidi/divergenti: LTC/DOGE 50-82% barre flat; XRP/BNB non certificabili). - Verdetto sul feed pulito: FADE / PAIRS / XS01 / TSM01 morti (ogni portafoglio Sharpe -2.3..-3.0, DD ~40%); solo SH01 e frammenti HONEST con segnale residuo, da ri-validare in isolamento. - Cleanup "restart pulito": strategie, stack live (src/live, src/portfolio, runner/executor, yml, docker), ~100 script ricerca/gate, waste/games/ portfolios, dati non certificati + cache e 60+ diari -> archiviati in Old/ (preservati, non cancellati). Diario consolidato in un unico documento. - Skeleton ricerca tenuto: Strategy ABC + indicatori + src/fractal + src/backtest/engine + load_data; tool dati certificati (rebuild_history, certify_feed, audit_feed, multi_source_check). - Universo dati ATTIVO: solo BTC/ETH (5m/15m/1h); guardrail fisico (load_data su alt -> FileNotFoundError). Esecuzione DISABILITATA, conto flat. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
73 lines
3.1 KiB
Python
73 lines
3.1 KiB
Python
import numpy as np
|
|
import pandas as pd
|
|
from src.live.rotation_worker import RotationWorker
|
|
|
|
|
|
def _df(n=200, slope=1.0):
|
|
c = np.linspace(100, 100 + slope * n, n)
|
|
ts = (pd.date_range("2023-01-01", periods=n, freq="1D", tz="UTC").astype("int64") // 10**6)
|
|
return pd.DataFrame({"timestamp": ts, "open": c, "high": c, "low": c, "close": c, "volume": 1.0})
|
|
|
|
|
|
def test_rotation_picks_top_momentum_when_risk_on(tmp_path):
|
|
w = RotationWorker(universe=["BTC", "AAA", "BBB"], top_k=2, gross=0.45, data_dir=tmp_path)
|
|
data = {"BTC": _df(slope=1.0), "AAA": _df(slope=3.0), "BBB": _df(slope=0.1)}
|
|
w.tick(data)
|
|
assert w.weights["AAA"] > 0
|
|
assert abs(sum(w.weights.values()) - 0.45) < 1e-9
|
|
|
|
|
|
def test_rotation_flat_when_risk_off(tmp_path):
|
|
# BTC in downtrend -> risk_off -> nessuna posizione
|
|
w = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
|
|
data = {"BTC": _df(slope=-1.0), "AAA": _df(slope=3.0)}
|
|
w.tick(data)
|
|
assert sum(w.weights.values()) == 0.0
|
|
|
|
|
|
def test_rotation_persists_and_resumes(tmp_path):
|
|
w = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=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_panel_drops_forming_daily_bar(tmp_path):
|
|
# serie 1d che termina ADESSO -> l'ultima riga e' la candela 1d in formazione
|
|
import time
|
|
from src.live.rotation_worker import _panel
|
|
n = 200
|
|
now = int(time.time() * 1000)
|
|
ts = now - 86_400_000 * np.arange(n - 1, -1, -1)
|
|
def mk(slope):
|
|
c = np.linspace(100, 100 + slope * n, n)
|
|
return pd.DataFrame({"timestamp": ts, "open": c, "high": c, "low": c,
|
|
"close": c, "volume": 1.0})
|
|
panel, cols = _panel({"BTC": mk(1.0), "AAA": mk(2.0)}, ["BTC", "AAA"])
|
|
assert len(panel) == n - 1 # barra in formazione scartata
|
|
assert int(panel["timestamp"].iloc[-1]) == int(ts[-2])
|
|
|
|
|
|
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 == []
|