【量化交易实践】Python 实现股票箱体突破选股策略（完整代码 + 结果验证）

本文为《如何使用箱体突破做股票》下篇，承接上篇的策略思路讲解，本文将给出完整可运行的 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 未超买、均线多头排列等条件做二次筛选，能有效提升信号质量。

五、写在最后

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

风险提示：

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