From a7ada9f36c0c56d5e4f360822fde77a6c9ecf99d Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Fri, 29 May 2026 17:45:39 +0200 Subject: [PATCH] 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) --- scripts/portfolios/_defs.py | 14 ++- src/portfolio/runner.py | 130 +++++++++++++++++++------- tests/portfolio/test_runner_honest.py | 39 ++++++++ 3 files changed, 147 insertions(+), 36 deletions(-) create mode 100644 tests/portfolio/test_runner_honest.py diff --git a/scripts/portfolios/_defs.py b/scripts/portfolios/_defs.py index 6404a86..50470df 100644 --- a/scripts/portfolios/_defs.py +++ b/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="single", name="ROT02", sid="ROT02_rot", cluster="rotation"), + SleeveSpec(kind="basket", name="TR01", sid="TR01_basket", cluster="trend", + 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")] diff --git a/src/portfolio/runner.py b/src/portfolio/runner.py index 298eed6..d5ad9e8 100644 --- a/src/portfolio/runner.py +++ b/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} " - f"(honest DIP01/TR01/ROT02 richiedono worker dedicati, fase 2)") + raise ValueError(f"sleeve live non supportato: {spec.name} (kind={spec.kind})") 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. - Gli sleeve senza worker live (honest DIP01/TR01/ROT02) vengono SALTATI con warning - (restano solo in backtest); i pesi sono rinormalizzati sugli sleeve eseguibili.""" + """Loop live a portafoglio (tutti i tipi di sleeve). Data layer Cerbero v2 con resample; + ribilancio a cambio giornata UTC.""" 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) - - import yaml as _yaml - _ov = (_yaml.safe_load(__import__("pathlib").Path(config_path).read_text()) or {}).get("overrides", {}) + _ov = (yaml.safe_load(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() - for s in live_specs: - if s.kind == "pairs": - keys.add((s.a, s.tf)); keys.add((s.b, s.tf)) - else: - keys.add((s.asset, s.tf)) - cache = {} - end = datetime.now(timezone.utc); start = end - timedelta(days=60) - for asset, tf in keys: + # fetch 1h per asset al lookback massimo richiesto + raw1h: dict[str, pd.DataFrame] = {} + end = datetime.now(timezone.utc) + for asset, days in asset_days.items(): 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) - if ka in cache and kb in cache: - w.tick(cache[ka], cache[kb]) + w.tick(res[s.a], res[s.b]) + elif s.kind in _MULTI_KINDS: + w.tick(res) else: - key = (s.asset, s.tf) - if key in cache: - inner = getattr(w, "worker", w) - if hasattr(w, "needs_training") and w.needs_training(): - w.train(cache[key], hold=inner.hold_bars) - w.tick(cache[key]) + df = res[s.asset] + inner = getattr(w, "worker", w) + if hasattr(w, "needs_training") and w.needs_training(): + w.train(df, hold=inner.hold_bars) + w.tick(df) ledger.update_equity({sid: _worker_equity(wk) for sid, wk in workers.items()}) diff --git a/tests/portfolio/test_runner_honest.py b/tests/portfolio/test_runner_honest.py new file mode 100644 index 0000000..692cb18 --- /dev/null +++ b/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 == []