Coverage for src/stable_yield_lab/analytics/risk.py: 77%

1from __future__ import annotations 

2 

3from typing import Any 

4 

5import pandas as pd 

6 

7 

8def _require_riskfolio() -> None: 

9 try: 

10 import riskfolio as _ # noqa: F401 

11 except Exception as exc: # pragma: no cover - only raised when missing 

12 raise RuntimeError( 

13 "riskfolio-lib is required for these metrics. Install with: \n" 

14 " pip install 'riskfolio-lib'\n" 

15 "or enable the optional extra if using Poetry." 

16 ) from exc 

17 

18 

19def _call_assets_stats(portfolio: Any, *, method_mu: str, method_cov: str) -> None: 

20 """Invoke ``assets_stats`` with graceful fallback across riskfolio versions.""" 

21 

22 try: 

23 portfolio.assets_stats(method_mu=method_mu, method_cov=method_cov, d=0.94) 

24 except TypeError: 

25 portfolio.assets_stats(method_mu=method_mu, method_cov=method_cov) 

26 

27 

28def summary_statistics(returns: pd.DataFrame) -> pd.DataFrame: 

29 """Compute basic portfolio statistics per asset using riskfolio-lib formulas.""" 

30 

31 _require_riskfolio() 

32 import riskfolio as rp 

33 

34 stats_obj = rp.Sharpe_Risk(returns) 

35 if isinstance(stats_obj, pd.DataFrame): 

36 stats = stats_obj 

37 elif isinstance(stats_obj, pd.Series): 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true

38 stats = stats_obj.to_frame().T 

39 else: 

40 stats = pd.DataFrame([stats_obj]) 

41 

42 if isinstance(returns, pd.DataFrame) and not stats.empty: 42 ↛ 45line 42 didn't jump to line 45 because the condition on line 42 was always true

43 if stats.shape[1] == returns.shape[1]: 43 ↛ 45line 43 didn't jump to line 45 because the condition on line 43 was always true

44 stats.columns = list(returns.columns) 

45 return stats 

46 

47 

48def efficient_frontier( 

49 returns: pd.DataFrame, 

50 freq: int = 52, 

51 model: str = "Classic", 

52 risk_measure: str = "MV", 

53 obj: str = "Sharpe", 

54 rf: float = 0.0, 

55 l: float = 0.0, 

56) -> pd.DataFrame: 

57 """Compute an efficient frontier set of portfolios using riskfolio-lib. 

58 

59 Returns a DataFrame with weights per asset for frontier points. 

60 """ 

61 _require_riskfolio() 

62 import riskfolio as rp 

63 

64 Y = returns.copy() 

65 port = rp.Portfolio(returns=Y) 

66 method_mu = "hist" 

67 method_cov = "hist" 

68 _call_assets_stats(port, method_mu=method_mu, method_cov=method_cov) 

69 frontier = port.efficient_frontier(model=model, rm=risk_measure, obj=obj, rf=rf, l=l, points=20) 

70 if not isinstance(frontier, pd.DataFrame): 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 frontier = pd.DataFrame(frontier) 

72 return frontier 

73 

74 

75def risk_contributions(weights: pd.Series, returns: pd.DataFrame) -> pd.Series: 

76 """Compute marginal risk contributions for a given weights vector.""" 

77 _require_riskfolio() 

78 import riskfolio as rp 

79 

80 port = rp.Portfolio(returns=returns) 

81 _call_assets_stats(port, method_mu="hist", method_cov="hist") 

82 rc = port.risk_contribution(weights=weights, rm="MV") 

83 return rc