你有没有想过，能不能写一段 Python 代码，让它自动帮你盯盘、判断趋势、发出买卖信号，甚至回测 25 年的历史数据来验证效果？今天我们要深入探讨的，就是一个完整的量化交易策略实战项目——CRL-SuperTrend Pullback Reversal Strategy。这个策略专门针对美股 Charles River Laboratories（CRL），利用 SuperTrend 和 OSMA 动量指标，构建了一套纯规则驱动的交易系统。没有主观判断，没有情绪干扰，一切由代码说了算。
对于正在学习 Python 的你来说，这个项目是一个绝佳的练手案例：它串联了数据获取、技术指标计算、信号生成、回测模拟等多个实战环节。
免责声明： 本文仅供学习交流，不构成任何投资建议。回测结果基于历史数据，不代表未来收益。
一、策略核心思路
这个策略的设计哲学可以用一句话概括：在下跌中入场，在动量恢复时离场。
听起来有点反直觉，对吧？大多数人习惯追涨杀跌，而这个策略偏偏要做“逆势交易者”——在市场处于下行趋势时买入，等到动量指标确认回暖时卖出。
它依赖两个技术指标：
SuperTrend（超级趋势指标）：负责判断当前市场处于上升趋势还是下降趋势。当 SuperTrend 判定为下降趋势时，策略发出入场信号。
OSMA（MACD 振荡器）：即 MACD 线与信号线的差值。当 OSMA 从负值穿越零线变为正值时，说明动量由弱转强，策略发出离场信号。
二、动手实现：从数据到信号
2.1 获取历史数据
第一步是用 yfinance 库下载 CRL 的日线数据，时间跨度从 2000 年到 2025 年底，覆盖了多个市场周期。
import pandas as pd
import numpy as np
import yfinance as yf
import vectorbt as vbt
# -------------------------
# 下载数据
# -------------------------
symbol = "CRL" # 股票代码
start_date = "2000-01-01" # 起始日期
end_date = "2026-01-01" # 截止日期
interval = "1d" # 日线级别
# 通过 yfinance 下载历史行情数据
df = yf.download(symbol, start=start_date, end=end_date, interval=interval)
# 保存为 CSV 文件，方便后续复用
df.to_csv("CRL_clean.csv", index=False)
这段代码会生成一个包含开盘价、最高价、最低价、收盘价和成交量的 DataFrame。后续所有的分析都基于这份数据。
2.2 计算 SuperTrend 指标
SuperTrend 是策略的“趋势裁判”，它通过 ATR（真实波幅均值）动态计算上下轨，然后判定当前处于上升还是下降趋势。
# SuperTrend 参数
SUPERTREND_MULTIPLIER = 3 # ATR 乘数
SUPERTREND_PERIOD = 14 # ATR 计算周期
def calculate_supertrend(df, period=SUPERTREND_PERIOD, multiplier=SUPERTREND_MULTIPLIER):
"""计算 SuperTrend 指标，包括上轨、下轨和趋势方向"""
hl2 = (df['High'] + df['Low']) / 2 # 最高价与最低价的中间值
# 计算真实波幅（True Range）
tr1 = df['High'] - df['Low']
tr2 = abs(df['High'] - df['Close'].shift())
tr3 = abs(df['Low'] - df['Close'].shift())
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
# 计算平均真实波幅（ATR）
atr = tr.rolling(period).mean()
# 计算基础上下轨
upper_basic = hl2 + multiplier * atr
lower_basic = hl2 - multiplier * atr
# 初始化最终上下轨
upper_band = upper_basic.copy()
lower_band = lower_basic.copy()
# 趋势列表：True 表示上升趋势，False 表示下降趋势
supertrend = [True]
for i in range(1, len(df)):
# 上轨调整：如果前一天收盘价低于上轨，则取较小值
if df['Close'].iloc[i-1] <= upper_band.iloc[i-1]:
upper_band.iloc[i] = min(upper_basic.iloc[i], upper_band.iloc[i-1])
# 下轨调整：如果前一天收盘价高于下轨，则取较大值
if df['Close'].iloc[i-1] >= lower_band.iloc[i-1]:
lower_band.iloc[i] = max(lower_basic.iloc[i], lower_band.iloc[i-1])
# 判断趋势方向
if df['Close'].iloc[i] > upper_band.iloc[i-1]:
supertrend.append(True) # 突破上轨，转为上升趋势
elif df['Close'].iloc[i] < lower_band.iloc[i-1]:
supertrend.append(False) # 跌破下轨，转为下降趋势
else:
supertrend.append(supertrend[-1]) # 维持前一天的趋势
df['SuperTrend'] = supertrend
df['Upper_Band'] = upper_band
df['Lower_Band'] = lower_band
return df
def st_downtrend(df):
"""返回 SuperTrend 下降趋势的布尔序列（True = 下降趋势）"""
df = calculate_supertrend(df)
return ~df['SuperTrend']
关键点： 这个策略的入场条件是 SuperTrend 判定为下降趋势，这是一种逆向思维——在别人恐惧时贪婪。
2.3 计算 OSMA 动量指标
OSMA 是 MACD 与其信号线之间的差值，用来衡量动量的加速或减速。
# OSMA 参数
OSMA_FAST = 12 # 快速 EMA 周期
OSMA_SIGNAL = 9 # 信号线 EMA 周期
OSMA_SLOW = 26 # 慢速 EMA 周期
def calculate_macd_osma(df, fast=OSMA_FAST, slow=OSMA_SLOW, signal_period=OSMA_SIGNAL):
"""
计算 MACD、信号线和 OSMA（MACD - 信号线）
"""
df = df.copy()
# 计算快速和慢速指数移动平均线
df['EMA_fast'] = df['Close'].ewm(span=fast, adjust=False).mean()
df['EMA_slow'] = df['Close'].ewm(span=slow, adjust=False).mean()
# MACD = 快速 EMA - 慢速 EMA
df['MACD'] = df['EMA_fast'] - df['EMA_slow']
# 信号线 = MACD 的 EMA
df['Signal'] = df['MACD'].ewm(span=signal_period, adjust=False).mean()
# OSMA = MACD - 信号线
df['OSMA'] = df['MACD'] - df['Signal']
return df
def osma_cross_above_0(df):
"""判断 OSMA 是否从负值上穿零线（动量由弱转强）"""
df = calculate_macd_osma(df)
return (df['OSMA'].shift(1) <= 0) & (df['OSMA'] > 0)
当 OSMA 从负值穿越到正值，意味着动量恢复，策略选择在此时获利了结。
三、生成交易信号并回测
3.1 入场与离场条件
信号逻辑非常简洁：
# -------------------------
# 入场条件：SuperTrend 处于下降趋势
# -------------------------
df["Supertrend_DownTrend"] = st_downtrend(df)
entry_conditions = [
'Supertrend_DownTrend',
]
df['entry_signal'] = df[entry_conditions].all(axis=1)
# -------------------------
# 离场条件：OSMA 上穿零线
# -------------------------
df["OSMA_Cross_Above_0"] = osma_cross_above_0(df)
exit_conditions = [
'OSMA_Cross_Above_0',
]
df['exit_signal'] = df[exit_conditions].all(axis=1)
3.2 回测执行
为了避免“未来函数”，所有信号都向后移动一根 K 线再执行。同时加入了手续费和滑点，模拟真实交易环境。
# -------------------------
# 回测设置
# -------------------------
# 信号延迟 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()
# 使用 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()
3.3 对比买入持有策略
为了评估策略是否真的有效，还需要和最简单的“买入持有”进行对比：
# -------------------------
# 买入持有基准回测
# -------------------------
df_holding = df['Open']
pf_holding = vbt.Portfolio.from_holding(df_holding, init_cash=100_000, freq='1d')
print(pf_holding.stats())
四、回测结果一览
根据原文给出的回测数据（2000 年 6 月 23 日至 2025 年 12 月 31 日）：
| 指标 |
数值 |
| 总收益率 |
1,255% |
| 最大回撤 |
74% |
| 总交易次数 |
57 次 |
| 胜率 |
82% |
| 盈亏比（Profit Factor） |
1.50 |
这个策略在总收益上跑赢了买入持有，胜率高达 82%，但最大回撤也达到了 74%，说明在某些极端行情下策略仍然会经历较大的浮亏。
五、从这个项目中你能学到什么
对 Python 学习者而言，这个项目涵盖了以下实用技能：
数据处理：使用 Pandas 进行时间序列数据的清洗、计算和转换。移动平均、指数平滑、布尔索引等操作贯穿始终。
技术指标实现：从零手写 SuperTrend 和 MACD/OSMA 的计算逻辑，比直接调库更能帮你理解指标的本质，这涉及到核心的算法思想。
回测框架使用：vectorbt 是一个高性能的向量化回测库，适合快速验证策略想法。
工程化思维：信号延迟执行、手续费和滑点模拟、与基准策略对比——这些细节体现了从“玩具代码”到“可信回测”的进阶。
六、策略的局限性
原文作者也坦诚指出，仅仅在历史数据上跑通 26 年的回测，并不足以证明策略在真实市场中能存活。还需要进行更深入的稳健性测试，比如：
Walk-Forward 分析：将数据分成多个训练/测试窗口，验证策略在“未见数据”上的表现。
多标的测试：目前策略只针对 CRL 一只股票，是否能迁移到其他标的需要验证。
风控增强：74% 的最大回撤在实盘中很难承受，加入止损、仓位管理等风控层是必要的改进方向。
总结
这个 CRL-SuperTrend 回调反转策略用极简的规则（1 个入场条件 + 1 个离场条件）实现了一个完整的量化交易系统。它的核心理念是“逆势入场、顺势离场”——在 SuperTrend 判定下跌时买入，在 OSMA 动量恢复时卖出。
对于 Python 学习者来说，这个项目的价值不仅在于量化交易本身，更在于它展示了如何用 Python 构建一个完整的“数据驱动决策系统”：获取数据、计算特征、生成信号、模拟执行、评估结果。这套思路可以迁移到很多智能与数据科学领域。
当然，请务必记住：回测不等于实盘，历史不代表未来。如果你对量化交易感兴趣，一定要在充分学习和测试之后再做决策。如果你希望找到更多类似的实战项目与开发者交流，不妨来云栈社区看看，那里有丰富的技术资源和活跃的讨论氛围。
发表于 昨天 06:31
|
查看: 3|
回复: 0
