
用 Python 揭秘均值回归策略:你的收益从何而来?
2026年重磅升级已全面落地!欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球!您将获取持续更新的《财经数据宝典》与《量化投研宝典》,双典协同提供系统化指引;星球内含 350 篇以上独有高质量文章,深度覆盖策略开发、因子分析、风险管理等核心领域,内容基本每日更新;同步推出的「量化因子专题教程」系列(含完整可运行代码与实战案例),系统详解因子构建、回测与优化全流程,并实现日更迭代。我们持续扩充独家内容资源,全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者,这里都是助您少走弯路、事半功倍的理想伙伴,携手共探数据驱动的投资未来!
在量化交易的世界里,大多数人追逐的是 MACD、RSI、布林带这些耳熟能详的指标。但有时候,真正有效的信号恰恰来自那些"冷门"的技术指标。
今天要介绍的这套策略,只用了两个核心思路:当卖压减弱时入场,当趋势变强时离场。听起来简单得不可思议,但在 Caterpillar(CAT)这只股票上回测了 25 年,总收益高达 6521%,远超同期买入持有的 4581%。
这套策略的核心武器是两个相对冷门的技术指标——DeMarker 指标和 ADX 指标。接下来,我们用 Python 一步步实现它。
DeMarker 指标是由 Thomas DeMark 提出的震荡指标,它关注的不是价格动量,而是价格极值的变化。简单来说,它衡量的是当前最高价与前一日最高价、当前最低价与前一日最低价之间的关系。
DeMarker 的取值范围在 0 到 1 之间。当它处于低位时,说明卖压较重;当它从低位回升时,意味着卖方力量正在衰竭。
ADX(Average Directional Index,平均趋向指数)由 Welles Wilder 提出,专门用来衡量趋势的强度,而不关心趋势的方向。ADX 值越高,说明当前趋势越强。通常以 20 作为分界线:低于 20 表示市场震荡,高于 20 表示趋势正在形成。
这套策略的哲学非常简洁:
入场条件:DeMarker 指标在 5 日前的值小于 10 日前的值。这意味着卖压虽然还在,但已经在减弱——这往往发生在价格明显反转之前。
出场条件:ADX 从低于 20 上穿到 20 以上。这说明一个强趋势正在形成,此时策略选择离场,避免在波动放大时被反噬。
import pandas as pd
import numpy as np
import yfinance as yf
import vectorbt as vbt
# -------------------------
# 下载数据
# -------------------------
symbol = "CAT"
start_date = "2000-01-01"
end_date = "2026-01-01"
interval = "1d"
# 使用 yfinance 下载 CAT 的日线数据
df = yf.download(symbol, start=start_date, end=end_date, interval=interval)
# 保存为 CSV 文件,方便后续使用
df.to_csv("CAT_clean.csv", index=False)
print(df.head())# DeMarker 指标参数
DEM_PERIOD = 14 # 计算周期
DEM_SHIFT_1 = 5 # 近期回看天数
DEM_SHIFT_2 = 10 # 远期回看天数
def calculate_demarker(df, period=DEM_PERIOD):
"""
计算 DeMarker 指标。
DeMax:当今日最高价高于昨日最高价时,取差值;否则为 0。
DeMin:当今日最低价低于昨日最低价时,取差值;否则为 0。
DeMarker = DeMax 之和 / (DeMax 之和 + DeMin 之和)
"""
df = df.copy()
# 计算 DeMax 和 DeMin
df['DeMax'] = np.where(
df['High'] > df['High'].shift(1),
df['High'] - df['High'].shift(1),
0
)
df['DeMin'] = np.where(
df['Low'] < df['Low'].shift(1),
df['Low'].shift(1) - df['Low'],
0
)
# 计算滚动窗口内的求和
dem_max = df['DeMax'].rolling(window=period).sum()
dem_min = df['DeMin'].rolling(window=period).sum()
# 计算 DeMarker 值
df['DeMarker'] = dem_max / (dem_max + dem_min)
return df策略的入场逻辑是判断 DeMarker 是否在"减速下跌"——即 5 天前的 DeMarker 值小于 10 天前的值。这表示卖方力量虽然仍占上风,但衰减速度在放缓,市场可能即将反转。
def demarker_is_falling(df, shift_1=DEM_SHIFT_1, shift_2=DEM_SHIFT_2):
"""
判断 DeMarker 是否在缓慢下降。
比较 5 日前与 10 日前的 DeMarker 值,
如果 5 日前的值更低,说明下跌趋势在延续但力度减弱。
"""
df = calculate_demarker(df)
return df['DeMarker'].shift(shift_1) < df['DeMarker'].shift(shift_2)
# 生成入场信号
df["Demarker_Is_Falling"] = demarker_is_falling(df)
# 定义入场条件列表(可扩展)
entry_conditions = [
'Demarker_Is_Falling',
]
# 当所有入场条件同时满足时,发出入场信号
df['entry_signal'] = df[entry_conditions].all(axis=1)ADX_LEVEL = 20 # ADX 的趋势判断阈值
ADX_PERIOD = 14 # ADX 计算周期
def calculate_adx(df, period=14):
"""
计算 ADX(平均趋向指数)及相关列。
包括真实波幅(TR)、方向运动(+DM/-DM)、
方向指标(+DI/-DI)、以及最终的 ADX。
"""
# 真实波幅(True Range)
df['H-L'] = df['High'] - df['Low']
df['H-PC'] = abs(df['High'] - df['Close'].shift(1))
df['L-PC'] = abs(df['Low'] - df['Close'].shift(1))
df['TR'] = df[['H-L', 'H-PC', 'L-PC']].max(axis=1)
# 方向运动(Directional Movement)
df['+DM'] = np.where(
(df['High'] - df['High'].shift(1)) > (df['Low'].shift(1) - df['Low']),
np.maximum(df['High'] - df['High'].shift(1), 0),
0
)
df['-DM'] = np.where(
(df['Low'].shift(1) - df['Low']) > (df['High'] - df['High'].shift(1)),
np.maximum(df['Low'].shift(1) - df['Low'], 0),
0
)
# 滚动窗口平滑
df['TRn'] = df['TR'].rolling(window=period).sum()
df['+DMn'] = df['+DM'].rolling(window=period).sum()
df['-DMn'] = df['-DM'].rolling(window=period).sum()
# 计算 +DI 和 -DI
df['+DI'] = 100 * (df['+DMn'] / df['TRn'])
df['-DI'] = 100 * (df['-DMn'] / df['TRn'])
# 计算 DX 和 ADX
df['DX'] = 100 * abs(df['+DI'] - df['-DI']) / (df['+DI'] + df['-DI'])
df['ADX'] = df['DX'].rolling(window=period).mean()
return dfdef adx_cross_above_level(df, period=ADX_PERIOD, level=ADX_LEVEL):
"""
判断 ADX 是否从下方上穿指定阈值(默认 20)。
当 ADX 上穿 20 时,说明趋势正在形成,策略选择离场。
"""
df = calculate_adx(df, period)
return (df['ADX'].shift(1) <= level) & (df['ADX'] > level)
# 生成出场信号
df["ADX_XAbove20"] = adx_cross_above_level(df)
# 定义出场条件列表
exit_conditions = [
'ADX_XAbove20',
]
# 当所有出场条件同时满足时,发出出场信号
df['exit_signal'] = df[exit_conditions].all(axis=1)# -------------------------
# 回测
# -------------------------
# 将信号向后移一位,避免未来数据偷看(look-ahead bias)
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()
# 使用 vectorbt 构建投资组合
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()# 买入持有基准的表现
df_holding = df['Open']
pf_holding = vbt.Portfolio.from_holding(
df_holding,
init_cash=100_000,
fees=0.001,
slippage=0.002
)
print(pf_holding.stats())以下是该策略从 2000 年到 2025 年底的关键回测指标:
可以看到,策略在总收益、Sharpe 比率和 Sortino 比率上均优于简单的买入持有,同时最大回撤也略低。
这套策略并不追求抓住爆发性行情,而是专注于早期定位和均值回归的那段平静期。它的边际优势来自于行为层面的时机把握:在恐惧尚未消退但已在减弱时入场,在波动率扩大之前离场。
整个系统只有 2 个条件、2 个指标,没有复杂的参数优化,也不需要高频交易的基础设施。对于 Python 量化入门者来说,这是一个非常好的学习案例。
需要特别提醒的是,回测结果仅基于历史数据,不代表未来收益。在实际应用之前,还需要进行更多的稳健性测试,例如:走步前进优化(Walk-Forward Optimization)、跨品种验证、参数敏感性分析等。任何交易策略都伴随风险,请务必做好自己的研究。
这篇文章介绍了一套基于 DeMarker 和 ADX 指标的量化交易策略,并用 Python 完整实现了从数据下载到回测输出的全部流程。策略的核心理念是"不预测价格方向,只观察市场状态":当卖压衰竭时入场,当趋势爆发时离场。
通过这个案例,我们可以学到以下几点:
与其问"价格会去哪里",不如问"压力在减弱吗?趋势在形成吗?"有时候,这就够了。
2026年全面升级已落地!【数据科学实战】知识星球核心权益如下:
星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者,还是深耕领域的从业者,这里都是助您少走弯路、高效成长的理想平台。诚邀加入,共探数据驱动的投资未来!
好文推荐
1. 用 Python 打造股票预测系统:Transformer 模型教程(一)
2. 用 Python 打造股票预测系统:Transformer 模型教程(二)
3. 用 Python 打造股票预测系统:Transformer 模型教程(三)
4. 用 Python 打造股票预测系统:Transformer 模型教程(完结)
6. YOLO 也能预测股市涨跌?计算机视觉在股票市场预测中的应用
9. Python 量化投资利器:Ridge、Lasso 和 Elastic Net 回归详解
好书推荐