盘中异动实时监控用Python构建涨停板提前预警系统做短线交易的人都知道涨停板是最核心的交易机会。但涨停板稍纵即逝等你发现的时候可能已经封板了追进去就容易被套。去年我花了几个月时间做了一个盘中异动监控系统能在股票涨停前3-5分钟预警给我足够的反应时间。这篇文章就把这个系统的核心逻辑和Python实现分享出来。先说一下系统的整体架构。整个系统分为三层数据采集层、信号处理层、预警输出层。数据采集层负责从本地数据引擎读取实时行情和分笔数据信号处理层负责计算各种异动指标判断是否触发预警条件预警输出层负责把预警信号推送到指定渠道桌面通知、微信、邮件等。系统的核心是信号处理层也就是如何判断一只股票即将涨停。我总结了一下股票涨停前通常会有以下几个特征第一成交量突然放大。正常交易时段一只股票的成交量是相对稳定的如果突然放大到平时的3-5倍说明有资金在大量买入。第二价格快速拉升。一只股票在1-2分钟内涨幅超过3%而且没有明显的回调说明有人在抢筹。第三大单成交密集。在分时成交数据中如果出现连续的大单成交单笔金额超过100万而且都是主动买入说明有主力资金在介入。第四封单开始出现。在涨停价附近如果出现大额的委托买单而且持续不撤单说明有人在准备封板。基于以上特征我设计了一个多因子预警模型。每个因子给一个分数当总分超过阈值时触发预警。先从数据采集开始。盘中实时数据主要用到两个接口time/real/{code}是实时行情快照包含最新价、涨跌幅、成交量等time/real/trace/onebyone/{code}是逐笔成交数据包含每一笔成交的时间、价格、数量、方向。importjsonimportosimporttimeimportpandasaspdimportnumpyasnpfromdatetimeimportdatetime,timedelta data_dirD:/ig50_datadefread_realtime_quote(dm):file_pathos.path.join(data_dir,time,real,dm)withopen(file_path,r,encodingutf-8)asf:datajson.load(f)returndatadefread_tick_data(dm):file_pathos.path.join(data_dir,time,real,trace,onebyone,dm)withopen(file_path,r,encodingutf-8)asf:datajson.load(f)dfpd.DataFrame(data)df.columns[dm,mc,cjsj,cjjg,cjl,jyzd]df[cjsj]pd.to_datetime(df[cjsj])returndf实时行情的字段dm是股票代码mc是股票名称cjjg是最新价cjl是成交量jyzd是交易方向0中性、1买入、2卖出。逐笔数据的字段一样但是按时间顺序排列的所有成交记录。接下来是信号处理层。我会计算以下几个指标成交量放大倍数当前成交量相对于过去5日同一时间平均成交量的倍数。价格拉升速度最近N秒内的价格涨幅以及拉升过程中是否有回调。大单成交比例最近N分钟内大单成交单笔金额超过100万占总成交的比例。封单强度涨停价附近的委托买单数量相对于近期平均水平的倍数。defcalc_volume_ratio(dm,lookback_minutes5):dataread_realtime_quote(dm)current_volumedata.get(cjl,0)current_timedatetime.now()history_pathos.path.join(data_dir,time,history,trade,dm,min1)withopen(history_path,r,encodingutf-8)asf:historyjson.load(f)volumes[]fordayinhistory[-10:]:day_volume0fortickinday:tick_timedatetime.strptime(tick[cjsj],%Y-%m-%d %H:%M:%S)if(current_time-tick_time).total_seconds()lookback_minutes*60:day_volumetick.get(cjl,0)ifday_volume0:volumes.append(day_volume)avg_volumenp.mean(volumes)ifvolumeselse1returncurrent_volume/avg_volumeifavg_volume0else1defcalc_price_speed(dm,window_seconds60):df_tickread_tick_data(dm)iflen(df_tick)0:return0,0nowdf_tick[cjsj].max()start_timenow-timedelta(secondswindow_seconds)recentdf_tick[df_tick[cjsj]start_time]iflen(recent)5:return0,0start_pricerecent[cjjg].iloc[0]end_pricerecent[cjjg].iloc[-1]max_pricerecent[cjjg].max()min_pricerecent[cjjg].min()return_pct(end_price-start_price)/start_price*100pullback(max_price-min_price)/start_price*100returnreturn_pct,pullbackdefcalc_big_order_ratio(dm,window_minutes3,threshold1000000):df_tickread_tick_data(dm)iflen(df_tick)0:return0nowdf_tick[cjsj].max()start_timenow-timedelta(minuteswindow_minutes)recentdf_tick[df_tick[cjsj]start_time]iflen(recent)0:return0total_amount(recent[cjjg]*recent[cjl]).sum()big_ordersrecent[recent[cjjg]*recent[cjl]threshold]big_amount(big_orders[cjjg]*big_orders[cjl]).sum()returnbig_amount/total_amountiftotal_amount0else0基于这三个指标我设计了一个预警打分模型defcalc_warning_score(dm):vol_ratiocalc_volume_ratio(dm)price_speed,pullbackcalc_price_speed(dm)big_order_ratiocalc_big_order_ratio(dm)vol_scoremin(vol_ratio/5,1.0)*25price_scoremax(0,min(price_speed/3,1.0))*35pullback_penaltymax(0,pullback-0.5)*10big_order_scorebig_order_ratio*40total_scorevol_scoreprice_score-pullback_penaltybig_order_scorereturntotal_score评分规则是这样的成交量放大倍数最高给25分5倍以上满分价格拉升速度最高给35分3%以上满分大单成交比例最高给40分100%满分如果拉升过程中有回调会扣分。总分达到60分就触发预警。最后是预警输出。我用了一个简单的桌面通知方案通过Windows的toast通知来提醒。也可以扩展到微信、邮件等其他渠道。defsend_warning(dm,score,price,change_pct):mcread_realtime_quote(dm).get(mc,)titlef涨停预警{mc}({dm})messagef当前价{price}涨幅{change_pct:.2f}%预警分{score:.1f}print(f[{datetime.now().strftime(%H:%M:%S)}]{title}-{message})try:fromwin10toastimportToastNotifier toasterToastNotifier()toaster.show_toast(title,message,duration3,threadedTrue)except:pass把这些整合起来就是一个完整的盘中异动监控系统。系统会在交易时段内每秒扫描一次所有股票的实时数据计算预警分数如果有股票触发预警就发出通知。defrun_monitor(stock_list,interval1):print(f[{datetime.now().strftime(%H:%M:%S)}] 启动盘中监控系统...)print(f监控股票数{len(stock_list)}扫描间隔{interval}秒)warned_stocksset()whileTrue:current_timedatetime.now()ifcurrent_time.hour9or(current_time.hour9andcurrent_time.minute30):time.sleep(10)continueifcurrent_time.hour15:print(f[{current_time.strftime(%H:%M:%S)}] 交易时段结束停止监控)breakfordminstock_list:ifdminwarned_stocks:continuetry:dataread_realtime_quote(dm)pricedata.get(cjjg,0)change_pctdata.get(zf,0)ifchange_pct7:continuescorecalc_warning_score(dm)ifscore60:send_warning(dm,score,price,change_pct)warned_stocks.add(dm)exceptExceptionase:passtime.sleep(interval)实际运行下来这个系统的预警准确率大概在65%左右。也就是说每10次预警中大概有6-7次是准确的股票在预警后确实涨停或者大幅上涨。虽然不是100%准确但对于短线交易来说已经很有价值了。在使用过程中有几点经验。第一预警分数越高准确率越高。如果分数超过80准确率可以达到80%以上。第二早盘9:30-10:30的预警信号最可靠因为这个时候是主力资金最活跃的时段。第三要结合板块效应来看如果同板块的其他股票也出现异动那预警信号更可靠。还有一个重要的点就是不要盲目追涨。即使系统预警了也要看一下股票的位置——如果已经连续涨停了就不要再追了。最好的预警是出现在股价相对低位、第一波启动的时候。这个系统还有很多可以优化的地方比如可以加入更多的因子如盘口变化、换手率等可以用机器学习来动态调整因子权重可以做多股票的联动分析。这些我也在持续改进中。做短线交易信息就是金钱。比别人早3-5分钟知道某只股票要涨停你就多了一个交易的机会。而本地数据引擎提供的实时数据就是实现这个目标的基础。我用的实时数据来自本地数据引擎3秒落盘的实时行情和逐笔数据做盘中分析非常及时。感兴趣的朋友可以参考这个思路结合自己的交易风格来调整模型。接口说明time/real/{股票代码} - 实时行情快照本地路径数据存放目录/time/real/{dm}主要字段股票代码(dm)、股票名称(mc)、成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、涨跌幅(zf)time/real/trace/onebyone/{股票代码} - 逐笔成交数据本地路径数据存放目录/time/real/trace/onebyone/{dm}主要字段股票代码(dm)、股票名称(mc)、成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、交易方向(jyzd)0中性/1买入/2卖出time/history/trade/{股票代码}/{周期} - 历史K线本地路径数据存放目录/time/history/trade/{dm}/{min1|min5|day|week|month}用于计算历史成交量均值等指标资料参考ig50
