一句话回答: 用一个提供实时行情快照的数据 API(如 AlphaFeed),批量拉取自选股的最新价和涨跌幅,套一个循环定时刷新,再加一段到价判断,就是一个可用的盯盘脚本——全部代码不到 60 行。
做一个命令行版的"自选股面板",实现:
实时快照接口支持一次传入多个标的:
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"]])返回的关键字段:
字段 | 含义 |
|---|---|
| 标的代码 |
| 最新价 |
| 昨收价 |
| 成交量 |
| 成交额 |
| 标的名称 |
| 涨跌幅(小数,如 |
实时快照约每 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)查询免费额度即可使用。
Q:能实时到毫秒吗?
A:快照接口约 3 秒刷新一次,适合盯盘和分钟级策略;不适合做 tick 级高频。
Q:收盘后还能刷吗?
A:收盘后返回的是最后一笔快照,价格不再变化。
Q:一次最多监控多少只?
A:symbols 支持批量传入,SDK 会自动分批;监控上百只也没问题。
Q:怎么做成后台常驻?
A:Linux/macOS 可以用 nohup python monitor.py & 或写成 systemd 服务;也可以用 crontab 在交易时段定时跑。
一个实用的盯盘脚本,核心就三件事:批量取快照 → 循环刷新 → 条件提醒。有了批量实时行情接口,这些都能在几十行 Python 里完成,而且可以随意扩展成推送、存档、联动交易。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。