聊一聊箱体突破法，附代码

子晓聊技术

发布于 2026-04-23 17:23:03

1430

很多同学喜欢玩箱体突破，这里聊一聊。

首先说一说，什么是箱体突破法？

想象一下，一只股票的价格在一段时间内，就像被困在一个无形的箱子里反复震荡。这个"箱子"的上沿是阻力位（价格多次上涨至此回落），下沿是支撑位（价格多次下跌至此反弹）。当股价经过充分盘整后，突然以强劲的势头冲破这个箱子的上沿，我们就称之为向上突破。

这种形态背后蕴含着深刻的市场心理：多空双方在箱体内经过反复拉锯，最终多头力量占据绝对优势，推动股价突破阻力位，开启新一轮上涨行情。反之，若跌破支撑位，则形成向下突破。

箱体突破法的核心就是识别这种"盘整-突破"的模式，并在突破发生时及时介入，捕捉后续的上涨空间。

如何识别有效的箱体突破？

并非所有突破都值得追涨。一个有效的箱体突破通常需要满足两个关键条件：

价格突破：股价必须明确收于前期阻力位之上，而不是盘中短暂触及后回落。
成交量确认：突破时必须有显著放大的成交量配合，通常是近期平均成交量的1.5倍以上。这表明有大量资金积极入场，为突破提供了动力，增加了突破的可靠性。

没有成交量配合的突破往往是"假突破"，价格很可能很快回落至箱体内，形成诱多陷阱。

这里说下技术方案：

数据获取：首先，代码从数据库中获取指定时间范围内的股票日线数据，包括交易日期、股票代码、名称、收盘价、最高价、最低价和成交量。
确定阻力位：对于每只股票，代码通过以下方式确定阻力位：

historical_high = historical_data['high'].tail(breakout_days).max()

这行代码获取了最近breakout_days（默认为20）个交易日的最高价，这就是我们所说的"箱体上沿"或阻力位。

计算平均成交量

avg_volume = historical_data['volume'].tail(5).mean()

代码计算了最近5个交易日的平均成交量，作为判断成交量是否放大的基准。

判断突破条件

price_breakout = current_close > historical_high
volume_breakout = current_volume > avg_volume * volume_multiplier

计算突破强度

breakout_strength = round((current_close - historical_high) / historical_high * 100, 2)
volume_ratio = round(current_volume / avg_volume, 2)

完整代码如下：

import pandas as pd
import numpy as np
from sqlalchemy import create_engine
import sys
# 数据库配置
engine = create_engine("mysql://root:12345678@localhost/db_stock?charset=utf8")
def get_breakout_stocks(breakout_days=20, volume_multiplier=1.5):
    """
    筛选向上突破股票
    参数:
        breakout_days: 突破周期（默认20日）
        volume_multiplier: 成交量放大倍数（默认1.5倍）
    """
    # 获取最近的有效交易日
    recent_dates_query = """
    SELECT DISTINCT trade_date 
    FROM stock_daily 
    ORDER BY trade_date DESC 
    LIMIT 2
    """
    recent_dates = pd.read_sql(recent_dates_query, engine)
    if len(recent_dates) < 2:
        print("数据量不足，需要至少2个交易日的数据")
        return []
    latest_date = recent_dates.iloc[0]['trade_date']
    # 计算回看开始日期
    lookback_days = breakout_days + 10  # 额外增加缓冲天数
    start_date_query = f"""
    SELECT DISTINCT trade_date 
    FROM stock_daily 
    WHERE trade_date <= '{latest_date}' 
    ORDER BY trade_date DESC 
    LIMIT {lookback_days}
    """
    all_dates = pd.read_sql(start_date_query, engine)
    if len(all_dates) < breakout_days + 5:
        print("历史数据不足")
        return []
    lookback_start_date = all_dates.iloc[-1]['trade_date']
    data_query = f"""
    SELECT sd.trade_date, sd.ts_code, sb.name, sd.close, sd.high, sd.low, sd.vol as volume
    FROM stock_daily sd
    LEFT JOIN stock_basic sb ON sd.ts_code = sb.ts_code
    WHERE sd.trade_date >= '{lookback_start_date}' AND sd.trade_date <= '{latest_date}'
    ORDER BY sd.ts_code, sd.trade_date
    """
    df = pd.read_sql(data_query, engine)
    if df.empty:
        print("未找到数据")
        return []
    # 筛选向上突破的股票
    results = []
    for ts_code in df['ts_code'].unique():
        try:
            stock_data = df[df['ts_code'] == ts_code].copy()
            stock153_data = stock_data.sort_values('trade_date')
            if len(stock_data) < breakout_days + 5:
                continue
            # 获取最新交易日数据
            latest_data = stock_data[stock_data['trade_date'] == latest_date]
            if latest_data.empty:
                continue
            current_close = latest_data['close'].iloc[0]
            current_high = latest_data['high'].iloc[0]
            current_volume = latest_data['volume'].iloc[0]  # 使用vol字段
            stock_name = latest_data['name'].iloc[0] if pd.notna(latest_data['name'].iloc[0]) else 'N/A'
            # 计算阻力位（breakout_days日最高点）
            historical_data = stock_data[stock_data['trade_date'] < latest_date]
            if len(historical_data) < breakout_days:
                continue
            historical_high = historical_data['high'].tail(breakout_days).max()
            # 计算平均成交量（前5日）
            avg_volume = historical_data['volume'].tail(5).mean()
            if avg_volume == 0:  # 避免除零错误
                continue
            # 向上突破条件
            price_breakout = current_close > historical_high
            volume_breakout = current_volume > avg_volume * volume_multiplier
            if price_breakout and volume_breakout:
                # 计算突破强度
                breakout_strength = round((current_close - historical_high) / historical_high * 100, 2)
                volume_ratio = round(current_volume / avg_volume, 2)
                results.append({
                    'ts_code': ts_code,
                    'name': stock_name,
                    'trade_date': latest_date,
                    'close_price': round(current_close, 2),
                    'resistance_level': round(historical_high, 2),
                    'breakout_strength': f"{breakout_strength}%",
                    'volume_ratio': f"{volume_ratio}倍",
                    'breakout_days': breakout_days
                })
        except Exception as e:
            print(f"处理 {ts_code} 时出错: {str(e)}")
            continue
    return results
