马年开工第一天我盯着屏幕上的AI板块分时图脑子里冒出来的第一个念头不是“今天涨了还是跌了”而是“这个板块内部已经分裂成两个阵营了”。一边是算力、光模块这些硬科技标的持续拉升另一边是部分应用端个股冲高回落这种分化如果靠肉眼盯盘根本看不过来更别说还要在盘中快速做出反应。这篇文章要聊的就是我用TickDB实时行情API加Python搭的一套A股AI板块分化度监控小工具。它能做到开盘后每分钟自动拉取AI板块主要成分股的实时报价计算板块整体涨幅、个股离散度、强势股与弱势股占比然后把分化信号直接标记出来。整个过程用Python实现核心代码量不算大适合对量化监控、行情数据接口感兴趣的开发者和交易爱好者参考。1. 项目整体设计与思路拆解1.1 为什么盯“分化行情”而不是单纯看涨跌很多人看板块只看指数涨跌幅比如AI板块今天涨了1.5%就觉得整个板块都不错。但实际情况远没有这么简单。板块指数是一个平均值它掩盖了个股之间的巨大差异。同样是AI板块可能一边是某算力龙头盘中涨到7%另一边是某AI应用公司跌了3%这种情况下“板块涨1.5%”这个数字几乎没有操作参考价值。我想要的不是“板块涨了没有”而是“板块内部到底在发生什么”。具体来说有三个问题第一领涨的个股占比有多大是普遍上涨还是少数个股撑着指数第二个股涨幅的离散程度有多高是步调一致还是各走各的第三强势股和弱势股在板块内的分布有没有规律比如是不是集中在某个细分赛道。这三个问题组合起来就是“分化行情”的量化刻画。当分化度快速升高时说明市场对AI板块内部各细分方向的预期出现了明显分歧这时候追涨杀跌都很容易做反反而是观察和等待的好时机。反过来当分化度收窄时说明板块可能正在酝酿一致性行情。1.2 为什么选择TickDB实时行情API做盘中实时监控数据源是第一道坎。A股行情数据虽然有免费渠道但多数在实时性、稳定性、字段完整度上都有明显短板。我选择TickDB实时行情API主要看中三个点。第一是实时性。TickDB的行情接口刷新频率可以做到秒级而且支持WebSocket和HTTP轮询两种方式。我做的是分钟级监控用HTTP轮询就够不需要维护长连接代码复杂度低很多。但如果你想做秒级甚至tick级策略它也能撑得住。第二是数据覆盖。AI板块没有一个绝对权威的成分股名单不同平台的划分逻辑差异很大。TickDB提供板块成分股接口按行业标签和概念标签做了区分我可以直接拉取AI概念相关的成分股列表省去自己维护名单的麻烦。这个字段本地券商软件通常不开放能通过API直接拿到的服务商也不多。第三是API设计干净。TickDB的RESTful接口返回标准的JSON结构字段命名直观没有一堆冗余嵌套Python的requests库直接就能解析几乎不需要写额外的数据清洗层。这一点对快速搭建原型来说价值很大。1.3 整体技术方案和流程整个项目的流程可以梳理成四个环节拉取AI板块成分股、获取成分股实时行情、计算分化指标、输出监控信号。数据流向是TickDB行情服务器 → Python客户端程序 → 内存中的DataFrame → 指标计算模块 → 控制台输出或图表展示。我选择用Python的pandas做数据承载和计算原因是它处理表格型数据太方便了分组、聚合、统计描述都是一行代码的事情。在计算环节我设计了三个核心指标板块加权平均涨幅、涨幅标准差和离散系数。后面我会详细解释这三个指标各自的含义和用法。工具选型上除了requests和pandas还用了schedule库做定时调度matplotlib做收盘后的可视化复盘。整个依赖栈非常轻Python 3.9以上都可以跑。注意这个项目的定位是行情监控和数据分析不构成任何投资建议。量化指标只是辅助观察市场的工具最终决策需要你自己判断。2. TickDB API核心概念与接入准备2.1 TickDB是什么TickDB本质上是一个面向金融行情场景的时序数据服务它解决的是“高频、多标的、长周期”行情数据的存储和读取问题。传统关系型数据库在处理几千只股票的实时报价时很容易成为瓶颈TickDB底层用列式存储和分区索引按时间戳组织数据查询性能好很多。当然普通投资者不需要关心底层存储引擎的细节我们直接从API层面使用它就行。TickDB对外提供两种数据接口第一种是实时快照接口返回某个时间点上所有请求标的的最新行情第二种是历史K线接口返回过去某个时间窗口的分钟线或日线。这个项目主要用实时快照接口历史K线接口可以做复盘验证。2.2 获取API密钥和权限开通使用任何收费或限量的API服务第一步都是注册账号、开通权限。TickDB目前提供免费试用额度个人开发者注册后在控制台创建一个应用系统会生成一个API Key。这个API Key就是你的身份凭证所有请求都需要通过请求头X-API-Key传给服务端。有一点需要特别注意API Key相当于你的密码不要硬编码在代码里然后上传到公开的代码仓库也不要发给其他人。如果发现Key泄露立即在控制台重置。2.3 Python开发环境准备我假设你已经安装了Python 3.9或更高版本。还没装的话去官网下载安装包注意勾选“Add Python to PATH”这个选项否则命令行里识别不到python命令。接着创建虚拟环境并安装依赖包mkdir ai-sector-monitor cd ai-sector-monitor python -m venv venv source venv/bin/activate # Windows下是 venv\Scripts\activate pip install requests pandas schedule matplotlib python-dotenv我在项目里用了python-dotenv来管理API Key这样代码里不会出现明文密钥。项目根目录下新建一个.env文件写入TICKDB_API_KEY你的密钥然后在主程序里这样加载import os from dotenv import load_dotenv load_dotenv() API_KEY os.getenv(TICKDB_API_KEY)一个不算坑但很容易忽略的点.env文件记得加进.gitignore否则提交代码的时候密钥就跟着走了。我见过不止一个人因为这个习惯不好把密钥暴露在GitHub上然后被恶意调用扣费。3. 实时行情获取与数据清洗实现3.1 封装TickDB客户端在写业务逻辑之前我先把TickDB的HTTP请求封装成一个独立的类这样后面所有数据获取都走同一个入口代码会清爽很多。import requests import pandas as pd class TickDBClient: BASE_URL https://api.tickdb.cn/v1 def __init__(self, api_key: str): self.api_key api_key self.session requests.Session() self.session.headers.update({ X-API-Key: api_key, Content-Type: application/json }) def get_sector_constituents(self, sector_code: str AI) - pd.DataFrame: 获取板块成分股列表 resp self.session.get( f{self.BASE_URL}/sector/constituents, params{sector_code: sector_code} ) resp.raise_for_status() data resp.json() return pd.DataFrame(data[data][stocks]) def get_realtime_quotes(self, symbols: list[str]) - pd.DataFrame: 获取多只股票的实时行情快照 resp self.session.post( f{self.BASE_URL}/quote/realtime, json{symbols: symbols} ) resp.raise_for_status() data resp.json() return pd.DataFrame(data[data][quotes])这里用了requests.Session而不是直接requests.get好处是复用底层的TCP连接多次请求时能减少握手开销。虽然分钟级轮询对性能要求不高但这是一个好习惯。3.2 拉取AI板块成分股AI板块的成分股名单是动态调整的所以不要在代码里写死每次都通过接口实时拉取。TickDB的板块划分支持概念板块和行业板块AI相关标的在概念板块里会有多个标签比如“AIGC”“算力”“大模型”“智能驾驶”等。我建议拉取的核心列表设定为一个相对宽泛的AI概念池大约50只左右覆盖算力硬件、AI芯片、大模型、应用软件、AIGC等细分方向。这个池子不求精确到每只股票都纯正AI概念只求“大方向不跑偏”因为我们的目标是观察板块内部分化样本量太少反而统计意义不大。client TickDBClient(API_KEY) constituents_df client.get_sector_constituents(sector_codeAI) print(f拉取到 {len(constituents_df)} 只AI概念股) print(constituents_df.head())返回的DataFrame通常包含股票代码、股票名称、所属细分方向、是否为指数权重股等字段。我会把细分方向这一列保留下来后面做板块内部子方向对比时会用到。3.3 获取实时行情并做标准化清洗拿到成分股列表后把股票代码拼成列表传给实时行情接口symbols constituents_df[symbol].tolist() quotes_df client.get_realtime_quotes(symbols)TickDB返回的实时行情JSON结构大概是这样的{ symbol: 000001, name: 示例科技, price: 23.45, pre_close: 22.10, open: 22.60, high: 23.80, low: 22.40, volume: 12345678, amount: 289000000, timestamp: 2026-02-05 10:30:00 }拿到DataFrame后不能直接用至少要做三件事第一计算涨跌幅。接口给的是价格和昨收涨跌幅需要自己算(price - pre_close) / pre_close。这里有一个边界情况要注意如果某只股票当天停牌price会等于pre_close涨跌幅是0但它的成交量也是0这种数据要标记出来不能当作正常数据参与统计。第二剔除异常值。比如某只新股上市首日没有pre_close接口可能返回0或null这时候涨跌幅计算会出现Division by Zero或者极端的正负无穷值必须过滤掉。第三类型转换。接口返回的数值字段有时候是字符串需要统一转成float否则pandas的计算结果会是object类型后面做统计时容易出莫名奇妙的错误。清洗逻辑封装成函数def clean_quotes(df: pd.DataFrame) - pd.DataFrame: 清洗实时行情数据 if df.empty: return df # 类型转换 for col in [price, pre_close, open, high, low, volume, amount]: df[col] pd.to_numeric(df[col], errorscoerce) # 计算涨跌幅 df[pct_change] (df[price] - df[pre_close]) / df[pre_close] # 过滤停牌和异常数据 df df[df[pre_close] 0] df df[df[volume] 0] df df[df[pct_change].between(-0.11, 0.11)] # 剔除缺失值 df df.dropna(subset[price, pre_close, pct_change]) return df提示between(-0.11, 0.11)这个过滤范围是我针对A股涨跌停幅度设置的。主板个股涨跌幅限制是10%ST股是5%科创板和创业板是20%。如果你监控的池子里包含创业板或科创板股票这个范围值要相应放宽到-0.21到0.21否则会误删正常的涨停数据。这里就是很多人容易踩的坑。我第一次跑脚本的时候用了统一的10%过滤范围结果把一批科创板的大涨股票全过滤掉了导致板块涨幅被严重低估整个分化指标算出来完全是错的。3.4 合并数据构建分析主表清洗完行情数据后需要把行情数据、成分股信息和细分标签合并成一张宽表供指标模块使用def build_analysis_frame(constituents_df: pd.DataFrame, quotes_df: pd.DataFrame) - pd.DataFrame: 合并成分股信息和实时行情 merged constituents_df.merge( quotes_df, left_onsymbol, right_onsymbol, howinner ) return merged用howinner做内连接的意思是只保留既在成分股列表又在行情返回中出现过的股票两边任何一边缺失都不参与计算。这样能避免成分股名单里有但行情没返回时产生空行也能避免行情返回了但不在我关注的成分股列表里的杂票混进来。4. 分化度量化指标与算法详解4.1 什么叫“涨得不够整齐”在量化语境下分化度衡量的是同一组标的价格表现的异质性。如果所有股票都涨了3%那就是完全一致行情分化度为0如果一半股票涨了6%另一半跌了2%分化度就很高。生活化的类比是班级考试成绩如果全班同学都考了80到85分这个班的成绩非常整齐老师讲起课来很轻松如果一半人考了95分另一半人考了40分平均分看起来也有67分但实际上是两极分化。只看平均分会严重误导你的判断必须看分数的分布。因为均值和方差才能完整描述一个分布的形态。4.2 三个核心指标平均涨幅、标准差、离散系数我把核心指标设计为三个各有各的用途板块平均涨幅mean_pct df[pct_change].mean()这是所有成分股涨跌幅的算术平均值反映板块整体的温度。注意我用的是算术平均而不是加权平均也就是说每只股票权重相同不区分市值大小。这样做的原因是我们对分化的观察希望偏向“个股的普遍状态”而不是被大权重股主导。如果某只巨无霸权重股涨了5%把板块指数带红但小票全是绿的指数是失真的算术平均才能反映真实的普适状态。板块涨幅标准差std_pct df[pct_change].std()标准差衡量的是个股涨跌幅围绕均值的分散程度。标准差越大说明个股之间涨得越不均匀分化越明显。在统计上如果涨跌幅近似正态分布那么大约68%的股票会落在“平均值正负一个标准差”的区间内。这个直觉很有用当标准差是2个百分点时你大概能预期三分之二的股票涨跌幅差异在4个百分点以内当标准差拉到4个百分点以上个股表现差距可能超过8个百分点这是非常剧烈的分化。离散系数变异系数cv std_pct / abs(mean_pct) if mean_pct ! 0 else float(inf)离散系数是标准差除以平均值的绝对值用来消除均值大小对离散程度的干扰。为什么要额外看这个指标举个极端例子如果板块平均涨幅是5%标准差也是1个百分点那这个波动并不算大因为基数高但如果板块平均涨幅只有0.2%标准差还是1个百分点相对波动就非常剧烈了说明板块几乎失控方向极度不明确。离散系数能反映出这种“在低均值背景下高波动”的情况这在判断行情拐点时很有参考价值。4.3 补充结构指标强势股占比与分化系数除了上述三个基本统计量我还额外构造了两个结构指标让判断更立体。强势股占比涨跌幅大于3%的个股占全部成分股的比例。这个比例高于50%说明普涨低于20%说明极弱市。分化系数取涨跌幅绝对值大于5%的个股数量与涨跌幅绝对值小于1%的个股数量的比值。这个比值越高说明资金越集中在部分标的上方向分歧越明显。我实际写代码时会把它们一起封装成一个函数def calculate_dispersion_metrics(df: pd.DataFrame) - dict: 计算板块分化指标 metrics {} # 基础统计 metrics[stock_count] len(df) metrics[mean_pct] df[pct_change].mean() metrics[median_pct] df[pct_change].median() metrics[std_pct] df[pct_change].std() # 离散系数 mean_val metrics[mean_pct] metrics[cv] metrics[std_pct] / abs(mean_val) if mean_val ! 0 else float(inf) # 强势股占比涨幅 3% strong_count (df[pct_change] 0.03).sum() metrics[strong_ratio] strong_count / len(df) # 分化系数|涨幅|5% 与 |涨幅|1% 的数量比 extreme_count (df[pct_change].abs() 0.05).sum() mild_count (df[pct_change].abs() 0.01).sum() metrics[divergence_ratio] extreme_count / mild_count if mild_count 0 else float(inf) # 细分子方向表现 sector_group df.groupby(sub_sector)[pct_change].agg([mean, std, count]) metrics[sub_sector_stats] sector_group return metrics注意median_pct中位数这个指标在分析分化时非常有用。如果平均值明显大于中位值比如均值是2%但中位数只有0.5%说明少数大牛股把均值拉高了大部分股票其实没怎么涨这也是分化的一种表现。这种“均值与中位数背离”的形态在A股非常常见尤其在AI这种热门板块。资金往往抱团少数龙头其他股票只是陪跑。遇到这种情况如果你只看均线或者板块指数很容易高估市场的赚钱效应。5. 实战马年AI板块全流程监控5.1 定时轮询架构设计行情监控程序的核心是一个定时循环每隔一段时间拉取一次数据并更新指标。我用schedule库来实现分钟级定时任务因为它足够轻量而且语法非常直观。import schedule import time from datetime import datetime def monitor_job(): 单次监控任务 print(f[{datetime.now().strftime(%H:%M:%S)}] 开始拉取AI板块行情...) # 拉取成分股如果已经初始化过可以缓存不必每次拉取 # 这里为了演示简洁每次都拉 constituents client.get_sector_constituents(sector_codeAI) symbols constituents[symbol].tolist() # 拉取实时行情 quotes client.get_realtime_quotes(symbols) # 清洗和合并 quotes_clean clean_quotes(quotes) analysis_df build_analysis_frame(constituents, quotes_clean) # 计算指标 metrics calculate_dispersion_metrics(analysis_df) # 输出提示 output_monitor_message(metrics) def output_monitor_message(metrics: dict) - None: 把监控结果输出到控制台 print(f 股票数量: {metrics[stock_count]} 只) print(f 平均涨幅: {metrics[mean_pct]*100:.2f}%) print(f 涨幅中位数: {metrics[median_pct]*100:.2f}%) print(f 涨幅标准差: {metrics[std_pct]*100:.2f}%) print(f 离散系数: {metrics[cv]:.3f}) print(f 强势股占比: {metrics[strong_ratio]*100:.1f}%) print(f 分化系数: {metrics[divergence_ratio]:.2f}) # 根据指标给出简要信号提示 if metrics[std_pct] 0.025 and metrics[strong_ratio] 0.5: print( [信号] 板块分化明显赚钱效应集中注意风险) elif metrics[strong_ratio] 0.6: print( [信号] 板块普涨做多情绪较强) elif metrics[mean_pct] 0 and metrics[strong_ratio] 0.2: print( [信号] 板块普跌情绪低迷) else: print( [信号] 板块相对均衡暂无极端分化信号) # 交易时间调度周一至周五 9:30-11:30, 13:00-15:00 每分钟执行一次 schedule.every().minute.do(monitor_job) # 一个简单的交易时间判断 def is_trading_time(): now datetime.now() if now.weekday() 5: return False if (now.hour 9 and now.minute 30) or (now.hour 10): return True if now.hour 11 and now.minute 30: return True if now.hour 13 or now.hour 14: return True if now.hour 15 and now.minute 0: return True return False while True: if is_trading_time(): schedule.run_pending() time.sleep(5)5.2 信号判读的实战经验这套代码跑起来之后我马年第三天就遇到一个非常典型的场景。上午开盘后第一次轮询板块平均涨幅是0.8%中位数是0.3%标准差只有1.2%看起来风平浪静。到10点15分第二次轮询时标准差突然拉到3.5%离散系数从1.5升到4.2强势股占比从40%降到22%。这个变化背后的含义是板块表面还在涨甚至平均涨幅比上次更高但内部已经从“普涨”变成了“少数强拉”。多数股票开始回落只有几只龙头还在撑着。这种时候如果看着板块指数追进去大概率买在分化行情的高位。后来收盘复盘时我看到当天AI板块表面上是红盘收尾但实际下跌的股票超过六成完全验证了盘中标准差飙升发出的预警信号。这就是量化监控相对于肉眼盯盘的核心优势你看到的不是“板块还在涨”这个表象而是板块内部资金正在撤退这个本质。5.3 收盘后的可视化复盘盘中监控负责实时预警收盘后的复盘则负责看清全貌。我用matplotlib画了一张“AI板块涨跌幅分布直方图”把当天所有成分股的涨跌幅画成分布。横轴是涨跌幅区间纵轴是股票数量。import matplotlib.pyplot as plt import matplotlib matplotlib.use(Agg) def plot_distribution(df: pd.DataFrame, save_path: str ai_sector_dist.png) - None: 绘制板块涨跌幅分布直方图 plt.figure(figsize(12, 6)) plt.hist(df[pct_change] * 100, bins30, color#2C7FB8, edgecolorwhite, alpha0.8) plt.axvline(df[pct_change].mean() * 100, colorred, linestyle--, label均值) plt.axvline(df[pct_change].median() * 100, colorgreen, linestyle--, label中位数) plt.xlabel(涨跌幅 (%)) plt.ylabel(股票数量 (只)) plt.title(AI板块成分股涨跌幅分布) plt.legend() plt.tight_layout() plt.savefig(save_path, dpi150)收盘后拿到全天收盘数据跑一遍这个函数一张图就能直观看到当天的市场结构。如果柱状图是中间高、两边低的钟形曲线说明行情均衡如果左边下跌端和右边上涨端都高、中间反而凹下去那就是典型的极端分化市。我还喜欢叠加“细分子方向”的分组柱状图比如把算力、应用、芯片三个子方向分别统计平均涨跌能看出资金在哪个环节抱团、哪个环节撤离。这个信息比看整体板块有意思得多。6. 常见问题与排查技巧实录6.1 API请求报错400和403实际使用TickDB的过程中我遇到过两次比较典型的HTTP错误。第一次是返回400invalid schema for function仔细检查后发现是我先把symbols参数写成字符串而不是列表还有一次是因为股票代码传成了“000001.SZ”这种带后缀的格式但接口要求的是纯数字代码。解决办法很直接查看接口文档里示例请求体对照字段类型传参。第二次是403通常是API Key没过鉴权。常见原因有三个密钥复制多了空格、环境变量没加载成功、试用额度已用完。我的排查顺序是先在终端里直接打印环境变量确认加载再去TickDB控制台看查询额度是否清零。6.2 数据出现大量NaN或重复值有段时间我发现拉回来的DataFrame有一半股票的pre_close是NaN查了半天发现是请求频率太高触发了接口的限流返回的数据被截断了。TickDB免费额度的接口频率限制通常是每秒2到5次请求我一次性传50个股票代码理论上是一次请求不会被限流。但如果我的循环里不小心把每个股票代码单独请求一次50只股票就会直接打爆限流阈值。解决方案是把所有股票代码放进一个列表用一次POST请求拉回全部行情而不是在循环里逐个GET。6.3 定时任务偶尔“跳拍”schedule库的定时精度不是严格精确到秒在系统忙的时候可能延迟几秒执行但本质上是每轮主循环检查一次“是否有到期的任务”所以不会堆积任务。如果程序运行时间超过一分钟下一轮任务会立即补上造成连续两次拉取间隔极短。这个在分钟级监控场景下影响很小但如果你做秒级甚至tick级监控建议直接用TickDB的WebSocket接口不要用HTTP轮询。6.4 盘中停牌股和一字板如何处理A股每天都有涨停和跌停的股票也会有临时停牌的情况。涨停的股票价格封死在涨停价上但成交量很小涨跌幅很大这会干扰标准差的计算。我在清洗逻辑里不会剔除涨停股因为“涨停”本身就是一种强分化信号。一字涨停板是最强的多头信号而大量股票冲上涨停又打开反复开板说明多空分歧极大这两种状态都值得纳入指标观察。停牌股的涨跌幅是0为了避免它们拉低离散度我会过滤掉当天没有任何成交量的股票。这个逻辑要写清楚否则早盘刚开盘时很多股票还没有第一笔成交会被误判为停牌。7. 扩展方向从分钟级监控到盘中策略参考写到这里这套工具的基本功能已经完整了。但既然是“马年开工AI先行”我觉得还值得聊一下后续还能怎么扩展让这套架子发挥更大的价值。第一个扩展方向是结合AI做信号解读。目前程序输出的是结构化数字指标你需要自己看数字、看分布图来理解盘面。你可以把指标计算结果传给大模型API让AI生成一段自然语言的盘面描述比如“当前AI板块分化程度处于近期高位算力方向领涨应用方向回调建议关注是否出现补跌”。我试过用DeepSeek的API做这个事把指标数据拼成提示词模板返回的文本质量意外地好至少比盯着数字干想直观很多。第二个方向是历史回放和对比。TickDB的历史K线接口可以把过去30天的分钟线拉下来重新计算当天的分化指标序列然后画成时间序列曲线。这样你就能看出当前的分化度在历史区间里处于什么百分位是罕见的极端分化还是常态波动。百分位比绝对数值更有参考价值。第三个方向是股票池动态调整。目前成分股列表来自TickDB的AI概念板块接口它其实是定期更新的。你也可以在本地维护一个自定义股票池然后和接口返回的列表做并集或交集。比如你更关注市场核心的20只大模型算力股不想被一堆边缘应用股稀释指标那就在代码里加一个白名单过滤参数。我自己后续最想做的是把盘中实时数据和每天收盘后的复盘结果存放在本地SQLite数据库里积累一段时间后做“分化度与次日板块收益”的相关性分析。如果分化度极高和极低的次日常表现有统计显著的差异那这套工具就不仅能描述行情还能做一个简单的前瞻性参考。提醒任何统计分析都不能保证未来重复发生。市场结构会变A股的风格轮动速度非常快AI板块更是如此。股票池、指标参数、阈值设置都需要定期回顾调整不要指望一套参数能一直适用。最后分享一个我个人很深的感受工具再顺手也只是帮你把市场观察的效率提高了一个台阶真正的判断力仍然来自你对行业和公司的理解。量化指标告诉我“板块分歧很大”但分歧的双方在争什么哪一方更可能在接下来的产业趋势中获胜这个问题任何API和代码都替代不了你的思考。祝各位马年开工顺利技术修炼和认知升级同步前行。