Add some portfolio optimization routines.

Add lagrangian solver for min variance.
Create PositionalBasketOrder, fix logic around book percent mode.

pyproject.toml

@@ -8,16 +8,20 @@ readme = "README.md"
 repository = "https://github.com/bsdz/yabte"
 
 [tool.poetry.dependencies]
-python = "^3.10"
+python = "^3.10,<3.12"
 pandas = "^1.5.2"
 
-[tool.poetry.dev-dependencies]
+[tool.poetry.group.dev.dependencies]
 black = "^22.12.0"
 mypy = "^0.991"
 matplotlib = "^3.6.2"
-
-[tool.poetry.group.dev.dependencies]
 isort = "^5.11.4"
+ipykernel = "^6.20.2"
+
+
+[tool.poetry.group.portopt.dependencies]
+scipy = "^1.10.0"
+scikit-learn = "^1.2.1"
 
 [tool.isort]
 profile = "black"

tests/_helpers.py

@@ -0,0 +1,18 @@
+from pathlib import Path
+
+import pandas as pd
+
+data_dir = Path(__file__).parent / "data"
+
+
+def generate_nasdaq_dataset():
+    asset_meta = {}
+    dfs = []
+    for csv_pth in (data_dir / "nasdaq").glob("*.csv"):
+        name = csv_pth.stem
+        asset_meta[name] = {"denom": "USD"}
+        df = pd.read_csv(csv_pth, index_col=0, parse_dates=[0])
+        df.columns = pd.MultiIndex.from_tuples([(name, f) for f in df.columns])
+        dfs.append(df)
+
+    return asset_meta, pd.concat(dfs, axis=1)

tests/_unittest_numpy_extensions.py

@@ -0,0 +1,34 @@
+""" monkey patch unittest to include numpy assertions.
+Maps numpy assert_func_name to numpyAssertFuncName.
+
+Example:
+
+    class NumpyWrapperTest(unittest.TestCase):
+        def test_allclose_example(self):
+            a1 = np.array([1.,2.,3.])
+            self.numpyAssertAllclose(a1, np.array([1.,2.,3.1]))
+            
+    suite = unittest.TestSuite()
+    suite.addTest(NumpyWrapperTest("test_allclose_example"))
+    unittest.TextTestRunner().run(suite)
+"""
+
+import unittest
+
+import numpy.testing as nptu
+
+
+def make_test_wrapper(fn):
+    def test_wrapper(self, *args, **kwargs):
+        try:
+            nptu.__dict__[fn](*args, **kwargs)
+        except AssertionError as err:
+            self.fail(err)
+
+    return test_wrapper
+
+
+for fn in nptu.__dict__:
+    if fn.startswith("assert") and not fn.endswith("_"):
+        new_name = "numpy" + fn.title().replace("_", "")
+        setattr(unittest.TestCase, new_name, make_test_wrapper(fn))

tests/test_portopt.py

@@ -0,0 +1,71 @@
+import unittest
+
+import numpy as np
+import numpy.linalg as la
+
+import tests._unittest_numpy_extensions  # noqa
+import yabte.portopt.pandas_extension  # noqa
+from tests._helpers import generate_nasdaq_dataset
+from yabte.portopt.lagrangian import Lagrangian
+from yabte.portopt.minimum_variance import (
+    minimum_variance,
+    minimum_variance_numeric,
+    minimum_variance_numeric_slsqp,
+)
+
+
+class LagrangianTestCase(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.asset_meta, cls.df_combined = generate_nasdaq_dataset()
+        cls.closes = cls.df_combined.loc[:, (slice(None), "Close")].droplevel(axis=1, level=1)
+
+
+    def test_lagrangian(self):
+        Sigma = self.closes.price.log_returns.cov()
+        mu = self.closes.price.capm_returns()
+        r = 0.1
+
+        # solve algebraically
+        m = len(mu)
+        ones = np.ones(m)
+        SigmaInv = la.inv(Sigma)
+        A = mu.T @ SigmaInv @ ones
+        B = mu.T @ SigmaInv @ mu
+        C = ones.T @ SigmaInv @ ones
+        D = B*C - A*A
+        l1 = (C*r - A)/D
+        l2 = (B - A*r)/D
+        w = SigmaInv@(l1 * mu + l2 * ones)
+
+        # sanity checks
+        self.numpyAssertAllclose(w.sum(), 1)
+        self.numpyAssertAllclose(w @ mu, r)
+
+        # test numerical
+        L = Lagrangian(
+            objective=lambda x: x.T @ Sigma @ x / 2,
+            constraints=[
+                lambda x: r - x.T @ mu,
+                lambda x: 1 - x.T @ ones,
+            ],
+            x0=np.ones(m)/m
+        )
+        wn = L.fit()
+
+        self.numpyAssertAllclose(wn, w)
+
+    def test_min_var(self):
+        Sigma = self.closes.price.log_returns.cov()
+        mu = self.closes.price.capm_returns()
+        r = 0.1
+
+        mv1 = minimum_variance(Sigma, mu, r)
+        mv2 = minimum_variance_numeric(Sigma, mu, r)
+        mv3 = minimum_variance_numeric_slsqp(Sigma, mu, r)
+
+        x = 1
+
+
+if __name__ == "__main__":
+    unittest.main()