Change in portfolio turnover calculation and test, Ledoit-Wolfe cov, base for Github actions

.github/workflows/run-tests.yml

@@ -0,0 +1,33 @@
+# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Run tests
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11"]
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+    - name: Test with unittest
+      run: |
+        python test/test_quadratic_program.py

requirements.txt

@@ -0,0 +1,8 @@
+cvxopt==1.3.2
+matplotlib==3.9.2
+numpy==2.1.3
+pandas==2.2.3
+qpsolvers==4.4.0
+qpsolvers[highs]
+scipy==1.14.1
+statsmodels==0.14.4

src/backtest.py

@@ -133,19 +133,19 @@ def build_selection(self, rebdate: str) -> None:
             item_builder(self, rebdate)
         return None
 
-    def build_optimization(self, rebdate: str) -> None:
+    def build_optimization(self, rebdate: str, init_weight: dict) -> None:
 
         # Initialize the optimization constraints
         self.optimization.constraints = Constraints(selection = self.selection.selected)
-
+        self.optimization.x0 = init_weight
         # Loop over the optimization_item_builders items
         for item_builder in self.optimization_item_builders.values():
             item_builder(self, rebdate)
         return None
 
-    def prepare_rebalancing(self, rebalancing_date: str) -> None:
-        self.build_selection(rebdate = rebalancing_date)
-        self.build_optimization(rebdate = rebalancing_date)
+    def prepare_rebalancing(self, rebalancing_date: str, init_weight: dict) -> None:
+        self.build_selection(rebalancing_date)
+        self.build_optimization(rebalancing_date, init_weight = init_weight)
         return None
 
 
@@ -187,7 +187,8 @@ def rebalance(self,
                   rebalancing_date: str) -> None:
 
         # Prepare the rebalancing, i.e., the optimization problem
-        bs.prepare_rebalancing(rebalancing_date = rebalancing_date)
+        prev_weight = self.strategy.portfolios[-1].weights if self.strategy.portfolios else {}
+        bs.prepare_rebalancing(rebalancing_date = rebalancing_date, init_weight = prev_weight)
 
         # Solve the optimization problem
         try:
@@ -205,8 +206,7 @@ def run(self, bs: BacktestService) -> None:
             if not bs.settings.get('quiet'):
                 print(f'Rebalancing date: {rebalancing_date}')
 
-            self.rebalance(bs = bs,
-                           rebalancing_date = rebalancing_date)
+            self.rebalance(bs = bs, rebalancing_date = rebalancing_date)
 
             # Append portfolio to strategy
             weights = bs.optimization.results['weights']
@@ -251,19 +251,18 @@ def append_custom(backtest: Backtest,
         what = ['w_dict', 'objective']
 
     for key in what:
+        if key not in bs.optimization.results.keys():
+            continue
+
         if key == 'w_dict':
             w_dict = bs.optimization.results['w_dict']
-            for key in w_dict.keys():
-                weights = w_dict[key]                    
-                if hasattr(weights, 'to_dict'):
-                    weights = weights.to_dict()
+            for w_key, w_val in w_dict.items():
+                weights = w_val.to_dict() if hasattr(w_val, 'to_dict') else w_val
                 portfolio = Portfolio(rebalancing_date = rebalancing_date, weights = weights)
                 backtest.append_output(date_key = rebalancing_date,
-                                        output_key = f'weights_{key}',
+                                        output_key = f'weights_{w_key}',
                                         value = pd.Series(portfolio.weights))
         else:
-            if not key in bs.optimization.results.keys():
-                continue
             backtest.append_output(date_key = rebalancing_date,
                                     output_key = key,
                                     value = bs.optimization.results[key])

src/builders.py

@@ -285,3 +285,18 @@ def bibfn_box_constraints(bs, rebdate: str, **kwargs) -> None:
                                         lower = lower,
                                         upper = upper)
     return None
+
+
+def bibfn_l1_constraints(bs, rebdate: str, **kwargs) -> None:
+
+    '''
+    Backtest item builder function for setting the L1 constraints.
+    '''
+
+    # Arguments
+    name = kwargs.get('name')
+    rhs = kwargs.get('rhs')
+
+    # Constraints
+    bs.optimization.constraints.add_l1(name = name, rhs = rhs)
+    return None

src/covariance.py

@@ -47,6 +47,8 @@ def estimate(self, X: pd.DataFrame) -> pd.DataFrame:
         elif estimation_method == 'linear_shrinkage':
             lambda_covmat_regularization = self.spec.get('lambda_covmat_regularization')
             covmat = cov_linear_shrinkage(X, lambda_covmat_regularization)
