Files

T

Adriano 753d786bb5 docs(portfolios): piano di implementazione TDD (10 task)

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

2026-05-29 15:29:41 +02:00

45 KiB

Raw Blame History

Cartella `portfolios/` — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Introdurre portafogli come oggetti di prima classe (capitale-pool condiviso) capaci sia di backtest/report sia di gestione live (sizing, ribilancio, ledger PnL), usando il data layer Cerbero v2.

Architecture: Una classe Portfolio (definizione: sleeve + schema pesi) con due facce sulla stessa definizione: backtest() riusa l'unico builder di equity-per-sleeve esistente (parità per costruzione col report); il live (PortfolioRunner) costruisce i worker esistenti come esecutori, alloca peso×capitale, ribilancia giornalmente e aggrega nel PortfolioLedger. Codice nuovo in src/portfolio/, definizioni concrete in scripts/portfolios/, config live in portfolios.yml.

Tech Stack: Python 3.11, uv, pandas/numpy, scipy (clustering già usato), pytest, requests (Cerbero MCP v2). Riusa scripts/analysis/{combine_portfolio,report_families,pairs_research,tsmom_research,shape_ml_validate}.py e src/live/{multi_runner,strategy_worker,pairs_worker,cerbero_client}.py.

Spec di riferimento: docs/superpowers/specs/2026-05-29-portfolios-design.md

File structure

File	Responsabilità
`src/portfolio/__init__.py`	package marker
`src/portfolio/sleeves.py`	`all_sleeve_equities()` — unico builder di equity-per-sleeve (delega a `report_families.build_everything`), `sleeve_returns_df()`
`src/portfolio/weighting.py`	`family_of`, `equal`, `manual`, `cap`, `inverse_vol`, `cluster_rp`, `weight_vector`
`src/portfolio/base.py`	`SleeveSpec`, `PortfolioResult`, `Portfolio` (`.backtest()`, `.weight_vector()`)
`src/portfolio/ledger.py`	`PortfolioLedger` (capitale, alloc, equity, PnL, peak/DD, persistenza/resume)
`src/portfolio/runner.py`	`PortfolioRunner` (live: data v2, build worker, sizing, ribilancio, aggregazione)
`src/live/cerbero_client.py`	modifica: aggiunge metodi v2 `get_historical_v2`, `get_instruments`, `get_ticker_batch`
`scripts/portfolios/PORT0{1..6}_*.py`	definizioni concrete + `run()` report
`portfolios.yml`	config live (portafoglio attivo, capitale, pesi, cap, leva, cadenza)
`tests/portfolio/test_*.py`	unit + parità + smoke

Sleeve id canonici (devono combaciare con le chiavi di build_everything per la parità): MR01_BTC MR02_BTC MR07_BTC MR01_ETH MR02_ETH MR07_ETH · DIP01_BTC TR01_basket ROT02_rot · PR_ETHBTC PR_LTCETH PR_ADAETH PR_BTCLTC PR_ETHSOL · TSM01 · SH_BTC SH_ETH.

Task 1: Metodi Cerbero v2 nel client

Files:

Modify: src/live/cerbero_client.py
Test: tests/portfolio/test_cerbero_v2.py (richiede rete; marcato network)
Step 1: Crea il package di test

mkdir -p tests/portfolio && touch tests/portfolio/__init__.py

Step 2: Aggiungi i metodi v2 al client

In src/live/cerbero_client.py, dentro la classe CerberoClient, dopo get_historical (riga ~50) aggiungi:

    def get_historical_v2(self, instrument: str, start_date: str, end_date: str,
                          interval: str = "1h", exchange: str = "deribit") -> list[dict]:
        """Endpoint unificato v2: /mcp/tools/get_historical (exchange deribit|hyperliquid).
        Stesso shape candele del legacy: [{timestamp(ms), open, high, low, close, volume}]."""
        data = self._post("/mcp/tools/get_historical", {
            "exchange": exchange, "instrument": instrument,
            "interval": interval, "start_date": start_date, "end_date": end_date,
        })
        return data.get("candles", [])

    def get_instruments(self, currency: str, kind: str = "future",
                        exchange: str = "deribit", limit: int = 100) -> list[dict]:
        """Enumera gli strumenti reali (v2). Usato per risolvere il naming senza hardcoding."""
        data = self._post("/mcp/tools/get_instruments", {
            "exchange": exchange, "currency": currency, "kind": kind, "limit": limit,
        })
        return data.get("instruments", data if isinstance(data, list) else [])

    def get_ticker_batch(self, instruments: list[str]) -> dict:
        """Prezzi correnti di N strumenti in una sola chiamata (v2, Deribit)."""
        return self._post("/mcp-deribit/tools/get_ticker_batch", {"instruments": instruments})

Step 3: Scrivi lo smoke test di rete

tests/portfolio/test_cerbero_v2.py:

import pytest
from src.live.cerbero_client import CerberoClient


@pytest.mark.network
def test_get_historical_v2_shape():
    cli = CerberoClient()
    candles = cli.get_historical_v2("BTC-PERPETUAL", "2026-05-25", "2026-05-27", "1h")
    assert len(candles) > 0
    c0 = candles[0]
    assert {"timestamp", "open", "high", "low", "close", "volume"} <= set(c0)


@pytest.mark.network
def test_get_instruments_returns_list():
    cli = CerberoClient()
    inst = cli.get_instruments("ETH", "future")
    assert isinstance(inst, list) and len(inst) > 0

Step 4: Esegui (con rete)

