Coverage for src/ifunnel/models/CVaRtargets.py: 100%

1import numpy as np 

2import pandas as pd 

3from loguru import logger 

4 

5from .ScenarioGeneration import ScenarioGenerator 

6 

7 

8# Primal CVaR formula 

9def CVaR(alpha: float, p: np.array, q: np.array) -> tuple[float, float]: 

10 """ 

11 Computes CVaR using primal formula. 

12 NOTE: Inputs p and q should be numpy arrays. 

13 """ 

14 # We need to be careful that math index starts from 1 but numpy starts from 0 

15 # (matters in formulas like ceil(alpha * T)) 

16 # T = q.shape[0] 

17 sort_idx = np.argsort(q) 

18 sorted_q = q[sort_idx] 

19 sorted_p = p[sort_idx] 

20 

21 # Starting index 

22 i_alpha = np.sort(np.nonzero(np.cumsum(sorted_p) >= alpha)[0])[0] 

23 

24 # Weight of VaR component in CVaR 

25 lambda_alpha = (np.sum(sorted_p[: (i_alpha + 1)]) - alpha) / (1 - alpha) 

26 

27 # CVaR 

28 var = sorted_q[i_alpha] 

29 cvar = lambda_alpha * sorted_q[i_alpha] + np.dot(sorted_p[(i_alpha + 1) :], sorted_q[(i_alpha + 1) :]) / (1 - alpha) 

30 

31 return var, cvar 

32 

33 

34# FUNCTION RUNNING THE OPTIMIZATION 

35# ---------------------------------------------------------------------- 

36def portfolio_risk_target(scenarios: pd.DataFrame, cvar_alpha: float) -> float: 

37 # Fixed equal weight x 

38 x = pd.Series(index=scenarios.columns, data=1 / scenarios.shape[1]) 

39 

40 # Number of scenarios 

41 scenario_n = scenarios.shape[0] 

42 

43 # Portfolio loss scenarios 

44 losses = (-scenarios @ x).to_numpy() 

45 

46 # Probabilities 

47 probs = np.ones(scenario_n) / scenario_n 

48 

49 # CVaR 

50 _, portfolio_cvar = CVaR(1 - cvar_alpha, probs, losses) 

51 

52 return portfolio_cvar 

53 

54 

55# ---------------------------------------------------------------------- 

56# Mathematical Optimization: TARGETS GENERATION 

57# ---------------------------------------------------------------------- 

58def get_cvar_targets( 

59 test_date: str, 

60 benchmark: list, 

61 budget: int, 

62 cvar_alpha: float, 

63 data: pd.DataFrame, 

64 scgen: ScenarioGenerator, 

65 n_simulations: int, 

66) -> tuple[pd.DataFrame, pd.DataFrame]: 

67 logger.info(f"🎯 Generating CVaR targets for {benchmark}") 

68 

69 # Define Benchmark 

70 tickers = benchmark 

71 # Get weekly return of our benchmark 

72 whole_dataset_benchmark = data[tickers].copy() 

73 

74 # Get weekly data just for testing period 

75 test_dataset_benchmark = whole_dataset_benchmark[whole_dataset_benchmark.index >= test_date] 

76 

77 # Number of weeks for testing 

78 weeks_n = len(test_dataset_benchmark.index) 

79 

80 # Get scenarios 

81 # The Monte Carlo Method 

82 target_scenarios = scgen.bootstrapping( 

83 data=whole_dataset_benchmark, # subsetMST or subsetCLUST 

84 n_simulations=n_simulations, 

85 n_test=weeks_n, 

86 ) 

87 

88 # Compute the optimal portfolio outperforming zero percentage return 

89 # ---------------------------------------------------------------------- 

90 p_points = len(target_scenarios[:, 0, 0]) # number of periods 

91 s_points = len(target_scenarios[0, :, 0]) # number of scenarios 

92 

93 # COMPUTE CVaR TARGETS 

94 list_targets = [] 

95 for p in range(p_points): 

96 # create data frame with scenarios for a given period p 

97 scenario_df = pd.DataFrame(target_scenarios[p, :, :], columns=tickers, index=list(range(s_points))) 

98 

99 # run CVaR model to compute CVaR targets 

100 cvar_target = portfolio_risk_target(scenarios=scenario_df, cvar_alpha=cvar_alpha) 

101 # save the result 

102 list_targets.append(cvar_target) 

103 

104 # Generate new column so that dtype is set right. 

105 targets = pd.DataFrame(columns=["CVaR_Target"], data=list_targets) 

106 

107 # COMPUTE PORTFOLIO VALUE 

108 list_portfolio_values = [] 

109 for w in test_dataset_benchmark.index: 

110 budget_next = sum((budget / len(tickers)) * (1 + test_dataset_benchmark.loc[w, :])) 

111 list_portfolio_values.append(budget_next) 

112 budget = budget_next 

113 

114 # Generate dataframe so that dtype is set right. 

115 portfolio_value = pd.DataFrame( 

116 columns=["Benchmark_Value"], 

117 index=test_dataset_benchmark.index, 

118 data=list_portfolio_values, 

119 ) 

120 

121 return targets, portfolio_value