Black-Litterman模型实战：融合主观观点与量化定价的中国股市配置策略

import numpy as np
import pandas as pd
from scipy.optimize import minimize

class BlackLitterman:
    def __init__(self, returns, market_caps, risk_aversion=None):
        """
        returns: DataFrame, 资产收益率矩阵
        market_caps: Series, 市值权重
        risk_aversion: 风险厌恶系数（None则自动估计）
        """
        self.returns = returns
        self.market_caps = market_caps / market_caps.sum()  # 归一化
        self.Sigma = returns.cov() * 252  # 年化协方差
        
        if risk_aversion is None:
            # 用历史数据估计市场风险溢价
            mkt_return = (returns @ self.market_caps).mean() * 252
            self.risk_aversion = mkt_return / self.Sigma.diagonal().mean()
        else:
            self.risk_aversion = risk_aversion
        
        # 逆向优化：计算隐含均衡收益
        self.pi = self.risk_aversion * self.Sigma @ self.market_caps
        
    def add_view(self, P, q, view_confidence=None):
        """
        P: 观点矩阵 (K x N)
        q: 观点收益向量 (K,)
        view_confidence: 观点置信度 (K,) 或标量
        """
        if view_confidence is None:
            # 默认：观点不确定性 = tau * P @ Sigma @ P.T
            tau = 0.05
            Omega = np.diag(np.diag(tau * P @ self.Sigma.values @ P.T))
        else:
            # 用置信度倒数为权重
            Omega = np.diag(1 / np.array(view_confidence))
        
        self.P = P
        self.q = q
        self.Omega = Omega
        
    def compute_posterior(self, tau=0.05):
        """计算后验预期收益"""
        Sigma = self.Sigma.values
        P = self.P.values if hasattr(self.P, 'values') else self.P
        q = self.q.values if hasattr(self.q, 'values') else self.q
        Omega = self.Omega
        pi = self.pi.values
        
        # 后验公式
        M_inv = np.linalg.inv(tau * Sigma) + P.T @ np.linalg.inv(Omega) @ P
        posterior_ret = np.linalg.inv(M_inv) @ (np.linalg.inv(tau * Sigma) @ pi + P.T @ np.linalg.inv(Omega) @ q)
        
        self.posterior_ret = pd.Series(posterior_ret, index=self.returns.columns)
        return self.posterior_ret
    
    def optimize_portfolio(self, risk_free=0.025):
        """用后验收益优化组合"""
        ret = self.posterior_ret - risk_free  # 超额收益
        Sigma = self.Sigma
        
        # 均值方差优化
        def objective(w):
            port_ret = ret @ w
            port_risk = np.sqrt(w @ Sigma.values @ w)
            return -port_ret / port_risk  # 最大化夏普比率
        
        constraints = [{'type': 'eq', 'fun': lambda w: w.sum() - 1}]
        bounds = [(0, 0.3) for _ in range(len(ret))]  # 单资产上限30%
        
        result = minimize(objective, self.market_caps.values, 
                         method='SLSQP', bounds=bounds, constraints=constraints)
        
        self.optimal_weights = pd.Series(result.x, index=self.returns.columns)
        return self.optimal_weights

# 1. 准备数据
returns = get_industry_returns(start='2020-01-01')  # 获取行业收益率
market_caps = get_industry_market_cap()  # 获取行业市值

# 2. 初始化BL模型
bl = BlackLitterman(returns, market_caps)

# 3. 添加观点
P = pd.DataFrame({
    'banking': [1, 0],
    'renewable': [0, 1],
    'tech': [0, 0]
}, index=['view1', 'view2'])

q = pd.Series([0.05, 0.03], index=['view1', 'view2'])  # 银行5%超额，新能源3%超额

bl.add_view(P, q, view_confidence=[0.7, 0.5])  # 银行观点置信度更高

# 4. 计算后验收益并优化
posterior = bl.compute_posterior()
weights = bl.optimize_portfolio()

print("最优权重：\n", weights[weights > 0.01].sort_values(ascending=False))