【期货量化进阶】期货量化交易策略组合优化（Python量化）

一、前言

单一策略往往存在局限性，策略组合可以分散风险、平滑收益。如何优化策略组合的权重分配，实现风险调整后的最优收益，是量化交易的重要课题。本文将介绍策略组合优化的方法。

本文将介绍：

• 策略组合构建
• 权重优化方法
• 风险平价组合
• 组合绩效评估

二、为什么选择天勤量化（TqSdk）

TqSdk策略组合支持：

安装方法：

pip install tqsdk pandas numpy scipy

三、策略组合构建

3.1 多策略组合框架

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：多策略组合框架
说明：本代码仅供学习参考
"""

from typing import List, Dict
import pandas as pd
import numpy as np
from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask

class StrategyPortfolio:
"""策略组合"""

def __init__(self, api, strategies: List[Dict], weights: List[float] = None):
"""
初始化策略组合

参数:
api: TqApi对象
strategies: 策略列表，每个策略包含symbol和strategy对象
weights: 权重列表，默认等权重
"""
self.api = api
self.strategies = strategies
self.weights = weights if weights else [1.0 / len(strategies)] * len(strategies)

# 归一化权重
total_weight = sum(self.weights)
self.weights = [w / total_weight for w in self.weights]

# 创建目标持仓任务
self.target_pos_tasks = {}
for i, strategy_info in enumerate(self.strategies):
symbol = strategy_info['symbol']
self.target_pos_tasks[symbol] = TargetPosTask(api, symbol)

def update_positions(self, account_balance: float):
"""
更新各策略持仓

参数:
account_balance: 账户总资金
"""
for i, strategy_info in enumerate(self.strategies):
symbol = strategy_info['symbol']
strategy = strategy_info['strategy']
weight = self.weights[i]

# 获取策略信号
signal = strategy.get_signal()

# 计算目标持仓（根据权重分配资金）
allocated_balance = account_balance * weight
quote = self.api.get_quote(symbol)

# 简化：假设每手保证金为价格的10%
price = quote.last_price
margin_per_lot = price * 10 * 0.1 # 假设1手=10吨

if margin_per_lot > 0:
target_lots = int(allocated_balance / margin_per_lot) * signal
self.target_pos_tasks[symbol].set_target_volume(target_lots)

# 示例策略
class SimpleStrategy:
"""简单策略"""

def __init__(self, symbol, api):
self.symbol = symbol
self.api = api
self.klines = api.get_kline_serial(symbol, 3600, 100)
self.quote = api.get_quote(symbol)
self.signal = 0

def update(self):
"""更新策略"""
self.api.wait_update()

if self.api.is_changing(self.klines.iloc[–1], "datetime"):
# 简单均线策略
if len(self.klines) > 30:
ma_short = self.klines['close'].tail(10).mean()
ma_long = self.klines['close'].tail(30).mean()

if ma_short > ma_long:
self.signal = 1
else:
self.signal = –1

def get_signal(self):
"""获取信号"""
return self.signal

# 使用示例
api = TqApi(TqSim(init_balance=1000000), auth=TqAuth("快期账户", "快期密码"))

strategies = [
{'symbol': 'SHFE.rb2510', 'strategy': SimpleStrategy('SHFE.rb2510', api)},
{'symbol': 'DCE.i2509', 'strategy': SimpleStrategy('DCE.i2509', api)},
]

portfolio = StrategyPortfolio(api, strategies, weights=[0.6, 0.4])

print("=" * 50)
print("策略组合构建")
print("=" * 50)

print(f"策略数量: {len(portfolio.strategies)}")
print(f"权重分配: {portfolio.weights}")

api.close()

3.2 策略相关性分析

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：策略相关性分析
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np

def analyze_strategy_correlation(returns_df):
"""
分析策略收益相关性