Run: uv run pytest tests/portfolio/test_cerbero_v2.py -v -m network Expected: 2 passed (se la rete/token è disponibile). Senza rete: uv run pytest -m "not network" li salta.

Step 5: Registra il marker network

In pyproject.toml, sotto [tool.pytest.ini_options] (riga ~27) aggiungi:

markers = ["network: test che richiede Cerbero MCP (rete+token)"]

Step 6: Commit

git add src/live/cerbero_client.py tests/portfolio/ pyproject.toml
git commit -m "feat(portfolio): metodi Cerbero v2 (get_historical_v2, get_instruments, get_ticker_batch)"

Task 2: Schemi di peso (`weighting.py`)

Files:

Create: src/portfolio/__init__.py, src/portfolio/weighting.py
Test: tests/portfolio/test_weighting.py
Step 1: Crea il package

mkdir -p src/portfolio && touch src/portfolio/__init__.py

Step 2: Scrivi i test (falliscono)

tests/portfolio/test_weighting.py:

import numpy as np
import pandas as pd
import pytest
from src.portfolio import weighting as W


def test_family_of():
    assert W.family_of("PR_ETHBTC") == "PAIRS"
    assert W.family_of("SH_BTC") == "SHAPE"
    assert W.family_of("TSM01") == "TSM"
    assert W.family_of("MR01_BTC") == "FADE"
    assert W.family_of("DIP01_BTC") == "HONEST"


def test_equal_sums_to_one():
    w = W.equal(["a", "b", "c", "d"])
    assert pytest.approx(sum(w.values())) == 1.0
    assert all(abs(v - 0.25) < 1e-9 for v in w.values())


def test_manual_normalizes():
    w = W.manual(["a", "b"], {"a": 3, "b": 1})
    assert pytest.approx(w["a"]) == 0.75 and pytest.approx(w["b"]) == 0.25


def test_cap_limits_family_and_redistributes():
    ids = ["PR_ETHBTC", "PR_LTCETH", "MR01_BTC", "MR02_BTC"]
    w = W.cap(ids, caps={"PAIRS": 0.30})
    pairs_w = w["PR_ETHBTC"] + w["PR_LTCETH"]
    assert pytest.approx(pairs_w, abs=1e-9) == 0.30          # cap rispettato
    assert pytest.approx(sum(w.values())) == 1.0             # resto ridistribuito
    assert w["MR01_BTC"] > 0.25                              # non-pairs sovrappesati


def test_inverse_vol_prefers_low_vol():
    idx = pd.date_range("2024-01-01", periods=100, freq="D", tz="UTC")
    rng = np.random.default_rng(0)
    df = pd.DataFrame({"lo": rng.normal(0, 0.01, 100), "hi": rng.normal(0, 0.05, 100)}, index=idx)
    w = W.inverse_vol(["lo", "hi"], df, lookback=90)
    assert w["lo"] > w["hi"]
    assert pytest.approx(sum(w.values())) == 1.0

Step 3: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_weighting.py -v Expected: FAIL (ModuleNotFoundError: weighting).

Step 4: Implementa weighting.py

"""Schemi di peso per i portafogli. Ogni funzione ritorna {sleeve_id: peso} con somma 1."""
from __future__ import annotations

import numpy as np
import pandas as pd

_PREFIX = [("PR_", "PAIRS"), ("SH_", "SHAPE"), ("TSM", "TSM"), ("MR", "FADE")]


def family_of(sleeve_id: str) -> str:
    for pre, fam in _PREFIX:
        if sleeve_id.startswith(pre):
            return fam
    return "HONEST"


def _normalize(w: dict[str, float]) -> dict[str, float]:
    tot = sum(w.values())
    return {k: (v / tot if tot > 0 else 0.0) for k, v in w.items()}


def equal(ids: list[str]) -> dict[str, float]:
    n = len(ids)
    return {i: 1.0 / n for i in ids} if n else {}


def manual(ids: list[str], weights: dict[str, float]) -> dict[str, float]:
    return _normalize({i: float(weights.get(i, 0.0)) for i in ids})


def cap(ids: list[str], caps: dict[str, float]) -> dict[str, float]:
    """Equal-weight con tetto al peso AGGREGATO di una famiglia; l'eccesso ridistribuito
    pro-quota alle famiglie non cappate (iterativo finché tutti i cap sono rispettati)."""
    w = equal(ids)
    fam = {i: family_of(i) for i in ids}
    for _ in range(10):
        over = {}
        for f, lim in caps.items():
            members = [i for i in ids if fam[i] == f]
            cur = sum(w[i] for i in members)
            if cur > lim + 1e-12 and members:
                over[f] = (members, lim, cur)
        if not over:
            break
        free_ids = [i for i in ids if fam[i] not in caps]
        freed = 0.0
        for f, (members, lim, cur) in over.items():
            scale = lim / cur
            for i in members:
                freed += w[i] * (1 - scale)
                w[i] *= scale
        if free_ids and freed > 0:
            add = freed / len(free_ids)
            for i in free_ids:
                w[i] += add
        else:
            break
    return _normalize(w)


def inverse_vol(ids: list[str], returns_df: pd.DataFrame, lookback: int) -> dict[str, float]:
    sub = returns_df[ids].iloc[-lookback:]
    vol = sub.std()
    inv = {i: (1.0 / vol[i] if vol[i] and vol[i] > 0 else 0.0) for i in ids}
    return _normalize(inv)


