2026年期货量化交易信号过滤_提高策略胜率的方法

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

一、前言

策略信号的质量直接影响交易结果。如何过滤低质量信号，提高策略胜率？2026年了，今天分享一下我在信号过滤方面的实践经验。

二、信号过滤的重要性

1. 信号质量问题

原始策略信号往往存在以下问题：

• 假信号：市场噪音产生的无效信号
• 弱信号：信号强度不足，胜率低
• 重复信号：短时间内产生多个相似信号

# 原始信号示例
raw_signals = [
{"time": "09:00", "type": "BUY", "strength": 0.3}, # 弱信号
{"time": "09:05", "type": "BUY", "strength": 0.8}, # 强信号
{"time": "09:06", "type": "BUY", "strength": 0.4}, # 重复信号
{"time": "09:10", "type": "SELL", "strength": 0.9}, # 强信号
]

# 如果全部执行，可能产生亏损
# 通过过滤，只执行高质量信号

2. 信号过滤的效果

# 信号过滤效果对比
# 过滤前
before_filtering = {
"信号数量": 100,
"胜率": 0.45, # 45%
"平均收益": 0.001
}

# 过滤后
after_filtering = {
"信号数量": 30, # 减少70%
"胜率": 0.65, # 提高到65%
"平均收益": 0.002 # 收益提高
}

三、信号过滤方法

1. 信号强度过滤

只执行强度超过阈值的信号：

# 信号强度过滤
def filter_by_strength(signals, min_strength=0.6):
"""
根据信号强度过滤
min_strength: 最小信号强度（0.6）
"""
filtered_signals = [
signal for signal in signals
if signal['strength'] >= min_strength
]
return filtered_signals

# 示例
signals = [
{"type": "BUY", "strength": 0.3},
{"type": "BUY", "strength": 0.8},
{"type": "SELL", "strength": 0.5},
{"type": "SELL", "strength": 0.9},
]

filtered = filter_by_strength(signals, min_strength=0.6)
# 只保留强度>=0.6的信号

2. 信号确认过滤

需要多个指标同时确认才执行：

# 信号确认过滤
def filter_by_confirmation(signals, indicators):
"""
需要多个指标确认
indicators: 指标列表，如['ma', 'macd', 'rsi']
"""
confirmed_signals = []

for signal in signals:
confirmation_count = 0
for indicator in indicators:
if indicator_confirms_signal(indicator, signal):
confirmation_count += 1

# 需要至少2个指标确认
if confirmation_count >= 2:
confirmed_signals.append(signal)

return confirmed_signals

# 示例
def indicator_confirms_signal(indicator, signal):
"""检查指标是否确认信号"""
if indicator == 'ma':
# 均线确认逻辑
return check_ma_confirmation(signal)
elif indicator == 'macd':
# MACD确认逻辑
return check_macd_confirmation(signal)
# …
return False

3. 时间间隔过滤

避免短时间内重复信号：

# 时间间隔过滤
from datetime import datetime, timedelta

class SignalTimeFilter:
def __init__(self, min_interval_minutes=30):
self.min_interval = timedelta(minutes=min_interval_minutes)
self.last_signal_time = {}

def filter(self, signal):
"""过滤时间间隔过短的信号"""
signal_type = signal['type']
signal_time = signal['time']

if signal_type in self.last_signal_time:
time_diff = signal_time – self.last_signal_time[signal_type]
if time_diff < self.min_interval:
return False # 时间间隔太短，过滤掉

self.last_signal_time[signal_type] = signal_time
return True

4. 市场状态过滤

只在特定市场状态下执行：

# 市场状态过滤
def filter_by_market_state(signals, market_state):
"""
根据市场状态过滤信号
market_state: 'trending', 'ranging', 'volatile'
"""
filtered_signals = []

for signal in signals:
# 趋势策略只在趋势市场执行
if signal['strategy'] == 'trend' and market_state != 'trending':
continue

# 均值回归策略只在震荡市场执行
if signal['strategy'] == 'mean_reversion' and market_state != 'ranging':
continue

filtered_signals.append(signal)

return filtered_signals

# 市场状态识别
def identify_market_state(klines):
"""识别市场状态"""
# 计算ADX判断趋势
adx = calculate_adx(klines)

# 计算波动率
volatility = calculate_volatility(klines)

if adx > 25:
return 'trending' # 趋势市场
elif volatility < 0.01:
return 'ranging' # 震荡市场
else:
return 'volatile' # 波动市场

5. 成交量过滤

只在成交量充足时执行：

# 成交量过滤
def filter_by_volume(signals, klines, min_volume_ratio=1.5):
"""
根据成交量过滤信号
min_volume_ratio: 最小成交量倍数（相对于平均成交量）
"""
avg_volume = klines['volume'].rolling(20).mean().iloc[–1]
current_volume = klines['volume'].iloc[–1]

if current_volume < avg_volume * min_volume_ratio:
# 成交量不足，过滤掉所有信号
return []

