箱体理论及量化实战 

一 基本概念 
箱体理论（Box Theory）由美国分析师 尼古拉斯・达瓦斯 提出，
股价在一段时间内会在一个明确的价格区间（箱体）内波动，上方压力位为箱顶，下方支撑位为箱底。
当股价突破箱体边界且有效站稳后，会进入新的箱体运行，原支撑 / 压力位转换为新的压力 / 支撑位。
股价像在一个 “箱子” 里上下震荡，突破箱子顶部就会进入更高的箱子，跌破底部就会掉入更低的箱子。

二 核心原理
市场情绪周期驱动
股价在箱底时，买方认为价格低估，买入力量增强，推动股价回升；
股价到箱顶时，卖方认为价格高估，卖出力量增强，压制股价回落；
突破箱体则意味着多空力量失衡，趋势形成。

支撑与压力的相互转换
有效突破箱顶 → 原箱顶变为新箱体的支撑位；
有效跌破箱底 → 原箱底变为新箱体的压力位。

三、量化策略结合点
箱体自动识别：
通过编程计算股价的 历史高低点，自动标记箱顶、箱底和箱体区间
Python 可使用 TA-Lib 库计算高低点；

突破信号量化：
设定量化指标
如收盘价突破箱顶 + 成交量 > 5 日均量 1.5 倍 + 站稳 3 天，自动触发交易信号；
回测验证：
用 ETF 历史数据回测箱体策略的胜率、盈亏比，优化箱体参数（如震荡时间、突破确认天数）。


四、风险提示
假突破风险：
需严格执行 “收盘价站稳 + 放量” 的确认条件，避免追高假突破；
趋势反转风险：
当大盘出现系统性风险时，箱体支撑 / 压力位可能失效，需及时止损；

结合其他指标：
单独使用箱体理论胜率有限，建议搭配 MACD、RSI 等指标，提高信号可靠性。




python实现 
ETF 箱体策略的量化回测 Python 代码框架，
涵盖箱体识别、突破信号生成、回测分析核心模块


一、环境依赖

pip install tushare pandas numpy matplotlib backtrader  # 核心库
tushare：获取金融数据（需注册获取 token）；
backtrader：回测框架；
pandas/numpy：数据处理；
matplotlib：可视化。

二、完整代码框架

import tushare as ts
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import backtrader as bt
from datetime import datetime

# ===================== 1. 全局配置 =====================
# Tushare token（替换为自己的）
TS_TOKEN = "你的tushare_token"
ts.set_token(TS_TOKEN)
pro = ts.pro_api()

# 策略参数
BOX_TOP_BARS = 20  # 识别箱顶的周期（连续20天内的高点）
BOX_BOTTOM_BARS = 20  # 识别箱底的周期（连续20天内的低点）
BREAK_CONFIRM_DAYS = 3  # 突破确认天数（站稳N天）
VOL_MULTIPLE = 1.5  # 突破时成交量需大于5日均量的倍数
STOP_LOSS_RATIO = 0.03  # 止损比例（3%）
TAKE_PROFIT_RATIO = None  # 止盈比例（默认按新箱体高度）

# ===================== 2. 数据获取函数 =====================
def get_etf_data(ts_code, start_date, end_date):
    """
    获取ETF日线数据（Tushare接口）
    :param ts_code: ETF代码（如510050.SH）
    :param start_date: 开始日期（格式：20200101）
    :param end_date: 结束日期（格式：20250101）
    :return: 标准化后的DataFrame
    """
    df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
    # 数据格式转换
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    df = df.sort_values('trade_date').reset_index(drop=True)
    # 重命名列（适配backtrader）
    df.rename(columns={
        'trade_date': 'datetime',
        'open': 'open',
        'high': 'high',
        'low': 'low',
        'close': 'close',
        'vol': 'volume'
    }, inplace=True)
    # 补充5日均量
    df['vol_5ma'] = df['volume'].rolling(window=5).mean()
    return df

# ===================== 3. 箱体识别函数 =====================
def identify_box(df):
    """
    识别股价的箱体区间（箱顶/箱底）
    :param df: 包含ohlcv的DataFrame
    :return: 新增箱顶(box_top)、箱底(box_bottom)列的DataFrame
    """
    # 计算N周期内的最高收盘价（箱顶）、最低收盘价（箱底）
    df['box_top'] = df['close'].rolling(window=BOX_TOP_BARS).max()
    df['box_bottom'] = df['close'].rolling(window=BOX_BOTTOM_BARS).min()
    
    # 填充空值（前N天无箱体数据）
    df['box_top'].fillna(method='bfill', inplace=True)
    df['box_bottom'].fillna(method='bfill', inplace=True)
    
    # 计算箱体高度（用于止盈）
    df['box_height'] = df['box_top'] - df['box_bottom']
    return df

