
2026年重磅升级已全面落地!欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球!您将获取持续更新的《财经数据宝典》与《量化投研宝典》,双典协同提供系统化指引;星球内含300篇以上独有高质量文章,深度覆盖策略开发、因子分析、风险管理等核心领域,内容基本每日更新;同步推出的「量化因子专题教程」系列(含完整可运行代码与实战案例),系统详解因子构建、回测与优化全流程,并实现日更迭代。我们持续扩充独家内容资源,全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者,这里都是助您少走弯路、事半功倍的理想伙伴,携手共探数据驱动的投资未来!
你是否想过,一个简单的交易策略,在 25 年间能把 10 万美元变成 200 万美元?今天我们来拆解一个基于 SuperTrend 和 Ichimoku Cloud(一目均衡表)的量化交易策略。这个策略不预测市场,不追涨杀跌,只遵循两条简单的规则:什么时候进场,什么时候离场。
这篇文章将带你了解这个策略的核心逻辑,并附上完整的 Python 代码实现。
这个策略的哲学很简单:当市场发出清晰信号时行动,当市场发出危险信号时离场。
它结合了两个经典的技术指标:
当价格收盘价跌破 SuperTrend 的下轨时,触发入场信号。这意味着市场动能出现了结构性的突破,而非随机波动。
当 Ichimoku 的 Senkou Span A 下穿 Senkou Span B 时,触发离场信号。这通常标志着趋势耗尽或结构性弱势。
import pandas as pd
import numpy as np
import yfinance as yf
import vectorbt as vbt
# -------------------------
# 下载股票数据
# -------------------------
symbol = "CAH" # 股票代码:Cardinal Health
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)
df.to_csv("CAH_clean.csv", index=False) # 保存为 CSV 文件# 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()
# SuperTrend 方向列表(True 表示上升趋势)
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])
# 判断 SuperTrend 方向
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 close_below_st(df):
"""判断收盘价是否低于 SuperTrend 下轨"""
df = calculate_supertrend(df)
return df['Close'] < df['Lower_Band']# Ichimoku 参数设置
ICHIMOKU_DISPLACEMENT = 26 # 位移周期
ICHIMOKU_KIJUN_PERIOD = 26 # 基准线周期
ICHIMOKU_SENKOU_B_PERIOD = 52 # 先行带 B 周期
ICHIMOKU_TENKAN_PERIOD = 9 # 转换线周期
def calculate_ichimoku(df, tenkan_period=ICHIMOKU_TENKAN_PERIOD,
kijun_period=ICHIMOKU_KIJUN_PERIOD,
senkou_b_period=ICHIMOKU_SENKOU_B_PERIOD,
displacement=ICHIMOKU_DISPLACEMENT):
"""计算 Ichimoku 云图各条线"""
df = df.copy()
# 转换线(Tenkan-sen):9 日内最高价与最低价的平均值
df['tenkan_sen'] = (df['High'].rolling(tenkan_period).max() +
df['Low'].rolling(tenkan_period).min()) / 2
# 基准线(Kijun-sen):26 日内最高价与最低价的平均值
df['kijun_sen'] = (df['High'].rolling(kijun_period).max() +
df['Low'].rolling(kijun_period).min()) / 2
# 先行带 A(Senkou Span A):转换线与基准线的平均值,向前位移 26 天
df['senkou_span_a'] = ((df['tenkan_sen'] + df['kijun_sen']) / 2).shift(displacement)
# 先行带 B(Senkou Span B):52 日内最高价与最低价的平均值,向前位移 26 天
df['senkou_span_b'] = ((df['High'].rolling(senkou_b_period).max() +
df['Low'].rolling(senkou_b_period).min()) / 2).shift(displacement)
# 延迟线(Chikou Span):收盘价向后位移 26 天
df['chikou_span'] = df['Close'].shift(-displacement)
return df
def senkou_span_cross_bearish(df):
"""判断 Senkou Span A 是否下穿 Senkou Span B(看跌信号)"""
df = calculate_ichimoku(df)
return (df['senkou_span_a'].shift(1) > df['senkou_span_b'].shift(1)) & \
(df['senkou_span_a'] <= df['senkou_span_b'])# -------------------------
# 生成入场信号
# -------------------------
df["Supertrend_Close_Below"] = close_below_st(df)
entry_conditions = [
'Supertrend_Close_Below', # 收盘价低于 SuperTrend 下轨
]
df['entry_signal'] = df[entry_conditions].all(axis=1)
# -------------------------
# 生成离场信号
# -------------------------
df["Ichimoku_Senkou_Span_Cross_Bearish"] = senkou_span_cross_bearish(df)
exit_conditions = [
'Ichimoku_Senkou_Span_Cross_Bearish', # Senkou Span A 下穿 B
]
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()这个策略在 2000 年至 2025 年间的表现:
这不是一个高频策略,平均每年只交易 2 次左右。它经历了互联网泡沫破裂、2008 年金融危机、新冠疫情暴跌等多次市场动荡,依然保持了稳健的表现。
58% 的最大回撤是长期持仓的代价,但超过 5 的盈亏比说明:亏损被控制住了,盈利的交易被充分放大。
这个策略的核心理念是:在市场给出明确信号时行动,在市场发出警告时离场。它不追求预测市场顶部或底部,而是通过两个互补的技术指标构建了一个完整的交易系统。
对于学习 Python 量化交易的同学来说,这个策略是一个很好的入门案例。它展示了如何使用 pandas 计算技术指标,如何使用 vectorbt 进行回测,以及如何构建一个完整的交易系统。
当然,任何交易策略都有风险,历史回测不代表未来表现。如果你想使用这个策略,请务必进行充分的研究和测试。
2026年全面升级已落地!【数据科学实战】知识星球核心权益如下:
星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者,还是深耕领域的从业者,这里都是助您少走弯路、高效成长的理想平台。诚邀加入,共探数据驱动的投资未来!
好文推荐
1. 用 Python 打造股票预测系统:Transformer 模型教程(一)
2. 用 Python 打造股票预测系统:Transformer 模型教程(二)
3. 用 Python 打造股票预测系统:Transformer 模型教程(三)
4. 用 Python 打造股票预测系统:Transformer 模型教程(完结)
6. YOLO 也能预测股市涨跌?计算机视觉在股票市场预测中的应用
9. Python 量化投资利器:Ridge、Lasso 和 Elastic Net 回归详解
好书推荐