def analyze_volume_patterns(ts_code, days=30):
    """
    分析个股成交量模式
    """
    # 获取历史数据
    end_date_query = "SELECT MAX(trade_date) as max_date FROM stock_daily"
    end_date = pd.read_sql(end_date_query, engine).iloc[0]['max_date']
    start_date_query = f"""
    SELECT DISTINCT trade_date 
    FROM stock_daily 
    WHERE trade_date <= '{end_date}' 
    ORDER BY trade_date DESC 
    LIMIT {days}
    """
    dates_df = pd.read_sql(start_date_query, engine)
    start_date = dates_df.iloc[-1]['trade_date']
    data_query = f"""
    SELECT sd.trade_date, sd.close, sd.high, sd.low, sd.vol as volume
    FROM stock_daily sd
    WHERE sd.ts_code = '{ts_code}' AND sd.trade_date >= '{start_date}'
    ORDER BY sd.trade_date
    """
    df = pd.read_sql(data_query, engine)
    if df.empty:
        return None
    # 计算成交量指标
    df['vol_ma5'] = df['volume'].rolling(5).mean()  # 5日平均成交量
    df['vol_ma10'] = df['volume'].rolling(10).mean()  # 10日平均成交量
    df['price_change'] = df['close'].pct_change() * 100
    return df
if __name__ == "__main__":
    print("开始筛选向上突破股票...")
    # 查询符合条件的股票
    results = get_breakout_stocks(breakout_days=20, volume_multiplier=1.5)
    if results:
        print(f"\n🎯 找到 {len(results)} 只向上突破股票:")
        print("=" * 80)
        result_df = pd.DataFrame(results)
        for _, stock in result_df.iterrows():
            print(f"股票代码: {stock['ts_code']}")
            print(f"股票名称: {stock['name']}")
            print(f"突破日期: {stock['trade_date']}")
            print(f"收盘价: {stock['close_price']}")
            print(f"阻力位: {stock['resistance_level']}")
            print(f"突破强度: {stock['breakout_strength']}")
            print(f"量比: {stock['volume_ratio']}")
            print("-" * 50)
            # 分析成交量模式
            volume_analysis = analyze_volume_patterns(stock['ts_code'])
            if volume_analysis is not None:
                latest_vol = volume_analysis['volume'].iloc[-1]
                avg_vol = volume_analysis['vol_ma5'].iloc[-1]
                print(f"成交量分析: 最新{latest_vol:.0f}手, 5日均量{avg_vol:.0f}手")
        # 保存结果
        result_df.to_csv('upward_breakout_corrected.csv', index=False, encoding='utf-8-sig')
        print(f"✅ 结果已保存至 upward_breakout_corrected.csv")
    else:
        print("❌ 未找到符合条件的向上突破股票")