# ===================== 4. 突破信号生成 =====================
def generate_break_signal(df):
    """
    生成箱体突破/跌破信号
    买入信号：放量突破箱顶 + 连续N天站稳
    卖出信号：放量跌破箱底 + 连续N天站不稳 / 止损/止盈触发
    """
    # 初始化信号列（0:无信号, 1:买入, -1:卖出）
    df['signal'] = 0
    df['break_confirm'] = 0  # 突破确认标记
    
    # 1. 突破箱顶的初步判断（收盘价>箱顶 + 成交量>5日均量*VOL_MULTIPLE）
    df['top_break_pre'] = (df['close'] > df['box_top']) & (df['volume'] > df['vol_5ma'] * VOL_MULTIPLE)
    # 2. 跌破箱底的初步判断（收盘价<箱底 + 成交量>5日均量*VOL_MULTIPLE）
    df['bottom_break_pre'] = (df['close'] < df['box_bottom']) & (df['volume'] > df['vol_5ma'] * VOL_MULTIPLE)
    
    # 3. 确认突破（连续BREAK_CONFIRM_DAYS天站稳/站不稳）
    for i in range(BREAK_CONFIRM_DAYS, len(df)):
        # 买入确认：连续N天收盘价>箱顶
        if all(df['close'].iloc[i-BREAK_CONFIRM_DAYS:i] > df['box_top'].iloc[i-BREAK_CONFIRM_DAYS:i]):
            df.loc[df.index[i], 'break_confirm'] = 1
        # 卖出确认：连续N天收盘价<箱底
        elif all(df['close'].iloc[i-BREAK_CONFIRM_DAYS:i] < df['box_bottom'].iloc[i-BREAK_CONFIRM_DAYS:i]):
            df.loc[df.index[i], 'break_confirm'] = -1
    
    # 4. 生成最终交易信号
    df.loc[df['break_confirm'] == 1, 'signal'] = 1  # 买入
    df.loc[df['break_confirm'] == -1, 'signal'] = -1  # 卖出
    
    # 5. 止损/止盈信号（持仓时触发）
    hold_flag = False  # 持仓标记
    entry_price = 0  # 入场价
    for i in range(len(df)):
        if df['signal'].iloc[i] == 1 and not hold_flag:
            hold_flag = True
            entry_price = df['close'].iloc[i]
        elif hold_flag:
            # 止损：收盘价 < 入场价*(1-STOP_LOSS_RATIO)
            if df['close'].iloc[i] < entry_price * (1 - STOP_LOSS_RATIO):
                df.loc[df.index[i], 'signal'] = -1
                hold_flag = False
            # 止盈：收盘价 >= 入场价 + 原箱体高度
            elif TAKE_PROFIT_RATIO is None and df['close'].iloc[i] >= entry_price + df['box_height'].iloc[i]:
                df.loc[df.index[i], 'signal'] = -1
                hold_flag = False
            # 自定义止盈比例
            elif TAKE_PROFIT_RATIO and df['close'].iloc[i] >= entry_price * (1 + TAKE_PROFIT_RATIO):
                df.loc[df.index[i], 'signal'] = -1
                hold_flag = False
        elif df['signal'].iloc[i] == -1:
            hold_flag = False
    
    return df

# ===================== 5. Backtrader回测策略 =====================
class BoxStrategy(bt.Strategy):
    """箱体策略回测类"""
    params = (
        ('stop_loss', STOP_LOSS_RATIO),
        ('take_profit', TAKE_PROFIT_RATIO),
    )
    
    def __init__(self):
        self.dataclose = self.datas[0].close
        self.dataopen = self.datas[0].open
        self.signal = None  # 信号缓存
        self.order = None  # 订单缓存
        self.entry_price = 0  # 入场价
    
    def next(self):
        # 避免重复下单
        if self.order:
            return
        
        # 获取当前信号（从外部数据传入，需提前处理）
        current_date = self.datas[0].datetime.date(0)
        signal = self.df.loc[self.df['datetime'] == pd.to_datetime(current_date), 'signal'].values[0]
        
        # 买入信号：无持仓时触发
        if signal == 1 and not self.position:
            self.order = self.buy(size=10000)  # 买入10000份（可调整）
            self.entry_price = self.dataclose[0]
            print(f"买入：{current_date}，价格：{self.dataclose[0]}")
        
        # 卖出信号：持仓时触发（突破跌破/止损/止盈）
        elif signal == -1 and self.position:
            self.order = self.sell(size=self.position.size)
            print(f"卖出：{current_date}，价格：{self.dataclose[0]}，盈亏：{(self.dataclose[0]-self.entry_price)*self.position.size}")
            self.entry_price = 0

