大家好，我是量化老王。作为《Python量化交易开发100篇》的第9篇，咱们在上一篇实现了多股票等权组合回测，通过“分散投资”解决了单股票非系统性风险过高的问题。但等权组合存在一个明显缺陷：无论股票近期表现好坏、波动大小，都分配相同权重——比如给近期持续亏损、波动剧烈的股票和盈利稳定、波动小的股票同样20%权重，无疑会拉低组合整体收益。

今天这篇，咱们就升级组合策略的核心——实现“动态权重分配”。核心思路是：让权重“用脚投票”，给近期收益好、波动小的股票分配更高权重，给表现差、波动大的股票降低权重，甚至暂时剔除，在分散风险的基础上进一步提升组合收益率。全程Python代码落地，整合前文的行情识别、风险控制体系，形成“全行情适配+风险控制+动态权重”的完整多股票量化组合。

一、核心逻辑：动态权重 vs 等权，优势在哪？1. 等权组合的痛点

等权组合的核心问题是“一刀切”，完全忽略个股差异：

收益拖累：若组合中某只股票长期处于下跌趋势，等权会让其持续拖累组合收益；风险不均：波动大的股票（如新能源股）和波动小的股票（如消费龙头）权重相同，会放大组合整体波动；缺乏灵活性：无法及时捕捉个股短期趋势变化，错失优质股票的盈利机会。2. 动态权重的核心逻辑（新手易理解）

动态权重的核心是“顺势而为”，通过量化指标动态调整每只股票的权重占比。咱们选择新手最易落地、效果最稳定的“波动率加权+近期收益加权”双维度逻辑：

波动率维度：波动越小的股票，权重越高（降低组合整体风险）；收益维度：近期（如20个交易日）收益越高的股票，权重越高（强化盈利效应）；定期再平衡：每20个交易日（约1个月）重新计算权重，避免权重偏离过大。3. 动态权重的量化公式（可直接套用）

为了让逻辑可落地，咱们制定明确的量化公式，分两步计算动态权重：

第一步：计算单只股票的“评分”（综合波动率和近期收益）

评分 = （1 - 标准化波动率）× 0.5 + 标准化近期收益 × 0.5

说明：标准化是为了让波动率和收益处于同一量级（0-1之间），权重各占50%（可根据风险偏好调整，保守型可提高波动率权重，激进型提高收益权重）。

第二步：计算单只股票的动态权重

单只股票权重 = 该股票评分 ÷ 所有股票评分之和

逻辑：评分越高，权重越高；若某只股票评分过低（如持续亏损导致收益标准化后接近0），权重会趋近于0，相当于被“暂时剔除”出组合。

二、第一步：数据准备与前文策略衔接

复用第8篇的股票池（5只行业龙头）和数据获取逻辑，同时直接调用第8篇封装的“apply_strategy_with_risk_control”函数，获取每只股票的策略信号、收益率等核心数据，确保代码连贯性。

