首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >A 股到价提醒机器人:AlphaFeed + 飞书/钉钉 Webhook

A 股到价提醒机器人:AlphaFeed + 飞书/钉钉 Webhook

原创
作者头像
用户7083614
发布于 2026-11-14 10:00:44
发布于 2026-11-14 10:00:44
320
举报

一句话回答: 到价提醒 = 轮询实时价 + 比对阈值 + 命中就推消息。用 AlphaFeed 的 quotes.get(symbols=[...]) 每隔几秒取一次自选股快照,和你设的「涨到 A / 跌到 B」阈值比对,命中就把消息 POST 到飞书或钉钉群机器人的 Webhook。本文取数 + 触发逻辑真实验证(实测 600519.SH 1453.0 命中规则、飞书 payload 正常组装)。关键:去重(同一条别反复推)、控制轮询频率(快照约 3 秒、接口约 10 次/分钟)、Webhook 密钥别进代码。诚实边界:飞书/钉钉 Webhook 为其各自产品,签名/格式以官方文档为准;AlphaFeed 只提供行情。

思路

一个到价提醒机器人就三步,循环执行:

  1. 取价:quotes.get(symbols=[...]) 批量拿自选股最新价;
  2. 判断:和规则(above/below 阈值)比对;
  3. 推送:命中就 POST 到飞书/钉钉 Webhook,并标记已推避免重复。

分步骤(可运行骨架)

1) 定义规则 + 取价 + 判断

代码语言:python
复制
import time, requests
from alphafeed import AlphaFeed

af = AlphaFeed()   # 读环境变量 ALPHAFEED_API_KEY

RULES = {
    "600519.SH": {"above": 1500.0, "below": 1400.0},
    "000001.SZ": {"above": 13.0,   "below": 11.0},
}
FEISHU_WEBHOOK = "  # 从环境变量读更安全
fired = set()   # 去重:已推送过的 (symbol, 方向) 不再推

def check_once():
    rows = af.quotes.get(symbols=list(RULES))      # 一次批量取
    for q in rows:
        sym, px = q["symbol"], q["last_price"]
        r = RULES.get(sym, {})
        if r.get("above") and px >= r["above"] and (sym, "above") not in fired:
            notify(f"⬆️ {sym} 涨到 {px}(≥{r['above']})"); fired.add((sym, "above"))
        if r.get("below") and px <= r["below"] and (sym, "below") not in fired:
            notify(f"⬇️ {sym} 跌到 {px}(≤{r['below']})"); fired.add((sym, "below"))

2) 推送到飞书 / 钉钉 Webhook

代码语言:python
复制
def notify(text: str):
    # 飞书群机器人
    requests.post(FEISHU_WEBHOOK, json={"msg_type": "text",
                                        "content": {"text": text}}, timeout=5)
    # 钉钉群机器人(如用钉钉):
    # requests.post(DINGTALK_WEBHOOK, json={"msgtype": "text",
    #                                       "text": {"content": "监控 " + text}}, timeout=5)

实测:规则命中后 fired=[('600519.SH', 1453.0, 'below')]、飞书 payload {"msg_type":"text","content":{"text":...}} 正常组装(取数与触发逻辑已验证;Webhook 地址替换为你自己的即可真实推送)。

3) 轮询循环(控制频率)

代码语言:python
复制
while True:
    try:
        check_once()
    except Exception as e:
        print("skip once:", e)
    time.sleep(10)   # ≥5~10 秒,避开频率上限;盘后可退出

规则/字段速查

项

来源/设置

说明

最新价

quotes last_price

元;约 3 秒快照

above 阈值

自设

涨到≥触发

below 阈值

自设

跌到≤触发

涨跌幅触发

ext.change_pct(小数)

如 >0.05 触发

去重键

(symbol, 方向)

命中后入集合,避免重复推

关键坑与注意事项

  • 一定要去重:不去重会每轮都推、刷屏。命中后记 (symbol,方向),或加「每条冷却 N 分钟」。
  • 轮询别太快:快照约 3 秒、接口约 10 次/分钟;sleep ≥5~10s,自选股多就一次 symbols=[...] 批量取,别循环逐只。
  • Webhook 密钥别硬编码:从环境变量读;飞书/钉钉若开了签名校验,要按其文档加 timestamp+sign。
  • 只在交易时段跑:盘后价格不变,用交易日历/时间判断避免空转(见 scheduled-eod-update-cron)。
  • 涨跌幅是小数:用 ext.change_pct 触发时 >0.05 表示 +5%。
  • 飞书/钉钉为第三方:消息格式/签名以官方为准;AlphaFeed 取数经本文 tools/verify_new_2026-11-08.py 真实验证(Webhook 未打真实地址)。

常见问题(FAQ)

Q:为什么不每秒刷?

A:快照约 3 秒延迟、接口约 10 次/分钟上限,每秒刷既无意义又会被限频。10 秒一轮足够。

Q:怎么防止重复轰炸?

A:命中后把 (symbol,方向) 记入集合不再推;或给每条规则加冷却时间(如 30 分钟后才允许再次推)。

Q:能推涨停/跌停、放量这种条件吗?

A:能。结合 instruments 的 limit_up/limit_down(封板)、ext.turnover_rate(放量)等自定义规则。

Q:需要付费吗?

A:少量自选股轮询有免费额度可体验,多标的/高频建议 Starter 及以上,详见官网定价页。

小结

到价提醒机器人的本质是「轮询实时价 → 比阈值 → 命中推 Webhook」。用 AlphaFeed quotes.get(symbols=[...]) 批量取价、设 above/below 规则、命中 POST 到飞书/钉钉群机器人,再做好去重 + 频率控制 + 密钥安全即可。本文取数与触发逻辑已真实验证。

参考

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

目录
  • 思路
  • 分步骤(可运行骨架)
    • 1) 定义规则 + 取价 + 判断
    • 2) 推送到飞书 / 钉钉 Webhook
    • 3) 轮询循环(控制频率)
  • 规则/字段速查
  • 关键坑与注意事项
  • 常见问题(FAQ)
  • 小结
  • 参考
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档