Coverage for src/ifunnel/models/MVOmodel.py: 76%
74 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
6import scipy as sp
7from loguru import logger
10def cholesky_psd(m):
11 """
12 Computes the Cholesky decomposition of the given matrix, that is not positive definite, only semidefinite.
13 """
14 lu, d, perm = sp.linalg.ldl(m)
15 assert np.max(np.abs(d - np.diag(np.diag(d)))) < 1e-12, "Matrix 'd' is not diagonal!"
17 # Do non-negativity fix
18 min_eig = np.min(np.diag(d))
19 if min_eig < 0:
20 d -= 5 * min_eig * np.eye(*d.shape)
22 sqrtd = sp.linalg.sqrtm(d)
23 C = (lu @ sqrtd).T
24 return C
27# ----------------------------------------------------------------------
28# MODEL FOR OPTIMIZING THE BACKTEST PERIODS
29# ----------------------------------------------------------------------
30def rebalancing_model(
31 mu,
32 covariance,
33 vty_target,
34 cash,
35 x_old,
36 trans_cost,
37 max_weight,
38 solver,
39 inaccurate,
40 lower_bound,
41):
42 """This function finds the optimal enhanced index portfolio according to some benchmark.
43 The portfolio corresponds to the tangency portfolio where risk is evaluated according to
44 the volatility of the tracking error. The model is formulated using quadratic programming.
46 Parameters
47 ----------
48 mu : pandas.Series with float values
49 asset point forecast
50 covariance : pandas.DataFrame with covariances
51 Asset covariances
52 vty_target:
53 targets for our optimal portfolio
54 cash:
55 additional budget for our portfolio
56 x_old:
57 old portfolio allocation
58 trans_cost:
59 transaction costs
60 max_weight : float
61 Maximum allowed weight
62 solver: str
63 The name of the solver to use, as returned by cvxpy.installed_solvers()
64 inaccurate: bool
65 Whether to also use solution with status "optimal_inaccurate"
66 lower_bound: int
67 Minimum weight given to each selected asset.
69 Returns
70 -------
71 float
72 Asset weights in an optimal portfolio
73 """
74 # Number of assets
75 N = covariance.shape[1]
77 # Variable transaction costs
78 c = trans_cost
80 # Factorize the covariance
81 # G = cholesky_psd(covariance)
82 G = np.linalg.cholesky(covariance)
84 # Define variables
85 # - portfolio
86 x = cp.Variable(N, name="x", nonneg=True)
87 # - |x - x_old|
88 absdiff = cp.Variable(N, name="absdiff", nonneg=True)
89 # - cost
90 cost = cp.Variable(name="cost", nonneg=True)
92 # Define objective (max expected portfolio return)
93 objective = cp.Maximize(mu.to_numpy() @ x)
95 # Define constraints
96 constraints = [
97 # - Volatility limit
98 cp.norm(G @ x, 2) <= vty_target,
99 # - Cost of rebalancing
100 c * cp.sum(absdiff) == cost,
101 x - x_old <= absdiff,
102 x - x_old >= -absdiff,
103 # - Budget
104 x_old.sum() + cash - cp.sum(x) - cost == 0,
105 # - Concentration limits
106 x <= max_weight * cp.sum(x),
107 ]
109 if lower_bound != 0:
110 z = cp.Variable(N, boolean=True) # Binary variable indicates if asset is selected
111 upper_bound = 100
113 constraints.append(lower_bound * z <= x)
114 constraints.append(x <= upper_bound * z)
115 constraints.append(cp.sum(z) >= 1)
117 # Define model
118 model = cp.Problem(objective=objective, constraints=constraints)
120 # Solve
121 model.solve(solver=solver, verbose=False)
123 # Get positions
124 accepted_statuses = ["optimal"]
125 if inaccurate:
126 accepted_statuses.append("optimal_inaccurate")
127 if model.status in accepted_statuses:
128 opt_port = pd.Series(x.value, index=mu.index)
130 # Set floating data points to zero and normalize
131 opt_port[np.abs(opt_port) < 0.000001] = 0
132 port_val = np.sum(opt_port)
133 vty_result_p = np.linalg.norm(G @ x.value, 2) / port_val
134 opt_port = opt_port / port_val
136 # Remaining cash
137 cash = cash - (port_val + cost.value - x_old.sum())
139 # return portfolio, CVaR, and alpha
140 return opt_port, vty_result_p, port_val, cash
142 else:
143 # Save inputs, so that failing problem can be investigated separately e.g. in a notebook
144 inputs = {
145 "mu": mu,
146 "covariance": covariance,
147 "vty_target": vty_target,
148 "cash": cash,
149 "x_old": x_old,
150 "trans_cost": trans_cost,
151 "max_weight": max_weight,
152 "lower_bound": lower_bound,
153 }
154 file = open("rebalance_inputs.pkl", "wb")
155 pickle.dump(inputs, file)
156 file.close()
158 # Print an error if the model is not optimal
159 logger.exception(f"❌ Solver does not find optimal solution. Status code is {model.status}")
162# ----------------------------------------------------------------------
163# Mathematical Optimization: RUN THE MVO MODEL
164# ----------------------------------------------------------------------
165def mvo_model(
166 test_ret: pd.DataFrame,
167 mu_lst: list,
168 sigma_lst: list,
169 targets: pd.DataFrame,
170 budget: float,
171 trans_cost: float,
172 max_weight: float,
173 solver: str,
174 lower_bound: int,
175 inaccurate: bool = True,
176) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
177 """
178 Method to run the MVO model over given periods
179 """
180 p_points = len(mu_lst) # number of periods
182 assets = test_ret.columns # names of all assets
184 # LIST TO STORE VOLATILITY TARGETS
185 list_portfolio_vty = []
186 # LIST TO STORE VALUE OF THE PORTFOLIO
187 list_portfolio_value = []
188 # LIST TO STORE PORTFOLIO ALLOCATION
189 list_portfolio_allocation = []
191 x_old = pd.Series(0, index=assets)
192 cash = budget
193 portfolio_value_w = budget
195 logger.debug(f"🤖 Selected solver is {solver}")
196 for p in range(p_points):
197 logger.info(f"🚀 Optimizing period {p + 1} out of {p_points}.")
199 # Get MVO parameters
200 mu = mu_lst[p]
201 sigma = sigma_lst[p]
203 # run CVaR model
204 p_alloc, vty_val, port_val, cash = rebalancing_model(
205 mu=mu,
206 covariance=sigma,
207 vty_target=targets.loc[p, "Vty_Target"] * portfolio_value_w,
208 cash=cash,
209 x_old=x_old,
210 trans_cost=trans_cost,
211 max_weight=max_weight,
212 solver=solver,
213 inaccurate=inaccurate,
214 lower_bound=lower_bound,
215 )
217 # save the result
218 list_portfolio_vty.append(vty_val)
219 # save allocation
220 list_portfolio_allocation.append(p_alloc)
222 portfolio_value_w = port_val
223 # COMPUTE PORTFOLIO VALUE
224 for w in test_ret.index[(p * 4) : (4 + p * 4)]:
225 portfolio_value_w = sum(p_alloc * portfolio_value_w * (1 + test_ret.loc[w, assets]))
226 list_portfolio_value.append((w, portfolio_value_w))
228 x_old = p_alloc * portfolio_value_w
230 portfolio_vty = pd.DataFrame(columns=["Volatility"], data=list_portfolio_vty)
231 portfolio_value = pd.DataFrame(columns=["Date", "Portfolio_Value"], data=list_portfolio_value).set_index(
232 "Date", drop=True
233 )
234 portfolio_allocation = pd.DataFrame(columns=assets, data=list_portfolio_allocation)
236 return portfolio_allocation, portfolio_value, portfolio_vty