# 导入必备库（复用前文基础库）import tushare as tsimport pandas as pdimport matplotlib.pyplot as pltimport warningswarnings.filterwarnings('ignore')# 解决中文乱码plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = False# 1. 基础配置：Tushare Token+股票列表（复用第8篇股票池）ts.set_token('你的Tushare Token')pro = ts.pro_api()stock_list = [ ('600519.SH', '贵州茅台'), ('002594.SZ', '比亚迪'), ('300760.SZ', '迈瑞医疗'), ('000858.SZ', '五粮液'), ('601318.SH', '中国平安')]start_date = '20230101'end_date = '20240630' # 回测周期与前文一致，便于对比# 2. 批量获取多股票数据（复用第8篇函数）def get_multi_stock_data(stock_list, start_date, end_date): stock_data_dict = {} for ts_code, name in stock_list: df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date) df['trade_date'] = pd.to_datetime(df['trade_date'], format='%Y%m%d') df = df.sort_values('trade_date').set_index('trade_date').dropna() stock_data_dict[name] = df return stock_data_dictstock_data = get_multi_stock_data(stock_list, start_date, end_date)# 3. 复用第8篇核心策略函数（行情识别+风险控制）def apply_strategy_with_risk_control(df): # 3.1 行情识别（ATR+标准差） def calculate_atr(data, period=14): data['tr1'] = data['high'] - data['low'] data['tr2'] = abs(data['high'] - data['close'].shift(1)) data['tr3'] = abs(data['low'] - data['close'].shift(1)) data['tr'] = data[['tr1', 'tr2', 'tr3']].max(axis=1) return data['tr'].rolling(window=period).mean() df['ATR14'] = calculate_atr(df) df['ATR_std20'] = df['ATR14'].rolling(window=20).std() df['market_type'] = 'unknown' df.loc[df['ATR_std20'] < 0.5, 'market_type'] = 'shock' df.loc[df['ATR_std20'] >= 0.5, 'market_type'] = 'trend' # 3.2 多均线+RSI策略信号 df['MA_short'] = df['close'].rolling(window=5).mean() df['MA_mid'] = df['close'].rolling(window=10).mean() df['MA_long'] = df['close'].rolling(window=20).mean() df['ma_buy'] = ((df['MA_short'] > df['MA_mid']) & (df['MA_mid'] > df['MA_long'])) & ( ~((df['MA_short'].shift(1) > df['MA_mid'].shift(1)) & (df['MA_mid'].shift(1) > df['MA_long'].shift(1)))) df['ma_sell'] = ((df['MA_short'] < df['MA_mid']) & (df['MA_mid'] < df['MA_long'])) & ( ~((df['MA_short'].shift(1) < df['MA_mid'].shift(1)) & (df['MA_mid'].shift(1) < df['MA_long'].shift(1)))) delta = df['close'].diff() gain = delta.where(delta > 0, 0) loss = -delta.where(delta < 0, 0) avg_gain = gain.rolling(window=14).mean() avg_loss = loss.rolling(window=14).mean() rs = avg_gain / avg_loss df['RSI'] = 100 - (100 / (1 + rs)).fillna(100) df['rsi_buy'] = (df['RSI'] < 30) & (df['RSI'].shift(1) >= 30) df['rsi_sell'] = (df['RSI'] > 70) & (df['RSI'].shift(1) <= 70) # 3.3 动态仓位管理 def calculate_position(data, risk_tolerance=0.02, max_loss_pct=0.05): data['base_position'] = 0.7 data.loc[data['market_type'] == 'shock', 'base_position'] = 0.3 data['position_ratio'] = (risk_tolerance / max_loss_pct) * data['base_position'] return data['position_ratio'].clip(lower=0.1, upper=1.0) df['position_ratio'] = calculate_position(df) # 3.4 精细化止损止盈 df['stop_loss_price'] = 0.0 df['take_profit_price'] = 0.0 df['entry_price'] = 0.0 df['final_buy'] = False df['final_sell'] = False entry_price = 0.0 position = 0 for date in df.index: market_type = df.loc[date, 'market_type'] close = df.loc[date, 'close'] atr = df.loc[date, 'ATR14'] if (df.loc[date, 'ma_buy'] and market_type == 'trend') or (df.loc[date, 'rsi_buy'] and market_type == 'shock'): if position == 0: entry_price = close df.loc[date, 'entry_price'] = entry_price df.loc[date, 'final_buy'] = True position = 1 if position == 1: profit_pct = (close - entry_price) / entry_price atr_multiplier = 2.0 if market_type == 'trend' else 1.5 stop_loss = close - atr_multiplier * atr df.loc[date, 'stop_loss_price'] = stop_loss if profit_pct < 0.1: take_profit = entry_price * 1.2 elif 0.1 <= profit_pct < 0.2: take_profit = close - 1.0 * atr else: take_profit = close - 0.5 * atr df.loc[date, 'take_profit_price'] = take_profit if close < stop_loss or close > take_profit: df.loc[date, 'final_sell'] = True position = 0 entry_price = 0.0 # 3.5 计算单只股票收益率 df['daily_return'] = 0.0 position = 0 pos_ratio = 0.0 for date in df.index: if df.loc[date, 'final_buy']: position = 1 pos_ratio = df.loc[date, 'position_ratio'] elif df.loc[date, 'final_sell']: position = 0 pos_ratio = 0.0 if position == 1: df.loc[date, 'daily_return'] = pos_ratio * (df.loc[date, 'close'].shift(-1) / df.loc[date, 'close'] - 1) df['cum_return'] = (1 + df['daily_return']).cumprod() return df# 4. 批量获取各股票策略数据（与第8篇一致）stock_strategy_data = {}for name, df in stock_data.items(): stock_strategy_data[name] = apply_strategy_with_risk_control(df.copy())print("前文策略数据衔接完成，各股票累计收益率：")for name, df in stock_strategy_data.items(): print(f"{name}：{(df['cum_return'].iloc[-2]-1)*100:.2f}%")第二步：核心实现：动态权重分配模块（Python代码）

