feat(portfolio): integra worker honest/TSM01 nel runner (PORT06 live completo)

build_worker_for gestisce basket/rotation/tsmom + DIP01 via StrategyWorker; run()
fetcha 1h e resampla a 4h/1d, lookback dimensionato sui daily (TSM01 252g); tick
multi-asset per kind. _defs marca TR01/ROT02/TSM01 col kind+universo. Niente piu'
sleeve saltati in PORT06.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

scripts/portfolios/_defs.py

@@ -9,12 +9,18 @@ sys.path.insert(0, str(PROJECT_ROOT))
 
 from src.portfolio.base import Portfolio, SleeveSpec  # noqa: E402
 
+# Universo live tradabile (8 asset con feed Cerbero v2 + parquet). ROT02/TSM01 ci ruotano sopra.
+UNIVERSE8 = ["ADA", "BNB", "BTC", "DOGE", "ETH", "LTC", "SOL", "XRP"]
+
 FADE = [SleeveSpec(kind="single", name=c, sid=f"{c}_{a}", asset=a, cluster=f"{a}-rev")
         for a in ("BTC", "ETH") for c in ("MR01", "MR02", "MR07")]
 HONEST = [
+    # DIP01: single-asset 1h -> StrategyWorker (Strategy DIP01_dip_buy). TR01/ROT02: multi-asset.
     SleeveSpec(kind="single", name="DIP01", sid="DIP01_BTC", asset="BTC", cluster="BTC-rev"),
-    SleeveSpec(kind="single", name="TR01", sid="TR01_basket", cluster="trend"),
+    SleeveSpec(kind="basket", name="TR01", sid="TR01_basket", cluster="trend",
-    SleeveSpec(kind="single", name="ROT02", sid="ROT02_rot", cluster="rotation"),
+               params={"universe": ["BNB", "BTC", "DOGE", "SOL", "XRP"], "tf": "4h"}),
+    SleeveSpec(kind="rotation", name="ROT02", sid="ROT02_rot", cluster="rotation",
+               params={"universe": UNIVERSE8, "tf": "1d", "top_k": 3, "gross": 0.45}),
 ]
 PAIRS = [
     SleeveSpec(kind="pairs", name="PR01", sid="PR_ETHBTC", a="ETH", b="BTC", cluster="ETH-rev"),
@@ -23,7 +29,9 @@ PAIRS = [
     SleeveSpec(kind="pairs", name="PR01", sid="PR_BTCLTC", a="BTC", b="LTC", cluster="BTC-rev"),
     SleeveSpec(kind="pairs", name="PR01", sid="PR_ETHSOL", a="ETH", b="SOL", cluster="ETH-rev"),
 ]
-TSM = [SleeveSpec(kind="single", name="TSM01", sid="TSM01", cluster="trend")]
+TSM = [SleeveSpec(kind="tsmom", name="TSM01", sid="TSM01", cluster="trend",
+                  params={"universe": UNIVERSE8, "tf": "1d",
+                          "horizons": [63, 126, 252], "thr": 1.0, "gross": 0.30})]
 SHAPE = [SleeveSpec(kind="ml", name="SH01", sid=f"SH_{a}", asset=a, cluster="shape")
          for a in ("BTC", "ETH")]
 

src/portfolio/runner.py

@@ -1,24 +1,42 @@
 """PortfolioRunner: faccia live del portafoglio (capitale pool, sizing, ribilancio, ledger).
-Riusa i worker esistenti come esecutori e il data layer Cerbero v2."""
+Riusa i worker esistenti come esecutori e il data layer Cerbero v2.
+
+Worker per tipo di sleeve:
+  single (fade/dip) -> StrategyWorker        | ml (shape) -> MLWorkerWrapper(StrategyWorker)
+  pairs             -> PairsWorker (2 gambe)  | basket (TR01)   -> BasketTrendWorker
+  rotation (ROT02)  -> RotationWorker         | tsmom (TSM01)   -> TsmomWorker
+
+Feed: il runner fetcha candele 1h da Cerbero v2 e le RESAMPLA a 4h/1d (come get_df nel
+backtest) per i worker a cadenza piu' lenta. Il lookback per asset e' dimensionato sul
+worker piu' esigente (TSM01 usa 252 giorni)."""
 from __future__ import annotations
 
 from pathlib import Path
 
+import pandas as pd
+
 from src.portfolio.base import SleeveSpec, Portfolio
 from src.portfolio.ledger import PortfolioLedger
 from src.live.strategy_worker import StrategyWorker
 from src.live.pairs_worker import PairsWorker
+from src.live.basket_trend_worker import BasketTrendWorker
+from src.live.rotation_worker import RotationWorker
+from src.live.tsmom_worker import TsmomWorker
 from src.live.multi_runner import MLWorkerWrapper
 from src.live.strategy_loader import load_strategy
 
-# Codice-breve sleeve -> nome modulo Strategy in scripts/strategies/
+# Codice-breve sleeve -> nome modulo Strategy in scripts/strategies/ (worker single/ml)
 _STRAT_MODULE = {
     "MR01": "MR01_bollinger_fade", "MR02": "MR02_donchian_fade",
     "MR07": "MR07_return_reversal", "SH01": "SH01_shape_ml",
-    # DIP01/TR01/ROT02 sono honest a sé: vedi nota nel design (worker dedicati in fase 2)
+    "DIP01": "DIP01_dip_buy",
 }
+_MULTI_KINDS = ("basket", "rotation", "tsmom")
 DATA_DIR = Path("data/portfolio_paper")
 
+# giorni di storia da fetchare per timeframe (TSM01 1d usa 252 barre -> ~440 giorni col buffer)
+_LOOKBACK_DAYS = {"1h": 90, "4h": 220, "1d": 440}
+
 
 def build_worker_for(spec: SleeveSpec, alloc_capital: float, leverage: float,
                      data_dir: Path = DATA_DIR, position_size: float = 0.15):
@@ -29,10 +47,28 @@ def build_worker_for(spec: SleeveSpec, alloc_capital: float, leverage: float,
             capital=alloc_capital, position_size=position_size, leverage=leverage,
             fee_rt=0.001, name="PR01_pairs_reversion", data_dir=data_dir,
         )
+    if spec.kind == "basket":
+        pr = spec.params
+        return BasketTrendWorker(
+            universe=pr["universe"], tf=pr.get("tf", "4h"), capital=alloc_capital,
+            position_size=position_size, leverage=leverage, data_dir=data_dir,
+        )
+    if spec.kind == "rotation":
+        pr = spec.params
+        return RotationWorker(
+            universe=pr["universe"], top_k=pr.get("top_k", 3), gross=pr.get("gross", 0.45),
+            tf=pr.get("tf", "1d"), capital=alloc_capital, data_dir=data_dir,
+        )
+    if spec.kind == "tsmom":
+        pr = spec.params
+        return TsmomWorker(
+            universe=pr["universe"], horizons=tuple(pr.get("horizons", (63, 126, 252))),
+            thr=pr.get("thr", 1.0), gross=pr.get("gross", 0.30),
+            tf=pr.get("tf", "1d"), capital=alloc_capital, data_dir=data_dir,
+        )
     module = _STRAT_MODULE.get(spec.name)
     if module is None:
-        raise ValueError(f"sleeve live non ancora supportato: {spec.name} "
+        raise ValueError(f"sleeve live non supportato: {spec.name} (kind={spec.kind})")
-                         f"(honest DIP01/TR01/ROT02 richiedono worker dedicati, fase 2)")
     strategy = load_strategy(module)
     worker = StrategyWorker(
         strategy=strategy, asset=spec.asset, tf=spec.tf, capital=alloc_capital,
@@ -63,13 +99,36 @@ def rebalance_allocations(ledger: PortfolioLedger, workers: dict, weights: dict[
     ledger.save()
 
 
+def _resample(df: pd.DataFrame, tf: str) -> pd.DataFrame:
+    """Resampla candele 1h -> 4h/1d mantenendo timestamp ms reale (come get_df del backtest)."""
+    if tf == "1h":
+        return df
+    rule = {"4h": "4h", "1d": "1D"}[tf]
+    d = df.copy()
+    d["dt"] = pd.to_datetime(d["timestamp"], unit="ms", utc=True)
+    d = d.set_index("dt")
+    agg = d.resample(rule).agg({"open": "first", "high": "max", "low": "min",
+                                "close": "last", "volume": "sum"}).dropna()
+    epoch = pd.Timestamp("1970-01-01", tz="UTC")
+    agg["timestamp"] = ((agg.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
+    return agg.reset_index(drop=True)
+
+
+def _spec_assets_tf(spec: SleeveSpec):
+    """(lista asset, tf) coinvolti da uno sleeve."""
+    if spec.kind == "pairs":
+        return [spec.a, spec.b], spec.tf
+    if spec.kind in _MULTI_KINDS:
+        return list(spec.params["universe"]), spec.params.get("tf", "1d" if spec.kind != "basket" else "4h")
+    return [spec.asset], spec.tf
+
+
 def run(config_path: str = "portfolios.yml"):
-    """Loop live a portafoglio. Data layer Cerbero v2; ribilancio a fine giornata UTC.
+    """Loop live a portafoglio (tutti i tipi di sleeve). Data layer Cerbero v2 con resample;
-    Gli sleeve senza worker live (honest DIP01/TR01/ROT02) vengono SALTATI con warning
+    ribilancio a cambio giornata UTC."""
-    (restano solo in backtest); i pesi sono rinormalizzati sugli sleeve eseguibili."""
     import time
     from datetime import datetime, timezone, timedelta
-    import pandas as pd
+    import yaml
     from src.portfolio.base import load_active_portfolio
     from src.portfolio.sleeves import sleeve_returns_df
     from src.portfolio import weighting as W
@@ -77,13 +136,11 @@ def run(config_path: str = "portfolios.yml"):
     from src.live.multi_runner import INSTRUMENT_MAP
 
     p: Portfolio = load_active_portfolio(config_path)
-
+    _ov = (yaml.safe_load(Path(config_path).read_text()) or {}).get("overrides", {})
-    import yaml as _yaml
-    _ov = (_yaml.safe_load(__import__("pathlib").Path(config_path).read_text()) or {}).get("overrides", {})
     poll = int(_ov.get("poll_seconds", 60))
 
     def _supported(s):
-        return s.kind == "pairs" or s.name in _STRAT_MODULE
+        return s.kind in ("pairs",) + _MULTI_KINDS or s.name in _STRAT_MODULE
     live_specs = [s for s in p.sleeves if _supported(s)]
     skipped = [s.sid for s in p.sleeves if not _supported(s)]
     if skipped:
@@ -100,40 +157,47 @@ def run(config_path: str = "portfolios.yml"):
     alloc = ledger.allocate(weights)
     workers = {s.sid: build_worker_for(s, alloc[s.sid], p.leverage) for s in live_specs}
 
+    # lookback (giorni) richiesto per ogni asset = max sui worker che lo usano
+    asset_days: dict[str, int] = {}
+    for s in live_specs:
+        assets, tf = _spec_assets_tf(s)
+        for a in assets:
+            asset_days[a] = max(asset_days.get(a, 0), _LOOKBACK_DAYS.get(tf, 90))
+
     inst_map = dict(INSTRUMENT_MAP)
     last_day = ""
     while True:
         try:
-            keys = set()
+            # fetch 1h per asset al lookback massimo richiesto
-            for s in live_specs:
+            raw1h: dict[str, pd.DataFrame] = {}
-                if s.kind == "pairs":
+            end = datetime.now(timezone.utc)
-                    keys.add((s.a, s.tf)); keys.add((s.b, s.tf))
+            for asset, days in asset_days.items():
-                else:
-                    keys.add((s.asset, s.tf))
-            cache = {}
-            end = datetime.now(timezone.utc); start = end - timedelta(days=60)
-            for asset, tf in keys:
                 inst = inst_map.get(asset, f"{asset}-PERPETUAL")
+                start = end - timedelta(days=days)
                 candles = client.get_historical_v2(inst, start.strftime("%Y-%m-%d"),
-                                                   end.strftime("%Y-%m-%d"), tf)
+                                                   end.strftime("%Y-%m-%d"), "1h")
                 if candles:
                     df = pd.DataFrame(candles)
                     df["timestamp"] = df["timestamp"].astype("int64")
-                    cache[(asset, tf)] = df.sort_values("timestamp").reset_index(drop=True)
+                    raw1h[asset] = df.sort_values("timestamp").reset_index(drop=True)
 
+            # tick di ogni worker col suo timeframe (resample dal 1h)
             for s in live_specs:
                 w = workers[s.sid]
+                assets, tf = _spec_assets_tf(s)
+                if any(a not in raw1h for a in assets):
+                    continue
+                res = {a: _resample(raw1h[a], tf) for a in assets}
                 if s.kind == "pairs":
-                    ka, kb = (s.a, s.tf), (s.b, s.tf)
+                    w.tick(res[s.a], res[s.b])
-                    if ka in cache and kb in cache:
+                elif s.kind in _MULTI_KINDS:
-                        w.tick(cache[ka], cache[kb])
+                    w.tick(res)
                 else:
-                    key = (s.asset, s.tf)
+                    df = res[s.asset]
-                    if key in cache:
+                    inner = getattr(w, "worker", w)
-                        inner = getattr(w, "worker", w)
+                    if hasattr(w, "needs_training") and w.needs_training():
-                        if hasattr(w, "needs_training") and w.needs_training():
+                        w.train(df, hold=inner.hold_bars)
-                            w.train(cache[key], hold=inner.hold_bars)
+                    w.tick(df)
-                        w.tick(cache[key])
 
             ledger.update_equity({sid: _worker_equity(wk) for sid, wk in workers.items()})
 

tests/portfolio/test_runner_honest.py

@@ -0,0 +1,39 @@
+"""T5: integrazione worker honest/TSM01 nel PortfolioRunner."""
+from src.portfolio.runner import build_worker_for, _STRAT_MODULE, _MULTI_KINDS
+from src.portfolio.base import SleeveSpec
+from src.live.basket_trend_worker import BasketTrendWorker
+from src.live.rotation_worker import RotationWorker
+from src.live.tsmom_worker import TsmomWorker
+from src.live.strategy_worker import StrategyWorker
+from scripts.portfolios._defs import PORTFOLIOS
+
+
+def test_build_basket_worker(tmp_path):
+    spec = SleeveSpec(kind="basket", name="TR01", sid="TR01_basket",
+                      params={"universe": ["BNB", "BTC"], "tf": "4h"})
+    w = build_worker_for(spec, alloc_capital=120.0, leverage=2.0, data_dir=tmp_path)
+    assert isinstance(w, BasketTrendWorker) and w.capital == 120.0
+
+
+def test_build_rotation_and_tsmom(tmp_path):
+    rot = SleeveSpec(kind="rotation", name="ROT02", sid="ROT02_rot",
+                     params={"universe": ["BTC", "ETH"], "tf": "1d", "top_k": 1, "gross": 0.45})
+    tsm = SleeveSpec(kind="tsmom", name="TSM01", sid="TSM01",
+                     params={"universe": ["BTC", "ETH"], "tf": "1d", "gross": 0.30})
+    wr = build_worker_for(rot, 100.0, 2.0, data_dir=tmp_path)
+    wt = build_worker_for(tsm, 100.0, 2.0, data_dir=tmp_path)
+    assert isinstance(wr, RotationWorker) and wr.capital == 100.0
+    assert isinstance(wt, TsmomWorker) and wt.capital == 100.0
+
+
+def test_dip01_builds_as_strategy_worker(tmp_path):
+    spec = SleeveSpec(kind="single", name="DIP01", sid="DIP01_BTC", asset="BTC")
+    w = build_worker_for(spec, 80.0, 2.0, data_dir=tmp_path)
+    assert isinstance(w, StrategyWorker) and w.capital == 80.0
+
+
+def test_port06_has_no_unsupported_sleeves():
+    p = PORTFOLIOS["PORT06"]
+    unsupported = [s.sid for s in p.sleeves
+                   if not (s.kind in ("pairs",) + _MULTI_KINDS or s.name in _STRAT_MODULE)]
+    assert unsupported == []