Added types and removed legacy files.

backtest.py

@@ -39,7 +39,7 @@
     universe.date_today = date
     simulator = Simulator(universe)
     simulator.calibrate()
-    simulator.run_simulation(time_steps=5, number_of_simulations=1000)
+    simulator.run_simulation(time_steps=10, number_of_simulations=1000)
     optimizer = MPCMaximumReturn(universe, portfolio, simulator.return_tensor, gamma=2, l1_penalty=0, l2_penalty=0.02,
                                  covariance_matrix=simulator.covariance_matrix)
     optimizer.solve()

src/abacus/assessor/risk_assessor.py

@@ -1,7 +1,6 @@
 # -*- coding: utf-8 -*-
 import torch
 import numpy as np
-
 from scipy.optimize import minimize
 
 from src.abacus.config import EVT_THRESHOLD, GEV_INITIAL_SOLUTION
@@ -122,7 +121,6 @@ def _cost_function(self, parameters: np.array) -> tuple[float, 2]:
         log_loss.backward()
         return log_loss.data.cpu().numpy(), parameters.grad.data.cpu().numpy()
 
-
     def extreme_value_at_risk(self, confidence_level: float) -> float:
         """
         Calculates the Value at Risk of portfolio returns (either in percentage terms or in absolute value) as specified
@@ -146,7 +144,6 @@ def extreme_value_at_risk(self, confidence_level: float) -> float:
 
         return extreme_var.item()
 
-
     def extreme_expected_shortfall(self, confidence_level: float) -> float:
         """
         Calculates the Expected Shortfall of portfolio returns (either in percentage terms or in absolute value) as
@@ -167,7 +164,6 @@ def extreme_expected_shortfall(self, confidence_level: float) -> float:
 
         return extreme_es.item()
 
-
     def value_at_risk(self, confidence_level: float) -> float:
         """
         Calculates the Value at Risk of portfolio returns (either in percentage terms or in absolute value) as specified

src/abacus/models/garch.py

@@ -1,17 +1,19 @@
 # -*- coding: utf-8 -*-
-import time
-
-from cytoolz import memoize
+import logging
 
 import torch
 import numpy as np
-
+from cytoolz import memoize
 from scipy.optimize import minimize
 from torch.distributions import Normal
 
 from src.abacus.models.model import Model
 from src.abacus.config import INITIAL_VARIANCE_GARCH_OBSERVATIONS, INITIAL_GARCH_PARAMETERS
-from src.abacus.utils.exceptions import ParameterError
+
+
+logger = logging.getLogger(__name__)
+
+
 
 class GARCH(Model):
 
@@ -128,20 +130,6 @@ def _compute_inital_variance(self) -> torch.Tensor:
             return torch.square(torch.std(self._data[:INITIAL_VARIANCE_GARCH_OBSERVATIONS]))
         return self._initial_squared_returns
 
-    @memoize
-    def _compute_variance(self, parameters: torch.Tensor) -> torch.Tensor:
-        initial_variance = self._initial_variance
-        variances = torch.zeros(self._number_of_observations)
-        mu_corr, mu_ewma = self._intermediary_parameters(parameters=parameters)
-
-        for i in range(self._number_of_observations):
-            if i == 0:
-                variance = initial_variance
-            else:
-                variance = self._update_variance(variance, self._squared_returns[i-1], mu_corr, mu_ewma)
-            variances[i] = variance
-        return variances
-
     def _update_variance(self, variance: torch.Tensor, squared_return: torch.Tensor, mu_corr, mu_ewma):
         return self._long_run_variance + mu_corr * (mu_ewma * variance + (1 - mu_ewma) * squared_return - self._long_run_variance)
 
@@ -150,24 +138,31 @@ def _sanity_check(self):
         solution_check = self._check_solution()
 
         if not parameter_check:
-            # log.
-            ...
+            logger.debug(f"Garch model did not pass parameter check.")
 
         if not solution_check:
-            # log.
-            ...
-
-        if not parameter_check or not solution_check:
-            # TODO: Make more stable or catch this error when running loop!
-            # raise ParameterError("Parameters could not be asceratined succesfully.")
-            ...
+            logger.debug(f"Garch model did not pass solution check.")
 
     def _check_parameters(self) -> bool:
         return self._solution.success
 
     def _check_solution(self) -> bool:
         return np.sum(self._optimal_parameters) < 1
 
+    @memoize
+    def _compute_variance(self, parameters: torch.Tensor) -> torch.Tensor:
+        initial_variance = self._initial_variance
+        variances = torch.zeros(self._number_of_observations)
+        mu_corr, mu_ewma = self._intermediary_parameters(parameters=parameters)
+
+        for i in range(self._number_of_observations):
+            if i == 0:
+                variance = initial_variance
+            else:
+                variance = self._update_variance(variance, self._squared_returns[i-1], mu_corr, mu_ewma)
+            variances[i] = variance
+        return variances
+
     @staticmethod
     def _intermediary_parameters(parameters: torch.Tensor):
         """Computes mu_corr and mu_ewma from z_corr and z_ewma.

src/abacus/optimizer/optimizer.py