这是本文的核心部分，我们封装动态权重计算函数，实现“波动率+近期收益”双维度评分，以及定期再平衡逻辑。

# 5. 动态权重分配核心模块def calculate_dynamic_weights(stock_strategy_data, rebalance_period=20, lookback_period=20): """ 计算动态权重（波动率+近期收益加权） 参数： - stock_strategy_data: 含各股票策略数据的字典（来自前文） - rebalance_period: 权重再平衡周期（默认20个交易日，约1个月） - lookback_period: 计算近期收益的回溯周期（默认20个交易日） 返回： - weight_df: 动态权重DataFrame，index=日期，columns=股票名称，values=权重 """ # 5.1 统一日期索引，确保所有股票数据对齐 all_dates = pd.date_range(start=start_date, end=end_date, freq='B') stock_names = list(stock_strategy_data.keys()) # 初始化权重DataFrame（默认全0） weight_df = pd.DataFrame(0.0, index=all_dates, columns=stock_names) # 5.2 计算每只股票的波动率和近期收益（用于评分） # 存储各股票的波动率和近期收益 vol_data = pd.DataFrame(0.0, index=all_dates, columns=stock_names) recent_return_data = pd.DataFrame(0.0, index=all_dates, columns=stock_names) for name in stock_names: df = stock_strategy_data[name].reindex(all_dates, fill_value=0.0) # 计算滚动波动率（lookback_period个交易日的日收益率标准差） vol_data[name] = df['daily_return'].rolling(window=lookback_period).std() # 计算近期收益（lookback_period个交易日的累计收益） recent_return_data[name] = (1 + df['daily_return']).rolling(window=lookback_period).apply(lambda x: x[-1]/x[0] - 1) # 5.3 动态权重计算（按再平衡周期更新） for i in range(rebalance_period, len(all_dates), rebalance_period): # 确定当前再平衡日期 rebalance_date = all_dates[i] # 获取再平衡日之前lookback_period的数据（确保数据充足） start_idx = max(0, i - lookback_period) current_vol = vol_data.iloc[start_idx:i].iloc[-1] # 最新波动率 current_recent_return = recent_return_data.iloc[start_idx:i].iloc[-1] # 最新近期收益 # 5.3.1 数据标准化（将波动率和收益映射到0-1区间） # 波动率标准化：越小越好，所以用（1 - 标准化值） vol_normalized = (current_vol - current_vol.min()) / (current_vol.max() - current_vol.min()) vol_score = 1 - vol_normalized # 波动率越低，得分越高 # 近期收益标准化：越大越好，直接标准化 return_normalized = (current_recent_return - current_recent_return.min()) / (current_recent_return.max() - current_recent_return.min()) return_score = return_normalized # 收益越高，得分越高 # 5.3.2 综合评分（波动率和收益各占50%权重） total_score = 0.5 * vol_score + 0.5 * return_score # 5.3.3 计算动态权重（评分占比） dynamic_weights = total_score / total_score.sum() # 5.3.4 填充当前再平衡周期的权重（直到下一次再平衡） end_idx = min(i + rebalance_period, len(all_dates)) weight_df.iloc[i:end_idx] = dynamic_weights.values # 5.3.5 处理初始周期（第一个rebalance_period内）：用等权过渡 weight_df.iloc[:rebalance_period] = 1 / len(stock_names) # 权重修正：确保每行权重和为1（避免计算误差） weight_df = weight_df.div(weight_df.sum(axis=1), axis=0) return weight_df# 6. 计算动态权重dynamic_weight_df = calculate_dynamic_weights(stock_strategy_data)# 查看动态权重变化（以几个关键日期为例）print("\n=== 动态权重示例（关键日期）===")key_dates = [pd.Timestamp('2023-03-31'), pd.Timestamp('2023-06-30'), pd.Timestamp('2023-09-30'), pd.Timestamp('2024-03-31')]for date in key_dates: if date in dynamic_weight_df.index: print(f"\n{date.strftime('%Y-%m-%d')} 动态权重：") weight_row = dynamic_weight_df.loc[date].round(3) for name, weight in weight_row.items(): print(f"{name}：{weight:.1%}")# 对比：等权权重equal_weight = 1 / len(stock_names)print(f"\n等权组合权重（固定）：每只股票 {equal_weight:.1%}")第三步：动态权重组合回测与效果对比

