feat(live): TsmomWorker (TSM01) consenso TSMOM multi-orizzonte risk-gated

src/live/tsmom_worker.py

@@ -0,0 +1,92 @@
+"""TsmomWorker (TSM01): consenso TSMOM multi-orizzonte risk-gated, ribilancio giornaliero.
+Replica live di tsmom_research.tsmom_sim (horizons 63/126/252, thr 1.0, gross 0.30, SMA100 gate)."""
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from src.live.rotation_worker import _panel, FEE_RT
+
+
+class TsmomWorker:
+    def __init__(self, universe, horizons=(63, 126, 252), thr=1.0, gross=0.30,
+                 regime_n=100, tf="1d", capital=1000.0, fee_rt=FEE_RT,
+                 name="TSM01", data_dir=Path("data/portfolio_paper")):
+        self.universe = list(universe)
+        self.horizons = tuple(horizons)
+        self.thr = thr
+        self.gross = gross
+        self.regime_n = regime_n
+        self.tf = tf
+        self.initial_capital = capital
+        self.capital = capital
+        self.fee_rt = fee_rt
+        self.worker_id = f"{name}__{tf}"
+        self.work_dir = Path(data_dir) / self.worker_id
+        self.work_dir.mkdir(parents=True, exist_ok=True)
+        self.status_path = self.work_dir / "status.json"
+        self.trades_path = self.work_dir / "trades.jsonl"
+        self.weights = {a: 0.0 for a in self.universe}
+        self.last_bar_ts = 0
+        self.in_position = False
+        self._load()
+
+    def _load(self):
+        if self.status_path.exists():
+            s = json.loads(self.status_path.read_text())
+            self.capital = s.get("capital", self.capital)
+            self.weights = {**{a: 0.0 for a in self.universe}, **s.get("weights", {})}
+            self.last_bar_ts = s.get("last_bar_ts", 0)
+        self.in_position = any(v > 0 for v in self.weights.values())
+
+    def _save(self):
+        self.status_path.write_text(json.dumps({
+            "capital": round(self.capital, 2), "weights": self.weights,
+            "last_bar_ts": self.last_bar_ts,
+            "ts": datetime.now(timezone.utc).isoformat()}, indent=2))
+
+    def tick(self, data: dict):
+        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:
+            return
+        P = panel[cols].values
+        bar_ts = int(panel["timestamp"].iloc[-1])
+        if self.last_bar_ts and bar_ts > self.last_bar_ts:
+            day_ret = P[-1] / P[-2] - 1.0
+            port_r = sum(self.weights.get(cols[k], 0.0) * day_ret[k] for k in range(len(cols)))
+            self.capital = max(self.capital * (1.0 + float(port_r)), 10.0)
+        btc = P[:, cols.index("BTC")]
+        bma = pd.Series(btc).rolling(self.regime_n).mean().values
+        risk_on = btc[-1] > bma[-1] if not np.isnan(bma[-1]) else False
+        score = np.zeros(len(cols))
+        for h in self.horizons:
+            score += np.sign(P[-1] / P[-1 - h] - 1.0)
+        score /= len(self.horizons)
+        chosen = [k for k in range(len(cols)) if score[k] >= self.thr] if risk_on else []
+        nw = {a: 0.0 for a in self.universe}
+        for k in chosen:
+            nw[cols[k]] = self.gross / len(chosen)
+        turnover = sum(abs(nw[a] - self.weights.get(a, 0.0)) for a in self.universe)
+        self.capital -= self.capital * turnover * (self.fee_rt / 2)
+        if turnover > 0:
+            self._log(nw, float(self.capital))
+        self.weights = nw
+        self.last_bar_ts = bar_ts
+        self.in_position = any(v > 0 for v in nw.values())
+        self._save()
+
+    def _log(self, weights, cap):
+        with open(self.trades_path, "a") as f:
+            f.write(json.dumps({"ts": datetime.now(timezone.utc).isoformat(),
+                                "weights": {a: round(w, 4) for a, w in weights.items() if w > 0},
+                                "capital": round(cap, 2)}) + "\n")
+
+    @property
+    def status_summary(self):
+        held = {a: round(w, 3) for a, w in self.weights.items() if w > 0}
+        return f"{self.worker_id}: cap={self.capital:.0f} held={held}"

tests/portfolio/test_tsmom_worker.py

@@ -0,0 +1,33 @@
+import numpy as np
+import pandas as pd
+from src.live.tsmom_worker import TsmomWorker
+
+
+def _df(n=300, 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_tsmom_selects_full_consensus_uptrend(tmp_path):
+    # tutti gli orizzonti positivi -> score=1>=thr; BTC su -> risk_on
+    w = TsmomWorker(universe=["BTC", "AAA"], horizons=(63, 126, 252), thr=1.0,
+                    gross=0.30, data_dir=tmp_path)
+    data = {"BTC": _df(slope=1.0), "AAA": _df(slope=2.0)}
+    w.tick(data)
+    assert w.weights["BTC"] > 0 and w.weights["AAA"] > 0
+    assert abs(sum(w.weights.values()) - 0.30) < 1e-9
+
+
+def test_tsmom_flat_when_risk_off(tmp_path):
+    w = TsmomWorker(universe=["BTC", "AAA"], thr=1.0, gross=0.30, data_dir=tmp_path)
+    data = {"BTC": _df(slope=-1.0), "AAA": _df(slope=2.0)}
+    w.tick(data)
+    assert sum(w.weights.values()) == 0.0
+
+
+def test_tsmom_persists_and_resumes(tmp_path):
+    w = TsmomWorker(universe=["BTC", "AAA"], gross=0.30, data_dir=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