def cluster_rp(ids: list[str], clusters: dict[str, str],
               returns_df: pd.DataFrame, lookback: int) -> dict[str, float]:
    """Equal fra i cluster naturali, poi inverse-vol dentro ogni cluster."""
    groups: dict[str, list[str]] = {}
    for i in ids:
        groups.setdefault(clusters.get(i, i), []).append(i)
    per = 1.0 / len(groups) if groups else 0.0
    w: dict[str, float] = {}
    for members in groups.values():
        iv = inverse_vol(members, returns_df, lookback)
        for i in members:
            w[i] = per * iv[i]
    return _normalize(w)


def weight_vector(scheme: str, ids: list[str], returns_df: pd.DataFrame | None = None,
                  *, weights: dict | None = None, caps: dict | None = None,
                  clusters: dict | None = None, lookback: int = 90) -> dict[str, float]:
    if scheme == "equal":
        return equal(ids)
    if scheme == "manual":
        return manual(ids, weights or {})
    if scheme == "cap":
        return cap(ids, caps or {})
    if scheme == "inverse_vol":
        return inverse_vol(ids, returns_df, lookback)
    if scheme == "cluster_rp":
        return cluster_rp(ids, clusters or {}, returns_df, lookback)
    raise ValueError(f"schema peso sconosciuto: {scheme}")

Step 5: Run — verifica passaggio

Run: uv run pytest tests/portfolio/test_weighting.py -v Expected: 5 passed.

Step 6: Commit

git add src/portfolio/__init__.py src/portfolio/weighting.py tests/portfolio/test_weighting.py
git commit -m "feat(portfolio): schemi di peso (equal/manual/cap/inverse_vol/cluster_rp)"

Task 3: Builder unificato delle equity-per-sleeve (`sleeves.py`)

Files:

Create: src/portfolio/sleeves.py
Test: tests/portfolio/test_sleeves.py
Step 1: Scrivi il test (fallisce)

tests/portfolio/test_sleeves.py:

import pandas as pd
from src.portfolio import sleeves as S

ALL_IDS = {"MR01_BTC", "MR02_BTC", "MR07_BTC", "MR01_ETH", "MR02_ETH", "MR07_ETH",
           "DIP01_BTC", "TR01_basket", "ROT02_rot",
           "PR_ETHBTC", "PR_LTCETH", "PR_ADAETH", "PR_BTCLTC", "PR_ETHSOL",
           "TSM01", "SH_BTC", "SH_ETH"}


def test_all_sleeve_equities_keys_and_index():
    eq = S.all_sleeve_equities()
    assert ALL_IDS <= set(eq)
    s = eq["MR01_BTC"]
    assert isinstance(s, pd.Series) and len(s) > 100
    assert str(s.index.tz) == "UTC"


def test_returns_df_aligned():
    df = S.sleeve_returns_df(["MR01_BTC", "PR_ETHBTC", "SH_BTC"])
    assert list(df.columns) == ["MR01_BTC", "PR_ETHBTC", "SH_BTC"]
    assert df.isna().sum().sum() == 0

Step 2: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_sleeves.py -v Expected: FAIL (ModuleNotFoundError: sleeves).

Step 3: Implementa sleeves.py

"""Unico builder delle equity GIORNALIERE per sleeve (fonte di verità del backtest).

Delega a scripts/analysis/report_families.build_everything (che a sua volta usa
combine_portfolio + pairs_research + tsmom_research + shape_ml_validate), così le
metriche del Portfolio coincidono per costruzione con report_families."""
from __future__ import annotations

import pandas as pd

_CACHE: dict[str, pd.Series] | None = None


def all_sleeve_equities() -> dict[str, pd.Series]:
    """{sleeve_id: equity giornaliera normalizzata su IDX comune}. Cache di processo."""
    global _CACHE
    if _CACHE is None:
        from scripts.analysis.report_families import build_everything
        S, pairs, tsm, shape = build_everything()
        _CACHE = {**S, **pairs, **tsm, **shape}
    return _CACHE


def sleeve_returns_df(ids: list[str]) -> pd.DataFrame:
    """Rendimenti giornalieri allineati per gli sleeve richiesti."""
    eq = all_sleeve_equities()
    return pd.DataFrame({i: eq[i].pct_change().fillna(0.0) for i in ids})

Step 4: Run — verifica passaggio

Run: uv run pytest tests/portfolio/test_sleeves.py -v Expected: 2 passed (richiede i parquet in data/raw/; ~2-3 min per la build).

Step 5: Commit

git add src/portfolio/sleeves.py tests/portfolio/test_sleeves.py
git commit -m "feat(portfolio): builder unificato equity-per-sleeve (parità con report_families)"

Task 4: `SleeveSpec`, `Portfolio`, `PortfolioResult` + backtest (`base.py`)

Files:

Create: src/portfolio/base.py
Test: tests/portfolio/test_backtest_parity.py
Step 1: Scrivi il test di parità (fallisce)

tests/portfolio/test_backtest_parity.py:

import pytest
from src.portfolio.base import Portfolio, SleeveSpec
from scripts.analysis.report_families import build_everything
from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT


