📊 用Python打造A股市场实时监控系统：从零构建QMT量化风控工具

作者：[小树实盘笔记]关键词：QMT、量化交易、市场监控、涨停板分析、风控系统、Python

（建议配图：一张融合股市行情、涨停板数据与代码界面的科技感图片）

一、为什么我要写这样一段代码？——源于实战的风控需求

在A股这个以情绪驱动为主、政策与资金为辅的市场中，每日的市场情绪变化直接决定了交易策略的生死。我们常常听到“冰点期”、“回暖期”、“高潮期”，但这些词太主观，难以量化。

作为一名量化交易者，我一直在思考一个问题：

“如何用客观数据判断当前市场是否适合出手？什么时候该进攻，什么时候该防守？”

于是，我决定在 QMT（迅投量化平台） 上，用 Python 自建一个 实时A股市场情绪监控系统，自动统计以下关键指标：

• ✅ 市场整体涨跌情况（加权涨幅、中位数）
• ✅ 涨停/跌停数量
• ✅ 炸板数与封板率
• ✅ 上涨与下跌家数对比
• ✅ 成交额趋势

这些数据不仅是盘后复盘的依据，更是盘中风控与策略开关的重要参考。

例如：

• 当涨停数骤降、炸板率飙升时 → 可能是情绪退潮信号，应降低仓位；
• 当封板率持续高于70%且成交额放大 → 市场热度上升，可加大打板或追涨策略投入。

这，就是我写这段代码的初衷：让情绪可度量，让风控有依据。

二、我们需要统计哪些关键数据？——构建市场情绪指标体系

为了全面刻画A股市场状态，我们设计了如下六大核心指标：

这些指标结合使用，能形成对市场情绪的立体感知，远超单一“涨跌家数比”。

三、关键技术实现：QMT API 与关键函数解析

QMT 提供了丰富的底层接口，我们通过其 Python策略环境 实现数据获取与计算。以下是本系统用到的核心函数及其作用解析：

🔹 1. 获取股票池：get_stock_list_in_sector()

C.get_stock_list_in_sector('沪深A股')C.get_stock_list_in_sector('京市A股')

• 作用：获取指定板块的所有股票代码列表。
• 说明：A股市场包含沪市、深市、北交所，需合并处理。

🔹 2. 获取实时行情：get_full_tick()

full_tick = C.get_full_tick(A.hsa)

• 作用：批量获取所有股票的最新Tick数据（含最新价、昨收、最高价、成交额等）。
• 优势：高效、低延迟，适合高频监控。

🔹 3. 获取涨停跌停价：get_instrumentdetail()

detail = C.get_instrumentdetail(stock)up_stop_price = detail["UpStopPrice"]down_stop_price = detail["DownStopPrice"]

• 作用：获取个股的涨跌停价格，用于判断是否封板。
• 注意：不同股票（如ST、注册制新股）涨跌幅不同，必须动态获取。

🔹 4. 定时执行：run_time()

C.run_time("f", "3nSecond", "2019-10-14 13:20:00")

• 作用：每3秒调用一次 f(C) 函数，实现实时刷新监控。
• 参数说明：
• "f"：回调函数名
• "3nSecond"：每3秒执行一次
• 起始时间：避免策略启动时误触发

🔹 5. 数据处理与输出

• 使用字典缓存每只股票的成交量（vol_dict），用于市值加权；
• 使用 to_zw() 函数将金额转为“万/亿”单位，提升可读性；
• 使用 time.time() 统计函数执行耗时，优化性能。

四、代码演进：从想法到完善系统的完整过程

🧩 第一阶段：搭建框架

class a(): passA = a()

• 使用简单类对象 a 作为全局变量容器，存储股票池、成交量等中间数据。

🧩 第二阶段：初始化数据

def init(C):    A.hsa = 获取沪深京A股    A.vol_dict = {stock: C.get_last_volume(stock) for stock in A.hsa}    C.run_time("f", "3nSecond", start_time)

• 初始化时获取股票池和基础数据，并启动定时任务。

🧩 第三阶段：核心逻辑 f(C) 实现

在每次定时触发中：

1. 获取最新Tick数据；
2. 遍历所有股票，计算涨幅、判断封板状态；
3. 统计六大指标；
4. 打印格式化结果。

🧩 第四阶段：优化与健壮性处理

• 增加 if stock not in full_tick: continue 防止数据缺失报错；
• 排除 last_close <= 0 的异常情况（如停牌）；
• 使用 np.isnan() 判断问题数据；
• 封板率计算前判断分母是否为0。