参数:
returns_df: 策略收益DataFrame，每列为一个策略
"""
# 计算相关系数矩阵
corr_matrix = returns_df.corr()

# 找出高相关策略对
high_corr_pairs = []
for i in range(len(corr_matrix.columns)):
for j in range(i+1, len(corr_matrix.columns)):
corr = corr_matrix.iloc[i, j]
if abs(corr) > 0.7:
high_corr_pairs.append({
'strategy1': corr_matrix.columns[i],
'strategy2': corr_matrix.columns[j],
'correlation': corr
})

# 计算平均相关性
avg_correlation = corr_matrix.values[np.triu_indices_from(corr_matrix.values, k=1)].mean()

return {
'correlation_matrix': corr_matrix,
'high_corr_pairs': pd.DataFrame(high_corr_pairs),
'avg_correlation': avg_correlation
}

# 示例使用
# 模拟策略收益数据
np.random.seed(42)
n_periods = 252

# 策略1和策略2相关
strategy1_returns = np.random.randn(n_periods) * 0.02
strategy2_returns = strategy1_returns * 0.7 + np.random.randn(n_periods) * 0.01
strategy3_returns = np.random.randn(n_periods) * 0.02

returns_df = pd.DataFrame({
'策略1': strategy1_returns,
'策略2': strategy2_returns,
'策略3': strategy3_returns
})

analysis = analyze_strategy_correlation(returns_df)

print("=" * 50)
print("策略相关性分析")
print("=" * 50)

print("相关系数矩阵:")
print(analysis['correlation_matrix'])

print(f"\\n平均相关性: {analysis['avg_correlation']:.4f}")

if len(analysis['high_corr_pairs']) > 0:
print("\\n高相关策略对:")
print(analysis['high_corr_pairs'])

四、权重优化方法

4.1 均值方差优化

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：均值方差优化
说明：本代码仅供学习参考
"""

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

def mean_variance_optimize(returns_df, target_return=None, risk_free_rate=0.025):
"""
均值方差优化（Markowitz）

参数:
returns_df: 策略收益DataFrame
target_return: 目标收益率（可选）
risk_free_rate: 无风险利率
"""
# 计算期望收益和协方差矩阵
mean_returns = returns_df.mean() * 252 # 年化
cov_matrix = returns_df.cov() * 252 # 年化

n_strategies = len(returns_df.columns)

def portfolio_variance(weights):
"""组合方差"""
return np.dot(weights.T, np.dot(cov_matrix, weights))

def portfolio_return(weights):
"""组合收益"""
return np.sum(mean_returns * weights)

def negative_sharpe(weights):
"""负夏普比率（用于最小化）"""
port_return = portfolio_return(weights)
port_var = portfolio_variance(weights)
port_std = np.sqrt(port_var)
sharpe = (port_return – risk_free_rate) / port_std if port_std > 0 else 0
return –sharpe

# 约束条件
constraints = [{'type': 'eq', 'fun': lambda x: np.sum(x) – 1}] # 权重和为1

if target_return:
constraints.append({
'type': 'eq',
'fun': lambda x: portfolio_return(x) – target_return
})

# 边界条件（权重在0-1之间）
bounds = tuple((0, 1) for _ in range(n_strategies))

# 初始权重（等权重）
initial_weights = np.array([1.0 / n_strategies] * n_strategies)

# 优化（最大化夏普比率）
result = minimize(negative_sharpe, initial_weights, method='SLSQP',
bounds=bounds, constraints=constraints)

optimal_weights = result.x

# 计算最优组合指标
opt_return = portfolio_return(optimal_weights)
opt_var = portfolio_variance(optimal_weights)
opt_std = np.sqrt(opt_var)
opt_sharpe = (opt_return – risk_free_rate) / opt_std if opt_std > 0 else 0

return {
'weights': optimal_weights,
'expected_return': opt_return,
'volatility': opt_std,
'sharpe': opt_sharpe
}

# 示例使用
np.random.seed(42)
n_periods = 252

returns_df = pd.DataFrame({
'策略1': np.random.randn(n_periods) * 0.02 + 0.001,
'策略2': np.random.randn(n_periods) * 0.015 + 0.0008,
'策略3': np.random.randn(n_periods) * 0.018 + 0.0009,
})

