盘中异动实时监控：用Python构建涨停板提前预警系统

做短线交易的人都知道，涨停板是最核心的交易机会。但涨停板稍纵即逝，等你发现的时候可能已经封板了，追进去就容易被套。去年我花了几个月时间，做了一个盘中异动监控系统，能在股票涨停前3-5分钟预警，给我足够的反应时间。

这篇文章就把这个系统的核心逻辑和Python实现分享出来。

先说一下系统的整体架构。整个系统分为三层：数据采集层、信号处理层、预警输出层。数据采集层负责从本地数据引擎读取实时行情和分笔数据；信号处理层负责计算各种异动指标，判断是否触发预警条件；预警输出层负责把预警信号推送到指定渠道（桌面通知、微信、邮件等）。

系统的核心是信号处理层，也就是如何判断一只股票即将涨停。我总结了一下，股票涨停前通常会有以下几个特征：

第一，成交量突然放大。正常交易时段一只股票的成交量是相对稳定的，如果突然放大到平时的3-5倍，说明有资金在大量买入。

第二，价格快速拉升。一只股票在1-2分钟内涨幅超过3%，而且没有明显的回调，说明有人在抢筹。

第三，大单成交密集。在分时成交数据中，如果出现连续的大单成交（单笔金额超过100万），而且都是主动买入，说明有主力资金在介入。

第四，封单开始出现。在涨停价附近，如果出现大额的委托买单，而且持续不撤单，说明有人在准备封板。

基于以上特征，我设计了一个多因子预警模型。每个因子给一个分数，当总分超过阈值时触发预警。

先从数据采集开始。盘中实时数据主要用到两个接口：time/real/{code}是实时行情快照，包含最新价、涨跌幅、成交量等；time/real/trace/onebyone/{code}是逐笔成交数据，包含每一笔成交的时间、价格、数量、方向。

import json
import os
import time
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

data_dir = "D:/ig50_data"

def read_realtime_quote(dm):
    file_path = os.path.join(data_dir, "time", "real", dm)
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    return data

def read_tick_data(dm):
    file_path = os.path.join(data_dir, "time", "real", "trace", "onebyone", dm)
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    df = pd.DataFrame(data)
    df.columns = ["dm", "mc", "cjsj", "cjjg", "cjl", "jyzd"]
    df["cjsj"] = pd.to_datetime(df["cjsj"])
    return df

实时行情的字段，dm是股票代码，mc是股票名称，cjjg是最新价，cjl是成交量，jyzd是交易方向（0中性、1买入、2卖出）。逐笔数据的字段一样，但是按时间顺序排列的所有成交记录。

接下来是信号处理层。我会计算以下几个指标：

成交量放大倍数：当前成交量相对于过去5日同一时间平均成交量的倍数。

价格拉升速度：最近N秒内的价格涨幅，以及拉升过程中是否有回调。

大单成交比例：最近N分钟内大单成交（单笔金额超过100万）占总成交的比例。

封单强度：涨停价附近的委托买单数量相对于近期平均水平的倍数。

def calc_volume_ratio(dm, lookback_minutes=5):
    data = read_realtime_quote(dm)
    current_volume = data.get("cjl", 0)
    current_time = datetime.now()
    
    history_path = os.path.join(data_dir, "time", "history", "trade", dm, "min1")
    with open(history_path, "r", encoding="utf-8") as f:
        history = json.load(f)
    
    volumes = []
    for day in history[-10:]:
        day_volume = 0
        for tick in day:
            tick_time = datetime.strptime(tick["cjsj"], "%Y-%m-%d %H:%M:%S")
            if (current_time - tick_time).total_seconds() <= lookback_minutes * 60:
                day_volume += tick.get("cjl", 0)
        if day_volume > 0:
            volumes.append(day_volume)
    
    avg_volume = np.mean(volumes) if volumes else 1
    return current_volume / avg_volume if avg_volume > 0 else 1

def calc_price_speed(dm, window_seconds=60):
    df_tick = read_tick_data(dm)
    if len(df_tick) == 0:
        return 0, 0
    
    now = df_tick["cjsj"].max()
    start_time = now - timedelta(seconds=window_seconds)
    recent = df_tick[df_tick["cjsj"] >= start_time]
    
    if len(recent) < 5:
        return 0, 0
    
    start_price = recent["cjjg"].iloc[0]
    end_price = recent["cjjg"].iloc[-1]
    max_price = recent["cjjg"].max()
    min_price = recent["cjjg"].min()
    
    return_pct = (end_price - start_price) / start_price * 100
    pullback = (max_price - min_price) / start_price * 100
    
    return return_pct, pullback

def calc_big_order_ratio(dm, window_minutes=3, threshold=1000000):
    df_tick = read_tick_data(dm)
    if len(df_tick) == 0:
        return 0
    
    now = df_tick["cjsj"].max()
    start_time = now - timedelta(minutes=window_minutes)
    recent = df_tick[df_tick["cjsj"] >= start_time]
    
    if len(recent) == 0:
        return 0
    
    total_amount = (recent["cjjg"] * recent["cjl"]).sum()
    big_orders = recent[recent["cjjg"] * recent["cjl"] >= threshold]
    big_amount = (big_orders["cjjg"] * big_orders["cjl"]).sum()
    
    return big_amount / total_amount if total_amount > 0 else 0

