企业官网建设流程全解析

分时成交实战：本地数据引擎如何重构日内趋势跟踪体系

做日内交易的人都明白，分时成交数据是判断短期走势最直接的依据。每一分钟的成交量和平均价格变化，能反映出资金的进出节奏和市场情绪的演变。但想做好日内趋势跟踪，数据的实时性和连续性是关键。我之前用在线API做日内分析，数据经常断档，要么某几分钟的数据缺失，要么延迟太高跟不上行情变化。后来换成本地落盘方案，才算真正把日内分析的流程跑通了。今天想聊聊用ig50本地数据引擎做分时成交分析的实战经验。

在线API做日内分析的痛点很明显。免费的在线接口大多只能提供当天的分时数据，而且经常出现数据缺失的情况，比如上午10:30到10:35这五分钟的数据突然没了，导致趋势判断出现断层。付费接口的数据质量好一些，但频次限制摆在那，想每分钟刷新一次全市场的分时数据，根本做不到。更要命的是延迟，在线接口从数据产生到传到你手里，少说也有几十毫秒，遇到行情波动大的时候，延迟甚至能到几百毫秒。我之前写过一个日内动量策略，用在线API跑，经常因为数据延迟错过最佳入场时机。

后来换成ig50本地数据引擎，才发现日内分析原来可以这么顺畅。当天分时成交数据在time/real/trace/timedeal/{股票代码}路径下，记录了盘中以分钟为单位统计的成交总量和平均价格，每2分钟更新一次，10秒左右落盘。程序直接读本地文件，没有网络延迟，没有请求频次限制，想怎么分析就怎么分析。

先看分时成交数据的读取和基本分析，下面这段代码是我盘中日内监控的核心逻辑：

importjsonimportpandasaspdimportnumpyasnpfrompathlibimportPathfromdatetimeimportdatetime DATA_DIR=Path('/ig50-data')defload_timedeal(code):"""读取单只股票的当天分时成交数据"""path=DATA_DIR/'time'/'real'/'trace'/'timedeal'/codeifnotpath.exists():returnNonewithopen(path,'r',encoding='utf-8')asf:data=json.load(f)returndata# 读取股票列表withopen(DATA_DIR/'base'/'gplist','r',encoding='utf-8')asf:stock_list=json.load(f)df_stocks=pd.DataFrame(stock_list)# 过滤掉ST和新股df_pool=df_stocks[(df_stocks['isSt']==0)&(df_stocks['isNew']==0)]# 读取单只股票的分时成交数据（以贵州茅台为例）code='600519'timedeal_data=load_timedeal(code)iftimedeal_data:df_td=pd.DataFrame(timedeal_data)# 字段：数据更新时间、分时时间、成交量（股）、成交价df_td['分时时间']=pd.to_datetime(df_td['分时时间'],format='%H:%M:%S')df_td['成交量']=pd.to_numeric(df_td['成交量'],errors='coerce')df_td['成交价']=pd.to_numeric(df_td['成交价'],errors='coerce')df_td=df_td.set_index('分时时间').sort_index()print(f'==={code}分时成交数据 ===')print(f'数据时间范围：{df_td.index[0].time()}-{df_td.index[-1].time()}')print(f'总成交分钟数：{len(df_td)}')print(f'总成交量：{df_td["成交量"].sum()/10000:.2f}万股')print(f'平均每分钟成交量：{df_td["成交量"].mean()/10000:.2f}万股')# 计算每分钟成交额df_td['成交额']=df_td['成交量']*df_td['成交价']# 前20分钟和最近20分钟对比first_20=df_td.head(20)last_20=df_td.tail(20)print(f'\n前20分钟平均成交额：{first_20["成交额"].mean()/10000:.2f}万元')print(f'最近20分钟平均成交额：{last_20["成交额"].mean()/10000:.2f}万元')print(f'放量倍数：{last_20["成交额"].mean()/first_20["成交额"].mean():.2f}')# 打印最近10分钟的分时数据print('\n=== 最近10分钟分时数据 ===')print(df_td[['成交价','成交量']].tail(10))

这段代码最关键的地方在于，读取分时成交数据是纯本地文件IO，几乎没有延迟。换成在线API，同样的操作至少要等几十毫秒。

分时成交数据的价值远不止看每分钟的成交量和价格，还能用来做日内趋势分析。下面这段代码是我用来识别日内趋势反转的脚本：

