Merge pull request #25 from cvxgrp/regularization-experiments

Regularization examples

experiments/backtest.py

@@ -40,6 +40,10 @@ class OptimizationInput:
     cash: float
     risk_target: float
 
+    @property
+    def n_assets(self) -> int:
+        return self.prices.shape[1]
+
 
 def run_backtest(
     strategy: Callable, risk_target: float, verbose: bool = False
@@ -108,7 +112,7 @@ def run_backtest(
         cash += execute_orders(latest_prices, trade_quantities, latest_spread)
 
         post_trade_cash.append(cash)
-        post_trade_quantities.append(quantities)
+        post_trade_quantities.append(quantities.copy())
 
     post_trade_cash = pd.Series(post_trade_cash, index=prices.index[lookback:-1])
     post_trade_quantities = pd.DataFrame(

experiments/taming.py

@@ -1,25 +1,23 @@
-import os
 import numpy as np
 import pandas as pd
 import cvxpy as cp
 from backtest import BacktestResult, OptimizationInput, run_backtest
-from markowitz import Data, Parameters
+from markowitz import Data, Parameters, markowitz
 import matplotlib.pyplot as plt
 
 
 def unconstrained_markowitz(
     inputs: OptimizationInput, long_only: bool = False
 ) -> np.ndarray:
     """Compute the unconstrained (or long-only) Markowitz portfolio weights."""
-    n_assets = inputs.prices.shape[1]
 
     mu, Sigma = inputs.mean.values, inputs.covariance.values
 
     if long_only:
-        w = cp.Variable(n_assets, nonneg=True)
+        w = cp.Variable(inputs.n_assets, nonneg=True)
         c = cp.Variable(nonneg=True)
     else:
-        w = cp.Variable(n_assets)
+        w = cp.Variable(inputs.n_assets)
         c = cp.Variable()
     objective = mu @ w
 
@@ -48,91 +46,105 @@ def equal_weights(inputs: OptimizationInput) -> np.ndarray:
     return w, c
 
 
-def ewma_mean_covariance(
-    prices: pd.DataFrame, lamb: float = 0.94
-) -> tuple[np.ndarray, np.ndarray]:
-    returns = prices.pct_change().dropna()
-    n_assets = returns.shape[1]
-    alpha = 1 - lamb
-    mu = returns.ewm(alpha=alpha).mean().iloc[-1].values
-    Sigma = returns.ewm(alpha=alpha).cov().iloc[-n_assets:].values
-    return mu, Sigma
-
-
-def prepare_data(
-    prices: pd.DataFrame,
-    spread: pd.DataFrame,
-    volume: pd.DataFrame,
-    quantities: np.ndarray,
-    cash: float,
-) -> Data:
-    n_assets = prices.shape[1]
-    latest_prices = prices.iloc[-1]
-    portfolio_value = cash + quantities @ latest_prices
-
-    mu, Sigma = ewma_mean_covariance(prices)
-
-    return Data(
-        w_prev=quantities * latest_prices / portfolio_value,
-        c_prev=cash / portfolio_value,
-        idio_mean=mu,
-        factor_mean=np.zeros(n_assets),
+def weight_limits_markowitz(inputs: OptimizationInput) -> np.ndarray:
+    lb = np.ones(inputs.n_assets) * (-0.05)
+    ub = np.ones(inputs.n_assets) * 0.1
+
+    data, param = get_unconstrained_data_and_parameters(inputs)
+
+    param.w_lower = lb
+    param.w_upper = ub
+    param.c_lower = -0.05
+    param.c_upper = 1.0
+    return markowitz(data, param)
+
+
+def leverage_limit_markowitz(inputs: OptimizationInput) -> np.ndarray:
+    data, param = get_unconstrained_data_and_parameters(inputs)
+
+    param.L_max = 1.5
+    return markowitz(data, param)
+
+
+def get_unconstrained_data_and_parameters(
+    inputs: OptimizationInput,
+) -> tuple[Data, Parameters]:
+    n_assets = inputs.n_assets
+    latest_prices = inputs.prices.iloc[-1]
+    portfolio_value = inputs.cash + inputs.quantities @ latest_prices
+
+    data = Data(
+        w_prev=(inputs.quantities * latest_prices / portfolio_value),
+        c_prev=(inputs.cash / portfolio_value),
+        idio_mean=np.zeros(n_assets),
+        factor_mean=inputs.mean.values,
         risk_free=0,
-        factor_covariance_chol=np.linalg.cholesky(Sigma),
-        idio_volas=np.sqrt(np.diag(Sigma)),
+        factor_covariance_chol=np.linalg.cholesky(inputs.covariance.values),
+        idio_volas=np.zeros(n_assets),
         F=np.eye(n_assets),
         kappa_short=np.zeros(n_assets),
         kappa_borrow=0.0,
         kappa_spread=np.zeros(n_assets),
         kappa_impact=np.zeros(n_assets),
     )
-
-
-def get_parameters(data, risk_target):
-    return Parameters(
-        w_lower=np.zeros(data.n_assets),
-        w_upper=np.ones(data.n_assets),
-        c_lower=0.0,
-        c_upper=1.0,
-        z_lower=-np.ones(data.n_assets),
-        z_upper=np.ones(data.n_assets),
-        T_max=0.1,
-        L_max=1.5,
+    param = Parameters(
+        w_lower=-np.ones(data.n_assets) * 1e3,
+        w_upper=np.ones(data.n_assets) * 1e3,
+        c_lower=-1e3,
+        c_upper=1e3,
+        z_lower=-np.ones(data.n_assets) * 1e3,
+        z_upper=np.ones(data.n_assets) * 1e3,
+        T_max=1e3,
+        L_max=1e3,
         rho_mean=np.zeros(data.n_assets),
         rho_covariance=0.0,
         gamma_hold=0.0,
         gamma_trade=0.0,
         gamma_turn=0.0,
         gamma_risk=0.0,
-        risk_target=risk_target,
+        risk_target=inputs.risk_target,
     )
+    return data, param
+
 
+def main(from_checkpoint: bool = True):
+    annualized_target = 0.13
 
-def main(from_checkpoint: bool = False):
     if from_checkpoint:
-        unconstrained_files = [
-            f for f in os.listdir("checkpoints") if f.startswith("unconstrained")
-        ]
-        assert len(unconstrained_files) == 1
         unconstrained_result = BacktestResult.load(
-            f"checkpoints/{unconstrained_files[0]}"
+            f"checkpoints/unconstrained_{annualized_target}.pickle"
+        )
+        long_only_result = BacktestResult.load(
+            f"checkpoints/long_only_{annualized_target}.pickle"
+        )
+        weight_limited_result = BacktestResult.load(
+            f"checkpoints/weight_limited_{annualized_target}.pickle"
         )
-
-        long_only_files = [
-            f for f in os.listdir("checkpoints") if f.startswith("long_only")
-        ]
-        assert len(long_only_files) == 1
-        long_only_result = BacktestResult.load(f"checkpoints/{long_only_files[0]}")
-
         equal_weights_results = BacktestResult.load("checkpoints/equal_weights.pickle")
+        leverage_limit_result = BacktestResult.load(
+            f"checkpoints/leverage_limit_{annualized_target}.pickle"
+        )
     else:
         equal_weights_results = run_backtest(equal_weights, 0.0, verbose=True)
         equal_weights_results.save("checkpoints/equal_weights.pickle")
 
         adjustment_factor = np.sqrt(equal_weights_results.periods_per_year)
-        annualized_target = 0.13
         sigma_target = annualized_target / adjustment_factor
 
+        leverage_limit_result = run_backtest(
+            leverage_limit_markowitz, sigma_target, verbose=True
+        )
+        leverage_limit_result.save(
+            f"checkpoints/leverage_limit_{annualized_target}.pickle"
+        )
+
+        weight_limited_result = run_backtest(
+            weight_limits_markowitz, sigma_target, verbose=True
+        )
+        weight_limited_result.save(
+            f"checkpoints/weight_limited_{annualized_target}.pickle"
+        )
+
         unconstrained_result = run_backtest(
             unconstrained_markowitz, sigma_target, verbose=True
         )
@@ -143,28 +155,62 @@ def main(from_checkpoint: bool = False):
         long_only_result = run_backtest(long_only_markowitz, sigma_target, verbose=True)
         long_only_result.save(f"checkpoints/long_only_{annualized_target}.pickle")
 
-    generate_table(equal_weights_results, unconstrained_result, long_only_result)
+    generate_table(
+        equal_weights_results,
+        unconstrained_result,
+        long_only_result,
+        weight_limited_result,
+        leverage_limit_result,
+    )
+
     # plot_results(equal_weights_results, unconstrained_result, long_only_result)
 
 
 def generate_table(
     equal_weights_results: BacktestResult,
     unconstrained_results: BacktestResult,
     long_only_results: BacktestResult,
+    weight_limited_result: BacktestResult,
+    leverage_limit_result: BacktestResult,
 ) -> None:
     # Table 1
     df = pd.DataFrame(
-        index=["Equal weights", "Unconstrained Markowitz", "Long-only Markowitz"],
+        index=[
+            "Equal weights",
+            "Unconstrained Markowitz",
+            "Long-only Markowitz",
+            "Weight-limited Markowitz",
+            "Leverage-limited Markowitz",
+        ],
         columns=["Mean return", "Volatility", "Sharpe", "Turnover", "Max leverage"],
     )
-    strategies = [equal_weights_results, unconstrained_results, long_only_results]
+    strategies = [
+        equal_weights_results,
+        unconstrained_results,
+        long_only_results,
+        weight_limited_result,
+        leverage_limit_result,
+    ]
 
     df["Mean return"] = [result.mean_return for result in strategies]
     df["Volatility"] = [result.volatility for result in strategies]
     df["Sharpe"] = [result.sharpe for result in strategies]
     df["Turnover"] = [result.turnover for result in strategies]
     df["Max leverage"] = [result.max_leverage for result in strategies]
-    print(df.to_latex(float_format="%.2f"))
+
+    formatters = {
+        "Mean return": lambda x: rf"{100 * x:.2f}\%",
+        "Volatility": lambda x: rf"{100 * x:.2f}\%",
+        "Sharpe": lambda x: f"{x:.2f}",
+        "Turnover": lambda x: f"{x:.2f}",
+        "Max leverage": lambda x: f"{x:.2f}",
+    }
+
+    print(
+        df.to_latex(
+            formatters=formatters,
+        )
+    )
 
 
 def plot_results(

markowitz.py

@@ -115,11 +115,12 @@ def markowitz(data: Data, param: Parameters) -> tuple[np.ndarray, float]:
         param.z_lower <= z,
         z <= param.z_upper,
         L <= param.L_max,
+        T <= param.T_max,
     ]
 
     problem = cp.Problem(cp.Maximize(objective), constraints)
     problem.solve()
-
+    assert problem.status in {cp.OPTIMAL, cp.OPTIMAL_INACCURATE}, problem.status
     return w.value, c.value