分时数据深度挖掘：用Python构建日内T+0交易信号系统

分时数据深度挖掘：用Python构建日内T+0交易信号系统

做日内T+0交易，分时数据是最核心的数据源。但大部分人只是看个分时图的白线黄线，很多有价值的细节信息都被忽略了。去年我搭建了一个分时数据深度挖掘系统，用Python从分钟级K线和逐笔成交数据中提取交易信号，辅助日内T+0决策。这篇文章分享系统的设计思路和核心代码。

本地数据引擎提供了丰富的分时数据。1分钟K线在time/history/trade/{dm}/min1，5分钟K线在time/history/trade/{dm}/min5，15分钟K线在time/history/trade/{dm}/min15。更细的逐笔成交数据在time/real/trace/onebyone/{dm}，大单成交数据在time/real/trace/bigdeal/{dm}。实时行情快照在time/real/{dm}。

importjsonimportosimportpandasaspdimportnumpyasnpfromdatetimeimportdatetime,timedelta data_dir="D:/ig50_data"defread_min_kline(dm,period="min1"):file_path=os.path.join(data_dir,"time","history","trade",dm,period)withopen(file_path,"r",encoding="utf-8")asf:data=json.load(f)df=pd.DataFrame(data)df.columns=["dm","cjsj","cjjg","cjl","cje","zf"]df["cjsj"]=pd.to_datetime(df["cjsj"])returndfdefread_tick_data(dm):file_path=os.path.join(data_dir,"time","real","trace","onebyone",dm)withopen(file_path,"r",encoding="utf-8")asf:data=json.load(f)df=pd.DataFrame(data)df.columns=["dm","mc","cjsj","cjjg","cjl","jyzd"]df["cjsj"]=pd.to_datetime(df["cjsj"])returndfdefread_bigdeal(dm):file_path=os.path.join(data_dir,"time","real","trace","bigdeal",dm)withopen(file_path,"r",encoding="utf-8")asf:data=json.load(f)df=pd.DataFrame(data)df.columns=["dm","mc","cjsj","cjjg","cjl","jyzd"]df["cjsj"]=pd.to_datetime(df["cjsj"])returndfdefread_realtime(dm):file_path=os.path.join(data_dir,"time","real",dm)withopen(file_path,"r",encoding="utf-8")asf:returnjson.load(f)

字段方面，dm是股票代码，mc是股票名称，cjsj是成交时间，cjjg是成交价格，cjl是成交量，cje是成交额，zf是涨跌幅，jyzd是交易方向（0中性/1买入/2卖出）。

系统的第一个分析模块是早盘量比信号。通过对比早盘30分钟成交量与历史均值，判断资金活跃度。

defcalc_morning_volume_ratio(dm):df_today=read_min_kline(dm,"min1")today_date=df_today["cjsj"].dt.date.iloc[-1]morning_start=datetime.strptime(f"{today_date}09:30:00","%Y-%m-%d %H:%M:%S")morning_end=datetime.strptime(f"{today_date}10:00:00","%Y-%m-%d %H:%M:%S")today_morning=df_today[(df_today["cjsj"]>=morning_start)&(df_today["cjsj"]<=morning_end)]today_morning_vol=today_morning["cjl"].sum()df_history=read_min_kline(dm,"min1")history_dates=df_history["cjsj"].dt.date.unique()history_morning_vols=[]fordinhistory_dates[-20:]:h_start=datetime.strptime(f"{d}09:30:00","%Y-%m-%d %H:%M:%S")h_end=datetime.strptime(f"{d}10:00:00","%Y-%m-%d %H:%M:%S")h_data=df_history[(df_history["cjsj"]>=h_start)&(df_history["cjsj"]<=h_end)]iflen(h_data)>0:history_morning_vols.append(h_data["cjl"].sum())avg_morning_vol=np.mean(history_morning_vols)ifhistory_morning_volselse1returntoday_morning_vol/avg_morning_volifavg_morning_vol>0else0

第二个分析模块是分时量价背离检测。在分时图上，如果价格创新高但成交量萎缩，说明上涨动能不足。

defdetect_volume_price_divergence(dm,lookback_minutes=30):df=read_min_kline(dm,"min1")iflen(df)<lookback_minutes:returnNonerecent=df.tail(lookback_minutes)price_max_idx=recent["cjjg"].idxmax()price_max_time=recent.loc[price_max_idx,"cjsj"]vol_at_price_max=recent.loc[price_max_idx,"cjl"]avg_vol=recent["cjl"].mean()ifprice_max_time<recent["cjsj"].iloc[-5]:ifvol_at_price_max<avg_vol*0.7:return{"type":"顶背离","time":price_max_time,"strength":avg_vol/(vol_at_price_max+1)}price_min_idx=recent["cjjg"].idxmin()price_min_time=recent.loc[price_min_idx,"cjsj"]vol_at_price_min=recent.loc[price_min_idx,"cjl"]ifprice_min_time<recent["cjsj"].iloc[-5]:ifvol_at_price_min<avg_vol*0.7:return{"type":"底背离","time":price_min_time,"strength":avg_vol/(vol_at_price_min+1)}returnNone