@@ -1,11 +1,11 @@
 # -*- coding: utf-8 -*-
 import os
+from abc import ABC, abstractmethod
+from typing import ClassVar
 
 import torch
 import numpy as np
 import amplpy as ap
-from abc import ABC, abstractmethod
-from typing import ClassVar
 
 from src.abacus.utils.portfolio import Portfolio
 from src.abacus.utils.universe import Universe
@@ -145,15 +145,10 @@ def __init__(self, universe: Universe, portfolio: Portfolio, simulation_tensor:
         self._l2_penalty = l2_penalty
         self._covariance_matrix = covariance_matrix
 
-    def solve(self):
-        super().solve()
-
     @property
     def solution(self):
         self._check_solved()
-        # TODO: Should be general?
         ampl_output = self._ampl.get_variable("weights").to_pandas().loc[1].to_dict()["weights.val"]
-        # print(self._ampl.eval("display OBJECTIVE;"))
         return ampl_output
 
     @property

src/abacus/simulator/model_selector.py

@@ -13,7 +13,6 @@
 
 
 class ModelSelector:
-    # TODO: Add more models here.
 
     STOCK_ADMISSIBLE_MODELS = AR, GARCH
 
@@ -29,8 +28,6 @@ def instrument(self, other: Instrument):
         self._instrument = other
 
     def select_model(self):
-        # TODO: Add more supported instrument types here.
-        # TODO: Add sanity check of if instrument is None and instrument model is None.
         instrument_type = self.instrument.instrument_type
         if instrument_type == "Stock":
             self._select_stock_model()

src/abacus/utils/exceptions.py

@@ -1,8 +1,7 @@
 # -*- coding: utf-8 -*-
 
 class StationarityError(ValueError):
-    ...
-
+    pass
 
 class ParameterError(ValueError):
-    ...
+    pass

src/abacus/utils/instrument.py

@@ -3,6 +3,7 @@
 import numpy as np
 import pandas as pd
 
+from src.abacus.models.model import Model
 
 
 class Instrument:
@@ -37,15 +38,15 @@ def art_returns_tensor(self) -> torch.Tensor:
         return torch.Tensor(self.art_returns.values.flatten())
 
     @property
-    def model(self):
+    def model(self) -> Model:
         return self._model
 
     @model.setter
     def model(self, other):
         self._model = other
 
     @property
-    def price_history(self):
+    def price_history(self) -> pd.DataFrame:
         return self._price_history
 
     @price_history.setter

src/abacus/utils/portfolio.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+from typing import NoReturn
 
 
 
@@ -10,7 +11,7 @@ def __init__(self, holdings: dict[str:int]=None, weights: dict[str:float]=None,
         self._cash = cash
 
     @property
-    def weights(self):
+    def weights(self) -> dict[str:float] | NoReturn:
         if not self._weights:
             raise ValueError("No weights are available.")
         return self._weights
@@ -20,7 +21,7 @@ def weights(self, new):
         self._weights = new
 
     @property
-    def holdings(self):
+    def holdings(self) -> dict[str:int] | NoReturn:
         if not self._weights:
             raise ValueError("No holdings are available.")
         return self._holdings
@@ -30,7 +31,7 @@ def holdings(self, new):
         self._holdings = new
 
     @property
-    def cash(self):
+    def cash(self) -> float | NoReturn:
         if not self._cash:
             raise ValueError("No cash is available.")
         return self._cash
@@ -40,22 +41,22 @@ def cash(self, new):
         self._cash = new
 
     @property
-    def weights_from_holdings(self):
+    def weights_from_holdings(self) -> dict[str:float]:
         total_holdings = sum(map(abs, self._holdings.values()))
         if total_holdings == 0:
             return {ticker: 0 for ticker in self._holdings.items()}
         return {ticker: holding/total_holdings for ticker, holding in self._holdings.items()}
 
     @property
-    def indices(self):
+    def indices(self) -> list[int] | NoReturn:
         if self._holdings:
             return [instrument.id for instrument in self._holdings]
         elif self._weights:
             return [instrument.id for instrument in self._weights]
         raise ValueError("Portfolio has no instruments.")
 
     @property
-    def instruments(self):
+    def instruments(self) -> list[str] | NoReturn:
         if self._holdings:
             return self._holdings.keys()
         elif self._weights:

src/abacus/utils/universe.py

@@ -16,15 +16,15 @@ def __init__(self, instrument_specifications: dict[str:DataFrame], date_today: s
         self._instrument_build_date = None
 
     @property
-    def date_today(self):
+    def date_today(self) -> str:
         return self._date_today
 
     @date_today.setter
     def date_today(self, new):
         self._date_today = new
 
     @property
-    def instrument_identifiers(self):
+    def instrument_identifiers(self) -> list[str]:
         return sorted(self._instrument_specifications.keys())
 
     @property
@@ -40,7 +40,7 @@ def instruments(self) -> list[Instrument]:
         return built_instruments
 
     @property
-    def todays_returns(self):
+    def todays_returns(self) -> np.ndarray:
         return np.array([instrument.art_returns[-1] for instrument in self.instruments]).flatten()
 
     def has_updated_cache(self) -> bool: