02-投资组合构建

预计学习时间：1.5 小时

难度：⭐⭐⭐

核心问题：模型输出的预测信号，如何转化为具体的股票持仓？

从信号到持仓

基本概念

在量化投资中，模型的输出通常是预测值（Predicted Score）或信号（Signal），但最终需要的是具体持仓（Positions）。

┌────────────────────────────────────────────────────────────────┐
│                    从信号到持仓的转化                           │
├────────────────────────────────────────────────────────────────┤
│                                                                │
│  模型输出          预测值              排名             持仓     │
│  ├─ 股票A    →    0.85      →       第1名     →    15%       │
│  ├─ 股票B    →    0.62      →       第5名     →    8%        │
│  ├─ 股票C    →   -0.23      →       第48名    →    0%        │
│  └─ 股票D    →   -0.45      →       第50名    →    0%        │
│                                                                │
│  转化过程：预测 → 排序 → 选股 → 分配权重                         │
│                                                                │
└────────────────────────────────────────────────────────────────┘

为什么重要？

信号相同，组合不同 → 收益天差地别

# 假设模型预测结果相同，但组合构建方式不同
 
方式A：等权Top30  → 年化收益 12%，Sharpe 1.2
方式B：IC加权     → 年化收益 15%，Sharpe 1.5
方式C：均值方差   → 年化收益 10%，Sharpe 1.8

组合构建是策略的”最后一公里”，决定了信号能否真正转化为收益。

仓位权重计算方法

方法1：等权法 (Equal Weight)

原理

每只入选股票分配相同的权重。

$w_{i} = \frac{1}{N}, i = 1, 2, ..., N$

其中 $N$ 是股票池中股票的数量。

代码实现

import pandas as pd
import numpy as np
 
def equal_weight(stock_pool):
    """
    等权组合构建
 
    参数
    ----
    stock_pool : DataFrame
        每行表示一天，每列表示一只股票，值为1表示在池中，0表示不在
 
    返回
    ----
    weights : DataFrame
        每只股票的权重
    """
    # 计算每天在池中的股票数量
    n_stocks = stock_pool.sum(axis=1)
 
    # 每只股票分配相等权重
    weights = stock_pool.div(n_stocks, axis=0).fillna(0)
 
    return weights
 
# 示例
dates = pd.date_range('2023-01-01', '2023-01-05')
stocks = ['A', 'B', 'C', 'D', 'E']
 
# 假设股票池（每天可能不同）
pool = pd.DataFrame({
    'A': [1, 1, 1, 0, 1],
    'B': [1, 1, 1, 1, 1],
    'C': [1, 0, 1, 1, 1],
    'D': [1, 1, 0, 0, 1],
    'E': [1, 1, 1, 1, 0],
}, index=dates)
 
weights = equal_weight(pool)
print(weights)

输出：

              A         B         C         D         E
2023-01-01  0.25  0.250000  0.250000  0.250000  0.250000
2023-01-02  0.25  0.250000  0.000000  0.250000  0.250000
2023-01-03  0.25  0.250000  0.333333  0.000000  0.333333
2023-01-04  0.00  0.333333  0.333333  0.000000  0.333333
2023-01-05  0.25  0.250000  0.250000  0.250000  0.000000

优缺点

优点	缺点
✅ 简单直观	❌ 没有利用预测值的差异
✅ 分散风险	❌ 对所有股票一视同仁
✅ 换手率相对低	❌ 无法表达”看好程度”
✅ 容易实现	❌ 可能包含低质量股票

方法2：Top-K 等权法

原理

选择预测值最高的 K 只股票，然后等权分配。