第三个分析模块是大单流向分析。通过逐笔成交数据统计大单的买卖方向，判断机构意图。

defanalyze_big_order_flow(dm,window_minutes=15,threshold=500000):df_tick=read_tick_data(dm)iflen(df_tick)==0:returnNonenow=df_tick["cjsj"].max()start_time=now-timedelta(minutes=window_minutes)recent=df_tick[df_tick["cjsj"]>=start_time]iflen(recent)==0:returnNonerecent["amount"]=recent["cjjg"]*recent["cjl"]big_buys=recent[(recent["amount"]>=threshold)&(recent["jyzd"]==1)]big_sells=recent[(recent["amount"]>=threshold)&(recent["jyzd"]==2)]buy_amount=big_buys["amount"].sum()sell_amount=big_sells["amount"].sum()net_flow=buy_amount-sell_amount total_amount=recent["amount"].sum()big_order_ratio=(buy_amount+sell_amount)/total_amountiftotal_amount>0else0return{"net_flow":net_flow,"buy_amount":buy_amount,"sell_amount":sell_amount,"big_order_ratio":big_order_ratio,"direction":"买入"ifnet_flow>0else"卖出"}

第四个分析模块是午后异动检测。检测午后开盘5分钟内的价格异动。

defdetect_afternoon_anomaly(dm):df=read_min_kline(dm,"min1")today_date=df["cjsj"].dt.date.iloc[-1]afternoon_start=datetime.strptime(f"{today_date}13:00:00","%Y-%m-%d %H:%M:%S")afternoon_check=datetime.strptime(f"{today_date}13:05:00","%Y-%m-%d %H:%M:%S")noon_end=datetime.strptime(f"{today_date}11:30:00","%Y-%m-%d %H:%M:%S")morning_close=df[df["cjsj"]<=noon_end]afternoon_open=df[(df["cjsj"]>=afternoon_start)&(df["cjsj"]<=afternoon_check)]iflen(morning_close)==0orlen(afternoon_open)==0:returnNonemorning_price=morning_close["cjjg"].iloc[-1]afternoon_price=afternoon_open["cjjg"].iloc[-1]change=(afternoon_price-morning_price)/morning_price*100return{"morning_close":morning_price,"afternoon_open":afternoon_price,"change_pct":change,"signal":"午后拉升"ifchange>1else"午后跳水"ifchange<-1else"平稳"}

把这些模块整合起来，形成完整的T+0信号系统。

defgenerate_t0_signals(dm):signals=[]vol_ratio=calc_morning_volume_ratio(dm)ifvol_ratio>2.5:signals.append({"signal":"早盘放量","score":30,"direction":"多"})elifvol_ratio<0.5:signals.append({"signal":"早盘缩量","score":20,"direction":"空"})divergence=detect_volume_price_divergence(dm)ifdivergence:ifdivergence["type"]=="顶背离":signals.append({"signal":"分时顶背离","score":25,"direction":"空"})else:signals.append({"signal":"分时底背离","score":25,"direction":"多"})flow=analyze_big_order_flow(dm)ifflowandflow["net_flow"]>10000000:signals.append({"signal":"大单净买入","score":30,"direction":"多"})elifflowandflow["net_flow"]<-10000000:signals.append({"signal":"大单净卖出","score":30,"direction":"空"})anomaly=detect_afternoon_anomaly(dm)ifanomalyandanomaly["signal"]=="午后拉升":signals.append({"signal":"午后异动拉升","score":25,"direction":"多"})elifanomalyandanomaly["signal"]=="午后跳水":signals.append({"signal":"午后异动跳水","score":25,"direction":"空"})total_score=sum(s["score"]forsinsignalsifs["direction"]=="多")-\sum(s["score"]forsinsignalsifs["direction"]=="空")return{"dm":dm,"signals":signals,"total_score":total_score,"action":"买入"iftotal_score>=50else"卖出"iftotal_score<=-50else"观望"}

实际运行下来，这个系统的信号准确率大约在60%左右。早盘放量和大单净买入这两个信号的预测能力最强，午后异动信号次之，分时背离信号相对较弱。

在使用过程中有几点经验。第一，T+0交易对数据的时效性要求很高，本地数据引擎的3秒更新频率能满足需求。第二，不要依赖单一信号，多个信号同时确认时胜率更高。第三，T+0交易的利润薄、频率高，交易成本的影响很大，一定要选择佣金低的券商。

做日内T+0，数据就是你的眼睛。用好分时数据，你就能看到别人看不到的市场细节。

我用的分时数据来自本地数据引擎，分钟级K线和逐笔成交数据接口完整，做日内分析非常方便。感兴趣的朋友可以参考这个思路，结合自己的交易风格来调整信号模型。

接口说明：

1. time/history/trade/{股票代码}/{级别} - 分钟级历史K线
   本地路径：数据存放目录/time/history/trade/{dm}/{min1|min5|min15}
   主要字段：成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、涨跌幅(zf)

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

3. time/real/trace/bigdeal/{股票代码} - 大单成交数据
   本地路径：数据存放目录/time/real/trace/bigdeal/{dm}
   主要字段：成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、交易方向(jyzd)

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

资料参考：ig50