五、最终成果：完整可运行代码

# -*- coding: gbk -*-import timeimport numpy as npclass a(): passA = a()def init(C):# 获取沪深京A股列表    A.hsa = C.get_stock_list_in_sector('沪深A股') + C.get_stock_list_in_sector('京市A股')print('股票池大小', len(A.hsa))    A.vol_dict = {}for stock in A.hsa:        A.vol_dict[stock] = C.get_last_volume(stock)# 每3秒执行一次，起始时间可设为策略启动时间    C.run_time("f", "3nSecond", "2019-10-14 13:20:00")def to_zw(a):"""金额转中文格式：万/亿"""if np.isnan(a):return '问题数据'if abs(a) < 10000:return str(round(a, 2))if abs(a) < 100000000:return str(round(a / 10000, 2)) + "万"return f"{round(a / 100000000, 2)}亿"def f(C):    t0 = time.time()    full_tick = C.get_full_tick(A.hsa)    total_market_value = 0    total_ratio = 0    total_amount = 0    ratio_list = []    up_stop_count = 0    # 涨停    down_stop_count = 0  # 跌停    blown_count = 0      # 炸板for stock in A.hsa:if stock not in full_tick:continue        tick = full_tick[stock]        last_price = tick['lastPrice']        last_close = tick['lastClose']        high_price = tick['high']        amount = tick['amount']if last_close <= 0:continue# 排除异常# 获取涨跌停价        detail = C.get_instrumentdetail(stock)        up_stop_price = detail.get("UpStopPrice", 0)        down_stop_price = detail.get("DownStopPrice", 0)        ratio = last_price / last_close - 1        ratio_list.append(ratio)# 成交额与市值        total_amount += amount        market_value = last_price * A.vol_dict.get(stock, 0)        total_ratio += ratio * market_value        total_market_value += market_value# 涨停判断if up_stop_price > 0 and last_price >= up_stop_price:            up_stop_count += 1elif up_stop_price > 0 and high_price >= up_stop_price and last_price < up_stop_price:            blown_count += 1  # 炸板if down_stop_price > 0 and last_price <= down_stop_price:            down_stop_count += 1# 计算加权涨幅    weighted_return = (total_ratio / total_market_value) * 100 if total_market_value != 0 else 0# 涨幅中位数    ratio_list.sort()    middle_ratio = ratio_list[len(ratio_list) // 2] * 100 if ratio_list else 0# 封板率    total_up_touch = up_stop_count + blown_count    limit_up_rate = (up_stop_count / total_up_touch * 100) if total_up_touch > 0 else 0# 上涨/下跌家数    rise_num = len([r for r in ratio_list if r > 0])    down_num = len([r for r in ratio_list if r < 0])# 输出结果print(f"--- 市场概况 ({time.strftime('%H:%M:%S', time.localtime())}) ---")print(f"加权涨幅: {round(weighted_return,2)}%  中位数: {round(middle_ratio,2)}%")print(f"上涨: {rise_num}  下跌: {down_num}  成交额: {to_zw(total_amount)}")print(f"涨停: {up_stop_count}  跌停: {down_stop_count}  炸板: {blown_count}")print(f"封板率: {round(limit_up_rate, 2)}%  耗时: {round(time.time()-t0, 3)}秒")print("-" * 40)

六、未来优化方向

1. 可视化输出：结合 matplotlib 或 pyecharts 生成实时图表；
2. 数据持久化：将每日数据写入CSV或数据库，用于回测分析；
3. 策略联动：当封板率<50%时，自动降低打板策略仓位；
4. 消息推送：通过微信/钉钉推送极端行情预警；
5. 行业维度拆分：按行业统计涨停数，识别热点轮动。

七、写在最后：量化，是把经验变成系统

这段代码看似简单，但它背后是对市场理解的沉淀。我们不是在写“监控程序”，而是在构建自己的交易操作系统。

真正的量化，不是预测市场，而是感知市场、适应市场、管理风险。

希望这个小工具，能帮助你在每一个交易日，看得更清，走得更稳。

📌 获取源码：关注公众号，回复“市场监控”即可获取完整代码与使用教程。

💬 欢迎交流：在评论区留下你常用的市场情绪指标，我们一起完善这个系统！

风险提示：本文仅作为技术分享，不构成任何投资建议。市场有风险，投资需谨慎。

👉 点击关注，不错过每一篇干货