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

发布时间:2026/6/25 20:43:35
【量化交易实践】Python 实现股票箱体突破选股策略(完整代码 + 结果验证) 本文为《如何使用箱体突破做股票》下篇承接上篇的策略思路讲解本文将给出完整可运行的 Python 实现代码、全市场选股逻辑、实盘筛选结果以及策略风险提示适合量化入门者参考复用。一、策略核心逻辑回顾箱体突破策略采用价格破位 量能确认双重过滤规则核心逻辑如下价格突破条件当日收盘价突破过去 N 个交易日的最高价箱体上沿确认趋势向上破位量能验证条件当日成交量大于过去 5 日均量的指定倍数确认突破有资金承接降低假突破概率本文默认参数突破周期20 个交易日成交量放大阈值1.5 倍二、完整代码实现本方案基于 PythonMySQL 本地股票数据库开发可批量扫描全市场个股自动筛选当日符合箱体突破条件的标的。2.1 环境依赖与数据库配置import pandas as pd import numpy as np from sqlalchemy import create_engine # 本地股票数据库配置 - 替换为自身数据库信息 engine create_engine(mysql://root:12345678localhost/db_stock?charsetutf8)2.2 箱体突破选股核心函数def get_breakout_stocks(breakout_days20, volume_multiplier1.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 results2.3 个股成交量模式分析函数def analyze_volume_patterns(ts_code, days30): 深度分析个股成交量结构 参数: 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 df2.4 主程序入口if __name__ __main__: print(开始筛选向上突破股票..) # 可自定义突破周期与量能阈值 results get_breakout_stocks(breakout_days20, volume_multiplier1.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 倍量能参数筛选出的部分标的如下股票代码股票名称突破日期收盘价阻力位突破强度量比突破周期000100.SZTCL 科技202512054.554.471.79%2.31 倍20 日000415.SZ渤海租赁202512053.883.850.78%1.68 倍20 日000725.SZ京东方 A202512054.264.084.41%3.16 倍20 日000823.SZ超声电子2025120514.3914.161.62%2.63 倍20 日002023.SZ海特高新2025120513.5813.262.41%3.87 倍20 日002031.SZ巨轮智能202512058.047.882.03%5.10 倍20 日002268.SZ电科网安2025120518.3617.872.74%3.55 倍20 日002300.SZ太阳电缆2025120510.369.429.98%5.55 倍20 日002520.SZ日发精机202512056.626.403.44%7.43 倍20 日四、策略风险与优化方向箱体突破是经典的趋势跟踪策略但不存在 100% 胜率的交易方法实盘应用中需注意以下要点防范假突破风险震荡行情中极易出现价格短暂破位后快速回落的假突破。建议设置止损位通常放置在突破点下方 2%-3% 或箱体上沿内侧跌破则及时离场。适配市场环境牛市趋势行情下突破策略的成功率与盈亏比显著更高震荡市或熊市中建议降低仓位、提高量能过滤阈值或仅参与主线板块内的突破标的。多指标共振过滤单一突破信号容错率较低可结合 MACD 金叉、RSI 未超买、均线多头排列等条件做二次筛选能有效提升信号质量。五、写在最后箱体突破策略逻辑简单、可解释性强非常适合量化新手入门实践。大家可以基于这份基础代码叠加行业过滤、基本面筛选、动态止损止盈等模块迭代出适配自身交易风格的专属策略。风险提示本文只做教学不做任何投资建议投资有风险入市需谨慎本文作者享有一切解释权