From d99c9895bbab53eb051bd1121468ad0c871a12f2 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Fri, 29 May 2026 15:49:27 +0200 Subject: [PATCH] feat(portfolio): schemi di peso (equal/manual/cap/inverse_vol/cluster_rp) --- src/portfolio/__init__.py | 0 src/portfolio/weighting.py | 96 +++++++++++++++++++++++++++++++ tests/portfolio/test_weighting.py | 41 +++++++++++++ 3 files changed, 137 insertions(+) create mode 100644 src/portfolio/__init__.py create mode 100644 src/portfolio/weighting.py create mode 100644 tests/portfolio/test_weighting.py diff --git a/src/portfolio/__init__.py b/src/portfolio/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/portfolio/weighting.py b/src/portfolio/weighting.py new file mode 100644 index 0000000..295bb4a --- /dev/null +++ b/src/portfolio/weighting.py @@ -0,0 +1,96 @@ +"""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}") diff --git a/tests/portfolio/test_weighting.py b/tests/portfolio/test_weighting.py new file mode 100644 index 0000000..0281a69 --- /dev/null +++ b/tests/portfolio/test_weighting.py @@ -0,0 +1,41 @@ +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 + assert pytest.approx(sum(w.values())) == 1.0 + assert w["MR01_BTC"] > 0.25 + + +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