defanalyze_intraday_trend(code):"""分析日内趋势，识别趋势反转信号"""timedeal_data=load_timedeal(code)ifnottimedeal_data:returnNonedf_td=pd.DataFrame(timedeal_data)df_td['分时时间']=pd.to_datetime(df_td['分时时间'],format='%H:%M:%S')df_td['成交量']=pd.to_numeric(df_td['成交量'],errors='coerce')df_td['成交价']=pd.to_numeric(df_td['成交价'],errors='coerce')df_td=df_td.set_index('分时时间').sort_index()# 计算每分钟成交额df_td['成交额']=df_td['成交量']*df_td['成交价']# 计算均线（5分钟和20分钟）df_td['MA5']=df_td['成交价'].rolling(5).mean()df_td['MA20']=df_td['成交价'].rolling(20).mean()# 计算成交量均线df_td['VOL_MA5']=df_td['成交量'].rolling(5).mean()df_td['VOL_MA20']=df_td['成交量'].rolling(20).mean()# 计算价格变化率df_td['price_change']=df_td['成交价'].pct_change()# 计算成交量变化率df_td['vol_change']=df_td['成交量'].pct_change()# 识别趋势信号signals=[]# 金叉信号：MA5上穿MA20golden_cross=(df_td['MA5']>df_td['MA20'])&(df_td['MA5'].shift(1)<=df_td['MA20'].shift(1))ifgolden_cross.iloc[-5:].any():last_cross=df_td[golden_cross].index[-1]signals.append(f'金叉({last_cross.time()})')# 死叉信号：MA5下穿MA20death_cross=(df_td['MA5']<df_td['MA20'])&(df_td['MA5'].shift(1)>=df_td['MA20'].shift(1))ifdeath_cross.iloc[-5:].any():last_cross=df_td[death_cross].index[-1]signals.append(f'死叉({last_cross.time()})')# 放量信号：成交量突破20日均线1.5倍volume_surge=df_td['成交量']>df_td['VOL_MA20']*1.5ifvolume_surge.iloc[-3:].any():signals.append('放量')# 缩量信号：成交量低于20日均线0.5倍volume_drop=df_td['成交量']<df_td['VOL_MA20']*0.5ifvolume_drop.iloc[-3:].any():signals.append('缩量')# 强势上涨：连续3分钟价格上涨且成交量递增strong_up=(df_td['price_change']>0).rolling(3).sum()==3ifstrong_up.iloc[-1]:signals.append('强势上涨')# 强势下跌：连续3分钟价格下跌且成交量递增strong_down=(df_td['price_change']<0).rolling(3).sum()==3ifstrong_down.iloc[-1]:signals.append('强势下跌')# 计算关键统计数据total_minutes=len(df_td)up_minutes=(df_td['price_change']>0).sum()down_minutes=(df_td['price_change']<0).sum()result={'代码':code,'当前价格':df_td['成交价'].iloc[-1]iflen(df_td)>0else0,'开盘价':df_td['成交价'].iloc[0]iflen(df_td)>0else0,'最高价':df_td['成交价'].max()iflen(df_td)>0else0,'最低价':df_td['成交价'].min()iflen(df_td)>0else0,'MA5':df_td['MA5'].iloc[-1]iflen(df_td)>5else0,'MA20':df_td['MA20'].iloc[-1]iflen(df_td)>20else0,'上涨分钟数':up_minutes,'下跌分钟数':down_minutes,'上涨占比':up_minutes/total_minutesiftotal_minutes>0else0,'平均每分钟成交额':df_td['成交额'].mean()/10000iflen(df_td)>0else0,'最新每分钟成交额':df_td['成交额'].iloc[-1]/10000iflen(df_td)>0else0,'信号':', '.join(signals)ifsignalselse'无',}returnresult# 分析单只股票的日内趋势result=analyze_intraday_trend('600519')ifresult:print('\n=== 日内趋势分析结果 ===')fork,vinresult.items():ifisinstance(v,float):print(f'{k}:{v:.4f}')else:print(f'{k}:{v}')

这段代码通过计算均线金叉死叉、成交量变化、价格涨跌等指标，能帮助识别日内趋势反转信号。我之前用在线API做同样的分析，因为数据延迟和不全，经常误判；现在本地数据完整实时，分析结果靠谱多了。

把分时成交和实时行情结合起来，就能构建一个全市场日内趋势监控体系。下面这段代码是我实际在用的盘中日内扫描脚本：

defscan_intraday_trend(top_n=10):"""全市场日内趋势扫描，找出趋势异动股票"""withopen(DATA_DIR/'base'/'gplist','r',encoding='utf-8')asf:stock_list=json.load(f)df_stocks=pd.DataFrame(stock_list)df_pool=df_stocks[(df_stocks['isSt']==0)&(df_stocks['isNew']==0)]results=[]forcodeindf_pool['dm']:result=analyze_intraday_trend(code)ifresult:results.append(result)df_result=pd.DataFrame(results)# 筛选趋势异动股票# 条件：上涨占比>60% 或 有金叉信号 或 放量上涨anomaly_mask=(df_result['上涨占比']>0.6)|\(df_result['信号'].str.contains('金叉'))|\(df_result['信号'].str.contains('放量'))anomalies=df_result[anomaly_mask]anomalies=anomalies.sort_values('上涨占比',ascending=False).head(top_n)print(f'\n==={datetime.now().strftime("%H:%M:%S")}日内趋势扫描 ===')print(f'异动股票数量：{len(anomalies)}')print(anomalies[['代码','当前价格','MA5','MA20','上涨占比','平均每分钟成交额','最新每分钟成交额','信号']])returnanomalies# 运行全市场日内趋势扫描scan_intraday_trend(top_n=15)

用本地数据做日内分析，最大的体感差异是"连续性"和"实时性"。以前用在线API，数据经常出现断档，比如某几分钟的数据缺失，导致趋势判断出现断层；现在本地数据完整连续，不会出现这种问题。而且数据更新后10秒就能读到，几乎是实时的。

还有个容易被忽视的点，分时成交数据的时间精度。在线API经常出现时间戳不准的情况，比如10:30的数据被标记成10:31，这会严重影响趋势判断。本地数据引擎后台持续更新，时间戳准确，不会出现这种问题。

数据名称：沪深京A股-股票交易数据-股票行情-当天分时成交
本地路径：数据存放目录/time/real/trace/timedeal/{股票代码}
描述：根据《股票列表》得到的股票代码获取当天分时成交数据（盘中以分钟为单位统计的成交总量以及平均价格），按时间升序
更新频率：交易时间段每2分钟。完成更新耗时：约10秒。数据格式：[{},…]
关键字：[数据更新时间, 分时时间, 成交量（股）, 成交价]

资料参考ig50.com/index_realtime.html?maodian=rt-trace-timedeal-title，如有侵犯网站权利，请联系我撤回。