
2026年重磅升级已全面落地!欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球!您将获取持续更新的《财经数据宝典》与《量化投研宝典》,双典协同提供系统化指引;星球内含300篇以上独有高质量文章,深度覆盖策略开发、因子分析、风险管理等核心领域,内容基本每日更新;同步推出的「量化因子专题教程」系列(含完整可运行代码与实战案例),系统详解因子构建、回测与优化全流程,并实现日更迭代。我们持续扩充独家内容资源,全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者,这里都是助您少走弯路、事半功倍的理想伙伴,携手共探数据驱动的投资未来!
你是否想过,用 Python 写一个简单的量化交易策略,就能在 25 年的回测中实现 716% 的总收益?今天我们来拆解一篇技术文章,学习如何用 QQE 指标和随机指标(Stochastic)构建一个"安静的猎人"式交易系统。
这个策略的核心理念很简单:在市场疲软时入场,在市场过热时离场。不追涨杀跌,不预测未来,只是耐心等待市场自己暴露状态。
QQE(Quantitative Qualitative Estimation)是一个基于 RSI 的平滑动量指标。当 QQE Value1 低于 50 时,说明市场动量减弱,是潜在的入场时机。
随机指标(Stochastic Oscillator)用于判断价格在近期区间中的位置。当 Slow %D 高于 80 时,说明市场处于超买状态,是离场信号。
import pandas as pd
import numpy as np
import yfinance as yf
import vectorbt as vbt
# -------------------------
# 下载股票数据
# -------------------------
symbol = "COF" # Capital One Financial 股票代码
start_date = "2000-01-01" # 开始日期
end_date = "2026-01-01" # 结束日期
interval = "1d" # 日线数据
# 从 Yahoo Finance 下载数据
df = yf.download(symbol, start=start_date, end=end_date, interval=interval)
df.to_csv("COF_clean.csv", index=False) # 保存到本地# QQE 参数设置
QQE_FACTOR = 4.236 # QQE 因子
QQE_LEVEL = 50 # QQE 阈值
QQE_PERIOD = 14 # RSI 周期
QQE_SMOOTH = 5 # 平滑周期
def calculate_qqe(df, rsi_period=QQE_PERIOD, smooth=QQE_SMOOTH, factor=QQE_FACTOR):
"""
计算 QQE 指标的两条线(Value1 和 Value2)
基于 RSI 平滑和类 ATR 波动率模型
"""
df = df.copy()
# --- 计算 RSI ---
delta = df['Close'].diff() # 价格变化
up = delta.clip(lower=0) # 上涨部分
down = -delta.clip(upper=0) # 下跌部分
roll_up = up.ewm(alpha=1/rsi_period, adjust=False).mean() # 上涨均值
roll_down = down.ewm(alpha=1/rsi_period, adjust=False).mean() # 下跌均值
rsi = 100 - (100 / (1 + roll_up / roll_down)) # RSI 公式
# --- 平滑 RSI ---
rsi_ma = rsi.rolling(window=smooth).mean().fillna(method='bfill')
# --- RSI 变化率和类 ATR 平滑 ---
rsi_delta = rsi_ma.diff().abs().fillna(0)
atr_rsi = rsi_delta.ewm(alpha=1/smooth, adjust=False).mean()
# --- QQE 计算 ---
value1 = rsi_ma.copy() # QQE Value1 就是平滑后的 RSI
value2 = pd.Series(index=df.index, dtype=float) # QQE Value2 是动态阈值
value2.iloc[0] = value1.iloc[0]
for i in range(1, len(df)):
prev_trail = value2.iloc[i-1]
prev_value1 = value1.iloc[i-1]
atr = atr_rsi.iloc[i]
direction = 1 if prev_value1 > prev_trail else -1
value2.iloc[i] = prev_trail + direction * factor * atr
df['QQE_Value1'] = value1
df['QQE_Value2'] = value2
return df
def qqe_value1_lower_than_level(df, level=QQE_LEVEL):
"""判断 QQE Value1 是否低于阈值(入场条件)"""
df = calculate_qqe(df)
return df['QQE_Value1'] < level# 随机指标参数
STOCHASTIC_D_PERIOD = 3 # %D 周期
STOCHASTIC_K_PERIOD = 14 # %K 周期
STOCHASTIC_OVERBOUGHT = 80 # 超买阈值
def calculate_stochastic(df, k_period=STOCHASTIC_K_PERIOD, d_period=STOCHASTIC_D_PERIOD):
"""计算 Fast %K 和 Slow %D"""
low_min = df['Low'].rolling(window=k_period).min() # 周期内最低价
high_max = df['High'].rolling(window=k_period).max() # 周期内最高价
# Fast %K:当前价格在区间中的位置
df['Fast_%K'] = 100 * (df['Close'] - low_min) / (high_max - low_min)
# Slow %D:%K 的移动平均
df['Slow_%D'] = df['Fast_%K'].rolling(window=d_period).mean()
return df
def slow_d_higher_than_overbought(df, level=STOCHASTIC_OVERBOUGHT):
"""判断 Slow %D 是否高于超买阈值(出场条件)"""
df = calculate_stochastic(df)
return df['Slow_%D'] > level# -------------------------
# 入场条件
# -------------------------
df["QQE_Value1_LowerThan_Level"] = qqe_value1_lower_than_level(df)
entry_conditions = [
'QQE_Value1_LowerThan_Level', # QQE Value1 < 50
]
df['entry_signal'] = df[entry_conditions].all(axis=1)
# -------------------------
# 出场条件
# -------------------------
df["Stochastic_Slow_%D_Higher_Than_Overbought"] = slow_d_higher_than_overbought(df)
exit_conditions = [
'Stochastic_Slow_%D_Higher_Than_Overbought', # Slow %D > 80
]
df['exit_signal'] = df[exit_conditions].all(axis=1)
# -------------------------
# 执行回测
# -------------------------
# 信号延迟一天执行,避免未来数据偏差
shift_entries = df['entry_signal'].shift(1).astype(bool).fillna(False).to_numpy()
shift_exits = df['exit_signal'].shift(1).astype(bool).fillna(False).to_numpy()
pf = vbt.Portfolio.from_signals(
close=df['Open'], # 使用开盘价成交
entries=shift_entries,
exits=shift_exits,
init_cash=100_000, # 初始资金 10 万
fees=0.001, # 手续费 0.1%
slippage=0.002, # 滑点 0.2%
freq='1d'
)
# -------------------------
# 输出统计结果
# -------------------------
print(pf.stats())
pf.plot().show()经过 25 年(2000-2025)的回测,该策略表现如下:
对比买入持有策略(603% 收益,90% 最大回撤),该策略在收益略高的同时,显著降低了最大回撤。
这个策略代表了一种"状态感知型动量策略":
这不是一个试图预测市场的策略,而是一个等待市场自己暴露状态的策略。
通过这篇文章,我们学习了:
当然,历史回测不代表未来收益。这个策略在实盘中还需要考虑更多因素,比如交易成本、流动性、市场环境变化等。但作为学习量化交易的入门案例,它展示了如何用 Python 将一个交易思想转化为可执行的代码。
正如文章作者所说:"安静的猎人不会追逐,它只是等待。"
2026年全面升级已落地!【数据科学实战】知识星球核心权益如下:
星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者,还是深耕领域的从业者,这里都是助您少走弯路、高效成长的理想平台。诚邀加入,共探数据驱动的投资未来!
好文推荐
1. 用 Python 打造股票预测系统:Transformer 模型教程(一)
2. 用 Python 打造股票预测系统:Transformer 模型教程(二)
3. 用 Python 打造股票预测系统:Transformer 模型教程(三)
4. 用 Python 打造股票预测系统:Transformer 模型教程(完结)
6. YOLO 也能预测股市涨跌?计算机视觉在股票市场预测中的应用
9. Python 量化投资利器:Ridge、Lasso 和 Elastic Net 回归详解
好书推荐