问题描述
怎么在策略里用 Scipy 做投资组合优化？
解决方案
在 PTrade 策略中使用 Scipy 进行投资组合优化，通常是指利用现代投资组合理论（如马科维茨均值-方差模型），通过数学规划方法来计算在特定风险下收益最大化（或特定收益下风险最小化）的资产权重。
最常见的场景是最大化夏普比率（Max Sharpe Ratio）。
以下是一个完整的 PTrade 策略示例。该策略每月运行一次，选取上证50成分股，利用 scipy.optimize.minimize 计算最优权重，并进行调仓。
策略逻辑概述
- 获取数据：获取股票池过去一段时间的历史收盘价。
- 计算统计量：计算每只股票的预期收益率（年化）和协方差矩阵。
- 定义目标函数：定义“负夏普比率”函数（因为
minimize是求最小值，最大化夏普比率等同于最小化负夏普比率）。 - 设置约束与边界：
- 约束：所有股票权重之和为 1。
- 边界：单只股票权重在 0 到 1 之间（不做卖空）。
- 求解：使用
SLSQP算法求解最优权重。 - 执行交易：根据最优权重和当前总资产，计算目标持仓市值并下单。
策略代码实现
import numpy as np
import pandas as pd
from scipy.optimize import minimize
def initialize(context):
"""
初始化函数
"""
# 设置基准
set_benchmark('000016.SS')
# 设置滑点
set_slippage(0.002)
# 设置佣金
set_commission(commission_ratio=0.0003, min_commission=5.0)
# 定义全局变量：股票池（上证50）
g.index_code = '000016.SS'
# 设置无风险利率 (用于计算夏普比率，这里假设为3%)
g.risk_free_rate = 0.03
# 设定回测/交易频率：每月第一个交易日进行调仓
run_daily(context, rebalance_portfolio, time='09:35')
def before_trading_start(context, data):
"""
盘前处理：每日更新股票池，剔除ST和停牌
"""
# 获取指数成分股
stocks = get_index_stocks(g.index_code)
# 过滤ST、停牌、退市股票
# 注意：filter_stock_by_status 返回的是剔除后的列表
g.stocks = filter_stock_by_status(stocks, filter_type=["ST", "HALT", "DELISTING"])
def get_portfolio_stats(weights, mean_returns, cov_matrix):
"""
计算投资组合的预期收益和波动率
"""
# 投资组合年化收益
returns = np.sum(mean_returns * weights) * 252
# 投资组合年化波动率
std = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights))) * np.sqrt(252)
return returns, std
def min_func_sharpe(weights, mean_returns, cov_matrix, risk_free_rate):
"""
目标函数：负夏普比率 (用于最小化)
"""
p_returns, p_std = get_portfolio_stats(weights, mean_returns, cov_matrix)
# 避免分母为0
if p_std == 0:
return 1e6
# 夏普比率 = (预期收益 - 无风险利率) / 波动率
sharpe_ratio = (p_returns - risk_free_rate) / p_std
return -sharpe_ratio
def optimize_portfolio(data_df, risk_free_rate):
"""
使用 Scipy 进行优化
"""
# 计算日收益率
returns_df = data_df.pct_change().dropna()
# 如果数据太少，无法计算协方差，返回None
if len(returns_df) < 20:
return None
# 计算平均日收益和协方差矩阵
mean_returns = returns_df.mean()
cov_matrix = returns_df.cov()
num_assets = len(mean_returns)
# 初始猜测：等权重
init_guess = num_assets * [1. / num_assets,]
# 权重范围：0 <= weight <= 1 (不做卖空)
bounds = tuple((0.0, 1.0) for i in range(num_assets))
# 约束条件：权重之和为 1
constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
# 使用 SLSQP 方法进行优化
try:
result = minimize(min_func_sharpe, init_guess, args=(mean_returns, cov_matrix, risk_free_rate,),
method='SLSQP', bounds=bounds, constraints=constraints)
except Exception as e:
log.error("Optimization failed: %s" % str(e))
return None
if result['success']:
# 返回资产代码和对应的最优权重
return dict(zip(mean_returns.index, result['x']))
else:
return None
def rebalance_portfolio(context):
"""
调仓逻辑
"""
# 判断是否是月初 (这里简单判断：如果是该月的前几个交易日)
# 实际策略中可以使用更复杂的日期判断，这里为了演示方便，每次run_daily都检查
# 为了演示效果，我们设定每隔20个交易日调仓一次
# 在PTrade中，可以通过 context.blotter.current_dt 获取当前时间
# 获取当前日期
current_date = context.blotter.current_dt
# 这里简化逻辑：仅在每月1号或者回测开始时运行，或者利用全局计数器
# 为了确保代码简洁，这里假设每次调用该函数都尝试调仓（配合run_daily的逻辑）
# 实际生产中建议配合 date.day == 1 或类似逻辑
# 1. 获取历史数据 (过去120个交易日)
# 注意：get_history 返回 DataFrame，列是股票代码 (Python 3.5+ PTrade环境)
if len(g.stocks) == 0:
return
hist_data = get_history(120, '1d', 'close', g.stocks, fq='pre')
# 检查数据格式，确保是 DataFrame 且索引正确
if hist_data is None or hist_data.empty:
log.warning("No history data fetched.")
return
# PTrade get_history 返回格式处理
# 如果是多只股票，通常返回 DataFrame，index是时间，columns是股票代码
# 确保只取收盘价
if 'close' in hist_data.columns and 'code' in hist_data.columns:
# 这种格式通常是 Panel 转的 Long Format，需要 pivot
# 但 PTrade 文档中 get_history 多股单字段通常直接返回宽表 (index=date, col=code)
# 这里假设是宽表格式，如果不是，需根据实际环境调整
pass
# 2. 执行优化
optimal_weights = optimize_portfolio(hist_data, g.risk_free_rate)
if not optimal_weights:
log.info("Optimization did not return a result.")
return
# 3. 执行交易
log.info("Rebalancing portfolio...")
# 获取当前账户总资产
total_value = context.portfolio.portfolio_value
# 先卖出不在最优组合中的股票，或者权重极小的股票
current_positions = context.portfolio.positions
for stock in list(current_positions.keys()):
if stock not in optimal_weights or optimal_weights[stock] < 0.001:
order_target_value(stock, 0)
# 买入/调整股票
for stock, weight in optimal_weights.items():
# 忽略极小的权重
if weight < 0.001:
continue
target_value = total_value * weight
order_target_value(stock, target_value)
log.info("Optimization and rebalance complete.")
def handle_data(context, data):
"""
盘中运行函数 (本策略主要逻辑在定时任务中，此处留空)
"""
pass
代码关键点解析
-
scipy.optimize.minimize的使用：- 目标函数 (
min_func_sharpe)：我们需要最大化夏普比率，但minimize函数是寻找最小值，所以我们返回-(夏普比率)。 - 参数 (
args)：将计算好的mean_returns（预期收益向量）和cov_matrix（协方差矩阵）以及无风险利率传递给目标函数。 - 约束 (
constraints)：{'type': 'eq', 'fun': lambda x: np.sum(x) - 1}表示所有资产的权重之和必须等于 1（全额投资）。 - 边界 (
bounds)：((0, 1), ...)表示每个资产的权重必须在 0 到 1 之间，这意味着不允许卖空（Short Selling）。 - 方法 (
method)：使用'SLSQP'(Sequential Least SQuares Programming)，这是处理带约束非线性优化问题的常用算法。
- 目标函数 (
-
数据处理：
- 使用
get_history获取收盘价。 - 使用
pct_change()计算日收益率。 - 使用
cov()计算协方差矩阵，这是现代投资组合理论（MPT）中衡量风险的核心。
- 使用
-
交易执行：
- 计算出的
optimal_weights是一个字典{股票代码: 权重}。 - 使用
context.portfolio.portfolio_value获取当前总资产。 - 使用
order_target_value(stock, total_value * weight)将每只股票的持仓调整到目标市值。
- 计算出的
注意事项
- Python 版本兼容性：PTrade 的部分环境可能使用 Python 3.5 或 3.6。上述代码避免了 f-string 等新特性，使用了
%s格式化字符串，以确保兼容性。 - 数据清洗：在计算协方差矩阵前，使用了
.dropna()。如果某只股票停牌时间过长导致数据缺失，可能会影响矩阵计算，实际生产中可能需要更复杂的填充逻辑。 - 运行频率：上述代码在
initialize中使用了run_daily。如果你希望严格按月调仓，可以在rebalance_portfolio内部增加日期判断（例如if context.blotter.current_dt.day <= 5:且通过全局变量控制当月只运行一次）。 - 计算耗时：如果股票池非常大（例如全A股），
minimize的计算可能会比较慢。对于上证50或沪深300这样的规模，通常是可以接受的。
Q&A
Q: 为什么优化结果有时候会集中在少数几只股票上？
A: 这是均值-方差优化的常见现象（Corner Solution）。如果某些股票在历史数据中表现出极高的收益风险比，算法会倾向于重仓这些股票。可以通过添加额外的约束（例如单只股票权重不超过 10%）来解决，修改 bounds 为 ((0, 0.1), ...) 即可。
Q: 如何更换优化目标，例如最小化波动率？
A: 只需修改目标函数 min_func_sharpe。
def min_func_volatility(weights, mean_returns, cov_matrix, risk_free_rate):
# 仅返回波动率
_, p_std = get_portfolio_stats(weights, mean_returns, cov_matrix)
return p_std
然后在 minimize 调用时传入这个新函数即可。
Q: get_history 获取的数据包含停牌日的填充吗？
A: PTrade 的 get_history 通常会用前一交易日的数据填充停牌日（价格不变，成交量为0）。这会导致收益率为0，在计算协方差时可能会低估波动率。在 before_trading_start 中使用 filter_stock_by_status 剔除停牌股票是非常必要的步骤。