Coverage for src/ifunnel/models/lifecycle/MVOlifecycleModel.py: 0%
125 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-12 09:14 +0000
1import cvxpy as cp
2import numpy as np
3import pandas as pd
4from loguru import logger
6from ..MVOmodel import cholesky_psd
9def calculate_risk_metrics(
10 yearly_returns: pd.Series, risk_free_rate: float = 0.02
11) -> (float, float, float, float, float):
12 """Calculate risk metrics for a given set of yearly returns."""
13 annual_return = yearly_returns.mean()
14 annual_std_dev = yearly_returns.std()
16 sharpe_ratio = (annual_return - risk_free_rate) / annual_std_dev if annual_std_dev != 0 else None
18 # Calculate downside deviation
19 downside_diffs = [min(0, return_ - risk_free_rate) for return_ in yearly_returns]
20 downside_std_dev = np.std(downside_diffs)
22 # Calculate Sortino ratio
23 sortino_ratio = (annual_return - risk_free_rate) / downside_std_dev if downside_std_dev != 0 else None
25 return annual_return, annual_std_dev, sharpe_ratio, downside_std_dev, sortino_ratio
28def calculate_analysis_metrics(terminal_values: pd.Series) -> pd.DataFrame:
29 """Calculate various metrics from the terminal values of the portfolio."""
30 mean_terminal_value = np.mean(terminal_values)
31 stdev_terminal_value = np.std(terminal_values)
32 max_terminal_value = np.max(terminal_values)
33 min_terminal_value = np.min(terminal_values)
35 # Sorting for decile and quartile calculations
36 sorted_terminal_values = sorted(terminal_values)
37 lower_decile_count = len(sorted_terminal_values) // 10
38 upper_decile_count = len(sorted_terminal_values) - lower_decile_count
39 lower_quartile_count = len(sorted_terminal_values) // 4
40 upper_quartile_count = len(sorted_terminal_values) - lower_quartile_count
42 lower_decile_avg = np.mean(sorted_terminal_values[:lower_decile_count])
43 upper_decile_avg = np.mean(sorted_terminal_values[-upper_decile_count:])
44 lower_quartile_avg = np.mean(sorted_terminal_values[:lower_quartile_count])
45 upper_quartile_avg = np.mean(sorted_terminal_values[-upper_quartile_count:])
47 # Creating a DataFrame to store the metrics
48 metrics_df = pd.DataFrame(
49 {
50 "Mean Terminal Value": [mean_terminal_value],
51 "Standard Deviation Terminal Value": [stdev_terminal_value],
52 "Max Terminal Value": [max_terminal_value],
53 "Min Terminal Value": [min_terminal_value],
54 "Lower Decile Average": [lower_decile_avg],
55 "Upper Decile Average": [upper_decile_avg],
56 "Lower Quartile Average": [lower_quartile_avg],
57 "Upper Quartile Average": [upper_quartile_avg],
58 }
59 )
61 return metrics_df
64def lifecycle_rebalance_model(
65 mu: pd.DataFrame,
66 sigma: pd.DataFrame,
67 vol_target: float,
68 max_weight: float,
69 solver: str,
70 inaccurate: bool = True,
71 lower_bound: float = 0,
72) -> (pd.Series, float):
73 """
74 Optimizes asset allocations within a portfolio to maximize expected returns
75 while adhering to a risk budget glide path.
77 Parameters:
78 - mu: Expected returns for each asset.
79 - sigma: Covariance matrix of asset returns.
80 - vol_target: Target portfolio volatility for period.
81 - max_weight: Maximum weight allowed for any single asset (excluding cash).
82 - solver: Solver to be used by CVXPY for optimization.
83 - inaccurate: If True, accepts 'optimal_inaccurate' as a successful solve status.
85 Returns:
86 - port_nom: Nominal allocations for each asset in the optimized portfolio.
87 - port_val: Total value of the portfolio based on the allocations.
89 This function uses convex optimization to find the asset weights that maximize
90 expected returns subject to constraints on total weight, individual asset weights,
91 and portfolio volatility. It optionally includes binary selection variables to enforce
92 a minimum allocation to any selected asset.
93 """
95 # Prepare basic variables and indices
96 N = len(mu) # Number of assets
97 cash_index = mu.index.get_loc("Cash") # Identify the index of the 'Cash' asset
98 non_cash_indices = np.array([i for i in range(len(mu)) if i != cash_index])
100 # Optimization variables
101 x = cp.Variable(N, name="x", nonneg=True) # Asset weights
103 # Prepare matrix for volatility constraint
104 G = cholesky_psd(sigma) # Transform to standard deviation matrix
106 # Define the optimization problem
107 objective = cp.Maximize(x.T @ mu)
109 constraints = [
110 cp.sum(x) == 1, # Weights sum to 1
111 cp.norm(G @ x, 2) <= vol_target, # Portfolio volatility constraint
112 # cp.quad_form(x, sigma) <= vol_target ** 2,
113 x[non_cash_indices] <= max_weight * cp.sum(x[non_cash_indices]), # Max weight constraint for non-cash assets
114 ]
116 # Optional lower bound constraint
117 if lower_bound != 0:
118 z = cp.Variable(N, boolean=True) # Binary variable indicates if asset is selected
119 upper_bound = 100 # Arbitrary upper bound for asset weights
121 constraints += [
122 lower_bound * z <= x, # Lower bound constraint
123 x <= upper_bound * z, # Upper bound enables asset deselection
124 cp.sum(z) >= 1, # At least one asset must be selected
125 ]
127 # Solve the optimization problem
128 model = cp.Problem(objective, constraints)
129 model.solve(solver=solver, qcp=True)
131 # Process optimization results
132 accepted_statuses = ["optimal"]
133 if inaccurate:
134 accepted_statuses.append("optimal_inaccurate")
135 if model.status in accepted_statuses:
136 port_nom = pd.Series(x.value, index=mu.index) # Nominal allocations in each asset
137 port_val = np.sum(port_nom) # Total portfolio value
138 port_nom[np.abs(port_nom) < 0.00001] = 0 # Eliminate numerical noise by zeroing very small weights
139 return port_nom, port_val
141 else:
142 # Handle non-optimal solve status
143 logger.exception(
144 f"The model is {model.status}. Look into the constraints. It might be an issue of too low risk targets."
145 )
146 port_nom = pd.Series(np.nan, index=mu.index) # Use NaNs to indicate failure
147 port_val = np.sum(port_nom)
148 return port_nom, port_val
151def get_port_allocations(
152 mu_lst: pd.DataFrame,
153 sigma_lst: pd.DataFrame,
154 targets: pd.DataFrame,
155 max_weight: float,
156 solver: str,
157) -> pd.DataFrame:
158 """
159 Calculates optimal portfolio allocations for the glide paths.
161 Parameters:
162 - mu_lst, sigma_lst: Expected returns and standard deviations for each year
163 - targets: Target volatilities from the risk budget glide paths
164 - budget: Initial budget
165 - trans_cost: Transaction costs
166 - max_weight: Maximum weight for any asset
167 - withdrawal_lst: List of withdrawals for each year
168 - solver: Optimization solver to use
169 - interest_rate: Interest rate for borrowing
171 Returns:
172 - allocation_df: DataFrame showing the optimal asset allocations for each year.
174 This function iterates through each year, using the provided expected returns, covariance matrices,
175 and risk budget glide path to determine the optimal asset allocations.
176 """
177 # Initial setup
178 num_years = len(targets)
179 years = [str(i + 2023) for i in range(num_years)]
180 assets = mu_lst.index
182 # Initialize DataFrames
183 allocation_df = pd.DataFrame(index=years, columns=assets)
185 for year in range(num_years):
186 port_weights, _ = lifecycle_rebalance_model(
187 mu=mu_lst,
188 sigma=sigma_lst,
189 vol_target=targets.loc[year],
190 max_weight=max_weight,
191 solver=solver,
192 )
194 allocation_df.loc[allocation_df.index[year], :] = port_weights
196 logger.info(f"The optimal portfolio allocations has been obtained for the {num_years + 1} years.")
197 return allocation_df
200def portfolio_rebalancing(
201 budget: float,
202 targets: pd.DataFrame,
203 withdrawal_lst: list[float],
204 transaction_cost: float,
205 scenarios: pd.DataFrame,
206 interest_rate: float,
207) -> (pd.DataFrame, pd.DataFrame):
208 """
209 Simulates portfolio rebalancing over multiple years, accounting for withdrawals,
210 transaction costs, returns based on scenarios, and handling defaults when withdrawals exceed portfolio value.
212 Parameters:
213 - budget: Initial budget available for investment.
214 - targets: DataFrame containing target asset allocations for each year.
215 - withdrawal_lst: List of annual withdrawal amounts.
216 - transaction_cost: Transaction cost rate applied to rebalancing and withdrawals.
217 - scenarios: DataFrame containing asset return scenarios for each year.
218 - interest_rate: Interest rate applied to borrowed amounts in case of default.
220 Returns:
221 - ptf_performance: DataFrame for portfolio performance for each year.
222 - allocation_df: DataFrame showing asset allocations for each year.
223 """
225 # Initialize the number of years and assets from the targets DataFrame
226 n_years, n_assets = targets.shape
227 years = [i + 2023 for i in range(n_years)]
228 assets = targets.columns
230 # Prepare DataFrame to store portfolio performance
231 ptf_performance_columns = [
232 "Portfolio Value Primo",
233 "Portfolio Value Ultimo",
234 "Withdrawal",
235 "Returns in DKK",
236 "Yearly Returns",
237 "Transaction Costs",
238 "Absolute DKK Rebalanced",
239 "Borrowed Amount",
240 ]
241 ptf_performance = pd.DataFrame(index=years, columns=ptf_performance_columns)
243 # DataFrame for tracking asset allocations each year
244 allocation_df = pd.DataFrame(0, index=years, columns=assets, dtype=float)
246 # Initialize portfolio values
247 default_year, borrowed_amount, interest_for_the_year = 0, 0, 0
248 x_old, portfolio_value_ultimo_aw = pd.Series(0, index=targets.columns), budget
250 for year in range(n_years):
251 # Handle scenario where the portfolio is in default
252 if default_year > 0:
253 withdrawal_amount = withdrawal_lst[year]
254 borrowed_amount += withdrawal_amount
255 interest_for_the_year = borrowed_amount * interest_rate
256 borrowed_amount += interest_for_the_year # Accumulate interest
258 # Update performance DataFrame for periods in default
259 ptf_performance.loc[ptf_performance.index[year]] = {
260 "Portfolio Value Primo": ptf_performance["Portfolio Value Ultimo"][ptf_performance.index[year - 1]],
261 "Portfolio Value Ultimo": -borrowed_amount,
262 "Withdrawal": withdrawal_lst[year],
263 "Returns in DKK": -interest_for_the_year,
264 "Yearly Returns": -interest_rate,
265 "Transaction Costs": 0,
266 "Absolute DKK Rebalanced": 0,
267 "Borrowed Amount": borrowed_amount - interest_for_the_year,
268 }
269 ptf_performance["Default Year"] = default_year
270 continue
272 # Normal operation: calculate returns, rebalancing, and manage withdrawals
273 port_weights = targets.iloc[year]
274 absolute_rebalance = (port_weights - x_old).abs().sum() * portfolio_value_ultimo_aw
275 costs_rebalance = absolute_rebalance * transaction_cost
277 portfolio_value_primo = portfolio_value_ultimo_aw - costs_rebalance
279 year_return = np.dot(scenarios.loc[year], port_weights)
280 portfolio_value_ultimo_bw = portfolio_value_primo * (1 + year_return)
282 # Check if withdrawals exceed the portfolio value, leading to default
283 if withdrawal_lst[year] >= portfolio_value_ultimo_bw * (1 + transaction_cost):
284 withdrawal_amount = portfolio_value_ultimo_bw * (1 - transaction_cost)
285 default_year = year + 2023
286 borrowed_amount = withdrawal_lst[year] - withdrawal_amount
287 interest_for_the_year = borrowed_amount * interest_rate
288 borrowed_amount += interest_for_the_year
290 else:
291 withdrawal_amount = withdrawal_lst[year]
293 costs_withdraw = withdrawal_amount * transaction_cost
294 costs_total = costs_rebalance + costs_withdraw
295 portfolio_value_ultimo_aw = portfolio_value_ultimo_bw - (withdrawal_amount + costs_withdraw)
297 x_old = port_weights
299 # Update allocation DataFrame for the year
300 allocation_df.loc[allocation_df.index[year], :] = port_weights
302 # Update summary DataFrame for normal operation
303 ptf_performance.loc[ptf_performance.index[year]] = {
304 "Portfolio Value Primo": portfolio_value_primo,
305 "Portfolio Value Ultimo": portfolio_value_ultimo_aw - borrowed_amount,
306 "Withdrawal": withdrawal_lst[year],
307 "Returns in DKK": portfolio_value_primo * year_return,
308 "Yearly Returns": year_return,
309 "Transaction Costs": costs_total,
310 "Absolute DKK Rebalanced": absolute_rebalance,
311 "Borrowed Amount": borrowed_amount,
312 }
313 ptf_performance["Default Year"] = default_year
315 return ptf_performance, allocation_df
318def riskadjust_model_scen(
319 scen: np.ndarray,
320 targets: pd.DataFrame,
321 budget: float,
322 trans_cost: float,
323 withdrawal_lst: list[float],
324 interest_rate: float,
325) -> (pd.DataFrame, pd.DataFrame, pd.DataFrame):
326 """
327 Simulates portfolio performance across different scenarios, adjusting for risk and calculating
328 various financial metrics based on the portfolio's rebalancing strategy.
330 Parameters:
331 - scen: 3D numpy array containing return scenarios (scenarios, periods, assets).
332 - targets: DataFrame specifying target asset allocations for each period.
333 - budget: Initial budget for investment.
334 - trans_cost: Transaction costs as a fraction of the trade amount.
335 - withdrawal_lst: List of annual withdrawal amounts.
336 - interest_rate: Interest rate applied to borrowed amounts in case of default.
338 Returns:
339 - portfolio_df: DataFrame summarizing the performance metrics for each scenario.
340 - mean_allocations_df: DataFrame showing the average asset allocations across all scenarios for each period.
341 - analysis_metrics: Dictionary containing overall analysis metrics calculated from portfolio_df.
343 The function iterates through each scenario, rebalances the portfolio according to the targets,
344 and calculates performance metrics such as total returns, costs, withdrawals, and risk-adjusted measures.
345 It also tracks the occurrence of default events and calculates average allocations and other analysis metrics.
346 """
348 s_points, p_points, a_points = scen.shape # Scenario, periods, and assets dimensions
349 assets = targets.columns # Asset names from the targets DataFrame
351 # Initialize DataFrame to hold portfolio performance metrics for each scenario
352 portfolio_df = pd.DataFrame(
353 columns=[
354 "Terminal Wealth",
355 "Total Returns",
356 "Total Costs",
357 "Total Withdrawals",
358 "Default Year",
359 "Average Cash Hold",
360 "Annual StDev",
361 "Annual StDev_dd",
362 "Average Annual Return",
363 "Sharpe Ratio",
364 "Sortino Ratio",
365 "Total borrowed",
366 ],
367 index=range(s_points),
368 )
369 # Initialize array to hold asset allocations for all scenarios
370 all_allocations = np.zeros((s_points, p_points, a_points))
372 for scenario in range(s_points):
373 # Convert scenario data to DataFrame for processing
374 scenarios_df = pd.DataFrame(scen[scenario, :, :], columns=assets)
376 # Perform portfolio rebalancing for the scenario
377 res, res_alloc = portfolio_rebalancing(
378 budget=budget,
379 targets=targets,
380 withdrawal_lst=withdrawal_lst,
381 transaction_cost=trans_cost,
382 scenarios=scenarios_df,
383 interest_rate=interest_rate,
384 )
386 # Store the allocation results for this scenario
387 all_allocations[scenario] = res_alloc.to_numpy()
389 # Calculate risk metrics for the scenario
390 annual_return, annual_std_dev, sr, downside_std_dev, sortino_ratio = calculate_risk_metrics(
391 res["Yearly Returns"]
392 )
394 # Update the portfolio DataFrame with calculated metrics for the scenario
395 portfolio_df.loc[scenario] = {
396 "Terminal Wealth": res["Portfolio Value Ultimo"].iloc[-1],
397 "Total Returns": res["Returns in DKK"].sum(),
398 "Total Costs": res["Transaction Costs"].sum(),
399 "Total Withdrawals": res["Withdrawal"].sum(),
400 "Default Year": res["Default Year"].max(),
401 "Average Cash Hold": res_alloc["Cash"].mean(),
402 "Annual StDev": annual_std_dev,
403 "Annual StDev_dd": downside_std_dev,
404 "Average Annual Return": annual_return,
405 "Sharpe Ratio": sr,
406 "Sortino Ratio": sortino_ratio,
407 "Total borrowed": res["Borrowed Amount"].sum(),
408 }
410 if scenario % (s_points // 2) == 0 and scenario != 0:
411 logger.info(f"{scenario} out of {s_points} scenarios finished")
413 # Log progress and calculate default count
414 default_count = (portfolio_df["Default Year"] != 0).sum()
416 # Reshape and aggregate allocation data for analysis
417 index = pd.MultiIndex.from_product([range(s_points), range(p_points)], names=["scenario", "period"])
418 allocations_long = all_allocations.reshape(-1, a_points) # Reshape to long format
419 allocations_df = pd.DataFrame(allocations_long, index=index, columns=assets)
420 mean_allocations_df = allocations_df.groupby("period").mean() # Mean allocations by period
422 # Calculate overall analysis metrics from portfolio performance
423 analysis_metrics = calculate_analysis_metrics(portfolio_df["Terminal Wealth"])
425 logger.info(
426 f"{s_points} out of {s_points} scenarios has now been made. We saw a total of {default_count} "
427 f"scenarios where the risk budget glide path strategy defaulted over the period of {p_points} years."
428 )
429 return portfolio_df, mean_allocations_df, analysis_metrics