马年开工第一天AI板块早盘集体高开我打开行情软件扫了一眼发现一个有意思的现象几个被市场当作AI风向标的龙头冲得很猛但板块中不少后排标的只是勉强翻红甚至还有绿盘躺着的。当时心里就一个判断这种分化状态比普涨更值得记录。因为普涨行情里大家一齐涨看不出资金偏好一旦出现分化说明钱开始有选择地流动了。要把这种分化捕捉下来靠日线数据太迟靠手动切换窗口盯盘又太累。所以我用TickDB实时行情API把整个AI观察池的分时数据拉下来用Python做了几层过滤和量化刻画把“分化”这个词从感觉变成了一个可以连续观察、可以设置预警的数值指标。这篇文章就把这整套思路完整复盘一遍为什么需要用实时数据、TickDB怎么接、分化指标怎么设计、Python源码怎么落以及我在马年开工这几天的实际运行中观察到的几个信号和数据坑。1. 为什么捕捉分化行情必须用实时API以及TickDB的选型逻辑1.1 行情分化到底是什么信号先明确一个概念我说的“分化”不是简单指某只股票涨、某只跌而是指同一主题、同一板块内部的涨跌幅、量能、资金行为出现明显不一致。AI板块尤其适合观察这种分化因为它在A股更像一个主题群而不是一个传统行业板块。算力、模型、应用、数据、终端硬件每个细分方向都有自己的驱动逻辑资金在这些方向之间切换的速度非常快。分化的观察价值在于它往往发生在板块整体行情的转折点之前。比如板块普涨了一波之后如果龙头继续走强而多数个股开始走弱说明行情进入缩圈阶段反过来如果龙头滞涨而低位票放量补涨说明行情可能进入扩散阶段。这两种状态对应的操作思路完全不同但用日线数据去看往往要等收盘后才知道发生了什么而盘中资金切换的痕迹已经被洗掉了。只有实时行情API能把这些痕迹留下来。TickDB这类面向金融场景的时序数据库正好解决这个问题它既能存tick级或秒级的行情快照又支持按时间窗口做聚合分析不用自己再维护一套存储和查询引擎。1.2 为什么选TickDB而不是自己爬数据说实话能做实时行情数据采集的方式不少但我这些年试下来的感受是自己写爬虫去轮询行情接口短期看省了成本长期全是坑。主要体现在几个方面对比维度自建爬虫/轮询方案TickDB实时行情API方案数据完整性容易漏tick断线后补数据麻烦服务端落库断线可拉取区间数据补齐查询效率数据散落分析时临时过滤列式存储时间分区分钟级聚合秒出字段规范化各接口字段单位不统一要自己清洗字段标准单位明确稳定性自己维护接口一变就崩成熟服务有鉴权和限流机制时间成本第一版能跑后期维护量很大上手成本低专注策略本身这里不是吹TickDB多神而是说它解决的核心问题数据基础设施不用自己造了。实时行情分析难的不是算指标而是拿到干净、连续、带正确时间戳的数据。TickDB把数据接入和查询这两层做好了我就能把精力花在指标逻辑上。1.3 数据维度规划除了价格还需要什么在写代码之前我先列清楚要拉哪些字段。不是为了多拉数据显得专业而是每个字段都对应一个分析用途分钟级OHLCV开高低收和成交量计算涨跌幅、量比、换手率的底层输入。成交额单纯成交量在不同价位意义不同成交额能反映真实资金规模。最新价相对当日均线的偏离度衡量个股是否处于过热或超卖状态。时间戳这是最容易被忽略但最关键的字段所有聚合都依赖准确的时间对齐。很多人在首次接入实时API时只顾着拉价格忽略时间戳和字段单位校验结果后续指标全错这是我在实战中踩过最深的一个坑后面专门讲。2. 搭建AI板块观察池以及TickDB数据接入的完整流程2.1 观察池怎么圈定AI不是一个板块是一个主题群AI板块在A股不像银行、白酒那样有标准的申万行业分类它横跨软件服务、通信设备、电子元件、计算机设备等多个行业。所以我建观察池的逻辑是先圈出AI产业链上市场公认度比较高的几个细分方向再从每个方向里挑流动性好、辨识度高的标的作为代表。我目前维护的观察池大致分为四簇算力基建GPU/AI芯片、服务器、光模块、液冷温控等硬件方向。模型与应用大模型公司、AIGC应用、办公软件、营销工具等偏软件方向。数据要素与软件数据服务、数据库、国产软件、云计算等方向。终端硬件AI手机、AI PC、智能驾驶、机器人等方向。每簇里我放了8到12只标的整个观察池约40只。这个数量是经过权衡的太少缺乏统计意义太多会导致API请求频率被限流而且盯不过来。每个季度我会根据市场关注度调整一次成分剔掉流动性差的加入新走出来的标的。2.2 TickDB的接入准备工作使用TickDB之前需要先准备好三样东西账号Token、API访问地址、以及本机的Python环境。我用的Python版本是3.10以上依赖库包括requests、pandas、numpy、python-dateutil。安装依赖很简单pip install requests pandas numpy python-dateutilTickDB的实时行情接口走的是RESTful风格默认返回JSON格式。先做一次最基础的连通性测试确认Token和网络都正常import requests import os API_BASE https://api.tickdb.example.com TOKEN os.environ.get(TICKDB_TOKEN, your_token_here) headers {Authorization: fBearer {TOKEN}} def test_connection(): resp requests.get( f{API_BASE}/v1/time, headersheaders, timeout5 ) resp.raise_for_status() print(TickDB server time:, resp.json()[server_time]) if __name__ __main__: test_connection()建议用环境变量存Token不要硬编码到源码里避免不小心把密钥传到公开仓库。这一步虽然是小习惯但很重要。2.3 核心接入类封装拉取逻辑接下来我封装一个TickDBClient类把拉取最新报价、拉取分钟K线、以及后续的分化分析基础操作统一起来。这样主程序里不用到处写重复的请求代码。class TickDBClient: def __init__(self, base_url, token): self.base_url base_url self.headers {Authorization: fBearer {token}} def _get(self, endpoint, paramsNone): url f{self.base_url}{endpoint} resp requests.get(url, headersself.headers, paramsparams, timeout10) resp.raise_for_status() return resp.json() def get_latest_quotes(self, symbols): params {symbols: ,.join(symbols)} data self._get(/v1/quotes, paramsparams) rows [] for item in data[quotes]: rows.append({ symbol: item[symbol], price: item[last_price], open: item[open], high: item[high], low: item[low], volume: item[volume], amount: item[amount], timestamp: pd.to_datetime(item[timestamp], unitms, utcTrue).tz_convert(Asia/Shanghai), }) return pd.DataFrame(rows) def get_minute_bars(self, symbol, startNone, endNone): params {symbol: symbol, period: 1min} if start: params[start] start if end: params[end] end data self._get(/v1/bars, paramsparams) rows [] for bar in data[bars]: rows.append({ symbol: symbol, datetime: pd.to_datetime(bar[timestamp], unitms, utcTrue).tz_convert(Asia/Shanghai), open: bar[open], high: bar[high], low: bar[low], close: bar[close], volume: bar[volume], amount: bar[amount], }) return pd.DataFrame(rows)这个类有两个关键设计一是把时区统一转成Asia/Shanghai避免后续做时间窗口聚合时出现UTC和本地时间混用的诡异问题二是把每条记录里的timestamp从毫秒时间戳转成真正的datetime类型这样pandas做groupby和resample时才不会瞎报警告。2.4 连续拉数时的限流与重试策略实时行情API通常会有请求频率限制。TickDB默认是每秒最多60次请求如果观察池有40只标的一次性并发40个request没问题但如果是盘中持续轮询就要设计合理的频率。我的策略是每15秒更新一次最新报价每分钟更新一次分钟K线。这个频率对捕捉板块分化已经足够因为分化是相对缓慢的累积过程不是毫秒级的闪电行为。频率设太高反而容易触发限流。为了避免偶发的网络抖动导致数据缺失我建议在请求层加一个简单的重试机制import time def request_with_retry(func, *args, retries3, backoff2, **kwargs): for attempt in range(retries): try: return func(*args, **kwargs) except requests.exceptions.RequestException as e: if attempt retries - 1: raise wait_time backoff * (2 ** attempt) print(fRequest failed, retrying in {wait_time}s...) time.sleep(wait_time)在实际盘中运行时偶发一次超时是正常的重试两三次基本能解决。如果连续重试还失败就检查是不是Token过期了或者触发了限流而不是无限重试。3. 核心指标设计用Python把“分化”变成可观测的数值3.1 分化度板块内部涨幅的离散度要把“分化”量化最直接的办法是计算板块内部涨幅的标准差。标准差越大说明个股之间的涨跌幅差异越明显分化越剧烈。我在实盘中用的是这样一个公式板块平均涨幅 mean(个股涨跌幅) 涨幅标准差 std(个股涨跌幅) 分化度 涨幅标准差 / max(abs(板块平均涨幅), 0.01)分母加上一个最小值保护是为了防止板块平均涨幅接近0时计算出极端的无效值。这样得到的比值是一个无单位的相对指标能够在不同交易日之间做横向对比。对应代码如下def compute_divergence(quotes_df): if quotes_df.empty: return None avg_change quotes_df[change_pct].mean() std_change quotes_df[change_pct].std() if pd.isna(std_change): return None denominator max(abs(avg_change), 0.01) return std_change / denominator这里的change_pct需要根据拉取时的最新价和昨收价计算quotes_df[change_pct] (quotes_df[price] - quotes_df[pre_close]) / quotes_df[pre_close] * 100我在TickDB的quotes接口里取pre_close字段没有的话就取前一个交易日的收盘价。字段缺失时直接报错不要静默用0填充否则算出来的涨幅全错。3.2 强弱分层给每只标的算相对板块强度只看整体离散度是不够的还需要知道是谁在涨、谁在跌。我引入一个相对强度z-score的概念z_score (个股涨跌幅 - 板块平均涨跌幅) / 板块涨跌幅标准差z_score大于0说明个股强于板块小于0说明弱于板块。我按z_score把观察池分成三层强势层z_score ≥ 0.8大概率是带动板块情绪的核心标的。中性层-0.8 z_score 0.8跟随板块波动方向不明确。弱势层z_score ≤ -0.8资金明显流出的方向。层次划分之后再做两件事统计每一层的标的数量变化以及计算强势层平均涨幅和弱势层平均涨幅的差值。这个差值我习惯叫“板块剪刀差”剪刀差扩大说明龙头信仰强化缩小说明后排补涨或者龙头走弱。代码如下def assign_strength_tier(quotes_df, up_thresh0.8, down_thresh-0.8): df quotes_df.copy() sector_mean df[change_pct].mean() sector_std df[change_pct].std() if sector_std 0 or pd.isna(sector_std): df[z_score] 0 else: df[z_score] (df[change_pct] - sector_mean) / sector_std conditions [ df[z_score] up_thresh, df[z_score] down_thresh, ] choices [strong, weak] df[tier] np.select(conditions, choices, defaultneutral) return df3.3 量能配合找出“涨幅落后但资金进场”的标的分化除了看价格更要看量。有的标的涨幅一般但成交额连续放大这往往意味着有资金在悄悄建仓后续可能补涨反之有的标的涨幅靠前但成交额萎缩可能只是缩量拉升持续性存疑。我用的量能指标是分钟级成交额的移动平均比量能放大倍数 最近5分钟平均成交额 / 全天平均分钟成交额这个指标超过1.5就说明近5分钟资金活跃度显著高于全天平均水平。我会特别关注那些z_score在-0.8到0之间、但量能放大倍数大于1.5的标的它们处于“价格没动但资金在动”的状态。实现上非常简单def compute_volume_pulse(df, window5): df df.sort_values(datetime) df[avg_amount_5min] df[amount].rolling(window).mean() df[avg_amount_all] df[amount].expanding().mean() df[volume_pulse] df[avg_amount_5min] / df[avg_amount_all] return df这里的关键是分清楚expanding和rolling的语义expanding计算的是从开盘到当前时刻的累积均值rolling计算的只是最近5分钟的均值两者对比才能看出当前量能相对全天平均水平的异动。3.4 把整个流程串起来主程序源码上面这些函数拆开来都不复杂但组合起来就需要一条流水线。下面是我盘中定时运行的主程序简化版import pandas as pd import numpy as np from datetime import datetime, timedelta WATCH_POOL [000001, 000002, 300001, ...] # 你的AI观察池 POOL_NAME ai_sector def run_pipeline(client, symbols): # 1. 拉最新报价 quotes client.get_latest_quotes(symbols) quotes[change_pct] (quotes[price] - quotes[pre_close]) / quotes[pre_close] * 100 # 2. 计算分化度和强弱分层 divergence compute_divergence(quotes) layered assign_strength_tier(quotes) # 3. 对每个标的拉最近1分钟线计算量能脉冲 pulse_rows [] for sym in symbols: end datetime.now() start end - timedelta(minutes10) bars client.get_minute_bars(sym, startstart.isoformat(), endend.isoformat()) if not bars.empty: bars compute_volume_pulse(bars) latest bars.iloc[-1] pulse_rows.append({ symbol: sym, volume_pulse: latest[volume_pulse], }) pulse_df pd.DataFrame(pulse_rows) result layered.merge(pulse_df, onsymbol, howleft) # 4. 标记潜在补涨标的 result[potential_pickup] ( (result[z_score].between(-0.8, 0)) (result[volume_pulse] 1.5) ) return result, divergence if __name__ __main__: client TickDBClient(API_BASE, TOKEN) result, div run_pipeline(client, WATCH_POOL) print(fSector divergence: {div:.3f}) print(result[result[potential_pickup]][[symbol, change_pct, z_score, volume_pulse]])这套主程序跑起来之后我每隔15分钟手动看一眼输出或者直接接到告警服务里。盘中自动轮询之后把结果写成CSV收盘后还能做复盘非常方便。4. 马年开工实测三个典型分化信号和两个数据坑4.1 信号一开盘30分钟分化度快速拉升马年开工第一个交易日早上9:30到10:00这段时间分化度指标从开盘的0.52附近一路拉升到1.34这个幅度在过去的观察里属于非常剧烈的水平。再看强弱分层结果强势层集中在算力硬件方向弱势层基本被应用软件方向包揽。这说明开盘后半小时资金的态度非常鲜明先抢算力不碰应用。如果你只看板块指数很容易觉得AI整体很强但拉开内部结构会发现所谓的“AI行情”其实是算力方向的单兵突进。这种认知差异直接影响选股方向做板块beta的人会失望做结构性alpha的人反而能找到机会。4.2 信号二午盘前出现“隐性蓄水”标的第二个典型信号出现在开盘第二天。那天上午收盘前分化度已经从高点回落到0.8附近按常理说明行情趋于一致。但我用量能脉冲过滤了一遍发现有3只标的的z_score仍然在负值区间但量能放大倍数同时超过了2.0。这几只标的分属于终端硬件和数据软件方向价格没怎么动但最近5分钟的成交额大幅放大。我后来复盘时发现这种“价格不动但量先行”的状态往往是大资金建仓的初期特征对次日走势有较强的提示意义。如果没有量能脉冲这个过滤条件这几只标的基本会被当成弱势股直接忽略。4.3 数据坑一价格字段被放大10倍的排查过程马年开工第三天我突然发现有几只标的的涨跌幅显示异常离谱涨幅超过20%。第一反应是TickDB的数据错了但去行情软件上核对价格确实不一样。排查过程花了大半个小时。顺着代码一步一步查最后发现问题出在字段单位上TickDB的quotes接口里部分合约的price字段返回的是整数形式比如1250表示12.50元而部分合约直接返回浮点数。我统一按浮点数处理结果把前者的价格放大了100倍。这个坑的本质是接口字段在不同品种之间不一致没有任何报错只会让后续指标全部失真。修复方法很简单接入数据后先做一次单位校验把所有price/volume/amount字段统一除以对应的倍数字段或者按symbol映射规则处理。更重要的是在代码里加一条断言确保所有价格不超过该标的过去5日的最高价乘以1.2否则就报警提示而不是静默接受。def validate_price(row, max_history_price): if row[price] max_history_price * 1.2 or row[price] max_history_price * 0.5: raise ValueError(fPrice anomaly: {row[symbol]} {row[price]})4.4 数据坑二断线重连后分钟线出现重复段第二次踩坑是TickDB的WebSocket断线导致的。盘中有一段时间网络不稳定WebSocket自动重连之后TickDB为了保证数据完整会把断线期间的分钟线重新补推一次。但我的接收端没有做好去重导致内存里的DataFrame出现了两个完全相同的分钟段。重复数据最要命的地方在于做expanding均值计算时重复的5分钟数据会被当成10分钟计算量能脉冲指标直接偏差50%以上整个潜在补涨筛选结果全乱了。解决方法是给DataFrame加上基于symboldatetime的唯一性约束发现重复就直接去重并且保留最后一次接收到的数据def deduplicate_candles(df): df df.sort_values(datetime) df df.drop_duplicates(subset[symbol, datetime], keeplast) return df.reset_index(dropTrue)4.5 强校验和第三方行情软件做分钟K线对账经历过这两个坑之后我现在每次切换到新的数据源都会先做一个强制性的对账流程随机抽5只观察池标的拿TickDB拉出来的分钟K线和第三方行情软件的同分钟K线做逐项对比核对开盘价、最高价、最低价、收盘价、成交量和成交额。六个字段只要有一个对不上就停下来找原因不能带病运行。这个对账流程写成一个独立的脚本每次接入新接口或新合约时跑一遍能省去后面排查的大把时间。5. 从捕捉分化到辅助筛选这套分析框架还能怎么扩展5.1 自选股预警把指标阈值接到通知服务整套指标跑通之后最直接的应用是做实时预警。我目前设置了三个条件命中任意一个就推送提醒到手机分化度大于1.2板块内部剧烈分化需要警惕行情可能转向。分化度小于0.3板块走势高度一致可能是普涨或普跌注意观察方向。出现z_score低于-0.5但量能脉冲大于1.8的标的潜在的补涨候选。预警实现不复杂在run_pipeline的输出结果上做条件判断命中后调用微信或飞书机器人接口推送消息。我习惯只在条件从“不满足”变“满足”的瞬间推一次避免震荡行情下频繁轰炸。5.2 结合持仓做群体偏离检查如果你手里有持仓这个框架还能用来检查自己的持仓是否跑赢了所属细分方向。具体做法是把持仓标的分到对应的AI细分簇里计算该标的价格在自己细分簇内的排名百分位。如果连续两天排在最后20%说明它的相对强度一直在衰减需要重新审视持有逻辑是否还成立。这个用法本质上把“板块分化”从市场观察工具扩展成了持仓诊断工具不需要额外拉数据复用同一套观察池和实时行情API就能实现。5.3 与量化过滤条件叠加缩小人工复核范围对做中低频波段的人来说这套实时分析框架也可以做成一个初筛工具。TickDB提供的分钟级数据我可以在此基础上叠加量比、换手率、以及一些常用的量能指标做多条件过滤把几千只股票迅速缩减到十几只候选再靠人工做基本面和消息面复核。比如我自己常用的一组叠加条件AI观察池内。分化度处于0.6到1.0之间既不过热也不冷淡。个股z_score在0到0.8之间处于中性偏强尚未过度透支。量能放大倍数大于1.3有资金关注但不疯狂。最新价偏离20日均线小于5%排除已经连续拉升的高位标的。这组条件每次筛完基本只剩5到8只票人工复核成本极低。5.4 这套方案的边界什么场景下不适用最后说一些操作层面的体会和边界限制。TickDB这类实时行情API的强项是当前行情数据的拉取和分钟级聚合分析但如果你要做的是基于历史多年日线数据的策略回测它不是最合适的工具历史数据深度和应用生态上还需要搭配传统数据库或专门的回测平台。另外AI板块观察池有一个时效性问题。AI产业链变化很快今天代表性的标的可能三个月后就不再是市场主线。所以观察池必须定期更新否则分化指标会被一批失去关注度的老标的污染。我自己的节奏是每个季度末做一次全量复盘剔除连续一个月日均成交额排在观察池后20%的标的同时引入新活跃品种。还有一点要注意的是开盘和尾盘的极端分钟容易造成指标失真。开盘前几分钟集合竞价释放的波动非常大尾盘最后几分钟又经常出现资金抢跑。我在实际使用时会过滤掉开盘前5分钟和收盘前5分钟的数据再计算指标否则分化度和量能脉冲的噪音会非常明显。写在这篇实战分享的后面我自己的体会是实时行情API本身并不复杂真正决定这套分析有没有用的是对板块边界的理解和对指标含义的把控。TickDB把数据侧的活干了Python把计算侧的活干了但“什么算分化”“分化到什么程度需要警惕”“哪些细节需要过滤”这些判断还是要靠你对自己所跟踪板块的熟悉程度来定。如果你正在做类似的板块行情观察建议先小范围试点挑5到8只你最有把握的标的跑通链路再逐步扩到完整观察池。数据链路稳定了再上预警和自动筛选避免一上来就把一套复杂系统跑崩。马年开工先把工具磨利后面盯盘会轻松很多。