本文为《如何使用箱体突破做股票》下篇，承接上篇的策略思路讲解，本文将给出完整可运行的 Python 实现代码、全市场选股逻辑、实盘筛选结果以及策略风险提示，适合量化入门者参考复用。
一、策略核心逻辑回顾
箱体突破策略采用价格破位 + 量能确认双重过滤规则，核心逻辑如下：
- 价格突破条件：当日收盘价突破过去 N 个交易日的最高价（箱体上沿），确认趋势向上破位
- 量能验证条件：当日成交量大于过去 5 日均量的指定倍数，确认突破有资金承接，降低假突破概率
本文默认参数：
二、完整代码实现
本方案基于 Python+MySQL 本地股票数据库开发，可批量扫描全市场个股，自动筛选当日符合箱体突破条件的标的。
2.1 环境依赖与数据库配置
import pandas as pdimport numpy as npfrom 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% 胜率的交易方法，实盘应用中需注意以下要点：
防范假突破风险 震荡行情中极易出现价格短暂破位后快速回落的假突破。建议设置止损位，通常放置在突破点下方 2%-3% 或箱体上沿内侧，跌破则及时离场。
适配市场环境牛市趋势行情下，突破策略的成功率与盈亏比显著更高；震荡市或熊市中建议降低仓位、提高量能过滤阈值，或仅参与主线板块内的突破标的。
多指标共振过滤 单一突破信号容错率较低，可结合 MACD 金叉、RSI 未超买、均线多头排列等条件做二次筛选，能有效提升信号质量。
五、写在最后
箱体突破策略逻辑简单、可解释性强，非常适合量化新手入门实践。大家可以基于这份基础代码，叠加行业过滤、基本面筛选、动态止损止盈等模块，迭代出适配自身交易风格的专属策略。
图片展示仅为伪代码教学，真实测试代码需【888】发送！
风险提示：
本文只做教学，不做任何投资建议，投资有风险，入市需谨慎，本文作者享有一切解释权
关于作者
一个量化er，专注于QMT miniQMT PTrade量化与技术分享研究，兼顾量化平台信息分享，内容干货而实用，用数据量化市场，用代码构建策略👇关注+星标公众号 第一时间看到推文👇
如果文章对你有帮助，还请点「推荐」支持一下作者
--- End ---