+        elif estimation_method == 'ledoit_wolf':
+            covmat = cov_ledoit_wolf(X)
         else:
             raise NotImplementedError('This method is not implemented yet')
         if self.spec.get('check_positive_definite'):
@@ -82,3 +84,36 @@ def cov_linear_shrinkage(X, lambda_covmat_regularization = None):
             corrs.extend(np.diag(corrMat, k))
         sigmat = pd.DataFrame(sigmat.to_numpy() + lambda_covmat_regularization * np.mean(sig**2) * np.eye(d), columns=sigmat.columns, index=sigmat.index)
     return sigmat
+
+def cov_ledoit_wolf(X):
+    N, T = X.shape
+    Y = X - X.mean(axis=1, keepdims=True)
+
+    sample_cov = Y @ Y.T / T
+    var = np.diag(sample_cov).reshape(-1, 1)
+
+    sqrt_var = np.sqrt(var)
+    unit_cor_var = sqrt_var @ sqrt_var.T
+
+    sample_cor = sample_cov / unit_cor_var
+    avg_cor = (sample_cor.sum() - N) / (N * (N - 1))
+    F = avg_cor * unit_cor_var
+    np.fill_diagonal(F, var)
+
+    gamma = np.linalg.norm(sample_cov - F) ** 2
+    if gamma == 0:
+        return sample_cov
+
+    Y2 = Y * Y
+    pi_mat = (Y2 @ Y2.T) / T - sample_cov ** 2
+    pi = pi_mat.sum()
+
+    Y3 = Y2 * Y
+    rho_mat = ((Y3 @ Y.T) / T - var * sample_cov) * var.T / unit_cor_var
+    np.fill_diagonal(rho_mat, 0)
+
+    rho = np.diag(pi_mat).sum() + avg_cor * rho_mat.sum()
+
+    shrink = max(0, min(1, (pi - rho) / (gamma * T)))
+
+    return (1 - shrink) * sample_cov + shrink * F

src/data_loader.py

@@ -13,6 +13,7 @@
 from typing import Optional, Union, Any
 import pandas as pd
 import pickle
+import datetime as dt
 
 
 def load_pickle(filename: str,
@@ -35,24 +36,22 @@ def load_data_msci(path: str = None, n: int = 24) -> dict[str, pd.DataFrame]:
     '''Loads MSCI daily returns data from 1999-01-01 to 2023-04-18'''
 
     path = os.path.join(os.getcwd(), f'data{os.sep}') if path is None else path
+
     # Load msci country index return series
     df = pd.read_csv(os.path.join(path, 'msci_country_indices.csv'),
-                        sep=';',
                         index_col=0,
                         header=0,
-                        parse_dates=True)
-    df.index = pd.to_datetime(df.index, format='%d/%m/%Y')
+                        parse_dates=True,
+                        date_format='%d-%m-%Y')
     series_id = df.columns[0:n]
     X = df[series_id]
 
     # Load msci world index return series
     y = pd.read_csv(f'{path}NDDLWI.csv',
-                    sep=';',
                     index_col=0,
                     header=0,
-                    parse_dates=True)
-
-    y.index = pd.to_datetime(y.index, format='%d/%m/%Y')
+                    parse_dates=True,
+                    date_format='%d-%m-%Y')
 
     return {'return_series': X, 'bm_series': y}
 

src/optimization.py

@@ -64,6 +64,7 @@ def __init__(self,
         self.constraints = Constraints() if constraints is None else constraints
         self.model = None
         self.results = None
+        self.x0 = None
 
     @abstractmethod
     def set_objective(self, optimization_data: OptimizationData) -> None:
@@ -124,8 +125,9 @@ def model_qpsolvers(self) -> None:
 
         # Choose which reference position to be used
         tocon = self.constraints.l1.get('turnover')
-        x0 = tocon['x0'] if tocon is not None and tocon.get('x0') is not None else self.params.get('x0')
-        x_init = {asset: x0.get(asset, 0) for asset in universe} if x0 is not None else None
+        x0 = tocon['x0'] if tocon is not None and tocon.get('x0') is not None else self.x0
+        extra_var = x0.keys() - universe
+        x_init = {asset: x0.get(asset, 0) for asset in list(universe) + list(extra_var)} if x0 is not None else None
 
         # Transaction cost in the objective
         transaction_cost = self.params.get('transaction_cost')