AlphaFeed Blog

如何用 Python 做一个实时自选股监控脚本？

行情数据

如何用 Python 做一个实时自选股监控脚本？

一句话回答： 用一个提供实时行情快照的数据 API（如 AlphaFeed），批量拉取自选股的最新价和涨跌幅，套一个循环定时刷新，再加一段到价判断，就是一个可用的盯盘脚本——全部代码不到 60 行。

目标

做一个命令行版的”自选股面板”，实现：

  1. 批量获取多只股票的实时价格、涨跌幅、成交额；
  2. 每隔几秒自动刷新；
  3. 涨绿跌红（A 股习惯：红涨绿跌）终端着色；
  4. 价格突破设定的阈值时打印提醒。

第一步：批量获取实时行情

实时快照接口支持一次传入多个标的：

from alphafeed import AlphaFeed

af = AlphaFeed(api_key="your-api-key")

watchlist = ["600519.SH", "000001.SZ", "601318.SH", "300750.SZ"]

df = af.quotes.get(symbols=watchlist, to_dataframe=True)
print(df[["symbol", "last_price", "prev_close", "volume", "ext.name", "ext.change_pct"]])

返回的关键字段：

字段含义
symbol标的代码
last_price最新价
prev_close昨收价
volume成交量
amount成交额
ext.name标的名称
ext.change_pct涨跌幅（小数，如 -0.024 表示 -2.4%）

实时快照约每 3 秒刷新一次，因此监控脚本的刷新间隔设成 3 秒或更长最合理。

第二步：格式化 + 涨跌着色

def colorize(text, change_pct):
    """A 股习惯：红涨绿跌。"""
    if change_pct > 0:
        return f"\033[31m{text}\033[0m"   # 红
    elif change_pct < 0:
        return f"\033[32m{text}\033[0m"   # 绿
    return text

def render(df):
    lines = []
    for _, r in df.iterrows():
        name = r["ext.name"]
        pct = r["ext.change_pct"] * 100
        line = f"{name:<8s} {r['last_price']:>10.2f}  {pct:>+6.2f}%"
        lines.append(colorize(line, r["ext.change_pct"]))
    return "\n".join(lines)

第三步：加上定时刷新循环

import os
import time
from datetime import datetime

def monitor(watchlist, interval=3):
    af = AlphaFeed(api_key="your-api-key")
    while True:
        df = af.quotes.get(symbols=watchlist, to_dataframe=True)
        os.system("cls" if os.name == "nt" else "clear")
        print(f"自选股监控  {datetime.now():%H:%M:%S}\n")
        print(f"{'名称':<8s} {'最新价':>10s}  {'涨跌幅':>7s}")
        print("-" * 32)
        print(render(df))
        time.sleep(interval)

monitor(["600519.SH", "000001.SZ", "601318.SH"])

运行后终端里就是一个每 3 秒刷新、红涨绿跌的实时面板。

第四步：到价提醒

给每只股票设一个上/下阈值，触发时提醒（这里用打印，也可以换成发邮件、推送到微信/钉钉、调用系统通知）：

alerts = {
    "600519.SH": {"above": 1300, "below": 1200},
    "000001.SZ": {"above": 12.0, "below": 10.0},
}

def check_alerts(df):
    for _, r in df.iterrows():
        rule = alerts.get(r["symbol"])
        if not rule:
            continue
        price = r["last_price"]
        if "above" in rule and price >= rule["above"]:
            print(f"🔔 {r['ext.name']} 突破 {rule['above']}，现价 {price}")
        if "below" in rule and price <= rule["below"]:
            print(f"🔔 {r['ext.name']} 跌破 {rule['below']}，现价 {price}")

把 check_alerts(df) 加进循环即可。要避免重复提醒，可以记录已触发的标的，当天只提醒一次。

进阶：盯全市场而不只是自选股

如果你想监控的不是几只票，而是”全市场谁在异动”，可以用标的池查询一次拿到全 A 行情，再自己筛：

all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
all_cn["change_pct"] = (all_cn["last_price"] - all_cn["prev_close"]) / all_cn["prev_close"]

# 今日涨幅榜前 20
top = all_cn.sort_values("change_pct", ascending=False).head(20)
print(top[["symbol", "last_price", "change_pct", "amount"]])

标的池（universes）查询需要 Starter 及以上权限；自选股（symbols）查询免费额度即可使用。

常见问题（FAQ）

Q：能实时到毫秒吗？ A：快照接口约 3 秒刷新一次，适合盯盘和分钟级策略；不适合做 tick 级高频。

Q：收盘后还能刷吗？ A：收盘后返回的是最后一笔快照，价格不再变化。

Q：一次最多监控多少只？ A：symbols 支持批量传入，SDK 会自动分批；监控上百只也没问题。

Q：怎么做成后台常驻？ A：Linux/macOS 可以用 nohup python monitor.py & 或写成 systemd 服务；也可以用 crontab 在交易时段定时跑。

小结

一个实用的盯盘脚本，核心就三件事：批量取快照 → 循环刷新 → 条件提醒。有了批量实时行情接口，这些都能在几十行 Python 里完成，而且可以随意扩展成推送、存档、联动交易。

相关链接：