拒绝人肉盯盘！用 Python 快速撸一个「多市场个股突破」飞书/企微报警机器人

最近盯盘盯得眼睛疼。本来我们学写代码就是为了解放双手结果天天还在手动切窗口看 A 股、港股和美股的突破情况。俗话说能用代码解决的重复劳动就绝不人肉。昨天下午我花了半个多小时写了一个简单的 Python 脚本初始化每天开盘前自动拉取监控列表中个股的历史 K 线计算出 20 日阻力位最高价。盘中监控通过 API 批量轮询个股的最新实时价格。报警推送一旦盘中突破阻力位且当天尚未报警直接通过 Webhook 向飞书/企微群发送报警卡片。今天把完整代码整理出来分享给大家基本开箱即用。为什么不用免费爬虫而用专业 API在写这个监控机器人时最忌讳两点接口延迟和高频请求被封。如果用爬虫去扒新浪财经或者雅虎财经盘中每 10 秒轮询一次没几分钟 IP 就会直接被 403 封锁而且爬虫返回的数据格式极不规整。为了省心我这里用的是QuantDash (quantdash.net)的接口。它的 qd.quotes.get 支持直接传一个 list 批量获取 A股、港股和美股的实时行情返回的直接是 Pandas DataFrame数据清洗极其省事。核心代码实现直接上干货。你需要准备一个 QuantDash 的 api_key去官网注册拿个免费试用额度即可。一个飞书机器人或钉钉、企微的 Webhook 链接。import time import requests import pandas as pd from datetime import datetime from quantdash import QuantDash # 配置区 API_KEY your_quantdash_api_key # 替换为你的 QuantDash API KEY FEISHU_WEBHOOK https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxx # 替换为你的机器人Webhook # 监控列表支持跨市场A股/港股/美股 MONITOR_LIST [600519.SH, 00700.HK, AAPL.US] # qd QuantDash(api_keyAPI_KEY) RESISTANCE_LEVELS {} # 存储个股20日阻力位 ALERTED_TODAY set() # 记录当天已报警的股票防止重复轰炸 def update_resistance_levels(): 获取历史数据计算20日最高价作为阻力位 print(f[{datetime.now().strftime(%H:%M:%S)}] 开始初始化 20 日阻力位...) for symbol in MONITOR_LIST: try: # 获取最近30天的日K采用前复权 df qd.klines.get(symbolsymbol, period1d, adjustforward, to_dataframeTrue) if df is not None and not df.empty: # 取最近的20个交易日 recent_20 df.tail(20) max_high recent_20[high].max() RESISTANCE_LEVELS[symbol] max_high print(f - 标的: {symbol} | 20日阻力位: {max_high}) else: print(f - 标的: {symbol} 历史数据获取为空请检查代码格式。) except Exception as e: print(f - 标的: {symbol} 初始化异常: {e}) def send_webhook_notification(symbol, current_price, resistance): 向飞书群推送格式化报警信息 payload { msg_type: text, content: { text: ( f 策略预警个股价格突破\n f━━━━━━━━━━━━━━━━\n f 标的代码{symbol}\n f 当前价格{current_price}\n f 20日阻力{resistance}\n f⏰ 突破时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n f━━━━━━━━━━━━━━━━\n f请及时关注盘中交易机会。 ) } } try: resp requests.post(FEISHU_WEBHOOK, jsonpayload, timeout5) if resp.status_code 200: print(f - 报警推送成功: {symbol} 突破) else: print(f - 报警推送失败: {resp.text}) except Exception as e: print(f - 发送 Webhook 异常: {e}) def start_monitoring(): 盘中实时轮询监控主循环 print(f\n[{datetime.now().strftime(%H:%M:%S)}] 启动实时轮询监控...) while True: # 简单过滤非交易时间可根据实际需要加精准的交易日历判断 now datetime.now() # 这里仅作演示默认全天轮询 try: # 1. 一键批量获取最新行情大大减少 API 调用次数和延迟 df_quotes qd.quotes.get(symbolsMONITOR_LIST, to_dataframeTrue) if df_quotes is not None and not df_quotes.empty: for _, row in df_quotes.iterrows(): symbol row[symbol] current_price row[last_price] # 实时最新价 resistance RESISTANCE_LEVELS.get(symbol) if not resistance: continue # 2. 判断是否突破阻力位 if current_price resistance: alert_key f{symbol}_{now.date()} # 3. 避免日内重复报警 if alert_key not in ALERTED_TODAY: send_webhook_notification(symbol, current_price, resistance) ALERTED_TODAY.add(alert_key) except Exception as e: print(f轮询出错: {e}) # 每10秒轮询一次QuantDash 接口响应在毫秒级10s绰绰有余 time.sleep(10) if __name__ __main__: # 1. 盘前初始化阻力位 update_resistance_levels() # 2. 开启盘中监控 start_monitoring()核心实现逻辑拆解多市场统一接口的优势在编写 MONITOR_LIST 时我们混合了 A股600519.SH、港股00700.HK和美股AAPL.US。如果自己去拼凑爬虫需要写三套截然不同的解析规则。这里用 QuantDash 统一了返回结构直接用一行 row[last_price] 就能拿到最新价。防爆破机制Deduplication实盘中价格一旦突破阻力位可能会在阻力位附近反复横跳。如果不做去重群机器人一分钟就会推送几十条消息。我们使用 ALERTED_TODAY 集合通过 f{symbol}_{now.date()} 作为 Key确保每只股票在当天只触发一次预警。批量请求优化性能千万不要在 while 循环里对每只股票单独请求 API。我们通过 qd.quotes.get(symbolsMONITOR_LIST) 把需要监控的股票合并成一个 List 发送这样不仅网络开销小、策略延迟低对 API Key 的频次配额也最友好。进阶玩法拓展这个脚本只是一个基础骨架。如果你想让你的预警机器人更智能可以考虑在这个框架之上加入以下功能对接多重指标结合 pandas-ta在实时价格突破的同时判断 15 分钟线的 RSI 是否超买减少假突破的概率。飞书富文本卡片使用飞书的 Interactive Card 格式把通知信息做得更好看甚至可以带上直接跳转到交易软件的 URL 链接。定时任务集成使用 schedule 库在每天早上 9 点自动重新执行 update_resistance_levels() 更新阻力位。代码已经过本地调试有兴趣的同学可以直接 Copy 下去改改参数跑起来。大家在写盯盘脚本时有什么踩坑经历欢迎在评论区一起吹水。参考文档QuantDash - 专业金融数据平台