$w_{i} = {\frac{1}{K} 0 if r ank (scor e_{i}) \leq K otherwise$

K 值选择

K 值	特点	适用场景
10-20	集中度高，收益波动大	信号质量高，追求高Alpha
30-50	平衡集中度与分散度	大多数主动策略
80-100	高度分散，接近指数	信号质量一般，追求稳健

代码实现

def top_k_equal_weight(scores, k=30):
    """
    Top-K 等权组合构建
 
    参数
    ----
    scores : DataFrame
        每行表示一天，每列表示一只股票的预测分数
    k : int
        选择的股票数量
 
    返回
    ----
    weights : DataFrame
        每只股票的权重
    """
    # 对每天的分数进行排名
    ranks = scores.rank(axis=1, ascending=False)
 
    # 选择排名前K的股票
    selected = (ranks <= k).astype(int)
 
    # 等权分配
    weights = selected.div(k, axis=0).fillna(0)
 
    return weights
 
# 示例
np.random.seed(42)
dates = pd.date_range('2023-01-01', '2023-01-03')
stocks = [f'S{i}' for i in range(50)]
scores = pd.DataFrame(np.random.randn(len(dates), len(stocks)),
                     index=dates, columns=stocks)
 
weights = top_k_equal_weight(scores, k=10)
print(f"每日持仓数量: {weights.astype(bool).sum(axis=1).to_dict()}")
print(f"权重示例（第一天）: {weights.iloc[0][weights.iloc[0] > 0].head()}")

输出：

每日持仓数量: {Timestamp('2023-01-01 00:00:00'): 10, ...}
权重示例（第一天）:
S12    0.1
S15    0.1
S18    0.1
S25    0.1
S27    0.1

⚠️ 换手率问题

Top-K 策略的换手率可能很高：

# 计算换手率
def calculate_turnover(weights):
    """
    计算换手率
 
    换手率 = |本期权重 - 上期权重| 之和 / 2
    """
    weight_diff = weights.diff().abs()
    turnover = weight_diff.sum(axis=1) / 2
    return turnover
 
turnover = calculate_turnover(weights)
print(f"平均日换手率: {turnover.mean():.2%}")
print(f"年化换手率: {turnover.mean() * 252:.2%}")

降低换手率的方法：

增大 K 值
降低调仓频率
设置权重变动阈值（只有变动超过 5% 才调仓）

方法3：IC 加权法

原理

根据预测值（或 IC 值）分配权重，预测值越高的股票权重越大。

直接 IC 加权： $w_{i} = \frac{scor e _{i} - scor e ˉ}{\sum _{j = 1}^{N} ∣ scor e _{j} - scor e ˉ ∣}$

绝对值 IC 加权（用于多空）： $w_{i} = \frac{s i g n ( scor e _{i} ) \cdot ∣ scor e _{i} ∣ ^{p}}{\sum _{j = 1}^{N} ∣ scor e _{j} ∣ ^{p}}$

其中 $p$ 是幂次参数，控制权重集中度。

代码实现

def ic_weight(scores, method='direct', power=1, softmax_temp=1.0):
    """
    IC 加权组合构建
 
    参数
    ----
    scores : DataFrame
        每行表示一天，每列表示一只股票的预测分数
    method : str
        'direct' - 直接IC加权（只做多）
        'abs' - 绝对值加权（做多空）
        'softmax' - Softmax平滑
    power : float
        幂次，控制权重集中度
    softmax_temp : float
        Softmax温度参数，越小越集中
 
    返回
    ----
    weights : DataFrame
        每只股票的权重
    """
    weights = pd.DataFrame(0, index=scores.index, columns=scores.columns)
 
    for date in scores.index:
        daily_scores = scores.loc[date].dropna()
 
        if len(daily_scores) == 0:
            continue
 
        if method == 'direct':
            # 标准化后直接使用
            normalized = (daily_scores - daily_scores.mean()) / daily_scores.std()
            # 只做多：正负值分开
            positive = normalized[normalized > 0]
            if len(positive) > 0:
                w = positive / positive.sum()
                weights.loc[date, w.index] = w
 
        elif method == 'abs':
            # 绝对值加权，适合做多空
            abs_scores = np.abs(daily_scores) ** power
            w = abs_scores / abs_scores.sum()
            # 加入方向信息
            w = w * np.sign(daily_scores)
            weights.loc[date, w.index] = w
 
        elif method == 'softmax':
            # Softmax 平滑
            exp_scores = np.exp(daily_scores / softmax_temp)
            w = exp_scores / exp_scores.sum()
            weights.loc[date, w.index] = w
 
    return weights
 
# 示例对比
np.random.seed(42)
scores = pd.Series(np.random.randn(10), index=[f'S{i}' for i in range(10)])
 
print("=== IC 加权方法对比 ===")
print(f"\n原始预测值:\n{scores.sort_values(ascending=False)}")
 
w_direct = ic_weight(pd.DataFrame(scores).T, method='direct').iloc[0]
w_abs = ic_weight(pd.DataFrame(scores).T, method='abs', power=2).iloc[0]
w_softmax = ic_weight(pd.DataFrame(scores).T, method='softmax', softmax_temp=0.5).iloc[0]
 
print(f"\n直接IC加权（只做多）:\n{w_direct[w_direct > 0].sort_values(ascending=False)}")
print(f"\n绝对值IC加权（多空，p=2）:\n{w_abs.sort_values(ascending=False)}")
print(f"\nSoftmax加权（temp=0.5）:\n{w_softmax.sort_values(ascending=False)}")

IC 加权的优缺点

优点	缺点
✅ 充分利用预测值信息	❌ 对极端值敏感
✅ 可做多空	❌ 权重可能过于集中
✅ 理论上最优（线性假设下）	❌ 换手率通常较高
✅ 灵活性高（可调节参数）	❌ 需要额外处理负权重

方法4：均值方差优化 (Mean-Variance Optimization)

理论基础

基于 Markowitz 现代投资组合理论，寻找有效前沿（Efficient Frontier）上的最优组合。

优化目标：最大化 Sharpe 比率

$max_{w} \frac{w ^{T} μ - r _{f}}{w ^{T} Σ w}$

约束条件： $\sum_{i = 1}^{N} w_{i} = 1 (预算约束)$

$w_{i} \geq 0 (不允许做空，可选)$

其中：

$μ$ 是预期收益向量
$Σ$ 是协方差矩阵
$r_{f}$ 是无风险利率

数学推导

等价问题：给定目标收益 $μ_{p}$ ，最小化风险

$min_{w} \frac{1}{2} w^{T} Σ w$

$s . t . w^{T} μ = μ_{p}, w^{T} 1 = 1$

代码实现

from scipy.optimize import minimize
 
def mean_variance_optimization(returns, target_return=None, risk_free_rate=0.03,
                               allow_short=False, weight_bounds=(0, 1)):
    """
    均值方差优化
 
    参数
    ----
    returns : DataFrame
        历史收益率数据 (T x N)
    target_return : float
        目标年化收益率，None时最大化Sharpe
    risk_free_rate : float
        无风险利率（年化）
    allow_short : bool
        是否允许做空
    weight_bounds : tuple
        权重边界
 
    返回
    ----
    weights : Series
        最优权重
    """
    # 计算预期收益和协方差矩阵
    mu = returns.mean() * 252  # 年化
    Sigma = returns.cov() * 252  # 年化
 
    n_assets = len(mu)
 
    # 目标函数：负Sharpe（最小化负Sharpe = 最大化Sharpe）
    def neg_sharpe(weights):
        portfolio_return = weights @ mu
        portfolio_variance = weights @ Sigma @ weights
        portfolio_std = np.sqrt(portfolio_variance)
        return -(portfolio_return - risk_free_rate) / portfolio_std
 
    # 约束条件
    constraints = [
        {'type': 'eq', 'fun': lambda w: np.sum(w) - 1},  # 权重和为1
    ]
 
    if target_return is not None:
        constraints.append({
            'type': 'eq',
            'fun': lambda w: w @ mu - target_return
        })
 
    # 权重边界
    if allow_short:
        bounds = [(-1, 1) for _ in range(n_assets)]
    else:
        bounds = [(weight_bounds[0], weight_bounds[1]) for _ in range(n_assets)]
 
    # 初始值（等权）
    x0 = np.ones(n_assets) / n_assets
 
    # 优化
    result = minimize(
        neg_sharpe,
        x0=x0,
        method='SLSQP',
        bounds=bounds,
        constraints=constraints,
        options={'ftol': 1e-9}
    )
 
    if not result.success:
        print(f"优化失败: {result.message}")
        return pd.Series(np.ones(n_assets) / n_assets, index=returns.columns)
 
    weights = pd.Series(result.x, index=returns.columns)
 
    return weights
 
# 示例
np.random.seed(42)
n_stocks = 20
dates = pd.date_range('2020-01-01', '2022-12-31')
stocks = [f'S{i}' for i in range(n_stocks)]
 
# 模拟收益率数据
returns = pd.DataFrame(
    np.random.multivariate_normal(
        mean=np.zeros(n_stocks),
        cov=np.eye(n_stocks) * 0.02 + np.ones((n_stocks, n_stocks)) * 0.005,
        size=len(dates)
    ),
    index=dates,
    columns=stocks
)
 
# 计算最优权重
optimal_weights = mean_variance_optimization(returns, allow_short=False)
 
print("=== 均值方差优化结果 ===")
print(f"\n前10大权重:")
print(optimal_weights.nlargest(10))
print(f"\n权重统计:")
print(f"最大权重: {optimal_weights.max():.2%}")
print(f"最小权重: {optimal_weights.min():.2%}")
print(f"非零权重股票数: {(optimal_weights > 0.001).sum()}")

协方差矩阵估计

协方差矩阵的估计是均值方差优化的关键：

def estimate_covariance(returns, method='sample', **kwargs):
    """
    估计协方差矩阵
 
    参数
    ----
    returns : DataFrame
        收益率数据
    method : str
        'sample' - 样本协方差
        'ewma' - 指数加权
        'ledoit-wolf' - Ledoit-Wolf收缩
    **kwargs :
        method相关参数
 
    返回
    ----
    cov_matrix : DataFrame
        协方差矩阵
    """
    if method == 'sample':
        return returns.cov()
 
    elif method == 'ewma':
        # 指数加权移动平均
        span = kwargs.get('span', 60)
        return returns.ewm(span=span).cov().iloc[-len(returns.columns):]
 
    elif method == 'ledoit-wolf':
        # Ledoit-Wolf收缩估计
        from sklearn.covariance import LedoitWolf
        cov = LedoitWolf().fit(returns).covariance_
        return pd.DataFrame(cov, index=returns.columns, columns=returns.columns)
 
    else:
        raise ValueError(f"Unknown method: {method}")

实际问题

问题	说明	解决方案
输入敏感	$μ$ 和 $Σ$ 的小变化导致权重剧烈波动	使用收缩估计、添加正则化
极端权重	优化结果可能是少数股票权重极大	设置权重上限
角点解	很多股票权重为0	这是正常现象，说明不需要这些股票
采样误差	历史协方差不代表未来	使用EWMA、因子模型估计协方差

方法5：风险平价 (Risk Parity) 简介

原理

让每只资产对组合风险的贡献相等，而不是金额相等。

风险贡献： $R C_{i} = w_{i} \cdot \frac{( Σ w ) _{i}}{w ^{T} Σ w}$

目标： $R C_{1} = R C_{2} = ... = R C_{N}$

特点

特点	说明
风险分散	每个资产风险贡献相等
低波动资产权重高	债券等低波动资产会有更高权重
适合资产配置	常用于大类资产配置
不适合选股	股票间相关性高，效果有限

组合构建方法对比

┌─────────────────────────────────────────────────────────────────┐
│                     组合构建方法综合对比                         │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  方法           集中度   换手率   实现难度   适用场景            │
│  ─────────────────────────────────────────────────────────────  │
│  等权            低      低       ⭐        分散投资            │
│  Top-K等权       中      中       ⭐⭐       主动管理            │
│  IC加权          高      高       ⭐⭐⭐     信号质量高时        │
│  均值方差        中-高   低       ⭐⭐⭐⭐    风险控制重要        │
│  风险平价        低      低       ⭐⭐⭐⭐    资产配置            │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

代码对比

def compare_methods():
    """对比不同组合构建方法"""
    np.random.seed(42)
 
    # 生成模拟数据
    dates = pd.date_range('2020-01-01', '2022-12-31')
    stocks = [f'S{i}' for i in range(100)]
    scores = pd.DataFrame(np.random.randn(len(dates), len(stocks)),
                         index=dates, columns=stocks)
    returns = scores * 0.01  # 简单假设分数与收益正相关
 
    results = {}
 
    # 1. 等权（全市场）
    weights_equal = pd.DataFrame(1/len(stocks), index=dates, columns=stocks)
    results['等权'] = {
        'weights': weights_equal,
        'return': (returns * weights_equal).sum(axis=1).mean() * 252,
        'std': (returns * weights_equal).sum(axis=1).std() * np.sqrt(252)
    }
 
    # 2. Top-K等权
    weights_topk = top_k_equal_weight(scores, k=30)
    results['Top30等权'] = {
        'weights': weights_topk,
        'return': (returns * weights_topk).sum(axis=1).mean() * 252,
        'std': (returns * weights_topk).sum(axis=1).std() * np.sqrt(252)
    }
 
    # 3. IC加权
    weights_ic = ic_weight(scores, method='abs', power=2)
    results['IC加权'] = {
        'weights': weights_ic,
        'return': (returns * weights_ic).sum(axis=1).mean() * 252,
        'std': (returns * weights_ic).sum(axis=1).std() * np.sqrt(252)
    }
 
    # 输出对比
    print("=== 组合构建方法对比 ===")
    for name, result in results.items():
        sharpe = (result['return'] - 0.03) / result['std']
        turnover = calculate_turnover(result['weights']).mean() * 252
        print(f"\n{name}:")
        print(f"  年化收益: {result['return']:.2%}")
        print(f"  年化波动: {result['std']:.2%}")
        print(f"  Sharpe:   {sharpe:.2f}")
        print(f"  年化换手: {turnover:.2%}")
 
compare_methods()

行业/因子暴露约束

在实际应用中，我们通常希望组合在某些维度上保持中性或限制暴露。

行业中性

def industry_neutral_adjustment(weights, industry_exposure, target_exposure=None):
    """
    行业中性调整
 
    参数
    ----
    weights : Series
        原始权重
    industry_exposure : DataFrame
        每个股票的行业暴露（one-hot编码）
    target_exposure : Series or None
        目标行业暴露，None时市场中性（等权）
 
    返回
    ----
    adjusted_weights : Series
        调整后的权重
    """
    if target_exposure is None:
        target_exposure = pd.Series(1/len(industry_exposure.columns),
                                   index=industry_exposure.columns)
 
    # 计算当前行业暴露
    current_exposure = weights @ industry_exposure
 
    # 计算需要调整的幅度
    adjustment = target_exposure - current_exposure
 
    # 按行业等比例调整
    adjusted_weights = weights.copy()
    for industry in industry_exposure.columns:
        stocks_in_industry = industry_exposure[industry] > 0
        if adjustment[industry] != 0 and stocks_in_industry.sum() > 0:
            scale = (weights[stocks_in_industry].sum() + adjustment[industry]) / \
                    weights[stocks_in_industry].sum()
            adjusted_weights[stocks_in_industry] *= scale
 
    return adjusted_weights

风险因子约束

def factor_constraints(weights, factor_exposure, limits):
    """
    风险因子约束
 
    参数
    ----
    weights : Series
        原始权重
    factor_exposure : DataFrame
        每个股票的因子暴露
    limits : dict
        各因子的上下限，如 {'size': (-0.1, 0.1)}
 
    返回
    ----
    adjusted_weights : Series
        调整后的权重
    """
    from scipy.optimize import minimize
 
    n_assets = len(weights)
 
    # 目标：最小化与原始权重的偏差
    def objective(w):
        return np.sum((w - weights) ** 2)
 
    # 约束条件
    constraints = [
        {'type': 'eq', 'fun': lambda w: np.sum(w) - 1}  # 权重和为1
    ]
 
    # 因子暴露约束
    for factor, (lower, upper) in limits.items():
        if factor in factor_exposure.columns:
            exposure = factor_exposure[factor].values
 
            constraints.append({
                'type': 'ineq',
                'fun': lambda w, e=exposure, l=lower: w @ e - l
            })
            constraints.append({
                'type': 'ineq',
                'fun': lambda w, e=exposure, u=upper: u - w @ e
            })
 
    # 优化
    result = minimize(
        objective,
        x0=weights.values,
        method='SLSQP',
        bounds=[(0, 1) for _ in range(n_assets)],
        constraints=constraints
    )
 
    if result.success:
        return pd.Series(result.x, index=weights.index)
    else:
        print(f"优化失败: {result.message}")
        return weights

持仓约束

个股上限

def apply_single_stock_limit(weights, limit=0.05):
    """
    应用个股权重上限
 
    参数
    ----
    weights : Series
        原始权重
    limit : float
        单一股权重上限
 
    返回
    ----
    adjusted_weights : Series
        调整后的权重
    """
    adjusted = weights.copy()
 
    # 找出超过上限的权重
    mask = adjusted > limit
 
    while mask.any():
        # 超过上限的股票先设置为上限
        adjusted[mask] = limit
 
        # 计算剩余权重需要分配的总额
        excess = (weights[mask] - limit).sum()
 
        # 按比例分配给未超限的股票
        not_exceeded = ~mask
        if not_exceeded.sum() > 0:
            adjusted[not_exceeded] += excess * \
                (adjusted[not_exceeded] / adjusted[not_exceeded].sum())
 
        # 重新检查
        mask = adjusted > limit
 
    # 归一化
    adjusted = adjusted / adjusted.sum()
 
    return adjusted

行业上限

def apply_industry_limit(weights, industry_mapping, industry_limits):
    """
    应用行业权重上限
 
    参数
    ----
    weights : Series
        原始权重
    industry_mapping : Series
        股票到行业的映射
    industry_limits : dict
        各行业权重上限，如 {'金融': 0.3, '科技': 0.4}
 
    返回
    ----
    adjusted_weights : Series
        调整后的权重
    """
    adjusted = weights.copy()
 
    for industry, limit in industry_limits.items():
        stocks_in_industry = industry_mapping[industry_mapping == industry].index
        current_weight = adjusted[stocks_in_industry].sum()
 
        if current_weight > limit:
            scale = limit / current_weight
            adjusted[stocks_in_industry] *= scale
 
    # 归一化
    adjusted = adjusted / adjusted.sum()
 
    return adjusted

核心知识点总结

┌────────────────────────────────────────────────────────────────┐
│                  02-投资组合构建 核心要点                        │
├────────────────────────────────────────────────────────────────┤
│                                                                │
│  1. 从信号到持仓                                               │
│     预测值 → 排序 → 选股 → 分配权重                            │
│                                                                │
│  2. 主要方法                                                   │
│     ├─ 等权法：简单、分散、低换手                              │
│     ├─ Top-K等权：平衡集中与分散                               │
│     ├─ IC加权：充分利用预测信息                                │
│     ├─ 均值方差：理论上风险调整收益最优                         │
│     └─ 风险平价：风险贡献相等                                  │
│                                                                │
│  3. 实践要点                                                   │
│     ├─ 换手率是关键成本                                        │
│     ├─ 极端权重需要约束                                        │
│     ├─ 行业/因子中性很重要                                    │
│     └─ 没有万能方法，需根据策略选择                            │
│                                                                │
└────────────────────────────────────────────────────────────────┘

思考题

为什么IC加权的换手率通常比Top-K等权更高？
在什么情况下均值方差优化会产生极端权重？
如何平衡”集中持股获取高Alpha”和”分散持股降低风险”？

下一步

现在你知道如何从信号构建组合了。但真实的交易是有成本的——下一节我们将学习交易成本模型。

→ 前往：03-交易成本模型

MindCarver Blog

MindCarver

探索

02-投资组合构建

02-投资组合构建

从信号到持仓

基本概念

为什么重要？

仓位权重计算方法

方法1：等权法 (Equal Weight)

原理

代码实现

优缺点

方法2：Top-K 等权法

原理

K 值选择

代码实现

⚠️ 换手率问题

方法3：IC 加权法

原理

代码实现

IC 加权的优缺点

方法4：均值方差优化 (Mean-Variance Optimization)

理论基础

数学推导

代码实现

协方差矩阵估计

实际问题

方法5：风险平价 (Risk Parity) 简介

原理

特点

组合构建方法对比

代码对比

行业/因子暴露约束

行业中性

风险因子约束

持仓约束

个股上限

行业上限

核心知识点总结

思考题

下一步

关系图谱

目录

反向链接