def _master9_specs():
    fade = [SleeveSpec(kind="single", name=f"{c}", sid=f"{c}_{a}", asset=a, cluster=f"{a}-rev")
            for a in ("BTC", "ETH") for c in ("MR01", "MR02", "MR07")]
    honest = [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")]
    return fade + honest


def test_master9_backtest_matches_report():
    p = Portfolio(code="PORT03", label="Master", sleeves=_master9_specs(), weighting="equal")
    res = p.backtest()
    # riferimento: equal-weight degli stessi 9 sleeve via la macchina del report
    S, _, _, _ = build_everything()
    dr_ref = port_returns(S)
    ref_full, ref_oos = metrics(dr_ref), metrics(dr_ref, lo=SPLIT)
    assert res.full["sharpe"] == pytest.approx(ref_full["sharpe"], abs=1e-6)
    assert res.full["dd"] == pytest.approx(ref_full["dd"], abs=1e-6)
    assert res.oos["sharpe"] == pytest.approx(ref_oos["sharpe"], abs=1e-6)

Step 2: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_backtest_parity.py -v Expected: FAIL (ModuleNotFoundError: base).

Step 3: Implementa base.py

"""Portfolio: definizione (sleeve + schema pesi) con faccia di backtest.
La faccia live è in runner.py."""
from __future__ import annotations

from dataclasses import dataclass, field

import pandas as pd

from src.portfolio import weighting as W
from src.portfolio.sleeves import all_sleeve_equities, sleeve_returns_df
from scripts.analysis.combine_portfolio import port_returns, metrics, yearly_returns, SPLIT


@dataclass
class SleeveSpec:
    kind: str                       # "single" | "pairs" | "ml"
    name: str                       # codice strategia per il live (MR01/DIP01/PR01.../SH01)
    sid: str                        # id canonico (= chiave in all_sleeve_equities)
    asset: str | None = None
    a: str | None = None
    b: str | None = None
    tf: str = "1h"
    params: dict = field(default_factory=dict)
    cluster: str = ""


@dataclass
class PortfolioResult:
    code: str
    weights: dict
    full: dict                      # ret/cagr/dd/sharpe (FULL)
    oos: dict                       # ret/cagr/dd/sharpe (OOS)
    yearly: dict                    # anno -> ret%
    risk: dict                      # sid -> % contributo al rischio (equal informativo)


@dataclass
class Portfolio:
    code: str
    label: str
    sleeves: list[SleeveSpec]
    weighting: str = "equal"
    weights: dict | None = None
    caps: dict | None = None
    total_capital: float = 1000.0
    leverage: float = 3.0
    rebalance: str = "1D"
    vol_lookback: int = 90

    @property
    def sleeve_ids(self) -> list[str]:
        return [s.sid for s in self.sleeves]

    @property
    def clusters(self) -> dict[str, str]:
        return {s.sid: (s.cluster or s.sid) for s in self.sleeves}

    def weight_vector(self, returns_df: pd.DataFrame | None = None) -> dict[str, float]:
        return W.weight_vector(
            self.weighting, self.sleeve_ids, returns_df,
            weights=self.weights, caps=self.caps,
            clusters=self.clusters, lookback=self.vol_lookback,
        )

    def backtest(self) -> PortfolioResult:
        eq = all_sleeve_equities()
        members = {sid: eq[sid] for sid in self.sleeve_ids}
        dr = sleeve_returns_df(self.sleeve_ids)
        w = self.weight_vector(dr)
        port_dr = port_returns(members, w)
        full, oos = metrics(port_dr), metrics(port_dr, lo=SPLIT)
        # contributo al rischio (equal-weight, informativo)
        cov = dr.cov().values
        import numpy as np
        we = np.ones(len(self.sleeve_ids)) / len(self.sleeve_ids)
        pv = float(we @ cov @ we)
        rc = we * (cov @ we)
        risk = {sid: float(rc[k] / pv * 100) if pv > 0 else 0.0
                for k, sid in enumerate(self.sleeve_ids)}
        return PortfolioResult(self.code, w, full, oos, yearly_returns(port_dr), risk)

Step 4: Run — verifica passaggio

Run: uv run pytest tests/portfolio/test_backtest_parity.py -v Expected: 1 passed.

Step 5: Commit

git add src/portfolio/base.py tests/portfolio/test_backtest_parity.py
git commit -m "feat(portfolio): SleeveSpec/Portfolio/backtest con parità verso report_families"

Task 5: Definizioni concrete `scripts/portfolios/PORT01..06`

Files:

Create: scripts/portfolios/__init__.py, scripts/portfolios/_defs.py, PORT01..06_*.py
Test: tests/portfolio/test_definitions.py
Step 1: Test (fallisce)

tests/portfolio/test_definitions.py:

from scripts.portfolios._defs import PORTFOLIOS


def test_six_portfolios_defined():
    assert set(PORTFOLIOS) == {"PORT01", "PORT02", "PORT03", "PORT04", "PORT05", "PORT06"}


def test_port06_is_master_shape_cap():
    p = PORTFOLIOS["PORT06"]
    sids = set(p.sleeve_ids)
    assert {"SH_BTC", "SH_ETH", "TSM01", "PR_ETHBTC"} <= sids
    assert len(sids) == 17
    assert p.weighting == "cap" and p.caps == {"PAIRS": 0.33}


def test_default_leverage_sober():
    assert PORTFOLIOS["PORT06"].leverage == 2.0

Step 2: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_definitions.py -v Expected: FAIL (ModuleNotFoundError: _defs).

Step 3: Crea il package e le definizioni condivise

touch scripts/portfolios/__init__.py

scripts/portfolios/_defs.py:

"""Definizioni canoniche dei portafogli (tutti i tipi visti finora)."""
from __future__ import annotations

import sys
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from src.portfolio.base import Portfolio, SleeveSpec  # noqa: E402

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 = [
    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"),
]
PAIRS = [
    SleeveSpec(kind="pairs", name="PR01", sid="PR_ETHBTC", a="ETH", b="BTC", cluster="ETH-rev"),
    SleeveSpec(kind="pairs", name="PR01", sid="PR_LTCETH", a="LTC", b="ETH", cluster="ETH-rev"),
    SleeveSpec(kind="pairs", name="PR01", sid="PR_ADAETH", a="ADA", b="ETH", cluster="ETH-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"),
]
TSM = [SleeveSpec(kind="single", name="TSM01", sid="TSM01", cluster="trend")]
SHAPE = [SleeveSpec(kind="ml", name="SH01", sid=f"SH_{a}", asset=a, cluster="shape")
         for a in ("BTC", "ETH")]

PORTFOLIOS = {
    "PORT01": Portfolio("PORT01", "Honest", HONEST, weighting="equal"),
    "PORT02": Portfolio("PORT02", "Fade master", FADE, weighting="equal"),
    "PORT03": Portfolio("PORT03", "Master", FADE + HONEST, weighting="equal"),
    "PORT04": Portfolio("PORT04", "Master + pairs", FADE + HONEST + PAIRS,
                        weighting="cap", caps={"PAIRS": 0.33}),
    "PORT05": Portfolio("PORT05", "Master esteso", FADE + HONEST + PAIRS + TSM,
                        weighting="cap", caps={"PAIRS": 0.33}),
    "PORT06": Portfolio("PORT06", "Master + shape", FADE + HONEST + PAIRS + TSM + SHAPE,
                        weighting="cap", caps={"PAIRS": 0.33}, leverage=2.0),
}

Step 4: Crea i 6 script con run() (report)

Per ciascun code in PORT01..PORT06, crea scripts/portfolios/<code>_<slug>.py. Esempio scripts/portfolios/PORT06_master_shape.py:

"""PORT06 — Master + shape (default). Report backtest del portafoglio."""
import sys
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from scripts.portfolios._defs import PORTFOLIOS  # noqa: E402

CODE = "PORT06"


def run():
    p = PORTFOLIOS[CODE]
    r = p.backtest()
    print("=" * 80)
    print(f"  {p.code} — {p.label} | pesi={p.weighting} caps={p.caps} leva={p.leverage}x")
    print("=" * 80)
    print(f"  FULL ret {r.full['ret']:+.0f}%  CAGR {r.full['cagr']:.0f}%  "
          f"DD {r.full['dd']:.1f}%  Sharpe {r.full['sharpe']:.2f}")
    print(f"  OOS  ret {r.oos['ret']:+.0f}%  DD {r.oos['dd']:.1f}%  Sharpe {r.oos['sharpe']:.2f}")
    print("  per anno:", {y: round(v) for y, v in sorted(r.yearly.items())})
    print("  rischio % per sleeve:", {k: round(v, 1) for k, v in
          sorted(r.risk.items(), key=lambda x: -x[1])})


if __name__ == "__main__":
    run()

Gli altri 5 file sono identici cambiando solo CODE (PORT01..PORT05) e il nome file: PORT01_honest.py, PORT02_fade.py, PORT03_master.py, PORT04_master_pairs.py, PORT05_master_esteso.py.

Step 5: Run test + uno smoke report

Run: uv run pytest tests/portfolio/test_definitions.py -v Expected: 3 passed. Run: uv run python scripts/portfolios/PORT06_master_shape.py Expected: stampa FULL/OOS/per-anno coerenti col report (Sharpe FULL ~6, OOS più alto).

Step 6: Commit

git add scripts/portfolios/ tests/portfolio/test_definitions.py
git commit -m "feat(portfolio): definizioni PORT01-06 + report run() (default PORT06)"

Task 6: `PortfolioLedger` (stato/PnL/persistenza)

Files:

Create: src/portfolio/ledger.py
Test: tests/portfolio/test_ledger.py
Step 1: Test (fallisce)

tests/portfolio/test_ledger.py:

from pathlib import Path
from src.portfolio.ledger import PortfolioLedger


def test_alloc_split_by_weights(tmp_path):
    L = PortfolioLedger("PORTX", total_capital=1000.0, data_dir=tmp_path)
    alloc = L.allocate({"a": 0.6, "b": 0.4})
    assert alloc == {"a": 600.0, "b": 400.0}


def test_update_tracks_equity_and_dd(tmp_path):
    L = PortfolioLedger("PORTX", total_capital=1000.0, data_dir=tmp_path)
    L.update_equity({"a": 700.0, "b": 500.0})    # equity 1200
    assert L.equity == 1200.0 and L.peak == 1200.0 and L.max_dd == 0.0
    L.update_equity({"a": 500.0, "b": 400.0})    # equity 900 -> dd 25%
    assert L.equity == 900.0
    assert abs(L.max_dd - 25.0) < 1e-9


def test_persist_and_resume(tmp_path):
    L = PortfolioLedger("PORTX", total_capital=1000.0, data_dir=tmp_path)
    L.update_equity({"a": 1100.0})
    L.save()
    L2 = PortfolioLedger("PORTX", total_capital=1000.0, data_dir=tmp_path)
    assert L2.equity == 1100.0 and L2.peak == 1100.0
    assert (tmp_path / "PORTX" / "equity.jsonl").exists()

Step 2: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_ledger.py -v Expected: FAIL (ModuleNotFoundError: ledger).

Step 3: Implementa ledger.py

"""Ledger aggregato del portafoglio: capitale, allocazioni, equity, PnL, peak/DD, persistenza."""
from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path


class PortfolioLedger:
    def __init__(self, code: str, total_capital: float = 1000.0,
                 data_dir: Path = Path("data/portfolios")):
        self.code = code
        self.initial_capital = total_capital
        self.total_capital = total_capital
        self.work_dir = Path(data_dir) / code
        self.work_dir.mkdir(parents=True, exist_ok=True)
        self.status_path = self.work_dir / "status.json"
        self.equity_path = self.work_dir / "equity.jsonl"
        self.events_path = self.work_dir / "events.jsonl"
        self.equity = total_capital
        self.peak = total_capital
        self.max_dd = 0.0
        self.weights: dict[str, float] = {}
        self.alloc: dict[str, float] = {}
        self.last_rebalance = ""
        self._load()

    def _load(self):
        if not self.status_path.exists():
            return
        s = json.loads(self.status_path.read_text())
        self.total_capital = s.get("total_capital", self.total_capital)
        self.equity = s.get("equity", self.equity)
        self.peak = s.get("peak", self.peak)
        self.max_dd = s.get("max_dd", self.max_dd)
        self.weights = s.get("weights", {})
        self.alloc = s.get("alloc", {})
        self.last_rebalance = s.get("last_rebalance", "")

    def allocate(self, weights: dict[str, float]) -> dict[str, float]:
        self.weights = dict(weights)
        self.alloc = {sid: round(self.total_capital * w, 6) for sid, w in weights.items()}
        self.last_rebalance = datetime.now(timezone.utc).isoformat()
        self._append(self.events_path, {"event": "rebalance", "weights": self.weights,
                                        "total_capital": self.total_capital})
        return self.alloc

    def update_equity(self, sleeve_equity: dict[str, float], pnl_day: float = 0.0):
        self.equity = float(sum(sleeve_equity.values()))
        if self.equity > self.peak:
            self.peak = self.equity
        dd = (self.peak - self.equity) / self.peak * 100 if self.peak > 0 else 0.0
        self.max_dd = max(self.max_dd, dd)
        self._append(self.equity_path, {
            "ts": datetime.now(timezone.utc).isoformat(),
            "equity": round(self.equity, 2), "dd": round(dd, 3),
            "pnl_day": round(pnl_day, 2),
            "pnl_total": round(self.equity - self.initial_capital, 2),
        })

    def save(self):
        self.status_path.write_text(json.dumps({
            "code": self.code, "total_capital": round(self.total_capital, 2),
            "equity": round(self.equity, 2), "peak": round(self.peak, 2),
            "max_dd": round(self.max_dd, 3), "weights": self.weights,
            "alloc": self.alloc, "last_rebalance": self.last_rebalance,
            "ts": datetime.now(timezone.utc).isoformat(),
        }, indent=2))

    @staticmethod
    def _append(path: Path, row: dict):
        with open(path, "a") as f:
            f.write(json.dumps(row) + "\n")

Step 4: Run — verifica passaggio

Run: uv run pytest tests/portfolio/test_ledger.py -v Expected: 3 passed.

Step 5: Commit

git add src/portfolio/ledger.py tests/portfolio/test_ledger.py
git commit -m "feat(portfolio): PortfolioLedger (alloc, equity/DD, persistenza+resume)"

Task 7: `portfolios.yml` + loader della config live

Files:

Create: portfolios.yml
Modify: src/portfolio/base.py (aggiunge Portfolio.from_active_config)
Test: tests/portfolio/test_config.py
Step 1: Crea portfolios.yml

# Config LIVE del paper trader a portafoglio. Seleziona UN portafoglio attivo
# (definito in scripts/portfolios/_defs.py) e ne fa l'override dei parametri operativi.
active: PORT06            # default raccomandato: master + shape
overrides:
  total_capital: 1000
  weighting: cap          # equal | cap | inverse_vol | cluster_rp | manual
  caps: {PAIRS: 0.33}
  leverage: 2             # sobrio per il live reale
  rebalance: 1D
  poll_seconds: 60

Step 2: Test (fallisce)

tests/portfolio/test_config.py:

from src.portfolio.base import load_active_portfolio


def test_load_active_applies_overrides(tmp_path):
    cfg = tmp_path / "portfolios.yml"
    cfg.write_text("active: PORT06\noverrides:\n  leverage: 2\n  total_capital: 500\n")
    p = load_active_portfolio(cfg)
    assert p.code == "PORT06"
    assert p.leverage == 2.0
    assert p.total_capital == 500

Step 3: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_config.py -v Expected: FAIL (ImportError: load_active_portfolio).

Step 4: Implementa il loader in base.py

Aggiungi in fondo a src/portfolio/base.py:

def load_active_portfolio(config_path) -> "Portfolio":
    """Carica il portafoglio attivo da portfolios.yml applicando gli override."""
    import yaml
    from pathlib import Path
    from scripts.portfolios._defs import PORTFOLIOS

    cfg = yaml.safe_load(Path(config_path).read_text())
    p = PORTFOLIOS[cfg["active"]]
    ov = cfg.get("overrides", {})
    for k in ("total_capital", "weighting", "caps", "leverage", "rebalance", "vol_lookback"):
        if k in ov and ov[k] is not None:
            setattr(p, k, ov[k])
    return p

Step 5: Run — verifica passaggio

Run: uv run pytest tests/portfolio/test_config.py -v Expected: 1 passed.

Step 6: Commit

git add portfolios.yml src/portfolio/base.py tests/portfolio/test_config.py
git commit -m "feat(portfolio): portfolios.yml + load_active_portfolio (override operativi)"

Task 8: `PortfolioRunner` — costruzione worker + sizing pool

Files:

Create: src/portfolio/runner.py
Test: tests/portfolio/test_runner_build.py
Step 1: Test (fallisce)

tests/portfolio/test_runner_build.py:

from src.portfolio.runner import build_worker_for
from src.portfolio.base import SleeveSpec
from src.live.strategy_worker import StrategyWorker
from src.live.pairs_worker import PairsWorker


def test_build_single_worker_capital_from_alloc(tmp_path):
    spec = SleeveSpec(kind="single", name="MR01", sid="MR01_BTC", asset="BTC",
                      params={"bb_window": 50, "k": 2.5, "sl_atr": 2.0, "max_bars": 24})
    w = build_worker_for(spec, alloc_capital=300.0, leverage=2.0, data_dir=tmp_path)
    assert isinstance(w, StrategyWorker)
    assert w.capital == 300.0 and w.leverage == 2.0


def test_build_pairs_worker(tmp_path):
    spec = SleeveSpec(kind="pairs", name="PR01", sid="PR_ETHBTC", a="ETH", b="BTC",
                      params={"n": 50, "z_in": 2.0, "z_exit": 0.75, "max_bars": 72})
    w = build_worker_for(spec, alloc_capital=200.0, leverage=2.0, data_dir=tmp_path)
    assert isinstance(w, PairsWorker)
    assert w.capital == 200.0

Step 2: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_runner_build.py -v Expected: FAIL (ModuleNotFoundError: runner).

Step 3: Implementa la parte di build in runner.py

"""PortfolioRunner: faccia live del portafoglio (capitale pool, sizing, ribilancio, ledger).
Riusa i worker esistenti come esecutori e il data layer Cerbero v2."""
from __future__ import annotations

from pathlib import Path

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.multi_runner import MLWorkerWrapper
from src.live.strategy_loader import load_strategy

# Codice-breve sleeve -> nome modulo Strategy in scripts/strategies/
_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)
}
DATA_DIR = Path("data/paper_trades")