optimization = mean_variance_optimize(returns_df)

print("=" * 50)
print("均值方差优化")
print("=" * 50)

print("最优权重:")
for i, weight in enumerate(optimization['weights']):
print(f" 策略{i+1}: {weight:.2%}")

print(f"\\n预期收益: {optimization['expected_return']:.2%}")
print(f"波动率: {optimization['volatility']:.2%}")
print(f"夏普比率: {optimization['sharpe']:.4f}")

4.2 风险平价优化

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：风险平价优化
说明：本代码仅供学习参考
"""

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

def risk_parity_optimize(returns_df):
"""
风险平价优化

参数:
returns_df: 策略收益DataFrame
"""
# 计算协方差矩阵
cov_matrix = returns_df.cov() * 252

n_strategies = len(returns_df.columns)

def risk_parity_objective(weights):
"""风险平价目标函数"""
portfolio_vol = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))

# 各策略对组合风险的贡献
marginal_contrib = np.dot(cov_matrix, weights) / portfolio_vol
contrib = weights * marginal_contrib

# 风险贡献应该相等（风险平价）
target_contrib = portfolio_vol / n_strategies

# 最小化风险贡献的差异
diff = contrib – target_contrib
return np.sum(diff ** 2)

# 约束条件
constraints = [{'type': 'eq', 'fun': lambda x: np.sum(x) – 1}]

# 边界条件
bounds = tuple((0, 1) for _ in range(n_strategies))

# 初始权重
initial_weights = np.array([1.0 / n_strategies] * n_strategies)

# 优化
result = minimize(risk_parity_objective, initial_weights, method='SLSQP',
bounds=bounds, constraints=constraints)

optimal_weights = result.x

# 计算风险贡献
portfolio_vol = np.sqrt(np.dot(optimal_weights.T, np.dot(cov_matrix, optimal_weights)))
marginal_contrib = np.dot(cov_matrix, optimal_weights) / portfolio_vol
risk_contrib = optimal_weights * marginal_contrib

return {
'weights': optimal_weights,
'risk_contributions': risk_contrib,
'portfolio_volatility': portfolio_vol
}

# 示例使用
np.random.seed(42)
n_periods = 252

returns_df = pd.DataFrame({
'策略1': np.random.randn(n_periods) * 0.02,
'策略2': np.random.randn(n_periods) * 0.015,
'策略3': np.random.randn(n_periods) * 0.018,
})

optimization = risk_parity_optimize(returns_df)

print("=" * 50)
print("风险平价优化")
print("=" * 50)

print("最优权重:")
for i, weight in enumerate(optimization['weights']):
print(f" 策略{i+1}: {weight:.2%}")

print("\\n风险贡献:")
for i, contrib in enumerate(optimization['risk_contributions']):
print(f" 策略{i+1}: {contrib:.4f}")

print(f"\\n组合波动率: {optimization['portfolio_volatility']:.2%}")

五、组合绩效评估

5.1 组合绩效分析

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：组合绩效分析
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np

def analyze_portfolio_performance(returns_df, weights):
"""
分析组合绩效

