最近盯盘盯得眼睛疼。本来我们学写代码就是为了解放双手，结果天天还在手动切窗口看 A 股、港股和美股的突破情况。

俗话说，能用代码解决的重复劳动就绝不人肉。昨天下午我花了半个多小时，写了一个简单的 Python 脚本：

1. 初始化：每天开盘前，自动拉取监控列表中个股的历史 K 线，计算出 20 日阻力位（最高价）。

2. 盘中监控：通过 API 批量轮询个股的最新实时价格。

3. 报警推送：一旦盘中突破阻力位，且当天尚未报警，直接通过 Webhook 向飞书/企微群发送报警卡片。

今天把完整代码整理出来分享给大家，基本开箱即用。

为什么不用免费爬虫，而用专业 API？

在写这个监控机器人时，最忌讳两点：接口延迟和高频请求被封。
如果用爬虫去扒新浪财经或者雅虎财经，盘中每 10 秒轮询一次，没几分钟 IP 就会直接被 403 封锁；而且爬虫返回的数据格式极不规整。

为了省心，我这里用的是 QuantDash (quantdash.net) 的接口。它的 qd.quotes.get 支持直接传一个 list 批量获取 A股、港股和美股的实时行情，返回的直接是 Pandas DataFrame，数据清洗极其省事。

核心代码实现

直接上干货。你需要准备：

1. 一个 QuantDash 的 api_key（去官网注册拿个免费试用额度即可）。

2. 一个飞书机器人（或钉钉、企微）的 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_key=API_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(symbol=symbol, period="1d", adjust="forward", to_dataframe=True)
            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, json=payload, timeout=5)
        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(symbols=MONITOR_LIST, to_dataframe=True)
            
            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()

核心实现逻辑拆解

1. 多市场统一接口的优势：
   在编写 MONITOR_LIST 时，我们混合了 A股（600519.SH）、港股（00700.HK）和美股（AAPL.US）。如果自己去拼凑爬虫，需要写三套截然不同的解析规则。这里用 QuantDash 统一了返回结构，直接用一行 row['last_price'] 就能拿到最新价。

2. 防爆破机制（Deduplication）：
   实盘中，价格一旦突破阻力位，可能会在阻力位附近反复横跳。如果不做去重，群机器人一分钟就会推送几十条消息。我们使用 ALERTED_TODAY 集合，通过 f"{symbol}_{now.date()}" 作为 Key，确保每只股票在当天只触发一次预警。

3. 批量请求优化性能：
   千万不要在 while 循环里对每只股票单独请求 API。我们通过 qd.quotes.get(symbols=MONITOR_LIST) 把需要监控的股票合并成一个 List 发送，这样不仅网络开销小、策略延迟低，对 API Key 的频次配额也最友好。

进阶玩法拓展

这个脚本只是一个基础骨架。如果你想让你的预警机器人更智能，可以考虑在这个框架之上加入以下功能：

• 对接多重指标：结合 pandas-ta，在实时价格突破的同时，判断 15 分钟线的 RSI 是否超买，减少假突破的概率。

• 飞书富文本卡片：使用飞书的 Interactive Card 格式，把通知信息做得更好看，甚至可以带上直接跳转到交易软件的 URL 链接。

• 定时任务集成：使用 schedule 库，在每天早上 9 点自动重新执行 update_resistance_levels() 更新阻力位。

代码已经过本地调试，有兴趣的同学可以直接 Copy 下去改改参数跑起来。大家在写盯盘脚本时有什么踩坑经历，欢迎在评论区一起吹水。

参考文档：QuantDash - 专业金融数据平台