def build_worker_for(spec: SleeveSpec, alloc_capital: float, leverage: float,
                     data_dir: Path = DATA_DIR, position_size: float = 0.15):
    """Costruisce il worker esecutore per uno sleeve con capitale = quota allocata."""
    if spec.kind == "pairs":
        return PairsWorker(
            asset_a=spec.a, asset_b=spec.b, tf=spec.tf, params=spec.params,
            capital=alloc_capital, position_size=position_size, leverage=leverage,
            fee_rt=0.001, name="PR01_pairs_reversion", 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)")
    strategy = load_strategy(module)
    worker = StrategyWorker(
        strategy=strategy, asset=spec.asset, tf=spec.tf, capital=alloc_capital,
        position_size=position_size, leverage=leverage, params=spec.params, data_dir=data_dir,
    )
    if spec.kind == "ml":                       # SH01: retraining periodico
        return MLWorkerWrapper(worker, {"retrain_hours": 24})
    return worker

Step 4: Run — verifica passaggio

Run: uv run pytest tests/portfolio/test_runner_build.py -v Expected: 2 passed.

Step 5: Commit

git add src/portfolio/runner.py tests/portfolio/test_runner_build.py
git commit -m "feat(portfolio): build_worker_for (worker esecutori con capitale da alloc pool)"

