本文为《如何使用箱体突破做股票》下篇，承接上篇的策略思路讲解，本文将给出完整可运行的 Python 实现代码、全市场选股逻辑、实盘筛选结果以及策略风险提示，适合量化入门者参考复用。

一、策略核心逻辑回顾

箱体突破策略采用价格破位 + 量能确认双重过滤规则，核心逻辑如下：

1. 价格突破条件：当日收盘价突破过去 N 个交易日的最高价（箱体上沿），确认趋势向上破位
2. 量能验证条件：当日成交量大于过去 5 日均量的指定倍数，确认突破有资金承接，降低假突破概率

本文默认参数：

• 突破周期：20 个交易日
• 成交量放大阈值：1.5 倍

二、完整代码实现

本方案基于 Python+MySQL 本地股票数据库开发，可批量扫描全市场个股，自动筛选当日符合箱体突破条件的标的。

2.1 环境依赖与数据库配置

import pandas as pd
import numpy as np
from sqlalchemy import create_engine

# 本地股票数据库配置 - 替换为自身数据库信息
engine = create_engine("mysql://root:12345678@localhost/db_stock?charset=utf8")

2.2 箱体突破选股核心函数

def get_breakout_stocks(breakout_days=20, volume_multiplier=1.5):
    """
    筛选向上突破股票
    参数:
        breakout_days: 突破周期(默认20日)
        volume_multiplier: 成交量放大倍数(默认1.5倍)
    返回:
        符合条件的股票字典列表
    """
    # 1. 获取最近有效交易日
    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']

    # 2. 计算回看起始日期，增加缓冲天数保证数据完整性
    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']

    # 3. 拉取全市场个股日线数据
    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 []

    # 4. 逐股校验突破条件
    results = []
    for ts_code in df['ts_code'].unique():
        try:
            stock_data = df[df['ts_code'] == ts_code].copy()
            stock_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_volume = latest_data['volume'].iloc[0]
            stock_name = latest_data["name"].iloc[0] if pd.notna(latest_data["name"].iloc[0]) else ts_code

            # 计算箱体阻力位：突破周期内的最高价
            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()

            # 计算基准均量
            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

2.3 个股成交量模式分析函数

def analyze_volume_patterns(ts_code, days=30):
    """
    深度分析个股成交量结构
    参数:
        ts_code: 股票代码
        days: 分析周期(默认30日)
    返回:
        包含量能均线的个股行情DataFrame
    """
    # 获取市场最新交易日
    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

2.4 主程序入口

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} 手")
                print("-" * 50)
    else:
        print("当日无符合条件的突破股票")

三、选股结果展示

以 2025 年 12 月 5 日全市场行情数据为例，使用默认 20 日周期、1.5 倍量能参数，筛选出的部分标的如下：

四、策略风险与优化方向

箱体突破是经典的趋势跟踪策略，但不存在 100% 胜率的交易方法，实盘应用中需注意以下要点：

1. 防范假突破风险 震荡行情中极易出现价格短暂破位后快速回落的假突破。建议设置止损位，通常放置在突破点下方 2%-3% 或箱体上沿内侧，跌破则及时离场。

2. 适配市场环境 牛市趋势行情下，突破策略的成功率与盈亏比显著更高；震荡市或熊市中建议降低仓位、提高量能过滤阈值，或仅参与主线板块内的突破标的。

3. 多指标共振过滤 单一突破信号容错率较低，可结合 MACD 金叉、RSI 未超买、均线多头排列等条件做二次筛选，能有效提升信号质量。

五、写在最后

箱体突破策略逻辑简单、可解释性强，非常适合量化新手入门实践。大家可以基于这份基础代码，叠加行业过滤、基本面筛选、动态止损止盈等模块，迭代出适配自身交易风格的专属策略。

风险提示：

本文只做教学，不做任何投资建议，投资有风险，入市需谨慎，本文作者享有一切解释权