From dc63399cc71efa9b198c64b50d6092966a72ff5d Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Fri, 29 May 2026 17:35:10 +0200 Subject: [PATCH] docs(portfolio): piano fase 2-B worker honest/TSM01 dedicati (6 task) --- ...05-29-portfolios-phase2B-honest-workers.md | 377 ++++++++++++++++++ 1 file changed, 377 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-29-portfolios-phase2B-honest-workers.md diff --git a/docs/superpowers/plans/2026-05-29-portfolios-phase2B-honest-workers.md b/docs/superpowers/plans/2026-05-29-portfolios-phase2B-honest-workers.md new file mode 100644 index 0000000..d5e90e2 --- /dev/null +++ b/docs/superpowers/plans/2026-05-29-portfolios-phase2B-honest-workers.md @@ -0,0 +1,377 @@ +# Fase 2-B — Worker live honest/TSM01 (dedicati) — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: superpowers:subagent-driven-development o executing-plans. Steps con checkbox `- [ ]`. + +**Goal:** Costruire i worker live mancanti perché PORT06 giri live al completo (oltre a fade+pairs+shape già pronti): DIP01, TR01 (basket), ROT02 (rotation), TSM01 (tsmom rotation), e integrarli nel `PortfolioRunner`. + +**Architecture:** Worker DEDICATI per ogni strategia (scelta utente). DIP01 è single-asset → Strategy subclass + `StrategyWorker` esistente. TR01/ROT02/TSM01 sono multi-asset/rotation → tre classi worker nuove in `src/live/` con stato per-asset persistente, ciascuna fedele alla rispettiva funzione di backtest in `scripts/analysis/{honest_improve2,tsmom_research}.py`. Integrazione in `src/portfolio/runner.py::build_worker_for` + tick. + +**Tech Stack:** Python 3.11, pandas/numpy, pytest. Riusa CerberoClient v2 (multi-asset fetch), PortfolioLedger, e le funzioni di riferimento honest/tsm. + +**Branch:** `portfolio_phase2`. **Spec madre:** `docs/superpowers/specs/2026-05-29-portfolios-design.md` (§ scope live, fase 2). + +**Riferimenti di logica (NON modificare, sono la verità del backtest):** +- DIP01 → `honest_improve2.dip_market_gated` (z-score dip, gate BTC>SMA, TP=SMA/SL=ATR/max_bars, intrabar). +- TR01 → `honest_improve2._tr_basket_daily` (per asset 4h: EMA20>EMA100 long/flat; basket equal-weight). +- ROT02 → `honest_improve2._rot_daily_equity` (panel 1d, mom 60g, top-3 se mom>0 e BTC>SMA100, gross 0.45 split, ribilancio giornaliero). +- TSM01 → `tsmom_research.tsmom_sim` (panel 1d, Σ sign(P/P[-h]) h∈{63,126,252} ≥ thr=1.0, gate BTC>SMA100, gross 0.30 split). + +--- + +## File structure + +| File | Responsabilità | +|------|----------------| +| `scripts/strategies/DIP01_dip_buy.py` | Strategy `Dip01DipBuy` (single-asset; metadata tp/sl/max_bars + gate) | +| `src/live/basket_trend_worker.py` | `BasketTrendWorker` (TR01): N asset 4h, EMA cross, long/flat per asset | +| `src/live/rotation_worker.py` | `RotationWorker` (ROT02): panel 1d, dual-momentum top-k, gross split | +| `src/live/tsmom_worker.py` | `TsmomWorker` (TSM01): panel 1d, consenso segni multi-orizzonte | +| `src/live/strategy_loader.py` | **mod**: aggiungi `DIP01_dip_buy` a MODULE_MAP | +| `src/portfolio/runner.py` | **mod**: `build_worker_for` gestisce kind "basket"/"rotation"/"tsmom"; tick multi-asset | +| `src/portfolio/base.py` (`_defs.py`) | **mod**: SleeveSpec degli honest/tsm con `kind` e `universe` corretti | +| `tests/portfolio/test_honest_workers.py` | unit per ciascun worker + replay==backtest su finestra | + +**Universi:** TR01 = [BNB,BTC,DOGE,SOL,XRP] (4h); ROT02/TSM01 = `available_assets()` (1d). I worker multi-asset ricevono il dict {asset: df} dal runner. + +--- + +## Task 1: DIP01 come Strategy single-asset + +**Files:** Create `scripts/strategies/DIP01_dip_buy.py`; Modify `src/live/strategy_loader.py`; Test `tests/portfolio/test_dip01.py`. + +- [ ] **Step 1: Test (fallisce)** — `tests/portfolio/test_dip01.py`: + +```python +import pandas as pd +from src.data.downloader import load_data +from scripts.strategies.DIP01_dip_buy import Dip01DipBuy + + +def test_dip01_generates_long_signals_with_exits(): + df = load_data("BTC", "1h").iloc[-5000:].reset_index(drop=True) + ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + sigs = Dip01DipBuy().generate_signals(df, ts, asset="BTC", tf="1h") + assert len(sigs) > 0 + s = sigs[0] + assert s.direction == 1 # dip-buy è solo long + assert {"tp", "sl", "max_bars"} <= set(s.metadata) +``` + +- [ ] **Step 2:** `uv run pytest tests/portfolio/test_dip01.py -v` → FAIL (ModuleNotFoundError). + +- [ ] **Step 3: Implementa `scripts/strategies/DIP01_dip_buy.py`.** Replica ESATTA della logica di `dip_market_gated` (default `market_n=0` = senza gate, come lo sleeve DIP01_BTC del portafoglio: vedi combine_portfolio che usa `market_n=0`). Genera Signal long quando `z[i] <= -z_in and z[i-1] > -z_in`, con metadata `tp=SMA[i]`, `sl=c[i]-sl_atr*atr[i]`, `max_bars`. fee_rt=0.001, leverage 3, position 0.15. + +```python +"""DIP01 — Dip-buy mean-reversion single-asset (z-score sotto-banda). Honest family. + +Replica live della logica validata in scripts/analysis/honest_improve2.dip_market_gated +(con market_n=0, come lo sleeve DIP01_BTC del portafoglio): compra quando lo z-score del +prezzo rispetto a SMA(n) incrocia sotto -z_in; esce a TP=SMA, SL=close-sl_atr*ATR, o max_bars. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.strategies.base import Strategy, Signal # noqa: E402 + + +def _atr(df, n=14): + h, l, c = df["high"].values, df["low"].values, df["close"].values + pc = np.roll(c, 1); pc[0] = c[0] + tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc))) + return pd.Series(tr).rolling(n).mean().values + + +class Dip01DipBuy(Strategy): + name = "DIP01_dip_buy" + description = "Dip-buy mean-reversion single-asset (z-score), exit TP=SMA/SL=ATR/max_bars" + default_assets = ["BTC"] + default_timeframes = ["1h"] + fee_rt = 0.001 + leverage = 3.0 + position_size = 0.15 + + def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex, + n: int = 50, z_in: float = 2.5, sl_atr: float = 2.5, + max_bars: int = 24, **params) -> list[Signal]: + c = df["close"].values + ma = pd.Series(c).rolling(n).mean().values + sd = pd.Series(c).rolling(n).std().values + a = _atr(df, 14) + z = (c - ma) / np.where(sd == 0, np.nan, sd) + out: list[Signal] = [] + for i in range(n + 14, len(c)): + if np.isnan(z[i]) or np.isnan(a[i]) or np.isnan(ma[i]): + continue + if z[i] <= -z_in and z[i - 1] > -z_in: + out.append(Signal(idx=i, direction=1, entry_price=float(c[i]), + metadata={"tp": float(ma[i]), + "sl": float(c[i] - sl_atr * a[i]), + "max_bars": int(max_bars)})) + return out +``` + +- [ ] **Step 4: Registra nel loader.** In `src/live/strategy_loader.py` MODULE_MAP aggiungi: +```python + "DIP01_dip_buy": ("DIP01_dip_buy", "Dip01DipBuy"), +``` + +- [ ] **Step 5:** `uv run pytest tests/portfolio/test_dip01.py -v` → 1 passed. + +- [ ] **Step 6: Commit** +```bash +git add scripts/strategies/DIP01_dip_buy.py src/live/strategy_loader.py tests/portfolio/test_dip01.py +git commit -m "feat(live): DIP01 dip-buy come Strategy single-asset (worker via StrategyWorker)" +``` + +**Nota:** DIP01 nel runner usa lo StrategyWorker esistente (kind="single", name="DIP01"). Aggiorna `_STRAT_MODULE` in `runner.py` con `"DIP01": "DIP01_dip_buy"` e in `_defs.py` lo SleeveSpec DIP01_BTC resta kind="single". Il backtest dello sleeve DIP01_BTC continua a venire da `build_everything` (parità invariata). + +--- + +## Task 2: `BasketTrendWorker` (TR01) + +**Files:** Create `src/live/basket_trend_worker.py`; Test `tests/portfolio/test_basket_worker.py`. + +- [ ] **Step 1: Test (fallisce)** — verifica che, dato un dict {asset: df 4h}, il worker calcoli posizione long/flat per asset secondo EMA20>EMA100 e aggiorni il capitale equal-weight: + +```python +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 # EMA20>EMA100 in salita +``` + +- [ ] **Step 2:** `uv run pytest tests/portfolio/test_basket_worker.py -v` → FAIL. + +- [ ] **Step 3: Implementa `src/live/basket_trend_worker.py`.** Stato: capitale totale + dict `positions` (asset→0/1) + persistenza. `tick(data: dict[str,df])`: per ogni asset calcola EMA20/EMA100 sull'ultima barra; target = 1.0 se ef>es else 0.0; applica fee `FEE_RT/2*LEV` sul turnover |Δpos|; aggiorna capitale equal-weight col rendimento di barra di ogni asset attivo (`POS*LEV*ret*pos/len(universe)`... mantieni la convenzione di `_tr_basket_daily`: ogni asset è uno sleeve normalizzato, equal-weight → applica `mean` dei rendimenti per-asset). Persisti `status.json` (capitale, positions, last_bar_ts per asset) e logga `trades.jsonl`. fee_rt=0.001, leverage 3, position 0.15. + +```python +"""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 # per il ribilancio del runner (skip se True) + 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] + # rendimento di barra realizzato sulla posizione precedente (chiusa->aperta barra) + 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}" +``` + +- [ ] **Step 4:** `uv run pytest tests/portfolio/test_basket_worker.py -v` → 1 passed. + +- [ ] **Step 5: Commit** +```bash +git add src/live/basket_trend_worker.py tests/portfolio/test_basket_worker.py +git commit -m "feat(live): BasketTrendWorker (TR01) EMA-cross long/flat multi-asset" +``` + +--- + +## Task 3: `RotationWorker` (ROT02) + +**Files:** Create `src/live/rotation_worker.py`; Test `tests/portfolio/test_rotation_worker.py`. + +- [ ] **Step 1: Test (fallisce)** — dato {asset: df 1d}, sceglie i top-k per momentum 60g con gate BTC>SMA100 e imposta i pesi gross/k: + +```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) + # BTC in uptrend -> risk_on; top-2 momentum = AAA e BTC; pesi gross/2 + assert w.weights["AAA"] > 0 and abs(sum(w.weights.values()) - 0.45) < 1e-9 +``` + +- [ ] **Step 2:** `uv run pytest tests/portfolio/test_rotation_worker.py -v` → FAIL. + +- [ ] **Step 3: Implementa `src/live/rotation_worker.py`.** Replica di `_rot_daily_equity`: panel di close 1d allineato; `risk_on = BTC[-1] > SMA100(BTC)[-1]`; `mom = P[-1]/P[-61]-1`; `chosen = [top_k per mom con mom>0] se risk_on else []`; pesi `gross/len(chosen)`; turnover fee `FEE_RT/2 * Σ|Δw|`; capitale aggiornato col rendimento di portafoglio del giorno successivo (live: al tick si realizza il rendimento dell'ultima barra sui pesi correnti, poi si ricalcolano i pesi). Persisti capitale+weights+last_ts. `in_position = bool(weights)`. + +(Implementazione analoga a BasketTrendWorker: stato persistente, `tick(data)` allinea i panel per timestamp comune, calcola momentum/gate, applica fee sul turnover e rendimento di barra. Mantieni `top_k=3, gross=0.45` come default — i valori dello sleeve ROT02_rot del portafoglio.) + +- [ ] **Step 4:** test → 1 passed. + +- [ ] **Step 5: Commit** +```bash +git add src/live/rotation_worker.py tests/portfolio/test_rotation_worker.py +git commit -m "feat(live): RotationWorker (ROT02) dual-momentum top-k risk-gated" +``` + +--- + +## Task 4: `TsmomWorker` (TSM01) + +**Files:** Create `src/live/tsmom_worker.py`; Test `tests/portfolio/test_tsmom_worker.py`. + +- [ ] **Step 1: Test (fallisce)** — consenso segni multi-orizzonte: sceglie gli asset con `Σ sign(P/P[-h]) ≥ thr` (h∈{63,126,252}) sotto gate, pesi gross/k. + +- [ ] **Step 2-3: Implementa `src/live/tsmom_worker.py`** replicando `tsmom_sim`: `score[j] = mean_h sign(P[-1,j]/P[-1-h,j]-1)`; `chosen = [j: score>=thr] se risk_on`; pesi `gross/len(chosen)` con `gross=0.30`. Stessa struttura di RotationWorker (panel 1d, fee turnover, rendimento di barra, persistenza). Default `horizons=(63,126,252), thr=1.0, regime_n=100, gross=0.30`. + +- [ ] **Step 4:** test → passed. + +- [ ] **Step 5: Commit** +```bash +git add src/live/tsmom_worker.py tests/portfolio/test_tsmom_worker.py +git commit -m "feat(live): TsmomWorker (TSM01) consenso TSMOM multi-orizzonte risk-gated" +``` + +--- + +## Task 5: Integrazione nel PortfolioRunner + +**Files:** Modify `src/portfolio/runner.py`, `scripts/portfolios/_defs.py`, `src/portfolio/base.py`; Test `tests/portfolio/test_runner_honest.py`. + +- [ ] **Step 1:** In `_defs.py`, marca gli SleeveSpec multi-asset col `kind` giusto e l'universo: + - DIP01 → `kind="single", name="DIP01"` (resta StrategyWorker via _STRAT_MODULE["DIP01"]="DIP01_dip_buy"). + - TR01 → `kind="basket"`, aggiungi campo universo (riusa `params={"universe": ["BNB","BTC","DOGE","SOL","XRP"], "tf": "4h"}`). + - ROT02 → `kind="rotation"`, `params={"top_k":3, "gross":0.45, "tf":"1d"}`. + - TSM01 → `kind="tsmom"`, `params={"horizons":[63,126,252], "thr":1.0, "gross":0.30, "tf":"1d"}`. + (Aggiungi `universe`/campi a SleeveSpec se serve, default None.) + +- [ ] **Step 2:** In `runner.py::build_worker_for` aggiungi i rami `kind in ("basket","rotation","tsmom")` che costruiscono i rispettivi worker con `capital=alloc_capital` e `data_dir=DATA_DIR`. Aggiorna `_STRAT_MODULE` con `"DIP01": "DIP01_dip_buy"`. Rimuovi DIP01/TR01/ROT02/TSM01 dalla lista "saltati": ora sono supportati. + +- [ ] **Step 3:** In `runner.run()` il tick deve passare ai worker multi-asset un dict {asset: df} (fetch di tutti gli asset dell'universo). Estendi la raccolta `keys` e il dispatch del tick: per kind basket/rotation/tsmom costruisci `data = {a: cache[(a, tf)] for a in universe}` e chiama `w.tick(data)`. Per `_worker_equity` i nuovi worker espongono `.capital` (già ok). Per il ribilancio, espongono `.in_position` (skip se True). + +- [ ] **Step 4: Test** `tests/portfolio/test_runner_honest.py`: `build_worker_for` ritorna il tipo giusto per ogni kind con capitale = alloc; e `run()` con PORT06 non lascia più sleeve "saltati" (mocka il fetch o testa solo build). + +- [ ] **Step 5:** `uv run pytest tests/portfolio/ -m "not network" -v` → tutti verdi. + +- [ ] **Step 6: Commit** +```bash +git add src/portfolio/runner.py scripts/portfolios/_defs.py src/portfolio/base.py tests/portfolio/test_runner_honest.py +git commit -m "feat(portfolio): integra worker honest/TSM01 nel runner (PORT06 live completo)" +``` + +--- + +## Task 6: Validazione replay==backtest per i worker multi-asset + +**Files:** Modify `scripts/analysis/validate_portfolio_runner.py` (o nuovo `validate_honest_workers.py`). + +- [ ] **Step 1:** Per ogni worker multi-asset, replay bar-by-bar su dati storici (load_data) e confronto dell'equity finale con la funzione di riferimento (`_tr_basket_daily`, `_rot_daily_equity`, `tsmom_sim`) entro tolleranza. ROT02/TSM01 sono daily → replay veloce (poche migliaia di barre). TR01 4h → medio. Atteso: match stretto (differenze solo da bar-timing/cadenza). DIP01 ha il gap intrabar noto come le fade (documenta, non assert esatto). + +- [ ] **Step 2: Commit** +```bash +git add scripts/analysis/validate_honest_workers.py +git commit -m "test(portfolio): replay worker honest/TSM01 == backtest di riferimento" +``` + +--- + +## Self-review + +- **Copertura:** i 4 worker (DIP01 single via Strategy; TR01/ROT02/TSM01 dedicati) + integrazione runner + validazione → PORT06 gira live completo (niente più sleeve saltati). +- **Parità backtest:** invariata (gli sleeve del backtest vengono ancora da `build_everything`; i worker sono il path LIVE). La validazione replay==backtest (Task 6) certifica i worker live. +- **Gap noto:** DIP01, come le fade, ha exit intrabar nel backtest ma close-based nel live → gap strutturale documentato (non un bug). TR01/ROT02/TSM01 non hanno TP/SL intrabar (entry/exit a chiusura barra/giorno) → replay atteso stretto. +- **Tipi:** i nuovi worker espongono `.capital` e `.in_position` (richiesti da `_worker_equity`/`rebalance_allocations`); `tick(data: dict)` per i multi-asset vs `tick(df)`/`tick(dfa,dfb)` esistenti → il runner dispatcha per `kind`. +- **Rischio:** la convenzione di capitale/rendimento dei worker multi-asset deve combaciare con le funzioni di riferimento; la validazione Task 6 è il gate che lo verifica — se diverge, allineare la formula (non la reference). + +> **Punto aperto:** verificare la disponibilità su Cerbero v2 dei timeframe 4h/1d per tutti gli asset dell'universo (TR01 usa 4h; ROT02/TSM01 usano 1d, oggi resample da 1h in get_df). Il runner live dovrà resamplare 1h→4h/1d dal feed v2 o fetchare nativamente — da decidere in Task 5/Step 3.