feat: add the first implementation of the Black-Litterman model

pyproject.toml

@@ -38,6 +38,7 @@ dependencies = [
     "pydata-sphinx-theme>=0.16.1",
     "pyomo>=6.8.0",
     "scikit-learn>=1.6.1",
+    "scipy>=1.15.1",
     "streamlit>=1.40.0",
     "yfinance>=0.2.48",
 ]
@@ -80,6 +81,9 @@ ignore = [
     "F403",
     "FBT001",
     "FBT002",
+    "ISC001",
+    "N803",
+    "N806",
     "PD901",
     "PGH003",
     "T201",

pystock/portfolio.py

@@ -22,7 +22,6 @@ def __init__(self, assets: list[Asset], weights: list[float] | np.ndarray) -> No
         self.assets = assets
         self.weights = weights
         self._cov_matrix = None
-        self._expected_returns = None
         self.period_in_days = min([len(asset.daily_returns) for asset in self.assets])
         self._check_consistency()
 
@@ -70,7 +69,6 @@ def _check_consistency(self) -> None:
 
     def _reset_cache(self) -> None:
         self._cov_matrix = None
-        self._expected_returns = None
 
     @property
     def cov_matrix(self) -> np.ndarray:
@@ -96,9 +94,7 @@ def historical_expected_returns(self) -> np.ndarray:
             np.ndarray: The expected returns.
 
         """
-        if self._expected_returns is None:
-            self._expected_returns = np.array([asset.expected_return for asset in self.assets])
-        return self._expected_returns
+        return np.array([asset.expected_return for asset in self.assets])
 
     @property
     def historical_returns(self) -> np.ndarray:
@@ -148,7 +144,7 @@ def sharpe_ratio(self) -> float:
             float: The Sharpe ratio of the portfolio.
 
         """
-        return (self.historical_expected_return - cst.DEFAULT_RISK_FREE_RATE) / self.risk
+        return (self.historical_expected_return - cst.DEFAULT_RISK_FREE_RATE) / self.risk if self.risk > 0 else 0
 
     def add_asset(self, asset: Asset, weight: float) -> None:
         """Add an asset to the portfolio.

pystock/quantitative.py

@@ -4,6 +4,7 @@
 import plotly.graph_objects as go
 import pyomo.environ as pyo
 from joblib import Parallel, delayed
+from scipy.linalg import inv
 
 import pystock.constants as cst
 from pystock.portfolio import Portfolio
@@ -12,16 +13,17 @@
 class PortfolioOptimizer:
     """An optimizer for your Portfolio."""
 
-    def __init__(self, portfolio: Portfolio) -> None:
+    def __init__(self, portfolio: Portfolio, solver: str = cst.DEFAULT_SOLVER) -> None:
         """Initialize the PortfolioOptimizer.
 
         Args:
             portfolio (Portfolio): The portfolio that you want to optimize.
+            solver (str): The nonlinear solver to use for optimization.
 
         """
         self.portfolio = portfolio
         self.model = self._build_core_model()
-        self.solver = pyo.SolverFactory(cst.DEFAULT_SOLVER)
+        self.solver = pyo.SolverFactory(solver)
 
     def _build_core_model(self) -> pyo.ConcreteModel:
         model = pyo.ConcreteModel()
@@ -79,24 +81,24 @@ def __init__(self, portfolio: Portfolio) -> None:
     def _calculate_metrics(
         self,
         weights: np.ndarray,
-        risk_free_rate: float = 0.01,
     ) -> tuple[float, float, float, list[float]]:
-        portfolio_return = np.dot(weights, self.portfolio.historical_expected_returns)
-        portfolio_risk = np.sqrt(np.dot(weights, np.dot(self.portfolio.cov_matrix, weights)))
-        sharpe_ratio = (portfolio_return - risk_free_rate) / portfolio_risk if portfolio_risk > 0 else 0
-        return portfolio_return, portfolio_risk, sharpe_ratio, weights.tolist()
+        self.portfolio.weights = weights
+        return (
+            self.portfolio.historical_expected_return,
+            self.portfolio.risk,
+            self.portfolio.sharpe_ratio,
+            weights.tolist(),
+        )
 
     def simulation(
         self,
         num_simulations: int = 5000,
-        risk_free_rate: float = cst.DEFAULT_RISK_FREE_RATE,
         num_jobs: int = -1,
     ) -> pd.DataFrame:
         """Run a Monte Carlo simulation for portfolio optimization.
 
         Args:
             num_simulations (int): Number of simulations to run.
-            risk_free_rate (float): Risk-free rate.
             num_jobs (int): Number of parallel jobs to use (-1 for all available CPUs).
 
         Returns:
@@ -110,30 +112,17 @@ def simulation(
         simulations = np.random.dirichlet(np.ones(n_assets), size=num_simulations)  # noqa: NPY002
 
         results = Parallel(n_jobs=num_jobs, backend="threading")(
-            delayed(self._calculate_metrics)(weights, risk_free_rate) for weights in simulations
+            delayed(self._calculate_metrics)(weights) for weights in simulations
         )
 
         return pd.DataFrame(
             results,
             columns=["Returns", "Risk", "Sharpe Ratio", "Weights"],  # type: ignore
         )
 
-    @staticmethod
-    def get_gareto_front(df: pd.DataFrame) -> pd.DataFrame:
-        """Identify the Pareto front from the simulation results.
-
-        Args:
-            df (pd.DataFrame): Simulation results.
-
-        Returns:
-            pd.DataFrame: DataFrame containing Pareto-optimal portfolios.
-
-        """
-        return df.sort_values("Risk").drop_duplicates("Returns", keep="first")
-
     @staticmethod
     def create_efficient_frontier(df: pd.DataFrame) -> go.Figure:
-        """Plot the Pareto front using Plotly.
+        """Plot the Pareto front using Plotly with an optimal portfolio marker.
 
         Args:
             df (pd.DataFrame): DataFrame of simulation results.
@@ -143,6 +132,7 @@ def create_efficient_frontier(df: pd.DataFrame) -> go.Figure:
 
         """
         df["Weights"] = df["Weights"].apply(lambda x: str(x))
+
         fig = px.scatter(
             df,
             x="Risk",
@@ -152,9 +142,67 @@ def create_efficient_frontier(df: pd.DataFrame) -> go.Figure:
             title="Monte Carlo Simulation: Portfolio Risk vs. Return",
             labels={"Risk": "Risk (Standard Deviation)", "Returns": "Return"},
         )
+
         fig.update_layout(
             xaxis={"title": "Risk (Standard Deviation)"},
             yaxis={"title": "Return"},
             coloraxis_colorbar={"title": "Sharpe Ratio"},
         )
         return fig
+
+
+class BlackLitterman:
+    """Black-Litterman model for portfolio optimization.
+
+    This model combines market equilibrium returns with investor views to generate
+    a set of adjusted expected returns for portfolio optimization.
+    """
+
+    def __init__(self, portfolio: Portfolio, tau: float = 0.05) -> None:
+        """Initialize the Black-Litterman model.
+
+        Args:
+            portfolio (Portfolio): The portfolio object containing asset returns and covariance matrix.
+            tau (float, optional): Scaling factor for the market equilibrium returns. Defaults to 0.05.
+
+        """
+        self.portfolio = portfolio
+        self.tau = tau
+        self.market_implied_returns = self.compute_market_implied_returns()
+
+    def compute_market_implied_returns(self) -> np.ndarray:
+        """Compute the market-implied returns using the reverse optimization approach.
+
+        Returns:
+            numpy.ndarray: A 1D array of market-implied returns for each asset in the portfolio.
+
+        """
+        weights = self.portfolio.weights  # market capitalization weights
+        cov_matrix = self.portfolio.cov_matrix
+
+        return self.tau * cov_matrix @ weights
+
+    def adjust_returns_with_views(self, P: np.ndarray, Q: np.ndarray, omega: np.ndarray | None = None) -> np.ndarray:
+        """Adjust market-implied returns using investor views.
+
+        Args:
+            P (numpy.ndarray): A matrix linking views to assets (k x n).
+            Q (numpy.ndarray): A vector of expected returns based on views (k x 1).
+            omega (numpy.ndarray, optional): A diagonal covariance matrix for the views. If None,
+                                            it is set to tau * P @ cov_matrix @ P.T.
+
+        Returns:
+            numpy.ndarray: The adjusted expected returns incorporating investor views.
+
+        """
+        cov_matrix = self.portfolio.cov_matrix
+        pi = self.market_implied_returns
+
+        if omega is None:
+            omega = self.tau * P @ cov_matrix @ P.T
+
+        # Black-Litterman formula
+        M1 = inv(inv(self.tau * cov_matrix) + P.T @ inv(omega) @ P)
+        M2 = inv(self.tau * cov_matrix) @ pi + P.T @ inv(omega) @ Q
+
+        return M1 @ M2