集合竞价数据挖掘实战：用Python构建开盘信号识别系统

集合竞价数据挖掘实战：用Python构建开盘信号识别系统

集合竞价是每个交易日开盘前的15分钟，很多人觉得这段时间没什么信息量，开盘价出来再操作也不迟。但我实际测试下来，集合竞价阶段的数据包含了大量有价值的信号，特别是对日内短线交易来说。这篇文章分享我用Python构建的集合竞价信号识别系统。

集合竞价分两个时间段：9:15-9:20可以申报也可以撤单，9:20-9:25只能申报不能撤单，9:25产生开盘价。还有一个尾盘集合竞价是14:57-15:00。我主要分析的是早盘集合竞价。

数据用的是集合竞价接口time/real/trace/jhjj/{dm}，这个接口在集合竞价时间段每1分钟更新一次，返回的字段包括申报时间、匹配量、未匹配量、标志（1表示买单，-1表示卖单）、成交价。数据按时间升序排列。

import json
import os
import pandas as pd
import numpy as np
from collections import defaultdict

data_dir = "D:/stock_data"

def read_jhjj(dm):
file_path = os.path.join(data_dir, "time", "real", "trace", "jhjj", dm)
if not os.path.exists(file_path):
return None
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
df = pd.DataFrame(data)
df.columns = ["sbsj", "ppl", "wppl", "bz", "cjj"]
return df

字段名继续用简写，sbsj是申报时间，ppl是匹配量，wppl是未匹配量，bz是标志（1买单-1卖单），cjj是成交价。

第一个信号：集合竞价放量。集合竞价的成交量如果远超平时，说明市场对这只股票的关注度突然提高。我会把当天的集合竞价量和前5个交易日的平均集合竞价量做对比，如果超过2倍就标记为"放量"。

def calc_volume_ratio(df_today, history_volumes):
today_vol = df_today["ppl"].sum()
avg_vol = np.mean(history_volumes) if history_volumes else 0
ratio = today_vol / avg_vol if avg_vol > 0 else 0
return ratio, today_vol

def signal_volume(df_today, history_volumes, threshold=2.0):
ratio, vol = calc_volume_ratio(df_today, history_volumes)
if ratio >= threshold:
return {"signal": "集合竞价放量", "ratio": ratio, "volume": vol, "score": 65}
return None

第二个信号：集合竞价价格走势。9:20到9:25之间的成交价变化最有参考价值，因为这段时间不能撤单，申报都是真实的。如果这段期间成交价持续上升，说明买盘在加强；如果持续下降，说明卖盘占优。

def signal_price_trend(df_today):
df_late = df_today[df_today["sbsj"] >= "09:20:00"].copy()
if len(df_late) < 2:
return None

prices = df_late["cjj"].values
first_price = prices[0]
last_price = prices[–1]

if first_price <= 0:
return None

change = (last_price – first_price) / first_price

if change > 0.02:
return {"signal": "集合竞价尾段拉升", "change": change, "score": 70}
elif change < –0.02:
return {"signal": "集合竞价尾段下挫", "change": change, "score": 30}
else:
return None

第三个信号：未匹配量堆积。9:25时的未匹配量能反映买卖力量的失衡。如果未匹配买单远大于匹配量，说明买盘很强但没成交完，开盘后可能继续涨。反过来如果未匹配卖单堆积，开盘后可能跌。

def signal_unmatched(df_today):
df_last = df_today.tail(1)
if len(df_last) == 0:
return None

wppl = df_last["wppl"].values[0]
bz = df_last["bz"].values[0]
ppl = df_last["ppl"].values[0]

if ppl <= 0:
return None

ratio = abs(wppl) / ppl

if bz == 1 and ratio > 3:
return {"signal": "未匹配买单堆积", "ratio": ratio, "score": 68}
elif bz == –1 and ratio > 3:
return {"signal": "未匹配卖单堆积", "ratio": ratio, "score": 32}
return None

第四个信号：高开幅度。集合竞价形成开盘价，如果开盘价高于前日收盘价3%以上，属于大幅高开。大幅高开有两种情况：一种是利好消息驱动的，后续可能继续涨；另一种是主力拉高出货的，开盘后冲高回落。区分这两种情况需要结合成交量和未匹配量。

def signal_open_premium(df_today, prev_close):
if prev_close <= 0:
return None

open_price = df_today["cjj"].iloc[–1]
premium = (open_price – prev_close) / prev_close

vol_ratio, _ = calc_volume_ratio(df_today, [])
unmatched = df_today.tail(1)["wppl"].values[0]
unmatched_bz = df_today.tail(1)["bz"].values[0]

if premium > 0.03:
if unmatched_bz == 1 and unmatched > 0:
return {"signal": "放量高开+买盘堆积", "premium": premium, "score": 72}
else:
return {"signal": "高开但买盘不足", "premium": premium, "score": 45}
elif premium < –0.03:
return {"signal": "低开", "premium": premium, "score": 40}
return None

把这些信号汇总起来，就是一个完整的集合竞价分析系统。

def analyze_jhjj(dm, prev_close, history_volumes):
df = read_jhjj(dm)
if df is None or len(df) == 0:
return None

signals = []

s1 = signal_volume(df, history_volumes)
if s1:
signals.append(s1)

s2 = signal_price_trend(df)
if s2:
signals.append(s2)

s3 = signal_unmatched(df)
if s3:
signals.append(s3)

s4 = signal_open_premium(df, prev_close)
if s4:
signals.append(s4)

if not signals:
return {"dm": dm, "score": 50, "signals": ["无明显信号"]}

total_score = np.mean([s["score"] for s in signals])

return {
"dm": dm,
"score": round(total_score, 1),
"signals": [s["signal"] for s in signals]
}

实际使用的时候，我会在每天9:25运行这个系统，遍历全市场的股票，筛选出综合得分超过65分的股票作为当日观察池。然后配合开盘后的实时行情做最终决策。

几个使用经验。第一，集合竞价信号对次新股和小盘股效果更好，因为这类股票的集合竞价参与者少，几个大单就能改变走势。大盘股的集合竞价信号不太准。第二，集合竞价信号要配合前一日收盘状态来看，如果前一天涨停且今天集合竞价放量高开，信号更强。第三，9:20到9:25的信号比9:15到9:20的信号可靠，因为9:20之后不能撤单。

我用的数据来源是ig50的集合竞价接口，数据每分钟更新一次，做实时分析完全够用。

接口说明：

资料参考：ig50