
> 盘中实时监控全市场行情，自动筛选涨停、放量上涨、突破均线的股票 -- 这是很多量化交易者的刚需。本文用 Python 实现两种方案：REST 轮询方案（简单易懂）和 WebSocket 推送方案（低延迟），附完整可运行代码。

---

## 一、盘中行情监控能做什么？

实时行情监控是从"回测"走向"实盘"的关键一步。常见用途：

- **涨停板监控**：第一时间发现涨停或接近涨停的股票
- **放量突破筛选**：找出成交量异常放大且价格突破关键位的标的
- **自选股盯盘**：监控持仓或关注的股票，触发条件时提醒
- **日内交易信号**：结合分钟 K 线生成实时买卖信号

---

## 二、环境准备

### 安装依赖

```bash
pip install "tickflow[all]" --upgrade
```

实时行情功能需要注册获取 API key：

1. 访问 [tickflow.org](https://tickflow.org)
2. 登录后在控制台生成 API key
3. 配置环境变量（推荐）或直接传入

```bash
# Linux / Mac
export TICKFLOW_API_KEY="your-api-key"

# Windows
set TICKFLOW_API_KEY=your-api-key
```

---

## 三、方案一：REST API 轮询（入门推荐）

REST 方案最简单，通过定时请求获取全市场行情快照，适合对延迟要求不高的场景。

### 1. 获取全市场实时行情

```python
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

# 获取全部 A 股实时行情
quotes = tf.quotes.get(universes=["CN_Equity_A"], as_dataframe=True)
print(f"共获取 {len(quotes)} 只 A 股行情")
print(quotes[["symbol", "last_price", "volume", "amount"]].head())
```

一次调用即可拿到 5000+ 只 A 股的实时行情，包含最新价、开盘价、最高价、最低价、昨收、成交量、成交额等字段。

### 2. 涨跌幅排行

```python
quotes["change_pct"] = (quotes["last_price"] - quotes["prev_close"]) / quotes["prev_close"] * 100

# 涨幅榜 Top 10
print("涨幅榜 Top 10:")
top_gainers = quotes.nlargest(10, "change_pct")[["symbol", "last_price", "change_pct"]]
print(top_gainers.to_string(index=False))

# 跌幅榜 Top 10
print("\n跌幅榜 Top 10:")
top_losers = quotes.nsmallest(10, "change_pct")[["symbol", "last_price", "change_pct"]]
print(top_losers.to_string(index=False))
```

### 3. 涨停股监控

```python
# A 股主板涨跌停限制为 10%，创业板和科创板为 20%
limit_up = quotes[quotes["change_pct"] >= 9.9]
print(f"\n涨停或接近涨停: {len(limit_up)} 只")
print(limit_up[["symbol", "last_price", "change_pct", "amount"]].to_string(index=False))
```

### 4. 放量上涨股筛选

```python
# 成交额 > 5 亿 且 涨幅 > 3%
strong = quotes[(quotes["amount"] > 5e8) & (quotes["change_pct"] > 3)]
strong_sorted = strong.sort_values("change_pct", ascending=False)

print(f"\n放量上涨（成交额>5亿 且 涨幅>3%）: {len(strong_sorted)} 只")
print(strong_sorted[["symbol", "last_price", "change_pct", "amount"]].head(20).to_string(index=False))
```

### 5. 定时轮询监控

将上述逻辑封装为一个定时轮询程序：

```python
import time
import datetime
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")


def scan_market():
    """扫描全市场，返回符合条件的强势股"""
    quotes = tf.quotes.get(universes=["CN_Equity_A"], as_dataframe=True)
    quotes["change_pct"] = (
        (quotes["last_price"] - quotes["prev_close"]) / quotes["prev_close"] * 100
    )

    # 涨停股
    limit_up = quotes[quotes["change_pct"] >= 9.9]

    # 放量上涨
    strong = quotes[(quotes["amount"] > 5e8) & (quotes["change_pct"] > 3)]

    # 大跌预警（跌幅超过 7%）
    big_drop = quotes[quotes["change_pct"] <= -7]

    return {
        "total": len(quotes),
        "limit_up": limit_up,
        "strong": strong,
        "big_drop": big_drop,
        "avg_change": quotes["change_pct"].mean(),
    }


def run_monitor(interval=30):
    """每隔 interval 秒扫描一次"""
    print("开始监控全市场行情...")
    while True:
        now = datetime.datetime.now()
        hour = now.hour

        # 只在交易时间运行（9:25 - 15:05）
        if hour < 9 or hour >= 16:
            print(f"[{now:%H:%M:%S}] 非交易时间，等待中...")
            time.sleep(60)
            continue

        result = scan_market()

        print(f"\n[{now:%H:%M:%S}] 市场概况:")
        print(f"  全市场平均涨幅: {result['avg_change']:.2f}%")
        print(f"  涨停: {len(result['limit_up'])} 只")
        print(f"  放量上涨: {len(result['strong'])} 只")
        print(f"  大跌预警: {len(result['big_drop'])} 只")

        if len(result["limit_up"]) > 0:
            print("  --- 涨停股 ---")
            for _, row in result["limit_up"].head(5).iterrows():
                print(f"    {row['symbol']}: {row['last_price']} ({row['change_pct']:.2f}%)")

        time.sleep(interval)


run_monitor(interval=30)
```

---

## 四、方案二：WebSocket 实时推送（低延迟）

如果需要更低延迟，推荐使用 WebSocket 方案。服务端主动推送行情变动，无需轮询。

### 1. 基本用法

```python
import datetime
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")
stream = tf.stream


@stream.on_quotes
def on_quotes(quotes):
    for q in quotes:
        ts = datetime.datetime.fromtimestamp(q["timestamp"] / 1000)
        ext = q.get("ext", {})
        name = ext.get("name", "")
        change_pct = ext.get("change_pct")
        change_str = f"{change_pct:+.2%}" if change_pct is not None else "N/A"
        print(f"[{ts:%H:%M:%S}] {q['symbol']} {name} 最新:{q['last_price']} 涨跌:{change_str}")


@stream.on_error
def on_error(msg):
    print(f"错误: {msg}")


# 订阅感兴趣的标的
stream.subscribe("quotes", ["600519.SH", "000001.SZ", "600000.SH"])
stream.connect()  # 阻塞运行
```

输出示例：

```text
[09:31:05] 600519.SH 贵州茅台 最新:1580.00 涨跌:+0.82%
[09:31:05] 000001.SZ 平安银行 最新:11.12 涨跌:+0.54%
[09:31:06] 600000.SH 浦发银行 最新:9.89 涨跌:-0.10%
```

### 2. 带筛选条件的实时监控

```python
import datetime
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")
stream = tf.stream

alert_threshold = 5.0  # 涨幅超过 5% 时提醒


@stream.on_quotes
def on_quotes(quotes):
    for q in quotes:
        ext = q.get("ext", {})
        change_pct = ext.get("change_pct")
        if change_pct is None:
            continue

        change_pct_val = change_pct * 100
        if abs(change_pct_val) >= alert_threshold:
            name = ext.get("name", "")
            ts = datetime.datetime.fromtimestamp(q["timestamp"] / 1000)
            direction = "大涨" if change_pct_val > 0 else "大跌"
            print(
                f"[{ts:%H:%M:%S}] {direction} {q['symbol']} {name} "
                f"最新:{q['last_price']} 涨跌:{change_pct_val:+.2f}%"
            )


symbols = ["600519.SH", "000858.SZ", "601318.SH", "000001.SZ", "600036.SH"]
stream.subscribe("quotes", symbols)
stream.connect()
```

### 3. 非阻塞模式 + 动态订阅

```python
import time
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")
stream = tf.stream

watched = {}


@stream.on_quotes
def on_quotes(quotes):
    for q in quotes:
        watched[q["symbol"]] = q


# 后台运行 WebSocket
stream.subscribe("quotes", ["600519.SH", "000001.SZ"])
stream.connect(block=False)

# 主线程可以做其他事
time.sleep(5)

# 动态追加订阅
stream.subscribe("quotes", ["600036.SH", "601318.SH"])

# 定时打印最新行情
for _ in range(10):
    time.sleep(3)
    for symbol, q in watched.items():
        ext = q.get("ext", {})
        print(f"{symbol}: {q['last_price']} ({ext.get('name', '')})")
    print("---")

stream.close()
```

### 4. 异步 WebSocket

```python
import asyncio
import datetime
from tickflow import AsyncTickFlow


async def main():
    async with AsyncTickFlow(api_key="your-api-key") as tf:
        stream = tf.stream

        @stream.on_quotes
        def on_quotes(quotes):
            for q in quotes:
                ts = datetime.datetime.fromtimestamp(q["timestamp"] / 1000)
                ext = q.get("ext", {})
                name = ext.get("name", "")
                print(f"[{ts:%H:%M:%S}] {q['symbol']} {name}: {q['last_price']}")

        await stream.subscribe("quotes", ["600519.SH", "000001.SZ"])
        await stream.connect()


asyncio.run(main())
```

---

## 五、进阶：结合日内 K 线做实时选股

不仅可以看行情快照，还可以结合日内分钟 K 线做更精细的分析：

### 日内 VWAP 偏离度筛选

```python
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

symbols = ["600519.SH", "000858.SZ", "601318.SH", "000001.SZ", "600036.SH"]

dfs = tf.klines.intraday_batch(symbols, period="5m", as_dataframe=True, show_progress=True)

for symbol, df in dfs.items():
    if len(df) == 0:
        continue
    df["vwap"] = (df["amount"] / df["volume"]).round(2)
    latest = df.iloc[-1]
    vwap_dev = (latest["close"] - latest["vwap"]) / latest["vwap"] * 100
    status = "高于VWAP" if vwap_dev > 0 else "低于VWAP"
    print(f"{symbol}: 最新价 {latest['close']}, VWAP {latest['vwap']}, {status} {abs(vwap_dev):.2f}%")
```

### 分钟级均线突破监控

```python
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

# 获取日内 5 分钟 K 线
df = tf.klines.intraday("600519.SH", period="5m", as_dataframe=True)
df["ma10"] = df["close"].rolling(10).mean()
df["ma30"] = df["close"].rolling(30).mean()

# 检查最新是否金叉
if len(df) >= 30:
    latest = df.iloc[-1]
    prev = df.iloc[-2]
    if latest["ma10"] > latest["ma30"] and prev["ma10"] <= prev["ma30"]:
        print(f"600519.SH 日内 5 分钟级别 MA10 上穿 MA30，关注！")
    else:
        print(f"600519.SH 日内暂无均线交叉信号")
```

---

## 六、ETF 行情监控

除了个股，ETF 也是很好的监控对象：

```python
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

# 获取全部 ETF 行情
etf_quotes = tf.quotes.get(universes=["CN_ETF"], as_dataframe=True)
etf_quotes["change_pct"] = (
    (etf_quotes["last_price"] - etf_quotes["prev_close"]) / etf_quotes["prev_close"] * 100
)

# 成交额最大的 ETF（通常是宽基 ETF）
top_volume = etf_quotes.nlargest(20, "amount")
print("成交额 Top 20 ETF:")
print(top_volume[["symbol", "last_price", "change_pct", "amount"]].to_string(index=False))

# 涨幅最大的 ETF（发现热门板块）
top_change = etf_quotes.nlargest(10, "change_pct")
print("\n涨幅 Top 10 ETF:")
print(top_change[["symbol", "last_price", "change_pct"]].to_string(index=False))
```

通过 ETF 涨幅排行，可以快速发现当天最热门的板块方向。

---

## 七、五档盘口数据

如果需要更细粒度的分析，TickFlow 还提供五档买卖盘口（需 Pro/Expert 套餐）：

```python
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

depth = tf.depth.get("600519.SH")

print(f"标的: {depth['symbol']}")
for i in range(5):
    bid_price = depth["bid_prices"][i]
    bid_vol = depth["bid_volumes"][i]
    ask_price = depth["ask_prices"][i]
    ask_vol = depth["ask_volumes"][i]
    print(f"  卖{5-i}: {ask_price} x {ask_vol}  |  买{i+1}: {bid_price} x {bid_vol}")
```

---

## 八、两种方案对比

| 维度 | REST 轮询 | WebSocket 推送 |
|------|----------|---------------|
| 延迟 | 取决于轮询间隔（秒级） | 毫秒级 |
| 实现难度 | 简单 | 稍复杂 |
| 适用场景 | 选股扫描、定时监控 | 盯盘、日内交易 |
| 资源消耗 | 每次请求有网络开销 | 长连接，效率更高 |
| 全市场支持 | 一次拉取全部行情 | 按标的订阅 |

建议：

- **全市场扫描选股** -> 用 REST，一次拿全部数据做筛选
- **自选股盯盘** -> 用 WebSocket，延迟更低
- **两者结合** -> REST 做定时全市场扫描，WebSocket 订阅筛选出的标的做实时监控

---

## 九、总结

本文介绍了两种 Python 实时监控 A 股行情的方案：

1. **REST 轮询**：一行代码拿到全市场 5000+ 只 A 股行情，适合定时扫描选股
2. **WebSocket 推送**：毫秒级延迟，适合盯盘和日内交易

核心功能包括：

- 涨跌幅排行
- 涨停股监控
- 放量上涨筛选
- ETF 板块监控
- 日内 VWAP 偏离分析
- 分钟级均线突破信号

TickFlow 提供了统一的 API，无论是 REST 还是 WebSocket，SDK 都做了良好的封装，几行代码即可实现复杂的监控逻辑。

---

## 相关链接

- 官网：[https://tickflow.org](https://tickflow.org?utm_source=tencent_dev)
- 文档：[https://docs.tickflow.org](https://docs.tickflow.org)

---

> 实盘交易的第一步，不是下单，而是看到实时的市场。


盘中实时监控全市场行情，自动筛选涨停、放量上涨、突破均线的股票 -- 这是很多量化交易者的刚需。本文用 Python 实现两种方案：REST 轮询方案（简单易懂）和 WebSocket 推送方案（低延迟），附完整可运行代码。

Python 实时监控 A 股行情并自动筛选强势股（REST + WebSocket 两种方案）

编程语言

Python实现股票实时行情监控：REST轮询与WebSocket推送两种方案，包含涨停监控、放量上涨筛选、ETF行情分析等功能。提供完整代码示例，帮助量化交易者快速搭建实盘监控系统，实现低延迟行情推送与全市场扫描。

Python

Linux

2026采购季 | AI焕新·智启新局

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Python 实时监控 A 股行情并自动筛选强势股（REST + WebSocket 两种方案）-腾讯云开发者社区-腾讯云

Python 实时监控 A 股行情并自动筛选强势股（REST + WebSocket 两种方案）

Python 实时监控 A 股行情并自动筛选强势股（REST + WebSocket 两种方案）

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