Task 9: `PortfolioRunner` — loop live (data v2, ribilancio, aggregazione)

Files:

Modify: src/portfolio/runner.py
Test: tests/portfolio/test_runner_rebalance.py, scripts/analysis/smoke_portfolio.py
Step 1: Test del ribilancio (fallisce)

tests/portfolio/test_runner_rebalance.py:

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)}   # equity 1200
    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

Step 2: Run — verifica fallimento

Run: uv run pytest tests/portfolio/test_runner_rebalance.py -v Expected: FAIL (ImportError: rebalance_allocations).

Step 3: Implementa ribilancio + loop in runner.py

Aggiungi a src/portfolio/runner.py:

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."""
    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.weighting import weight_vector
    from src.live.cerbero_client import CerberoClient

    p: Portfolio = load_active_portfolio(config_path)
    ledger = PortfolioLedger(p.code, total_capital=p.total_capital)
    client = CerberoClient()

    # pesi iniziali (vol-based dai rendimenti storici degli sleeve; statici per equal/cap/manual)
    dr = sleeve_returns_df(p.sleeve_ids)
    weights = p.weight_vector(dr)
    alloc = ledger.allocate(weights)

    # costruisci i worker esecutori con capitale = quota allocata
    workers = {s.sid: build_worker_for(s, alloc[s.sid], p.leverage) for s in p.sleeves}

    # risolvi i nomi strumento via get_instruments (fallback alla mappa legacy)
    from src.live.multi_runner import INSTRUMENT_MAP
    inst_map = dict(INSTRUMENT_MAP)            # TODO opzionale: arricchire via client.get_instruments

    last_day = ""
    poll = 60
    while True:
        try:
            # fetch candele (v2 unificato) per ogni asset/tf richiesto dagli sleeve
            keys = set()
            for s in p.sleeves:
                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)

            # tick di ogni worker (esecutore)
            for s in p.sleeves:
                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])

            # aggrega equity nel ledger
            ledger.update_equity({sid: _worker_equity(w) for sid, w in workers.items()})

            # ribilancio a cambio giorno UTC
            today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
            if today != last_day and last_day:
                dr = sleeve_returns_df(p.sleeve_ids)
                rebalance_allocations(ledger, workers, p.weight_vector(dr))
            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()

Step 4: Run — verifica passaggio

Run: uv run pytest tests/portfolio/test_runner_rebalance.py -v Expected: 1 passed.

Step 5: Smoke live (un tick reale, niente ordini)

scripts/analysis/smoke_portfolio.py:

"""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]:                       # 3 sleeve campione per lo smoke
        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()