我们计算动态权重组合的收益率，并与上一篇的等权组合、单股票策略对比，验证动态权重的优化效果。回测逻辑与前文一致，确保对比公平。

# 7. 动态权重组合回测# 7.1 计算动态权重组合的每日收益率dynamic_combo_daily_return = pd.Series(0.0, index=dynamic_weight_df.index)for date in dynamic_weight_df.index: # 获取当日各股票权重 daily_weights = dynamic_weight_df.loc[date] # 计算当日组合收益率：各股票收益率 × 当日权重 求和 for name in stock_names: if date in stock_strategy_data[name].index: daily_return = stock_strategy_data[name].loc[date, 'daily_return'] dynamic_combo_daily_return.loc[date] += daily_return * daily_weights[name]# 7.2 计算动态权重组合累计收益率dynamic_combo_cum_return = (1 + dynamic_combo_daily_return).cumprod()# 7.3 复用第8篇等权组合数据（用于对比）# 计算等权组合收益率（复现第8篇逻辑）equal_weight = 1 / len(stock_names)equal_combo_daily_return = pd.Series(0.0, index=dynamic_weight_df.index)for name in stock_names: daily_return_aligned = stock_strategy_data[name]['daily_return'].reindex(dynamic_weight_df.index, fill_value=0.0) equal_combo_daily_return += daily_return_aligned * equal_weightequal_combo_cum_return = (1 + equal_combo_daily_return).cumprod()# 7.4 计算核心评估指标（复用前文指标函数）def calculate_metrics(cum_return, daily_return): final_return = (cum_return.iloc[-1] - 1) * 100 cum_max = cum_return.cummax() drawdown = (cum_return - cum_max) / cum_max max_drawdown = drawdown.min() * 100 annual_return = final_return / 1.5 # 回测1.5年 daily_vol = daily_return.std() * (240 ** 0.5) # 年化波动率 sharpe = (annual_return - 3) / daily_vol if daily_vol != 0 else 0 return { '累计收益率(%)': round(final_return, 2), '最大回撤(%)': round(max_drawdown, 2), '年化波动率(%)': round(daily_vol*100, 2), '夏普比率': round(sharpe, 2) }# 计算各组合指标dynamic_metrics = calculate_metrics(dynamic_combo_cum_return, dynamic_combo_daily_return)equal_metrics = calculate_metrics(equal_combo_cum_return, equal_combo_daily_return)# 计算单股票平均指标（用于参考）stock_avg_metrics = { '累计收益率(%)': 0.0, '最大回撤(%)': 0.0, '年化波动率(%)': 0.0, '夏普比率': 0.0}for name in stock_names: cum_ret = stock_strategy_data[name]['cum_return'].reindex(dynamic_weight_df.index, fill_value=1.0) daily_ret = stock_strategy_data[name]['daily_return'].reindex(dynamic_weight_df.index, fill_value=0.0) metrics = calculate_metrics(cum_ret, daily_ret) for key in stock_avg_metrics.keys(): stock_avg_metrics[key] += metrics[key] / len(stock_names)# 7.5 输出对比结果print("\n=== 动态权重组合 vs 等权组合 vs 单股票平均 核心指标对比 ===")print(f"动态权重组合：{dynamic_metrics}")print(f"等权组合：{equal_metrics}")print(f"单股票平均：{stock_avg_metrics}")第四步：专业可视化：动态权重组合效果全景图

用3张子图直观展示动态权重的优势：① 动态权重vs等权vs单股票累计收益对比；② 动态权重变化趋势；③ 各组合风险收益指标对比（散点图）。

