当前位置：首页>python>用 Python 写一个回测收益 1492% 的混合交易策略

用 Python 写一个回测收益 1492% 的混合交易策略

2026-02-06 23:25:04

欢迎加入专注于财经数据与量化投研的【数据科学实战】知识星球！在这里，您将获取持续更新的《财经数据宝典》和《量化投研宝典》，这两部宝典相辅相成，为您在量化投研道路上提供明确指引。我们提供了精选的国内外量化投研的 250+ 篇高质量文章，并每日更新最新研究成果，涵盖策略开发、因子分析、风险管理等核心领域。无论您是量化投资新手还是经验丰富的研究者，星球社区都能帮您少走弯路，事半功倍，共同探索数据驱动的投资世界！

引言

在量化交易的世界里，很多人追求复杂的模型和高频的交易信号。但有时候，简单的策略反而能穿越牛熊，经受住时间的考验。

今天分享一个结合 Ichimoku 云图 和 布林带 的混合交易策略。这个策略在 2000 年至 2025 年的 25 年回测中，实现了 1492% 的总收益，胜率高达 **72%**。

这篇文章将带你了解这个策略的核心逻辑，并附上完整的 Python 代码实现。

策略核心思路

这个策略的设计理念非常清晰：

入场信号：使用 Ichimoku 云图的基准线（Kijun-sen）判断价格失去平衡的时刻
出场信号：使用布林带上轨判断波动率扩张的时刻

简单来说，当价格跌破基准线时入场做空，当价格从布林带下方突破到上方时出场。这是一个顺应市场结构的策略，而不是预测市场的策略。

数据准备

首先，我们需要获取历史价格数据。这里使用 yfinance 库下载 25 年的日线数据：

import pandas as pd
import numpy as np
import yfinance as yf
import vectorbt as vbt

# -------------------------
# 下载数据
# -------------------------
symbol = "TECH"# 股票代码
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("TECH_clean.csv", index=False)

Ichimoku 云图计算入场信号

Ichimoku 云图是一个经典的日本技术分析工具。我们主要使用其中的基准线（Kijun-sen）来判断入场时机。

当价格从基准线上方跌破到下方时，说明市场平衡被打破，这是一个看跌信号。

# Ichimoku 参数设置
ICHIMOKU_DISPLACEMENT = 26# 位移周期
ICHIMOKU_KIJUN_PERIOD = 26# 基准线周期
ICHIMOKU_SENKOU_B_PERIOD = 52# 先行带 B 周期
ICHIMOKU_TENKAN_PERIOD = 9# 转换线周期

defcalculate_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()

# 转换线：9 日内最高价和最低价的平均值
    df['tenkan_sen'] = (df['High'].rolling(tenkan_period).max() + 
                        df['Low'].rolling(tenkan_period).min()) / 2

# 基准线：26 日内最高价和最低价的平均值
    df['kijun_sen'] = (df['High'].rolling(kijun_period).max() + 
                       df['Low'].rolling(kijun_period).min()) / 2

# 先行带 A：转换线和基准线的平均值，向前位移 26 天
    df['senkou_span_a'] = ((df['tenkan_sen'] + df['kijun_sen']) / 2).shift(displacement)

# 先行带 B：52 日内最高价和最低价的平均值，向前位移 26 天
    df['senkou_span_b'] = ((df['High'].rolling(senkou_b_period).max() + 
                            df['Low'].rolling(senkou_b_period).min()) / 2).shift(displacement)

# 迟行带：收盘价向后位移 26 天
    df['chikou_span'] = df['Close'].shift(-displacement)

return df

defprice_cross_kijun_bearish(df):
"""
    判断价格是否向下穿越基准线（看跌信号）
    条件：前一天收盘价在基准线上方，当天收盘价跌破基准线
    """
    df = calculate_ichimoku(df)
return (df['Close'].shift(1) > df['kijun_sen'].shift(1)) & (df['Close'] < df['kijun_sen'])

布林带计算出场信号

布林带用于衡量价格波动率。当价格从布林带下方突破到上方时，往往意味着市场出现了恐慌性反弹或空头回补，这是平仓的好时机。

# 布林带参数设置
BB_PERIOD = 20# 均线周期
BB_SHIFT = 5# 回看周期
BB_STD = 2# 标准差倍数

defcalculate_bollinger_bands(df, period=BB_PERIOD, std_multiplier=BB_STD):
"""
    计算布林带：中轨、上轨、下轨
    """
    df = df.copy()

# 中轨：20 日简单移动平均线
    df['BB_MA'] = df['Close'].rolling(window=period).mean()

# 上轨：中轨 + 2 倍标准差
    df['BB_Upper'] = df['BB_MA'] + std_multiplier * df['Close'].rolling(window=period).std()

# 下轨：中轨 - 2 倍标准差
    df['BB_Lower'] = df['BB_MA'] - std_multiplier * df['Close'].rolling(window=period).std()

return df

defbb_open_above_upper_after_below(df, period=BB_PERIOD, std_multiplier=BB_STD, shift=BB_SHIFT):
"""
    判断开盘价是否突破布林带上轨
    条件：当天开盘价在上轨上方，且 5 天前开盘价在上轨下方
    """
    df = calculate_bollinger_bands(df, period, std_multiplier)
return (df['Open'] > df['BB_Upper']) & (df['Open'].shift(shift) < df['BB_Upper'].shift(shift))

完整回测代码

现在我们把入场和出场信号组合起来，使用 vectorbt 进行回测：

# -------------------------
# 生成入场信号
# -------------------------
df["Ichimoku_Price_Cross_Kijun_Bearish"] = price_cross_kijun_bearish(df)

entry_conditions = [
'Ichimoku_Price_Cross_Kijun_Bearish',  # 价格跌破基准线
]
df['entry_signal'] = df[entry_conditions].all(axis=1)

# -------------------------
# 生成出场信号
# -------------------------
df["BB_Open_Above_Upper_after_Below"] = bb_open_above_upper_after_below(df)

exit_conditions = [
'BB_Open_Above_Upper_after_Below',  # 开盘价突破布林带上轨
]
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()  # 绘制收益曲线