Run: uv run python scripts/analysis/smoke_portfolio.py Expected: stampa il portafoglio attivo (PORT06) e 3/3 sleeve con candele v2 > 0.

Step 6: Commit

git add src/portfolio/runner.py tests/portfolio/test_runner_rebalance.py scripts/analysis/smoke_portfolio.py
git commit -m "feat(portfolio): PortfolioRunner live (data v2, tick, ribilancio giornaliero, ledger)"

Task 10: Documentazione (CLAUDE.md, README, comandi)

Files:

Modify: CLAUDE.md, README.md
Step 1: Aggiorna CLAUDE.md

Nella struttura aggiungi src/portfolio/ e scripts/portfolios/; in "Comandi" aggiungi:

uv run python scripts/portfolios/PORT06_master_shape.py   # report backtest portafoglio
uv run python -m src.portfolio.runner                      # paper trading a PORTAFOGLIO (capitale pool)
uv run python scripts/analysis/smoke_portfolio.py          # smoke live data layer v2

Aggiungi una sezione "Portafogli" che riassume: oggetto Portfolio (pool, backtest+live), schemi pesi, default PORT06 (cap pairs 33%, leva 2x), data layer Cerbero v2, limite noto (posizioni aperte non travasate al ribilancio).

Step 2: Aggiorna README.md

