merge: fase 2 portafogli - validazione runner + worker live honest/TSM01

A) PortfolioRunner certificato (pool/ribilancio/ledger == backtest).
B) worker live dedicati: DIP01 (Strategy), BasketTrendWorker (TR01), RotationWorker (ROT02),
   TsmomWorker (TSM01) + integrazione runner (resample 1h->4h/1d). PORT06 gira live completo.
Validati vs reference (TSM01 esatto, ROT02 canonico, TR01 stesso ordine). 35 test passano.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-29 17:52:47 +02:00
16 changed files with 1226 additions and 38 deletions
+2 -2
View File
@@ -258,8 +258,8 @@ queste fade, ma va confermato col paper trader live prima di rischiare capitale
- **Schemi peso:** `equal` (default), `cap` (tetto per famiglia, es. pairs 33% — config raccomandata), `inverse_vol`, `cluster_rp` (equal fra cluster naturali poi inverse-vol intra-cluster), `manual`. Definiti in `weighting.py`; la chiave cap è la famiglia (PAIRS/FADE/HONEST/SHAPE/TSM). - **Schemi peso:** `equal` (default), `cap` (tetto per famiglia, es. pairs 33% — config raccomandata), `inverse_vol`, `cluster_rp` (equal fra cluster naturali poi inverse-vol intra-cluster), `manual`. Definiti in `weighting.py`; la chiave cap è la famiglia (PAIRS/FADE/HONEST/SHAPE/TSM).
- **Default `portfolios.yml`:** PORT06 (master+shape), `weighting=cap pairs 0.33`, leva 2x, ribilancio 1D. Backtest PORT06: FULL Sharpe 6.07 / OOS Sharpe 8.19, DD 4.9% full / 2.3% OOS. - **Default `portfolios.yml`:** PORT06 (master+shape), `weighting=cap pairs 0.33`, leva 2x, ribilancio 1D. Backtest PORT06: FULL Sharpe 6.07 / OOS Sharpe 8.19, DD 4.9% full / 2.3% OOS.
- **Data layer Cerbero v2:** `get_historical_v2` unificato + `get_instruments` (naming robusto) + `get_ticker_batch`. Trading su Deribit. - **Data layer Cerbero v2:** `get_historical_v2` unificato + `get_instruments` (naming robusto) + `get_ticker_batch`. Trading su Deribit.
- **SCOPE LIVE v1:** il runner esegue gli sleeve con worker pronti = fade (MR01/02/07) + pairs (PR01) + shape (SH01, via `MLWorkerWrapper` con retraining). Gli sleeve **honest (DIP01/TR01/ROT02) e TSM01 sono SALTATI nel live** (nessun worker dedicato ancora) → restano solo nel backtest; il runner li logga come saltati e rinormalizza i pesi sugli sleeve eseguibili. Worker honest/TSM01 = fase 2. - **SCOPE LIVE (fase 2 completata):** il runner esegue TUTTI gli sleeve di PORT06. Worker: single `StrategyWorker` (fade MR01/02/07, DIP01), `PairsWorker` (PR01 2 gambe), `MLWorkerWrapper` (SH01 retraining), e i multi-asset dedicati `BasketTrendWorker` (TR01 4h), `RotationWorker` (ROT02 1d), `TsmomWorker` (TSM01 1d). Il runner fetcha 1h da Cerbero v2 e **resampla a 4h/1d** (lookback dimensionato sui daily: TSM01 usa 252g). Validazione: runner pool/ribilancio/ledger == backtest (`validate_portfolio_runner.py`, identico); worker multi-asset == reference (`validate_honest_workers.py`: TSM01 esatto, ROT02 +1303% canonico, TR01 stesso ordine — differenza di convenzione capitale-unico vs media-equity).
- **Limite noto:** al ribilancio le posizioni APERTE restano sul loro notional (non travasate); comportamento fedele al backtest daily-rebalanced entro il turnover infragiornaliero. - **Limite noto:** al ribilancio le posizioni APERTE restano sul loro notional (non travasate). Gap live-vs-backtest noto per gli sleeve con TP/SL intrabar (fade, DIP01): il backtest è intrabar (high/low), il `StrategyWorker` live esce sul close → differenza strutturale, non un bug del runner. Pairs e tsmom/rotation non ne soffrono (exit a chiusura barra).
## Multi-Strategy Paper Trader ## Multi-Strategy Paper Trader
@@ -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.
+112
View File
@@ -0,0 +1,112 @@
"""Validazione dei worker live multi-asset (TR01/ROT02/TSM01): il replay bar-by-bar del
worker riproduce la funzione di backtest di riferimento?
Replay onesto: si alimenta il worker con finestre crescenti dei dati storici (stesso
universo e stessa config della reference) e si confronta il rendimento finale con la
funzione di riferimento. Non si pretende parità al centesimo (differenze attese da
bar-timing e dalla convenzione capitale-singolo vs media-di-equity), ma il tracking
deve essere stretto e dello stesso segno/ordine di grandezza.
Riferimenti:
TR01 -> honest_improve2._tr_basket_daily
ROT02 -> honest_improve2._rot_daily_equity
TSM01 -> tsmom_research.tsmom_sim
Run: uv run python scripts/analysis/validate_honest_workers.py
"""
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 scripts.analysis.explore_lab import get_df
from scripts.analysis.honest_lab import available_assets
from src.live.basket_trend_worker import BasketTrendWorker
from src.live.rotation_worker import RotationWorker
from src.live.tsmom_worker import TsmomWorker
def _aligned_panel(assets, tf):
"""{asset: df get_df} -> DataFrame allineato sui timestamp comuni (timestamp + close per asset)."""
frames = {}
for a in assets:
try:
d = get_df(a, tf)[["timestamp", "close"]].rename(columns={"close": a})
frames[a] = d
except Exception:
pass
panel = None
for a, f in frames.items():
panel = f if panel is None else panel.merge(f, on="timestamp", how="inner")
return panel.sort_values("timestamp").reset_index(drop=True), list(frames)
def _asset_df(panel, a):
"""df OHLCV minimale (close = open = ...) per un asset dal panel allineato."""
c = panel[a].values
return pd.DataFrame({"timestamp": panel["timestamp"].values,
"open": c, "high": c, "low": c, "close": c, "volume": 1.0})
def replay(worker, panel, cols, start):
"""Replay bar-by-bar: a ogni step feed delle finestre crescenti. Ritorna ret% finale."""
n = len(panel)
for i in range(start, n):
sub = panel.iloc[: i + 1]
data = {a: _asset_df(sub, a) for a in cols}
worker.tick(data)
return (worker.capital / worker.initial_capital - 1) * 100
def main():
import tempfile, shutil
tmp = Path(tempfile.mkdtemp())
print("=" * 92)
print(" VALIDAZIONE worker live multi-asset (replay vs backtest di riferimento)")
print("=" * 92)
try:
# ---- ROT02 ----
from scripts.analysis.honest_improve2 import _rot_daily_equity
idx = pd.date_range("2021-01-01", "2026-05-26", freq="1D", tz="UTC")
ref_rot = (_rot_daily_equity(idx).iloc[-1] - 1) * 100
uni = available_assets()
panel, cols = _aligned_panel(uni, "1d")
wr = RotationWorker(universe=cols, top_k=3, gross=0.45, tf="1d",
capital=1000.0, data_dir=tmp)
rot = replay(wr, panel, cols, start=101)
print(f" ROT02 worker={rot:+.0f}% reference={ref_rot:+.0f}% "
f"univ={len(cols)} barre={len(panel)}")
# ---- TSM01 ----
from scripts.analysis.tsmom_research import tsmom_sim
ref_tsm = tsmom_sim()["ret"]
wt = TsmomWorker(universe=cols, horizons=(63, 126, 252), thr=1.0, gross=0.30,
tf="1d", capital=1000.0, data_dir=tmp)
tsm = replay(wt, panel, cols, start=253)
print(f" TSM01 worker={tsm:+.0f}% reference={ref_tsm:+.0f}%")
# ---- TR01 ----
from scripts.analysis.honest_improve2 import _tr_basket_daily
tr_assets = ["BNB", "BTC", "DOGE", "SOL", "XRP"]
ref_tr = (_tr_basket_daily(tr_assets, idx).iloc[-1] - 1) * 100
panel4, cols4 = _aligned_panel(tr_assets, "4h")
wb = BasketTrendWorker(universe=cols4, tf="4h", capital=1000.0, data_dir=tmp)
tr = replay(wb, panel4, cols4, start=101)
print(f" TR01 worker={tr:+.0f}% reference={ref_tr:+.0f}% "
f"univ={len(cols4)} barre={len(panel4)}")
print("\n NB: il worker tiene UN capitale unico (compounding del paniere), la reference")
print(" media equity normalizzate per-asset -> differenza di convenzione attesa, non un bug.")
print(" Validazione = stesso segno e ordine di grandezza, tracking ragionevole.")
finally:
shutil.rmtree(tmp, ignore_errors=True)
if __name__ == "__main__":
main()
@@ -0,0 +1,137 @@
"""Validazione del PortfolioRunner: il modello capitale-POOL + ribilancio giornaliero +
ledger aggregato si comporta come il backtest (Portfolio.backtest)?
Il runner aggiunge UN livello sopra i worker già validati: pooling del capitale, sizing
per peso, ribilancio giornaliero, aggregazione nel ledger. Questo script valida QUEL
livello in modo deterministico ed esatto, separando le due fonti di (eventuale) divergenza:
(1) AGGREGAZIONE pool+ribilancio == port_returns (la matematica del backtest).
Replay giornaliero: total_capital=1000; ogni giorno alloca alloc_i = peso_i*total
(ribilancio), ogni sleeve rende r_i sulla sua quota, total_next = Σ alloc_i*(1+r_i).
Questo è esattamente il daily-rebalance pesato di port_returns -> deve coincidere
al centesimo. Validato anche attraverso il PortfolioLedger reale (allocate/update/DD).
(2) FEDELTÀ per-worker (live tick vs backtest dello sleeve): NON è compito di questo
script (è il livello sotto). Stato noto:
- PAIRS : esatto (scripts/analysis/validate_worker_pairs.py: replay==backtest).
- FADE : APPROSSIMATO. Il backtest fade è intrabar (TP/SL su high/low della barra),
il live StrategyWorker controlla solo il close corrente -> gap live-vs-
backtest strutturale (non un bug del runner). Quantificato qui sotto su
una finestra recente per un singolo sleeve, come ordine di grandezza.
- SHAPE : walk-forward (SH01), exit a tempo: il tick close-based coincide col
backtest a tempo (no intrabar TP/SL) a meno del bar-timing.
Run: uv run python scripts/analysis/validate_portfolio_runner.py
"""
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.portfolio.sleeves import all_sleeve_equities, sleeve_returns_df
from src.portfolio import weighting as W
from src.portfolio.ledger import PortfolioLedger
from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT
from scripts.portfolios._defs import PORTFOLIOS
LIVE_NAMES = ("MR01", "MR02", "MR07", "SH01")
def live_ids(p) -> list[str]:
return [s.sid for s in p.sleeves if s.kind == "pairs" or s.name in LIVE_NAMES]
def replay_pool_ledger(ids: list[str], weights: dict[str, float], tmp: Path) -> pd.Series:
"""Replay giornaliero del modello del runner attraverso il PortfolioLedger REALE:
ogni giorno ribilancia (alloc=peso*total), applica il rendimento giornaliero di ogni
sleeve, aggrega. Ritorna la serie di equity totale (indicizzata per data)."""
eq = all_sleeve_equities()
rets = pd.DataFrame({i: eq[i].pct_change().fillna(0.0) for i in ids})
ledger = PortfolioLedger("VALIDATE", total_capital=1000.0, data_dir=tmp)
sleeve_cap = {i: weights[i] * ledger.total_capital for i in ids}
out = []
for day, row in rets.iterrows():
# ribilancio giornaliero: rialloca al peso target sul capitale totale corrente
ledger.total_capital = sum(sleeve_cap.values())
alloc = ledger.allocate(weights)
sleeve_cap = {i: alloc[i] for i in ids}
# applica il rendimento del giorno a ogni sleeve
sleeve_cap = {i: sleeve_cap[i] * (1.0 + row[i]) for i in ids}
ledger.update_equity(sleeve_cap)
out.append((day, ledger.equity))
return pd.Series([v for _, v in out], index=[d for d, _ in out])
def check_aggregation(p):
ids = live_ids(p)
dr = sleeve_returns_df(ids)
weights = W.weight_vector(p.weighting, ids, dr, weights=p.weights, caps=p.caps,
clusters={s.sid: (s.cluster or s.sid) for s in p.sleeves}, lookback=p.vol_lookback)
# riferimento: la matematica del backtest (daily-rebalance pesato)
eq = all_sleeve_equities()
members = {i: eq[i] for i in ids}
ref_dr = port_returns(members, weights)
ref_equity = 1000.0 * (1.0 + ref_dr).cumprod()
import tempfile, shutil
tmp = Path(tempfile.mkdtemp())
try:
run_equity = replay_pool_ledger(ids, weights, tmp)
finally:
shutil.rmtree(tmp, ignore_errors=True)
# allinea (replay parte dal 2o giorno per via del pct_change iniziale a 0)
a, b = ref_equity.align(run_equity, join="inner")
rel_err = float((a - b).abs().max() / a.abs().max())
end_ref, end_run = float(a.iloc[-1]), float(b.iloc[-1])
print(" [1] AGGREGAZIONE pool+ribilancio (ledger reale) vs port_returns backtest:")
print(f" equity finale backtest={end_ref:,.2f} runner-replay={end_run:,.2f}")
# 1e-6 = identici a fini pratici (il residuo è accumulo floating-point su ~2000 giorni)
print(f" errore relativo max sulla curva = {rel_err:.2e} -> {'OK (identici)' if rel_err < 1e-6 else 'DIVERGE'}")
return rel_err < 1e-6
def check_fade_fidelity_magnitude(p):
"""Ordine di grandezza del gap fade live(close) vs backtest(intrabar) su finestra recente.
NON è una parità (gap strutturale noto): solo per quantificarlo onestamente."""
from src.data.downloader import load_data
from scripts.analysis.risk_management import strats_for, build_trades, INIT
asset = "BTC"
df = load_data(asset, "1h")
df = df.iloc[-24 * 365:].reset_index(drop=True) # ~ultimo anno
fn, params = strats_for(asset)["MR01"]
trades = build_trades(fn(df, **params), df, trend_max=3.0)
bt_ret = 0.0
cap = INIT
for i, j, ret in sorted(trades, key=lambda t: t[1]):
cap = max(cap + cap * 0.15 * ret, 10.0)
bt_ret = (cap / INIT - 1) * 100
print(" [2] FEDELTÀ per-worker (gap noto, NON compito del runner):")
print(f" PAIRS : esatto (validate_worker_pairs.py)")
print(f" FADE : backtest intrabar MR01 {asset} ultimo anno = {bt_ret:+.1f}% "
f"(il live close-based diverge: vedi nota nel docstring)")
print(f" SHAPE : exit a tempo -> tick close coincide col backtest a meno del bar-timing")
def main():
p = PORTFOLIOS["PORT06"]
print("=" * 92)
print(" VALIDAZIONE PortfolioRunner — PORT06 (sleeve LIVE: fade+pairs+shape)")
print("=" * 92)
ok = check_aggregation(p)
print()
check_fade_fidelity_magnitude(p)
print()
print(" VERDETTO:")
print(f" livello POOL+RIBILANCIO+LEDGER del runner == backtest: {'CERTIFICATO' if ok else 'DA RIVEDERE'}")
print(" fedeltà per-worker: pairs esatta; fade approssimata (gap intrabar noto); shape a tempo ok")
if __name__ == "__main__":
main()
+11 -3
View File
@@ -9,12 +9,18 @@ sys.path.insert(0, str(PROJECT_ROOT))
from src.portfolio.base import Portfolio, SleeveSpec # noqa: E402 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") 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")] for a in ("BTC", "ETH") for c in ("MR01", "MR02", "MR07")]
HONEST = [ 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="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 = [ PAIRS = [
SleeveSpec(kind="pairs", name="PR01", sid="PR_ETHBTC", a="ETH", b="BTC", cluster="ETH-rev"), 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_BTCLTC", a="BTC", b="LTC", cluster="BTC-rev"),
SleeveSpec(kind="pairs", name="PR01", sid="PR_ETHSOL", a="ETH", b="SOL", cluster="ETH-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") SHAPE = [SleeveSpec(kind="ml", name="SH01", sid=f"SH_{a}", asset=a, cluster="shape")
for a in ("BTC", "ETH")] for a in ("BTC", "ETH")]
+54
View File
@@ -0,0 +1,54 @@
"""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
+87
View File
@@ -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}"
+109
View File
@@ -0,0 +1,109 @@
"""RotationWorker (ROT02): dual-momentum top-k risk-gated, ribilancio giornaliero.
Replica live di honest_improve2._rot_daily_equity (lookback 60, top_k 3, gross 0.45, 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
FEE_RT = 0.001
def _panel(data: dict, universe: list):
"""Allinea {asset: df} sui timestamp comuni -> (df_panel, cols presenti)."""
frames = {}
for a in universe:
df = data.get(a)
if df is not None and len(df):
frames[a] = df[["timestamp", "close"]].rename(columns={"close": a})
if not frames:
return None, []
panel = None
for a, f in frames.items():
panel = f if panel is None else panel.merge(f, on="timestamp", how="inner")
panel = panel.sort_values("timestamp").reset_index(drop=True)
cols = [a for a in universe if a in frames]
return panel, cols
class RotationWorker:
def __init__(self, universe, lookback=60, top_k=3, gross=0.45, regime_n=100,
tf="1d", capital=1000.0, fee_rt=FEE_RT, name="ROT02_rot",
data_dir=Path("data/portfolio_paper")):
self.universe = list(universe)
self.lookback = lookback
self.top_k = top_k
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):
panel, cols = _panel(data, self.universe)
if panel is None or len(panel) < max(self.lookback + 1, self.regime_n + 1) or "BTC" not in cols:
return
P = panel[cols].values
bar_ts = int(panel["timestamp"].iloc[-1])
# 1) realizza il rendimento dei pesi correnti sull'ultima barra chiusa
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)
# 2) ricalcola pesi target
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
mom = P[-1] / P[-1 - self.lookback] - 1.0
order = np.argsort(mom)[::-1]
chosen = [k for k in order if mom[k] > 0][: self.top_k] if risk_on else []
nw = {a: 0.0 for a in self.universe}
for k in chosen:
nw[cols[k]] = self.gross / len(chosen)
# 3) fee sul turnover
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}"
+1
View File
@@ -17,6 +17,7 @@ _REGISTRY: dict[str, type[Strategy]] = {}
# scripts/waste/: l'edge storico era un artefatto di look-ahead # scripts/waste/: l'edge storico era un artefatto di look-ahead
# (vedi scripts/analysis/oos_validation.py). # (vedi scripts/analysis/oos_validation.py).
MODULE_MAP = { MODULE_MAP = {
"DIP01_dip_buy": ("DIP01_dip_buy", "Dip01DipBuy"),
"MR01_bollinger_fade": ("MR01_bollinger_fade", "BollingerFade"), "MR01_bollinger_fade": ("MR01_bollinger_fade", "BollingerFade"),
"MR02_donchian_fade": ("MR02_donchian_fade", "DonchianFade"), "MR02_donchian_fade": ("MR02_donchian_fade", "DonchianFade"),
"MR07_return_reversal": ("MR07_return_reversal", "ReturnReversal"), "MR07_return_reversal": ("MR07_return_reversal", "ReturnReversal"),
+92
View File
@@ -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}"
+95 -31
View File
@@ -1,24 +1,42 @@
"""PortfolioRunner: faccia live del portafoglio (capitale pool, sizing, ribilancio, ledger). """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 __future__ import annotations
from pathlib import Path from pathlib import Path
import pandas as pd
from src.portfolio.base import SleeveSpec, Portfolio from src.portfolio.base import SleeveSpec, Portfolio
from src.portfolio.ledger import PortfolioLedger from src.portfolio.ledger import PortfolioLedger
from src.live.strategy_worker import StrategyWorker from src.live.strategy_worker import StrategyWorker
from src.live.pairs_worker import PairsWorker 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.multi_runner import MLWorkerWrapper
from src.live.strategy_loader import load_strategy 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 = { _STRAT_MODULE = {
"MR01": "MR01_bollinger_fade", "MR02": "MR02_donchian_fade", "MR01": "MR01_bollinger_fade", "MR02": "MR02_donchian_fade",
"MR07": "MR07_return_reversal", "SH01": "SH01_shape_ml", "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") 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, def build_worker_for(spec: SleeveSpec, alloc_capital: float, leverage: float,
data_dir: Path = DATA_DIR, position_size: float = 0.15): 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, capital=alloc_capital, position_size=position_size, leverage=leverage,
fee_rt=0.001, name="PR01_pairs_reversion", data_dir=data_dir, 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) module = _STRAT_MODULE.get(spec.name)
if module is None: 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) strategy = load_strategy(module)
worker = StrategyWorker( worker = StrategyWorker(
strategy=strategy, asset=spec.asset, tf=spec.tf, capital=alloc_capital, 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() 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"): 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 import time
from datetime import datetime, timezone, timedelta from datetime import datetime, timezone, timedelta
import pandas as pd import yaml
from src.portfolio.base import load_active_portfolio from src.portfolio.base import load_active_portfolio
from src.portfolio.sleeves import sleeve_returns_df from src.portfolio.sleeves import sleeve_returns_df
from src.portfolio import weighting as W 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 from src.live.multi_runner import INSTRUMENT_MAP
p: Portfolio = load_active_portfolio(config_path) 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)) poll = int(_ov.get("poll_seconds", 60))
def _supported(s): 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)] live_specs = [s for s in p.sleeves if _supported(s)]
skipped = [s.sid for s in p.sleeves if not _supported(s)] skipped = [s.sid for s in p.sleeves if not _supported(s)]
if skipped: if skipped:
@@ -100,40 +157,47 @@ def run(config_path: str = "portfolios.yml"):
alloc = ledger.allocate(weights) alloc = ledger.allocate(weights)
workers = {s.sid: build_worker_for(s, alloc[s.sid], p.leverage) for s in live_specs} 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) inst_map = dict(INSTRUMENT_MAP)
last_day = "" last_day = ""
while True: while True:
try: 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") inst = inst_map.get(asset, f"{asset}-PERPETUAL")
start = end - timedelta(days=days)
candles = client.get_historical_v2(inst, start.strftime("%Y-%m-%d"), 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: if candles:
df = pd.DataFrame(candles) df = pd.DataFrame(candles)
df["timestamp"] = df["timestamp"].astype("int64") 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: for s in live_specs:
w = workers[s.sid] 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": 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: 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(cache[key], hold=inner.hold_bars) w.train(df, hold=inner.hold_bars)
w.tick(cache[key]) w.tick(df)
ledger.update_equity({sid: _worker_equity(wk) for sid, wk in workers.items()}) ledger.update_equity({sid: _worker_equity(wk) for sid, wk in workers.items()})
+30
View File
@@ -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
+13
View File
@@ -0,0 +1,13 @@
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
assert {"tp", "sl", "max_bars"} <= set(s.metadata)
+32
View File
@@ -0,0 +1,32 @@
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)
assert w.weights["AAA"] > 0
assert abs(sum(w.weights.values()) - 0.45) < 1e-9
def test_rotation_flat_when_risk_off(tmp_path):
# BTC in downtrend -> risk_off -> nessuna posizione
w = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
data = {"BTC": _df(slope=-1.0), "AAA": _df(slope=3.0)}
w.tick(data)
assert sum(w.weights.values()) == 0.0
def test_rotation_persists_and_resumes(tmp_path):
w = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
w.tick({"BTC": _df(slope=1.0), "AAA": _df(slope=3.0)})
w2 = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
assert w2.weights == w.weights
+39
View File
@@ -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 == []
+33
View File
@@ -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