# ===================== 6. 回测执行函数 =====================
def run_backtest(df):
    """
    执行回测并输出结果
    """
    # 初始化回测引擎
    cerebro = bt.Cerebro()
    # 添加策略
    cerebro.addstrategy(BoxStrategy)
    
    # 转换数据格式为backtrader兼容格式
    df_bt = df[['datetime', 'open', 'high', 'low', 'close', 'volume']].copy()
    df_bt.set_index('datetime', inplace=True)
    data = bt.feeds.PandasData(dataname=df_bt)
    cerebro.adddata(data)
    
    # 设置初始资金
    cerebro.broker.setcash(100000.0)
    # 设置手续费（ETF手续费极低，按0.0003计算）
    cerebro.broker.setcommission(commission=0.0003)
    
    # 运行回测
    print('初始资金：%.2f' % cerebro.broker.getvalue())
    cerebro.run()
    print('最终资金：%.2f' % cerebro.broker.getvalue())
    
    # 绘制回测曲线
    cerebro.plot(style='candlestick')

# ===================== 7. 主函数（流程执行） =====================
if __name__ == "__main__":
    # 步骤1：获取ETF数据（以沪深300ETF 510300.SH为例）
    etf_data = get_etf_data(
        ts_code="510300.SH",
        start_date="20200101",
        end_date="20250101"
    )
    
    # 步骤2：识别箱体
    etf_data = identify_box(etf_data)
    
    # 步骤3：生成突破信号
    etf_data = generate_break_signal(etf_data)
    
    # 步骤4：箱体可视化（可选）
    plt.figure(figsize=(12, 6))
    plt.plot(etf_data['datetime'], etf_data['close'], label='收盘价', color='blue')
    plt.plot(etf_data['datetime'], etf_data['box_top'], label='箱顶', color='red', linestyle='--')
    plt.plot(etf_data['datetime'], etf_data['box_bottom'], label='箱底', color='green', linestyle='--')
    # 标记买入/卖出信号
    buy_signals = etf_data[etf_data['signal'] == 1]
    sell_signals = etf_data[etf_data['signal'] == -1]
    plt.scatter(buy_signals['datetime'], buy_signals['close'], marker='^', color='red', s=100, label='买入')
    plt.scatter(sell_signals['datetime'], sell_signals['close'], marker='v', color='green', s=100, label='卖出')
    plt.title('沪深300ETF箱体策略信号')
    plt.xlabel('日期')
    plt.ylabel('价格')
    plt.legend()
    plt.show()
    
    # 步骤5：执行回测
    run_backtest(etf_data)



三、代码使用说明
1. 基础配置
替换 TS_TOKEN 为自己的 Tushare token
注册地址：https://tushare.pro/register
调整策略参数（如 BOX_TOP_BARS 箱体周期、BREAK_CONFIRM_DAYS 突破确认天数）适配不同 ETF


2. 核心模块解释
模块	功能
get_etf_data	
获取 ETF 日线数据，标准化格式并计算 5 日均量

identify_box	
滚动计算 N 周期内的高低点，识别箱顶 / 箱底，计算箱体高度

generate_break_signal	
结合成交量和站稳天数，生成买入 / 卖出信号，包含止损 / 止盈逻辑

BoxStrategy	
Backtrader 策略类，执行下单逻辑，输出交易日志

run_backtest	
初始化回测引擎，计算收益，绘制回测曲线



3. 输出结果
可视化图表：包含收盘价、箱顶 / 箱底、买入 / 卖出信号标记；
回测日志：输出每笔交易的买卖日期、价格、盈亏；
核心指标：初始资金、最终资金、收益率（最终资金 / 初始资金 - 1）



四、优化方向
多指标融合：
加入 MACD/RSI 指标过滤假突破（如突破时 RSI<70 避免超买）；

参数优化：
用网格搜索遍历不同箱体周期 / 确认天数，找到最优参数组合；

多标的回测：
循环遍历多只 ETF（如 510050、159915），输出策略在不同标的上的表现；

风险指标：
添加最大回撤、胜率、盈亏比等回测指标（可通过 backtrader 的 Analyzer 实现）



五、注意事项
Tushare 免费版数据有频次限制，如需高频回测可改用本地数据（如 CSV 格式）；
ETF 无印花税，手续费设置需贴近实际（代码中按 0.03% 计算）；
回测结果仅为历史参考，实盘需结合市场环境和资金管理。