Coverage for src/ifunnel/models/CVaRmodel.py: 86%
69 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-12 09:14 +0000
1import pickle
3import cvxpy as cp
4import numpy as np
5import pandas as pd
6from loguru import logger
9# ----------------------------------------------------------------------
10# MODEL FOR OPTIMIZING THE BACKTEST PERIODS
11# ----------------------------------------------------------------------
12def rebalancing_model(
13 mu,
14 scenarios,
15 cvar_targets,
16 cvar_alpha,
17 cash,
18 x_old,
19 trans_cost,
20 max_weight,
21 solver,
22 inaccurate,
23 lower_bound,
24):
25 """This function finds the optimal enhanced index portfolio according to some benchmark.
26 The portfolio corresponds to the tangency portfolio where risk is evaluated according to
27 the CVaR of the tracking error. The model is formulated using fractional programming.
29 Parameters
30 ----------
31 mu : pandas.Series with float values
32 asset point forecast
33 scenarios : pandas.DataFrame with float values
34 Asset scenarios
35 cvar_targets:
36 cvar targets for our optimal portfolio
37 cash:
38 additional budget for our portfolio
39 x_old:
40 old portfolio allocation
41 trans_cost:
42 transaction costs
43 max_weight : float
44 Maximum allowed weight
45 cvar_alpha : float
46 Alpha value used to evaluate Value-at-Risk one
47 solver: str
48 The name of the solver to use, as returned by cvxpy.installed_solvers()
49 inaccurate: bool
50 Whether to also use solution with status "optimal_inaccurate"
51 lower_bound: int
52 Minimum weight given to each selected asset.
54 Returns
55 -------
56 float
57 Asset weights in an optimal portfolio
58 """
60 # Define index
61 i_idx = scenarios.columns
62 N = i_idx.size
64 # Number of scenarios
65 T = scenarios.shape[0]
66 # Variable transaction costs
67 c = trans_cost
69 # Define variables
70 # - portfolio
71 x = cp.Variable(N, name="x", nonneg=True)
72 # - |x - x_old|
73 absdiff = cp.Variable(N, name="absdiff", nonneg=True)
74 # - cost
75 cost = cp.Variable(name="cost", nonneg=True)
76 # - loss deviation
77 vardev = cp.Variable(T, name="vardev", nonneg=True)
78 # - VaR and CVaR
79 var = cp.Variable(name="var", nonneg=True)
80 cvar = cp.Variable(name="cvar", nonneg=True)
82 # Define objective (max expected portfolio return)
83 objective = cp.Maximize(mu.to_numpy() @ x)
85 # Define constraints
86 constraints = [
87 # - VaR deviation
88 -scenarios.to_numpy() @ x - var <= vardev,
89 # - CVaR limit
90 var + 1 / (T * cvar_alpha) * cp.sum(vardev) == cvar,
91 cvar <= cvar_targets,
92 # - Cost of rebalancing
93 c * cp.sum(absdiff) == cost,
94 x - x_old <= absdiff,
95 x - x_old >= -absdiff,
96 # - Budget
97 x_old.sum() + cash - cp.sum(x) - cost == 0,
98 # - Concentration limits
99 x <= max_weight * cp.sum(x),
100 ]
102 if lower_bound != 0:
103 z = cp.Variable(N, boolean=True) # Binary variable indicates if asset is selected
104 upper_bound = 100
106 constraints.append(lower_bound * z <= x)
107 constraints.append(x <= upper_bound * z)
108 constraints.append(cp.sum(z) >= 1)
110 # Define model
111 model = cp.Problem(objective=objective, constraints=constraints)
113 # Solve
114 model.solve(solver=solver, verbose=False)
116 # Get positions
117 accepted_statuses = ["optimal"]
118 if inaccurate:
119 accepted_statuses.append("optimal_inaccurate")
120 if model.status in accepted_statuses:
121 opt_port = pd.Series(x.value, index=mu.index)
123 # Set floating data points to zero and normalize
124 opt_port[np.abs(opt_port) < 0.000001] = 0
125 port_val = np.sum(opt_port)
126 cvar_result_p = cvar.value / port_val
127 opt_port = opt_port / port_val
129 # Remaining cash
130 cash = cash - (port_val + cost.value - x_old.sum())
132 # return portfolio, CVaR, and alpha
133 return opt_port, cvar_result_p, port_val, cash
135 else:
136 # Save inputs, so that failing problem can be investigated separately e. g. in a notebook
137 inputs = {
138 "mu": mu,
139 "scenarios": scenarios,
140 "cvar_targets": cvar_targets,
141 "cvar_alpha": cvar_alpha,
142 "cash": cash,
143 "x_old": x_old,
144 "trans_cost": trans_cost,
145 "max_weight": max_weight,
146 }
147 file = open("rebalance_inputs.pkl", "wb")
148 pickle.dump(inputs, file)
149 file.close()
151 # Print an error if the model is not optimal
152 logger.exception(f"❌ Solver does not find optimal solution. Status code is {model.status}")
155# ----------------------------------------------------------------------
156# Mathematical Optimization: RUN THE CVAR MODEL
157# ----------------------------------------------------------------------
158def cvar_model(
159 test_ret: pd.DataFrame,
160 scenarios: np.ndarray,
161 targets: pd.DataFrame,
162 budget: float,
163 cvar_alpha: float,
164 trans_cost: float,
165 max_weight: float,
166 solver: str,
167 inaccurate: bool = True,
168 lower_bound=int,
169) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
170 """
171 Method to run the CVaR model over given periods
172 """
173 p_points, s_points, _ = scenarios.shape # number of periods, number of scenarios
174 prob = 1 / s_points # probability of each scenario
176 assets = test_ret.columns # names of all assets
178 # LIST TO STORE CVaR TARGETS
179 list_portfolio_cvar = []
180 # LIST TO STORE VALUE OF THE PORTFOLIO
181 list_portfolio_value = []
182 # LIST TO STORE PORTFOLIO ALLOCATION
183 list_portfolio_allocation = []
185 x_old = pd.Series(0, index=assets)
186 cash = budget
187 portfolio_value_w = budget
189 logger.info(f"🤖 Selected solver is {solver}")
190 for p in range(p_points):
191 logger.info(f"🚀 Optimizing period {p + 1} out of {p_points}.")
193 # Create dataframe with scenarios for a period p
194 scenarios_df = pd.DataFrame(scenarios[p, :, :], columns=test_ret.columns)
196 # compute expected returns of all assets (EP)
197 expected_returns = sum(prob * scenarios_df.loc[i, :] for i in scenarios_df.index)
199 # run CVaR model
200 p_alloc, cvar_val, port_val, cash = rebalancing_model(
201 mu=expected_returns,
202 scenarios=scenarios_df,
203 cvar_targets=targets.loc[p, "CVaR_Target"] * portfolio_value_w,
204 cvar_alpha=cvar_alpha,
205 cash=cash,
206 x_old=x_old,
207 trans_cost=trans_cost,
208 max_weight=max_weight,
209 solver=solver,
210 inaccurate=inaccurate,
211 lower_bound=lower_bound,
212 )
214 # save the result
215 list_portfolio_cvar.append(cvar_val)
216 # save allocation
217 list_portfolio_allocation.append(p_alloc)
219 portfolio_value_w = port_val
220 # COMPUTE PORTFOLIO VALUE
221 for w in test_ret.index[(p * 4) : (4 + p * 4)]:
222 portfolio_value_w = sum(p_alloc * portfolio_value_w * (1 + test_ret.loc[w, assets]))
223 list_portfolio_value.append((w, portfolio_value_w))
225 x_old = p_alloc * portfolio_value_w
227 portfolio_cvar = pd.DataFrame(columns=["CVaR"], data=list_portfolio_cvar)
228 portfolio_value = pd.DataFrame(columns=["Date", "Portfolio_Value"], data=list_portfolio_value).set_index(
229 "Date", drop=True
230 )
231 portfolio_allocation = pd.DataFrame(columns=assets, data=list_portfolio_allocation)
233 return portfolio_allocation, portfolio_value, portfolio_cvar