diff --git a/scripts/analysis/smoke_portfolio.py b/scripts/analysis/smoke_portfolio.py new file mode 100644 index 0000000..f9d7a89 --- /dev/null +++ b/scripts/analysis/smoke_portfolio.py @@ -0,0 +1,39 @@ +"""Smoke reale: un giro di fetch v2 + build worker + un tick del portafoglio attivo. +NON apre ordini reali (paper). Verifica data layer v2 + sizing + ledger.""" +import sys, shutil, tempfile +from pathlib import Path +from datetime import datetime, timezone, timedelta +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.portfolio.base import load_active_portfolio +from src.portfolio.ledger import PortfolioLedger +from src.portfolio.runner import build_worker_for, _worker_equity +from src.live.cerbero_client import CerberoClient +from src.live.multi_runner import INSTRUMENT_MAP + + +def main(): + tmp = Path(tempfile.mkdtemp()) + p = load_active_portfolio(PROJECT_ROOT / "portfolios.yml") + ledger = PortfolioLedger(p.code, total_capital=p.total_capital, data_dir=tmp) + alloc = ledger.allocate({s.sid: 1.0 / len(p.sleeves) for s in p.sleeves}) + client = CerberoClient() + print(f"Portafoglio attivo: {p.code} ({p.label}) — {len(p.sleeves)} sleeve, leva {p.leverage}x") + end = datetime.now(timezone.utc); start = end - timedelta(days=60) + ok = 0 + for s in p.sleeves[:3]: + asset = s.asset or s.a + inst = INSTRUMENT_MAP.get(asset, f"{asset}-PERPETUAL") + candles = client.get_historical_v2(inst, start.strftime("%Y-%m-%d"), + end.strftime("%Y-%m-%d"), s.tf) + print(f" {s.sid:<12s} {inst:<18s} candele={len(candles)}") + ok += len(candles) > 0 + print(f"OK: {ok}/3 sleeve con feed v2 fresco. Ledger equity iniziale={ledger.equity}") + shutil.rmtree(tmp, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/src/portfolio/runner.py b/src/portfolio/runner.py index 75020a9..9184d07 100644 --- a/src/portfolio/runner.py +++ b/src/portfolio/runner.py @@ -41,3 +41,111 @@ def build_worker_for(spec: SleeveSpec, alloc_capital: float, leverage: float, if spec.kind == "ml": # SH01: retraining periodico return MLWorkerWrapper(worker, {"retrain_hours": 24}) return worker + + +def _worker_equity(w) -> float: + inner = getattr(w, "worker", w) # smonta MLWorkerWrapper + return float(getattr(inner, "capital", 0.0)) + + +def rebalance_allocations(ledger: PortfolioLedger, workers: dict, weights: dict[str, float]): + """Ribilancio: total_capital = Σ equity sleeve; riallinea il capitale-base di ogni worker + a peso×total. Le posizioni APERTE restano sul loro notional (approssimazione dichiarata).""" + ledger.total_capital = sum(_worker_equity(w) for w in workers.values()) + alloc = ledger.allocate(weights) + for sid, w in workers.items(): + inner = getattr(w, "worker", w) + inner.capital = alloc.get(sid, inner.capital) + ledger.save() + + +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.""" + import time + from datetime import datetime, timezone, timedelta + import pandas as pd + from src.portfolio.base import load_active_portfolio + from src.portfolio.sleeves import sleeve_returns_df + from src.portfolio import weighting as W + from src.live.cerbero_client import CerberoClient + from src.live.multi_runner import INSTRUMENT_MAP + + p: Portfolio = load_active_portfolio(config_path) + + def _supported(s): + return s.kind == "pairs" 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: + print(f"[runner] sleeve saltati nel live (worker non disponibili): {skipped}") + live_ids = [s.sid for s in live_specs] + clusters = {s.sid: (s.cluster or s.sid) for s in live_specs} + + ledger = PortfolioLedger(p.code, total_capital=p.total_capital) + client = CerberoClient() + + dr = sleeve_returns_df(live_ids) + weights = W.weight_vector(p.weighting, live_ids, dr, weights=p.weights, + caps=p.caps, clusters=clusters, lookback=p.vol_lookback) + alloc = ledger.allocate(weights) + workers = {s.sid: build_worker_for(s, alloc[s.sid], p.leverage) for s in live_specs} + + inst_map = dict(INSTRUMENT_MAP) + last_day = "" + poll = 60 + 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: + inst = inst_map.get(asset, f"{asset}-PERPETUAL") + candles = client.get_historical_v2(inst, start.strftime("%Y-%m-%d"), + end.strftime("%Y-%m-%d"), tf) + if candles: + df = pd.DataFrame(candles) + df["timestamp"] = df["timestamp"].astype("int64") + cache[(asset, tf)] = df.sort_values("timestamp").reset_index(drop=True) + + for s in live_specs: + w = workers[s.sid] + 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]) + 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]) + + ledger.update_equity({sid: _worker_equity(wk) for sid, wk in workers.items()}) + + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + if today != last_day and last_day: + dr = sleeve_returns_df(live_ids) + weights = W.weight_vector(p.weighting, live_ids, dr, weights=p.weights, + caps=p.caps, clusters=clusters, lookback=p.vol_lookback) + rebalance_allocations(ledger, workers, weights) + last_day = today + ledger.save() + except KeyboardInterrupt: + ledger.save() + print("shutdown") + break + except Exception as e: + print(f"[runner] errore: {e}") + time.sleep(poll) + + +if __name__ == "__main__": + run() diff --git a/tests/portfolio/test_runner_rebalance.py b/tests/portfolio/test_runner_rebalance.py new file mode 100644 index 0000000..064184e --- /dev/null +++ b/tests/portfolio/test_runner_rebalance.py @@ -0,0 +1,13 @@ +from src.portfolio.runner import rebalance_allocations +from src.portfolio.ledger import PortfolioLedger + + +def test_rebalance_resizes_to_total(tmp_path): + L = PortfolioLedger("PX", total_capital=1000.0, data_dir=tmp_path) + + class FakeWorker: + def __init__(self, cap): self.capital = cap + workers = {"a": FakeWorker(700.0), "b": FakeWorker(500.0)} + rebalance_allocations(L, workers, {"a": 0.5, "b": 0.5}) + assert L.total_capital == 1200.0 + assert workers["a"].capital == 600.0 and workers["b"].capital == 600.0