Python期货量化策略优化技巧_提升策略表现的方法

天勤量化大唯粉

96人浏览 · 2026-03-03 11:19:08

天勤量化大唯粉 · 2026-03-03 11:19:08 发布

免责声明：本文基于个人使用体验，与任何厂商无商业关系。内容仅供技术交流参考，不构成投资建议。

一、前言

策略优化是量化交易中持续进行的工作。一个策略从初版到最终版本，往往需要经过多次优化。做了二十年期货量化，我在策略优化上积累了一些技巧。

今天分享一些Python期货量化策略优化的实用技巧，帮大家提升策略表现。

二、策略优化维度

1. 信号优化

# 信号优化的方法
signal_optimization = {
    "方法1": "信号过滤（成交量、波动率等）",
    "方法2": "多重确认（多个指标同时确认）",
    "方法3": "信号强度过滤（只接受强信号）",
    "方法4": "信号间隔控制（避免信号过于密集）"
}

2. 参数优化

# 参数优化的方法
parameter_optimization = {
    "方法1": "参数稳定性检验",
    "方法2": "样本外验证",
    "方法3": "滚动窗口优化",
    "方法4": "贝叶斯优化"
}

3. 风控优化

# 风控优化的方法
risk_optimization = {
    "方法1": "动态止损（ATR、移动止损）",
    "方法2": "仓位管理（固定比例、波动率调整）",
    "方法3": "资金管理（单笔风险控制）",
    "方法4": "多策略组合（分散风险）"
}

三、策略优化技巧

技巧一：信号过滤优化

def optimized_signal_filter(klines):
    """优化的信号过滤"""
    from tqsdk.ta import MA, MACD, RSI
    
    # 基础信号
    ma5 = MA(klines, 5)
    ma20 = MA(klines, 20)
    base_signal = None
    
    if ma5.ma.iloc[-2] > ma20.ma.iloc[-2] and ma5.ma.iloc[-1] <= ma20.ma.iloc[-1]:
        base_signal = "BUY"
    elif ma5.ma.iloc[-2] < ma20.ma.iloc[-2] and ma5.ma.iloc[-1] >= ma20.ma.iloc[-1]:
        base_signal = "SELL"
    
    if base_signal:
        # 多重确认
        macd = MACD(klines, 12, 26, 9)
        rsi = RSI(klines, 14)
        
        macd_confirm = macd['bar'].iloc[-2] > 0
        rsi_confirm = rsi['rsi'].iloc[-2] > 50
        
        # 成交量确认
        volume_avg = klines['volume'].rolling(20).mean()
        volume_confirm = klines['volume'].iloc[-1] > volume_avg.iloc[-1] * 1.5
        
        # 需要至少2个确认
        confirmations = sum([macd_confirm, rsi_confirm, volume_confirm])
        if confirmations >= 2:
            return base_signal
    
    return None

技巧二：参数稳定性优化

def optimize_with_stability(strategy_func, data, param_ranges):
    """带稳定性检验的参数优化"""
    import numpy as np
    
    best_params = None
    best_score = -float('inf')
    best_stability = 0
    
    # 网格搜索
    for params in generate_param_combinations(param_ranges):
        # 样本外验证
        train_data, test_data = split_data(data, 0.7)
        
        train_result = strategy_func(train_data, params)
        test_result = strategy_func(test_data, params)
        
        # 计算稳定性（测试集表现/训练集表现）
        stability = test_result['sharpe'] / train_result['sharpe'] if train_result['sharpe'] > 0 else 0
        
        # 综合评分（测试集表现 + 稳定性）
        score = test_result['sharpe'] * 0.7 + stability * 0.3
        
        if score > best_score:
            best_score = score
            best_params = params
            best_stability = stability
    
    return best_params, best_score, best_stability

技巧三：动态仓位优化

def dynamic_position_sizing(klines, account, base_position=1):
    """动态仓位调整"""
    import numpy as np
    
    # 计算波动率
    returns = klines['close'].pct_change()
    volatility = returns.rolling(20).std().iloc[-1]
    
    # 波动率低时加仓，波动率高时减仓
    if volatility < 0.01:  # 低波动
        position_multiplier = 1.5
    elif volatility > 0.03:  # 高波动
        position_multiplier = 0.5
    else:
        position_multiplier = 1.0
    
    # 根据资金调整
    capital_ratio = account.balance / 100000  # 相对于10万的比例
    if capital_ratio > 2.0:
        position_multiplier *= 1.2
    elif capital_ratio < 0.5:
        position_multiplier *= 0.8
    
    final_position = int(base_position * position_multiplier)
    return max(final_position, 1)  # 至少1手

四、在TqSdk中实现策略优化

from tqsdk import TqApi, TqAuth, TqBacktest
from datetime import date

def optimized_strategy():
    """优化后的策略"""
    api = TqApi(
        backtest=TqBacktest(start_dt=date(2024, 1, 1), end_dt=date(2025, 12, 31)),
        auth=TqAuth("账户", "密码")
    )
    
    symbol = "SHFE.rb2510"
    klines = api.get_kline_serial(symbol, 300, 200)
    position = api.get_position(symbol)
    account = api.get_account()
    
    while True:
        api.wait_update()
        
        if api.is_changing(klines):
            # 优化的信号生成
            signal = optimized_signal_filter(klines)
            
            if signal:
                # 动态仓位
                position_size = dynamic_position_sizing(klines, account)
                
                if signal == "BUY" and position.pos_long == 0:
                    api.insert_order(symbol, "BUY", "OPEN", position_size)
                elif signal == "SELL" and position.pos_long > 0:
                    api.insert_order(symbol, "SELL", "CLOSE", position.pos_long)
    
    api.close()

五、优化效果评估

def evaluate_optimization(before_metrics, after_metrics):
    """评估优化效果"""
    improvements = {
        'return_improvement': (after_metrics['return'] - before_metrics['return']) / before_metrics['return'],
        'sharpe_improvement': (after_metrics['sharpe'] - before_metrics['sharpe']) / before_metrics['sharpe'],
        'drawdown_improvement': (before_metrics['max_drawdown'] - after_metrics['max_drawdown']) / before_metrics['max_drawdown']
    }
    
    return improvements