Aggiungi la cartella portfolios/ alla struttura e una riga d'uso del nuovo paper trader a portafoglio. Prosa italiana completa (artefatto pubblico).

Step 3: Esegui l'intera suite

Run: uv run pytest -m "not network" -v Expected: tutti i test (weighting, sleeves, backtest_parity, definitions, ledger, config, runner_build, runner_rebalance) passano.

Step 4: Commit

git add CLAUDE.md README.md
git commit -m "docs(portfolio): documenta cartella portfolios, comandi e default PORT06"

Self-review (svolta in fase di scrittura)

Copertura spec: §3 layout → Task 2-9; §4 schema → Task 4; §5 backtest → Task 4-5; §6 live → Task 8-9; §7 persistenza → Task 6; §8 portafogli/default → Task 5,7; §9 test → ogni task TDD + suite finale; §2.6 data v2 → Task 1,9. Tutte coperte.
Limite noto (posizioni aperte non travasate): implementato in rebalance_allocations e documentato (Task 9 docstring + Task 10).
Honest DIP01/TR01/ROT02 nel live: build_worker_for solleva un errore esplicito (worker dedicati in fase 2) — coerente con lo scope: backtest li include, il live v1 esegue fade/pairs/shape che hanno worker pronti. Nota: se il default PORT06 deve girare live al primo colpo servono i worker honest; in alternativa per il primo avvio live usare PORT04-shape senza honest, oppure aggiungere i 3 worker honest come Task 8b. Da decidere in esecuzione.
Consistenza tipi: sid usato come chiave ovunque (definizioni ↔ all_sleeve_equities ↔ ledger ↔ workers); weight_vector firma identica in weighting.py e Portfolio.weight_vector; _worker_equity gestisce MLWorkerWrapper.
Placeholder: nessun TBD nel codice; l'unico TODO è opzionale (arricchire inst_map via get_instruments) e non blocca.

Punto aperto per l'esecuzione: il default PORT06 contiene gli sleeve honest (DIP01/TR01/ROT02) che NON hanno ancora un worker live. Decidere a inizio esecuzione se (a) aggiungere Task 8b coi worker honest, oppure (b) far girare il primo live con un portafoglio senza honest (fade+pairs+shape) e tenere PORT06 completo solo in backtest finché i worker honest non esistono.

45 KiB Raw Blame History Unescape Escape

Cartella portfolios/ — Implementation Plan

File structure

Task 1: Metodi Cerbero v2 nel client

Task 2: Schemi di peso (weighting.py)

Task 3: Builder unificato delle equity-per-sleeve (sleeves.py)

Task 4: SleeveSpec, Portfolio, PortfolioResult + backtest (base.py)

Task 5: Definizioni concrete scripts/portfolios/PORT01..06

Task 6: PortfolioLedger (stato/PnL/persistenza)

Task 7: portfolios.yml + loader della config live

Task 8: PortfolioRunner — costruzione worker + sizing pool

Task 9: PortfolioRunner — loop live (data v2, ribilancio, aggregazione)

Task 10: Documentazione (CLAUDE.md, README, comandi)

Self-review (svolta in fase di scrittura)

45 KiB

Raw Blame History

Cartella `portfolios/` — Implementation Plan

Task 2: Schemi di peso (`weighting.py`)

Task 3: Builder unificato delle equity-per-sleeve (`sleeves.py`)

Task 4: `SleeveSpec`, `Portfolio`, `PortfolioResult` + backtest (`base.py`)

Task 5: Definizioni concrete `scripts/portfolios/PORT01..06`

Task 6: `PortfolioLedger` (stato/PnL/persistenza)

Task 7: `portfolios.yml` + loader della config live

Task 8: `PortfolioRunner` — costruzione worker + sizing pool

Task 9: `PortfolioRunner` — loop live (data v2, ribilancio, aggregazione)