基于这三个指标，我设计了一个预警打分模型：

def calc_warning_score(dm):
    vol_ratio = calc_volume_ratio(dm)
    price_speed, pullback = calc_price_speed(dm)
    big_order_ratio = calc_big_order_ratio(dm)
    
    vol_score = min(vol_ratio / 5, 1.0) * 25
    price_score = max(0, min(price_speed / 3, 1.0)) * 35
    pullback_penalty = max(0, pullback - 0.5) * 10
    big_order_score = big_order_ratio * 40
    
    total_score = vol_score + price_score - pullback_penalty + big_order_score
    return total_score

评分规则是这样的：成交量放大倍数最高给25分（5倍以上满分），价格拉升速度最高给35分（3%以上满分），大单成交比例最高给40分（100%满分），如果拉升过程中有回调会扣分。总分达到60分就触发预警。

最后是预警输出。我用了一个简单的桌面通知方案，通过Windows的toast通知来提醒。也可以扩展到微信、邮件等其他渠道。

def send_warning(dm, score, price, change_pct):
    mc = read_realtime_quote(dm).get("mc", "")
    title = f"涨停预警：{mc}({dm})"
    message = f"当前价：{price}，涨幅：{change_pct:.2f}%，预警分：{score:.1f}"
    
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {title} - {message}")
    
    try:
        from win10toast import ToastNotifier
        toaster = ToastNotifier()
        toaster.show_toast(title, message, duration=3, threaded=True)
    except:
        pass

把这些整合起来，就是一个完整的盘中异动监控系统。系统会在交易时段内每秒扫描一次所有股票的实时数据，计算预警分数，如果有股票触发预警就发出通知。

def run_monitor(stock_list, interval=1):
    print(f"[{datetime.now().strftime('%H:%M:%S')}] 启动盘中监控系统...")
    print(f"监控股票数：{len(stock_list)}，扫描间隔：{interval}秒")
    
    warned_stocks = set()
    
    while True:
        current_time = datetime.now()
        if current_time.hour < 9 or (current_time.hour == 9 and current_time.minute < 30):
            time.sleep(10)
            continue
        if current_time.hour >= 15:
            print(f"[{current_time.strftime('%H:%M:%S')}] 交易时段结束，停止监控")
            break
        
        for dm in stock_list:
            if dm in warned_stocks:
                continue
            
            try:
                data = read_realtime_quote(dm)
                price = data.get("cjjg", 0)
                change_pct = data.get("zf", 0)
                
                if change_pct > 7:
                    continue
                
                score = calc_warning_score(dm)
                
                if score >= 60:
                    send_warning(dm, score, price, change_pct)
                    warned_stocks.add(dm)
                    
            except Exception as e:
                pass
        
        time.sleep(interval)

实际运行下来，这个系统的预警准确率大概在65%左右。也就是说，每10次预警中，大概有6-7次是准确的，股票在预警后确实涨停或者大幅上涨。虽然不是100%准确，但对于短线交易来说已经很有价值了。

在使用过程中有几点经验。第一，预警分数越高，准确率越高。如果分数超过80，准确率可以达到80%以上。第二，早盘（9:30-10:30）的预警信号最可靠，因为这个时候是主力资金最活跃的时段。第三，要结合板块效应来看，如果同板块的其他股票也出现异动，那预警信号更可靠。

还有一个重要的点，就是不要盲目追涨。即使系统预警了，也要看一下股票的位置——如果已经连续涨停了，就不要再追了。最好的预警是出现在股价相对低位、第一波启动的时候。

这个系统还有很多可以优化的地方，比如可以加入更多的因子（如盘口变化、换手率等），可以用机器学习来动态调整因子权重，可以做多股票的联动分析。这些我也在持续改进中。

做短线交易，信息就是金钱。比别人早3-5分钟知道某只股票要涨停，你就多了一个交易的机会。而本地数据引擎提供的实时数据，就是实现这个目标的基础。

我用的实时数据来自本地数据引擎，3秒落盘的实时行情和逐笔数据，做盘中分析非常及时。感兴趣的朋友可以参考这个思路，结合自己的交易风格来调整模型。

---

接口说明：

1. time/real/{股票代码} - 实时行情快照
   本地路径：数据存放目录/time/real/{dm}
   主要字段：股票代码(dm)、股票名称(mc)、成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、涨跌幅(zf)

2. time/real/trace/onebyone/{股票代码} - 逐笔成交数据
   本地路径：数据存放目录/time/real/trace/onebyone/{dm}
   主要字段：股票代码(dm)、股票名称(mc)、成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、交易方向(jyzd)（0中性/1买入/2卖出）

3. time/history/trade/{股票代码}/{周期} - 历史K线
   本地路径：数据存放目录/time/history/trade/{dm}/{min1|min5|day|week|month}
   用于计算历史成交量均值等指标

gitee开源地址
github开源地址