# 8. 可视化动态权重组合效果fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(14, 15), gridspec_kw={'height_ratios': [3, 2, 2]})# 子图1：动态权重组合 vs 等权组合 vs 单股票累计收益ax1.plot(dynamic_combo_cum_return.index, dynamic_combo_cum_return, label='动态权重组合', color='red', linewidth=2.0)ax1.plot(equal_combo_cum_return.index, equal_combo_cum_return, label='等权组合', color='blue', linewidth=1.5)# 叠加单股票收益（透明度降低，作为参考）for name, df in stock_strategy_data.items(): cum_ret_aligned = df['cum_return'].reindex(dynamic_combo_cum_return.index, fill_value=1.0) ax1.plot(cum_ret_aligned.index, cum_ret_aligned, label=name, linewidth=1.0, alpha=0.5)ax1.set_title('动态权重组合 vs 等权组合 vs 单股票累计收益率对比（2023-2024）', fontsize=14, pad=15)ax1.set_ylabel('累计收益（单位：1）', fontsize=12)ax1.legend(fontsize=9, loc='upper left')ax1.grid(True, alpha=0.3)# 子图2：动态权重变化趋势（展示前3只股票，避免图表拥挤）top3_stocks = dynamic_weight_df.iloc[-1].nlargest(3).index.tolist()for name in top3_stocks: ax2.plot(dynamic_weight_df.index, dynamic_weight_df[name], label=name, linewidth=1.5)ax2.set_title('动态权重变化趋势（权重最高的3只股票）', fontsize=14, pad=15)ax2.set_ylabel('权重占比', fontsize=12)ax2.set_xlabel('日期', fontsize=12)ax2.legend(fontsize=10)ax2.grid(True, alpha=0.3)# 子图3：风险收益散点图（越靠近右上角越好）# 计算各组合/股票的风险（年化波动率）和收益（累计收益率）risk_return_data = []# 动态权重组合risk_return_data.append(('动态权重组合', dynamic_metrics['年化波动率(%)'], dynamic_metrics['累计收益率(%)'], 'red', 'o'))# 等权组合risk_return_data.append(('等权组合', equal_metrics['年化波动率(%)'], equal_metrics['累计收益率(%)'], 'blue', 's'))# 各单股票for name in stock_names: cum_ret = stock_strategy_data[name]['cum_return'].reindex(dynamic_weight_df.index, fill_value=1.0) daily_ret = stock_strategy_data[name]['daily_return'].reindex(dynamic_weight_df.index, fill_value=0.0) metrics = calculate_metrics(cum_ret, daily_ret) risk_return_data.append((name, metrics['年化波动率(%)'], metrics['累计收益率(%)'], 'gray', '^'))# 绘制散点图for label, risk, return_, color, marker in risk_return_data: ax3.scatter(risk, return_, label=label, color=color, marker=marker, s=80)ax3.set_title('各策略/股票风险收益对比（散点图）', fontsize=14, pad=15)ax3.set_xlabel('年化波动率（%）（风险）', fontsize=12)ax3.set_ylabel('累计收益率（%）（收益）', fontsize=12)ax3.legend(fontsize=9, loc='upper left')ax3.grid(True, alpha=0.3)plt.gcf().autofmt_xdate()plt.tight_layout()plt.show()可视化结果解读

1. 子图1中，动态权重组合的累计收益率明显高于等权组合，且走势更平稳——这是因为它自动给盈利好、波动小的股票提高权重，降低了亏损股票的拖累；

2. 子图2中，动态权重并非固定不变，而是随个股表现动态调整——比如某只股票近期收益飙升、波动下降，其权重会明显上升，反之则下降；

3. 子图3的风险收益散点图中，动态权重组合的散点更靠近“右上角”（高收益、低风险），说明其风险收益比显著优于等权组合和大多数单股票，验证了动态权重的优化效果。

三、新手常见问题解答

1. 再平衡周期选20天合适吗？可以调整吗？—— 20天是新手最优选择：太短会增加交易成本，太长会导致权重偏离最优状态；激进型可缩短至10天，保守型可延长至30天；

2. 评分时波动率和收益的权重（各50%）可以改吗？—— 完全可以！保守型投资者可将波动率权重提高到60%-70%（优先控制风险），激进型可将收益权重提高到60%-70%（优先追求收益）；

3. 动态权重组合的交易成本会更高吗？—— 会略高，但可控：再平衡周期越长，交易成本越低；且动态权重会减少对差股票的配置，避免无效交易，长期来看利大于弊。

四、下一篇预告

今天咱们实现了“波动率+近期收益”双维度动态权重，让多股票组合收益再升级。下一篇（第10篇），咱们将聚焦组合策略的“实战落地细节”——讲解如何处理复权、手续费、滑点等实战问题，让咱们的量化策略从“回测好看”变成“实盘能用”。

如果这篇文章对你有帮助，麻烦点赞+关注，后续100篇量化干货持续更新！有任何问题，评论区留言，我会逐一解答～