我用迅投miniqmt写个A股涨跌监控，盘面异动秒级推钉钉

【股票监控】2025-05-16 14:23:15
📉 上证大跌提醒
红盘 680 家 | 绿盘 4100 家
上证涨跌幅: -1.37%

from xtquant import xtdata

stocks = xtdata.get_stock_list_in_sector("沪深京A股")
tick_data = xtdata.get_full_tick(stocks)

for code in stocks:
    tick = tick_data[code]
    last_close = tick.get("lastClose", 0)
    last_price = tick.get("lastPrice", 0)
    change_pct = (last_price - last_close) / last_close * 100
    
    if change_pct > 0.01:
        red += 1
    elif change_pct < -0.01:
        green += 1

payload = {
    "msgtype": "text",
    "text": {"content": "【股票监控】红盘 680 家 | 绿盘 4100 家\n上证涨跌幅: -1.37%"},
    "at": {"isAtAll": True},
}
urllib.request.urlopen(req, timeout=10)

while True:
    tick_data = xtdata.get_full_tick(all_codes)
    result = count_red_green(all_codes, tick_data)
    alerts = tracker.update(result["red"], result["green"], index_chg)
    for msg in alerts:
        send_dingtalk(msg)
    time.sleep(30)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import time
import urllib.request
import json
import logging
import datetime
from typing import Set
from xtquant import xtdata
# ---- 钉钉机器人（关键字模式）----
# 创建机器人时安全设置选"自定义关键词"，添加关键词"股票监控"
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxx"
DINGTALK_KEYWORD = "股票监控"  # 钉钉机器人设置的关键词，消息中必须包含此词才能发送
# ---- 告警阈值 ----
RED_LOW_THRESHOLD    = 500     # 红盘家数低于此值告警（极弱）
RED_HIGH_THRESHOLD   = 4000    # 红盘家数高于此值告警（极强）
INDEX_CHANGE_LIMIT   = 1.0     # 上证指数涨跌幅超过此百分比告警（正负双向）
# ---- 监控标的 ----
INDEX_CODE = "000001.SH"       # 上证指数
# ---- 扫描频率 ----
SCAN_INTERVAL = 30             # 扫描间隔（秒），交易时段内每N秒全推一次
# ---- 日志 ----
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)
# ============================================================
#  钉钉推送（关键字模式）
# ============================================================
def send_dingtalk(message: str, at_all: bool = True) -> bool:
    """
    发送钉钉文本消息（关键字模式，消息中须包含设置的关键词才能发送）
    message: 文本内容
    at_all: 是否@所有人
    """
    if "YOUR_TOKEN" in DINGTALK_WEBHOOK:
        logger.warning("钉钉 webhook 未配置，跳过推送（消息内容已打印到日志）")
        return False
    url = DINGTALK_WEBHOOK
    payload = {
        "msgtype": "text",
        "text": {"content": message},
        "at": {"isAtAll": at_all},
    }
    try:
        req = urllib.request.Request(
            url,
            data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            headers={"Content-Type": "application/json"},
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            result = json.loads(resp.read().decode("utf-8"))
            if result.get("errcode") == 0:
                logger.info("钉钉推送成功")
                return True
            else:
                logger.error(f"钉钉推送失败: {result}")
                return False
    except Exception as e:
        logger.error(f"钉钉推送异常: {e}")
        return False
# ============================================================
#  市场状态追踪（防止重复告警）
# ============================================================
class MarketStateTracker:
    """
    追踪市场状态，只在状态发生变化时才触发告警。
    """
    def __init__(self):
        self.alert_active: Set[str] = set()
    def update(self, red_count: int, green_count: int, index_change_pct: float) -> list:
        """
        根据最新数据判断是否需要告警
        返回需要推送的告警消息列表（空=无新告警）
        """
        alerts = []
        # ---- 红盘家数 < 500 ----
        condition_id = "red_low"
        if red_count < RED_LOW_THRESHOLD:
            if condition_id not in self.alert_active:
                self.alert_active.add(condition_id)
                alerts.append(
                    f"🔴 极弱盘面警告\n"
                    f"红盘 {red_count} 家 | 绿盘 {green_count} 家\n"
                    f"上证涨跌幅: {index_change_pct:+.2f}%"
                )
        else:
            self.alert_active.discard(condition_id)
        # ---- 红盘家数 > 4000 ----
        condition_id = "red_high"
        if red_count > RED_HIGH_THRESHOLD:
            if condition_id not in self.alert_active:
                self.alert_active.add(condition_id)
                alerts.append(
                    f"🟠 过热盘面警告\n"
                    f"红盘 {red_count} 家 | 绿盘 {green_count} 家\n"
                    f"上证涨跌幅: {index_change_pct:+.2f}%"
                )
        else:
            self.alert_active.discard(condition_id)
        # ---- 上证涨跌幅 > +1% ----
        condition_id = "index_up"
        if index_change_pct > INDEX_CHANGE_LIMIT:
            if condition_id not in self.alert_active:
                self.alert_active.add(condition_id)
                alerts.append(
                    f"📈 上证大涨提醒\n"
                    f"红盘 {red_count} 家 | 绿盘 {green_count} 家\n"
                    f"上证涨跌幅: {index_change_pct:+.2f}%"
                )
        else:
            self.alert_active.discard(condition_id)
        # ---- 上证涨跌幅 < -1% ----
        condition_id = "index_down"
        if index_change_pct < -INDEX_CHANGE_LIMIT:
            if condition_id not in self.alert_active:
                self.alert_active.add(condition_id)
                alerts.append(
                    f"📉 上证大跌提醒\n"
                    f"红盘 {red_count} 家 | 绿盘 {green_count} 家\n"
                    f"上证涨跌幅: {index_change_pct:+.2f}%"
                )
        else:
            self.alert_active.discard(condition_id)
        return alerts
# ============================================================
#  核心监控逻辑
# ============================================================
def get_all_a_stock_codes() -> list:
    """获取全市场 A 股代码列表（沪深京）"""
    stocks = xtdata.get_stock_list_in_sector("沪深京A股")
    return list(set(stocks))
def count_red_green(codes: list, tick_data: dict) -> dict:
    """
    统计涨跌家数
    tick_data: get_full_tick 返回的数据
    返回: {"red": 红盘数, "green": 绿盘数, "flat": 平盘数, "total": 总数}
    """
    red = 0
    green = 0
    flat = 0
    for code in codes:
        if code not in tick_data or not tick_data[code]:
            continue
        tick = tick_data[code]
        # get_full_tick 返回字段: lastPrice(最新价) / lastClose(昨收盘价)
        last_close = tick.get("lastClose", 0)    # 昨收盘价
        last_price = tick.get("lastPrice", 0)     # 最新价
        if last_close <= 0 or last_price <= 0:
            continue
        change_pct = (last_price - last_close) / last_close * 100
        if change_pct > 0.01:
            red += 1
        elif change_pct < -0.01:
            green += 1
        else:
            flat += 1
    return {"red": red, "green": green, "flat": flat, "total": red + green + flat}
def get_index_change_pct(tick_data: dict) -> float:
    """获取上证指数涨跌幅(%)"""
    tick = tick_data.get(INDEX_CODE)
    if not tick:
        return 0.0
    last_close = tick.get("lastClose", 0)    # 昨收盘价
    last_price = tick.get("lastPrice", 0)     # 最新价
    if last_close <= 0 or last_price <= 0:
        return 0.0
    return (last_price - last_close) / last_close * 100
def is_trading_time() -> bool:
    """判断当前是否为交易时段（周一至周五 9:25~15:00）"""
    now = datetime.datetime.now()
    if now.weekday() >= 5:
        return False
    morning_start = now.replace(hour=9, minute=25, second=0)
    afternoon_end = now.replace(hour=15, minute=0, second=0)
    return morning_start <= now <= afternoon_end
def run():
    """主监控循环"""
    # 连接行情
    xtdata.connect()
    logger.info("xtdata 行情连接成功")
    # 下载上证指数历史数据
    xtdata.download_history_data(INDEX_CODE, period="1d", start_time="", end_time="")
    # 获取全市场 A 股代码
    all_codes = get_all_a_stock_codes()
    logger.info(f"全市场 A 股数量: {len(all_codes)}")
    # 状态追踪器
    tracker = MarketStateTracker()
    # 上一次打印摘要的时间
    last_summary_time = 0
    while True:
        try:
            # 非交易时段等待
            # if not is_trading_time():
            #     now = datetime.datetime.now()
            #     next_check = now.replace(hour=9, minute=25, second=0)
            #     if now > next_check:
            #         next_check += datetime.timedelta(days=1)
            #     wait_sec = max(60, (next_check - now).total_seconds())
            #     logger.info(f"非交易时段，{wait_sec:.0f}秒后重试")
            #     time.sleep(min(wait_sec, 300))
            #     continue
            # 1. 获取全市场分笔行情
            tick_data = xtdata.get_full_tick(all_codes)
            # 2. 统计涨跌家数
            result = count_red_green(all_codes, tick_data)
            # 3. 获取上证指数涨跌幅
            index_tick = xtdata.get_full_tick([INDEX_CODE])
            index_chg = get_index_change_pct(index_tick)
            # 4. 打印摘要（每60秒打印一次）
            now_ts = time.time()
            if now_ts - last_summary_time >= 60:
                now_str = datetime.datetime.now().strftime("%H:%M:%S")
                logger.info(
                    f"[{now_str}] 红盘={result['red']} 绿盘={result['green']} "
                    f"平盘={result['flat']} 共{result['total']}家 | "
                    f"上证{INDEX_CODE} {index_chg:+.2f}%"
                )
                last_summary_time = now_ts
            # 5. 检查告警条件
            alerts = tracker.update(result["red"], result["green"], index_chg)
            for msg in alerts:
                now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                full_msg = f"【股票监控】{now_str}\n{msg}"
                logger.info(f"触发告警:\n{full_msg}")
                send_dingtalk(full_msg)
            # 6. 等待下一次扫描
            time.sleep(SCAN_INTERVAL)
        except KeyboardInterrupt:
            logger.info("用户中断，监控退出")
            break
        except Exception as e:
            logger.error(f"监控循环异常: {e}", exc_info=True)
            time.sleep(30)
if __name__ == "__main__":
    run()