参数:
returns_df: 策略收益DataFrame
weights: 权重数组
"""
# 计算组合收益
portfolio_returns = (returns_df * weights).sum(axis=1)

# 计算累计收益
cumulative_returns = (1 + portfolio_returns).cumprod()

# 总收益
total_return = cumulative_returns.iloc[–1] – 1

# 年化收益
n_periods = len(returns_df)
annual_return = (1 + total_return) ** (252 / n_periods) – 1

# 波动率
volatility = portfolio_returns.std() * np.sqrt(252)

# 最大回撤
running_max = cumulative_returns.cummax()
drawdown = (cumulative_returns – running_max) / running_max
max_drawdown = abs(drawdown.min())

# 夏普比率
sharpe = annual_return / volatility if volatility > 0 else 0

# 各策略贡献
strategy_contributions = {}
for i, col in enumerate(returns_df.columns):
strategy_contrib = (returns_df[col] * weights[i]).sum()
strategy_contributions[col] = strategy_contrib

return {
'total_return': total_return,
'annual_return': annual_return,
'volatility': volatility,
'max_drawdown': max_drawdown,
'sharpe': sharpe,
'strategy_contributions': strategy_contributions,
'cumulative_returns': cumulative_returns
}

# 示例使用
np.random.seed(42)
n_periods = 252

returns_df = pd.DataFrame({
'策略1': np.random.randn(n_periods) * 0.02 + 0.001,
'策略2': np.random.randn(n_periods) * 0.015 + 0.0008,
'策略3': np.random.randn(n_periods) * 0.018 + 0.0009,
})

weights = np.array([0.4, 0.3, 0.3])

performance = analyze_portfolio_performance(returns_df, weights)

print("=" * 50)
print("组合绩效分析")
print("=" * 50)

print(f"总收益: {performance['total_return']:.2%}")
print(f"年化收益: {performance['annual_return']:.2%}")
print(f"波动率: {performance['volatility']:.2%}")
print(f"最大回撤: {performance['max_drawdown']:.2%}")
print(f"夏普比率: {performance['sharpe']:.4f}")

print("\\n各策略贡献:")
for strategy, contrib in performance['strategy_contributions'].items():
print(f" {strategy}: {contrib:.2%}")

5.2 组合优化对比

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：组合优化方法对比
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np

def compare_optimization_methods(returns_df):
"""
对比不同优化方法

参数:
returns_df: 策略收益DataFrame
"""
results = {}

# 1. 等权重
n_strategies = len(returns_df.columns)
equal_weights = np.array([1.0 / n_strategies] * n_strategies)
equal_perf = analyze_portfolio_performance(returns_df, equal_weights)
results['等权重'] = {
'weights': equal_weights,
'sharpe': equal_perf['sharpe'],
'return': equal_perf['annual_return'],
'volatility': equal_perf['volatility']
}

# 2. 均值方差优化
mv_opt = mean_variance_optimize(returns_df)
mv_weights = mv_opt['weights']
mv_perf = analyze_portfolio_performance(returns_df, mv_weights)
results['均值方差'] = {
'weights': mv_weights,
'sharpe': mv_perf['sharpe'],
'return': mv_perf['annual_return'],
'volatility': mv_perf['volatility']
}

# 3. 风险平价
rp_opt = risk_parity_optimize(returns_df)
rp_weights = rp_opt['weights']
rp_perf = analyze_portfolio_performance(returns_df, rp_weights)
results['风险平价'] = {
'weights': rp_weights,
'sharpe': rp_perf['sharpe'],
'return': rp_perf['annual_return'],
'volatility': rp_perf['volatility']
}

return results

# 示例使用
np.random.seed(42)
n_periods = 252

returns_df = pd.DataFrame({
'策略1': np.random.randn(n_periods) * 0.02 + 0.001,
'策略2': np.random.randn(n_periods) * 0.015 + 0.0008,
'策略3': np.random.randn(n_periods) * 0.018 + 0.0009,
})

comparison = compare_optimization_methods(returns_df)

print("=" * 50)
print("组合优化方法对比")
print("=" * 50)

for method, result in comparison.items():
print(f"\\n{method}:")
print(f" 权重: {result['weights']}")
print(f" 夏普比率: {result['sharpe']:.4f}")
print(f" 年化收益: {result['return']:.2%}")
print(f" 波动率: {result['volatility']:.2%}")

六、总结

组合优化速查

# 均值方差优化
weights = mean_variance_optimize(returns_df)

# 风险平价
weights = risk_parity_optimize(returns_df)

# 组合绩效
performance = analyze_portfolio_performance(returns_df, weights)

免责声明：本文仅供学习交流使用，不构成任何投资建议。期货交易有风险，入市需谨慎。

更多资源：

• 天勤量化官网：https://www.shinnytech.com
• GitHub开源地址：https://github.com/shinnytech/tqsdk-python
• 官方文档：https://doc.shinnytech.com/tqsdk/latest