feat(live): BasketTrendWorker (TR01) EMA-cross long/flat multi-asset

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/live/basket_trend_worker.py

@@ -0,0 +1,87 @@
+"""BasketTrendWorker (TR01): EMA20>EMA100 long/flat su un paniere, equal-weight.
+Replica live di honest_improve2._tr_basket_daily."""
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+FEE_RT, LEV, POS = 0.001, 3.0, 0.15
+
+
+def _ema(x, n):
+    return pd.Series(x).ewm(span=n, adjust=False).mean().values
+
+
+class BasketTrendWorker:
+    def __init__(self, universe, tf="4h", capital=1000.0, position_size=POS,
+                 leverage=LEV, fee_rt=FEE_RT, name="TR01_basket",
+                 data_dir=Path("data/portfolio_paper")):
+        self.universe = list(universe)
+        self.tf = tf
+        self.initial_capital = capital
+        self.capital = capital
+        self.position_size = position_size
+        self.leverage = leverage
+        self.fee_rt = fee_rt
+        self.worker_id = f"{name}__{'-'.join(self.universe)}__{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.positions = {a: 0.0 for a in self.universe}
+        self.last_bar_ts = {a: 0 for a in self.universe}
+        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.positions = {**self.positions, **s.get("positions", {})}
+            self.last_bar_ts = {**self.last_bar_ts, **s.get("last_bar_ts", {})}
+        self.in_position = any(v > 0 for v in self.positions.values())
+
+    def _save(self):
+        self.status_path.write_text(json.dumps({
+            "capital": round(self.capital, 2), "positions": self.positions,
+            "last_bar_ts": self.last_bar_ts,
+            "ts": datetime.now(timezone.utc).isoformat()}, indent=2))
+
+    def tick(self, data: dict):
+        rets = []
+        for a in self.universe:
+            df = data.get(a)
+            if df is None or len(df) < 110:
+                continue
+            c = df["close"].values
+            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])
+            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)
+                self._log(a, prev, target, float(c[-1]))
+            self.positions[a] = target
+            self.last_bar_ts[a] = bar_ts
+        if rets:
+            self.capital = max(self.capital * (1 + float(np.mean(rets))), 10.0)
+        self.in_position = any(v > 0 for v in self.positions.values())
+        self._save()
+
+    def _log(self, asset, frm, to, price):
+        with open(self.trades_path, "a") as f:
+            f.write(json.dumps({"ts": datetime.now(timezone.utc).isoformat(),
+                                "asset": asset, "from": frm, "to": to,
+                                "price": round(price, 6), "capital": round(self.capital, 2)}) + "\n")
+
+    @property
+    def status_summary(self):
+        longs = [a for a, v in self.positions.items() if v > 0]
+        return f"{self.worker_id}: cap={self.capital:.0f} long={longs}"

tests/portfolio/test_basket_worker.py

@@ -0,0 +1,30 @@
+import numpy as np
+import pandas as pd
+from src.live.basket_trend_worker import BasketTrendWorker
+
+
+def _ramp_df(n=300, slope=1.0):
+    c = np.linspace(100, 100 + slope * n, n)
+    ts = (pd.date_range("2024-01-01", periods=n, freq="4h", tz="UTC").astype("int64") // 10**6)
+    return pd.DataFrame({"timestamp": ts, "open": c, "high": c, "low": c, "close": c, "volume": 1.0})
+
+
+def test_basket_goes_long_in_uptrend(tmp_path):
+    w = BasketTrendWorker(universe=["AAA", "BBB"], tf="4h", capital=1000.0, data_dir=tmp_path)
+    data = {"AAA": _ramp_df(slope=1.0), "BBB": _ramp_df(slope=1.0)}
+    w.tick(data)
+    assert w.positions["AAA"] == 1.0 and w.positions["BBB"] == 1.0
+
+
+def test_basket_flat_in_downtrend(tmp_path):
+    w = BasketTrendWorker(universe=["AAA"], tf="4h", capital=1000.0, data_dir=tmp_path)
+    data = {"AAA": _ramp_df(slope=-1.0)}
+    w.tick(data)
+    assert w.positions["AAA"] == 0.0
+
+
+def test_basket_persists_and_resumes(tmp_path):
+    w = BasketTrendWorker(universe=["AAA"], tf="4h", capital=1000.0, data_dir=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