Coverage for src/stable_yield_lab/analytics/portfolio.py: 80%
92 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-04 20:38 +0000
1from __future__ import annotations
3import math
5import pandas as pd
7from . import performance
8from .risk import _require_riskfolio
11def _normalise_weights(returns: pd.DataFrame, weights: pd.Series | None) -> pd.Series:
12 """Align and normalise weight vector against ``returns`` columns."""
14 if returns.empty: 14 ↛ 15line 14 didn't jump to line 15 because the condition on line 14 was never true
15 return pd.Series(dtype=float)
17 if weights is None: 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true
18 return pd.Series(1.0 / returns.shape[1], index=returns.columns, dtype=float)
20 aligned = weights.reindex(returns.columns).fillna(0.0)
21 total = float(aligned.sum())
22 if total == 0.0: 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true
23 raise ValueError("weights sum to zero")
24 return aligned / total
27def allocate_mean_variance(
28 returns: pd.DataFrame,
29 *,
30 bounds: dict[str, tuple[float, float]] | None = None,
31 risk_measure: str = "MV",
32 rf: float = 0.0,
33 l: float = 0.0,
34) -> pd.Series:
35 """Optimize portfolio weights using mean-variance framework.
37 Parameters
38 ----------
39 returns: pd.DataFrame
40 Wide DataFrame of historical returns (rows=observations, cols=pools).
41 bounds: dict[str, tuple[float, float]] | None, optional
42 Mapping of asset -> (min_weight, max_weight). Bounds are applied
43 post-optimization and weights re-normalized. Defaults to None.
44 risk_measure: str, optional
45 Risk measure understood by riskfolio-lib (e.g. "MV", "MAD").
46 rf: float, optional
47 Risk-free rate for the optimizer. Defaults to 0.0.
48 l: float, optional
49 Risk aversion factor used by riskfolio-lib. Defaults to 0.0.
50 """
51 _require_riskfolio()
52 import riskfolio as rp
54 port = rp.Portfolio(returns=returns)
55 port.assets_stats(method_mu="hist", method_cov="hist")
56 weights_df = port.optimization(model="Classic", rm=risk_measure, obj="Sharpe", rf=rf, l=l)
57 weights = weights_df.squeeze()
59 if bounds: 59 ↛ 64line 59 didn't jump to line 64 because the condition on line 59 was always true
60 lo = pd.Series({k: v[0] for k, v in bounds.items()})
61 hi = pd.Series({k: v[1] for k, v in bounds.items()})
62 weights = weights.clip(lower=lo, upper=hi)
64 weights = weights / weights.sum()
65 return weights.reindex(returns.columns)
68def expected_apy(returns: pd.DataFrame, weights: pd.Series, *, freq: int = 52) -> float:
69 """Compute portfolio expected APY given periodic returns and weights."""
70 weights = weights.reindex(returns.columns).fillna(0)
71 mean_returns = returns.mean()
72 apy_assets = (1 + mean_returns) ** freq - 1
73 return float((weights * apy_assets).sum())
76def tvl_weighted_risk(returns: pd.DataFrame, weights: pd.Series, *, rm: str = "MV") -> float:
77 """Compute TVL-weighted risk of a portfolio using riskfolio risk measures."""
78 _require_riskfolio()
79 import riskfolio as rp
81 weights = weights.reindex(returns.columns).fillna(0)
82 cov = returns.cov()
83 rc = rp.Risk_Contribution(weights, returns, cov, rm=rm)
84 return float(rc.sum())
87def tracking_error(
88 returns: pd.DataFrame | pd.Series,
89 weights: pd.Series | None = None,
90 *,
91 freq: int = 52,
92 target_periodic_return: float | None = None,
93) -> tuple[float, float]:
94 """Compute periodic and annualised tracking error for a rebalanced portfolio.
96 Parameters
97 ----------
98 returns:
99 Either a wide DataFrame of asset returns or a Series of portfolio
100 returns. The returns must be periodic simple returns expressed as
101 decimal fractions.
102 weights:
103 Target weights per asset when ``returns`` is a DataFrame. Missing
104 weights default to zero and the vector is re-normalised to sum to one.
105 freq:
106 Number of compounding periods per year. Must be positive.
107 target_periodic_return:
108 Optional benchmark periodic return expressed as a decimal fraction. If
109 omitted, the realised mean of the portfolio returns is used.
111 Returns
112 -------
113 tuple[float, float]
114 Periodic tracking error followed by its annualised counterpart. When
115 insufficient observations are available the function returns
116 ``(nan, nan)``.
117 """
119 if freq <= 0:
120 raise ValueError("freq must be positive")
122 if isinstance(returns, pd.DataFrame):
123 if returns.empty: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 return float("nan"), float("nan")
125 if weights is None: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 raise ValueError("weights are required when returns is a DataFrame")
127 norm_weights = _normalise_weights(returns, weights)
128 portfolio_returns = returns.fillna(0.0).mul(norm_weights, axis=1).sum(axis=1)
129 else:
130 portfolio_returns = returns.dropna()
132 if portfolio_returns.empty: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 return float("nan"), float("nan")
135 benchmark = (
136 float(target_periodic_return)
137 if target_periodic_return is not None
138 else float(portfolio_returns.mean())
139 )
140 active = portfolio_returns - benchmark
141 active = active.dropna()
142 if active.size < 2:
143 return float("nan"), float("nan")
145 periodic_te = float(active.std(ddof=1))
146 annualised_te = periodic_te * math.sqrt(freq)
147 return periodic_te, annualised_te
150def apy_performance_summary(
151 returns: pd.DataFrame,
152 weights: pd.Series | None = None,
153 *,
154 freq: int = 52,
155 initial_nav: float = 1.0,
156 nav: pd.Series | None = None,
157) -> tuple[pd.Series, pd.Series]:
158 """Summarise expected versus realised APY for a rebalanced portfolio.
160 The function compares the static expectation from :func:`expected_apy`
161 against the realised performance implied by a rebalance-aware NAV path.
163 Parameters
164 ----------
165 returns:
166 Wide DataFrame of periodic simple returns, indexed by timestamp.
167 weights:
168 Target weights per asset. When ``None`` an equal-weight portfolio is
169 assumed.
170 freq:
171 Compounding periods per year used to annualise results.
172 initial_nav:
173 Starting NAV used for the aggregated portfolio path.
174 nav:
175 Optional externally computed NAV path. When provided it must align with
176 ``returns.index``.
178 Returns
179 -------
180 tuple[pandas.Series, pandas.Series]
181 Tuple containing (metrics, nav_path). ``metrics`` is a Series with
182 expected APY, realised APY, realised total return, active APY and
183 tracking error figures. ``nav_path`` is the NAV trajectory used for the
184 realised calculations.
185 """
187 if freq <= 0:
188 raise ValueError("freq must be positive")
190 if returns.empty: 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true
191 empty_nav = pd.Series(dtype=float)
192 metrics = pd.Series(
193 {
194 "expected_apy": float("nan"),
195 "realized_apy": float("nan"),
196 "realized_total_return": float("nan"),
197 "active_apy": float("nan"),
198 "tracking_error_periodic": float("nan"),
199 "tracking_error_annualized": float("nan"),
200 "horizon_periods": 0.0,
201 "horizon_years": float("nan"),
202 "final_nav": float(initial_nav),
203 },
204 dtype=float,
205 )
206 return metrics, empty_nav
208 norm_weights = _normalise_weights(returns, weights)
209 clean_returns = returns.fillna(0.0)
210 portfolio_returns = clean_returns.mul(norm_weights, axis=1).sum(axis=1)
212 if nav is None:
213 nav_path = performance.nav_series(clean_returns, norm_weights, initial=initial_nav)
214 else:
215 nav_path = nav.reindex(clean_returns.index)
216 if nav_path.isna().any(): 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 raise ValueError("nav path contains NaN after aligning with returns index")
219 if nav_path.empty: 219 ↛ 220line 219 didn't jump to line 220 because the condition on line 219 was never true
220 final_nav = float(initial_nav)
221 realised_total = float("nan")
222 realised_apy = float("nan")
223 periods = 0
224 horizon_years = float("nan")
225 else:
226 final_nav = float(nav_path.iloc[-1])
227 periods = nav_path.shape[0]
228 realised_total = final_nav / float(initial_nav) - 1.0
229 realised_apy = (1.0 + realised_total) ** (freq / periods) - 1.0
230 horizon_years = periods / freq
232 expected = expected_apy(returns, norm_weights, freq=freq)
233 expected_periodic = (1.0 + expected) ** (1.0 / freq) - 1.0
234 te_periodic, te_annualised = tracking_error(
235 portfolio_returns,
236 freq=freq,
237 target_periodic_return=expected_periodic,
238 )
240 metrics = pd.Series(
241 {
242 "expected_apy": expected,
243 "realized_apy": realised_apy,
244 "realized_total_return": realised_total,
245 "active_apy": realised_apy - expected,
246 "tracking_error_periodic": te_periodic,
247 "tracking_error_annualized": te_annualised,
248 "horizon_periods": float(periods),
249 "horizon_years": horizon_years if periods else float("nan"),
250 "final_nav": final_nav,
251 },
252 dtype=float,
253 )
255 return metrics, nav_path