return signals

四、使用量化工具实现信号过滤

1. 使用TqSdk实现信号过滤

我目前主要使用TqSdk进行策略开发，可以方便地实现信号过滤：

from tqsdk import TqApi, TqAuth
import pandas as pd

api = TqApi(auth=TqAuth("账户", "密码"))

klines = api.get_kline_serial("SHFE.rb2401", 60, 200)
position = api.get_position("SHFE.rb2401")

# 信号过滤类
class SignalFilter:
def __init__(self):
self.last_signal_time = None
self.min_strength = 0.6
self.min_interval_minutes = 30

def filter_signal(self, signal, current_time):
"""综合过滤信号"""
# 1. 强度过滤
if signal['strength'] < self.min_strength:
return False

# 2. 时间间隔过滤
if self.last_signal_time:
time_diff = (current_time – self.last_signal_time).total_seconds() / 60
if time_diff < self.min_interval_minutes:
return False

# 3. 市场状态过滤
market_state = self.identify_market_state(klines)
if not self.is_suitable_market_state(signal, market_state):
return False

# 4. 成交量过滤
if not self.check_volume(klines):
return False

self.last_signal_time = current_time
return True

def identify_market_state(self, klines):
"""识别市场状态"""
# 简化实现
ma20 = klines.close.iloc[–20:].mean()
current_price = klines.close.iloc[–1]

if abs(current_price – ma20) / ma20 > 0.02:
return 'trending'
else:
return 'ranging'

def is_suitable_market_state(self, signal, market_state):
"""检查市场状态是否适合"""
if signal['type'] == 'trend' and market_state != 'trending':
return False
return True

def check_volume(self, klines):
"""检查成交量"""
avg_volume = klines.volume.iloc[–20:].mean()
current_volume = klines.volume.iloc[–1]
return current_volume >= avg_volume * 1.5

# 创建过滤器
signal_filter = SignalFilter()

# 策略逻辑
while True:
api.wait_update()
if api.is_changing(klines):
# 生成信号
ma5 = klines.close.iloc[–5:].mean()
ma20 = klines.close.iloc[–20:].mean()

signal = None
if ma5 > ma20 and position.pos_long == 0:
signal = {
'type': 'BUY',
'strength': 0.7, # 计算信号强度
'strategy': 'trend'
}
elif ma5 < ma20 and position.pos_long > 0:
signal = {
'type': 'SELL',
'strength': 0.7,
'strategy': 'trend'
}

# 过滤信号
if signal and signal_filter.filter_signal(signal, datetime.now()):
if signal['type'] == 'BUY':
api.insert_order("SHFE.rb2401", "BUY", "OPEN", 1)
else:
api.insert_order("SHFE.rb2401", "SELL", "CLOSE", position.pos_long)

TqSdk的优势：

• 可以方便地获取市场数据
• 支持自定义过滤逻辑
• 代码结构清晰

需要注意的：

• 需要Python基础
• 需要合理设计过滤规则

2. 对比其他工具

VnPy：

• 提供了信号管理模块
• 可以自定义过滤规则
• 配置相对复杂

文华财经WH8：

• 信号过滤功能有限
• 主要依赖内置指标

五、信号过滤最佳实践

1. 过滤规则设计

# 过滤规则设计原则
filtering_principles = {
"平衡性": "不能过度过滤，否则会错过机会",
"适应性": "根据市场环境动态调整",
"可测试性": "过滤规则应该可以回测验证",
"简洁性": "规则不要太复杂，避免过拟合"
}

2. 回测验证

# 回测验证过滤效果
def backtest_filtering_effect(strategy, filter_rules):
"""回测验证过滤效果"""
# 无过滤的回测
results_no_filter = backtest(strategy, use_filter=False)

# 有过滤的回测
results_with_filter = backtest(strategy, use_filter=True, filter_rules=filter_rules)

# 对比结果
comparison = {
"信号数量": {
"无过滤": results_no_filter['signal_count'],
"有过滤": results_with_filter['signal_count']
},
"胜率": {
"无过滤": results_no_filter['win_rate'],
"有过滤": results_with_filter['win_rate']
},
"收益": {
"无过滤": results_no_filter['return'],
"有过滤": results_with_filter['return']
}
}

return comparison

六、总结

信号过滤是提高策略胜率的重要手段。通过强度过滤、确认过滤、时间间隔过滤等方法，可以有效提高信号质量。

我目前主要使用TqSdk进行策略开发和信号过滤，它的灵活性让我能够方便地实现各种过滤规则。当然，这只是我个人的选择，每个人需求不同，建议多试用、多对比。

最后再次强调：量化交易有风险，信号过滤只是提高胜率的手段，不能保证盈利。本文仅从技术角度介绍相关方法，不构成任何投资建议。

声明：本文基于个人学习经验整理，仅供技术交流参考，不构成任何投资建议。文中提及的工具和方法请自行评估是否适合自己的需求。