前言

这几天懂王那老头子不老实乱说话，造成全球股市涨跌幅度特别大；MJJ们上班时不能看股票，最多只能看一眼手机消息。今天闲着没事用gpt写了一个小脚本来监控推送涨跌通知，现分享出来。

青龙部署教程

• 1、进青龙面板-脚本管理里面，新建脚本或者上传脚本。命名：stock_monitor.py，如图示：

• 2、添加环境变量 STOCK_CODES ，PUSHDEER_PUSHKEY，THRESHOLD_PERCENT和PUSHDEER_PUSHKEY。说明：
  • STOCK_CODES=”sh601006,sz000001,sz300750″ # 监控多个股票用逗号隔开
  • PUSHDEER_PUSHKEY=”your_pushkey” #你的pushdeer秘钥
  • THRESHOLD_PERCENT=”2.0″ # 设置涨跌阈值
• 添加好后如图：

• 3、添加定时任务，自定义名称，脚本task stock_monitor.py，定时规则：
  • 25-59/5 9 1-5
  • /5 10-11 * 1-5
  • /5 13-14 * 1-5

添加好后如下图：

• 最后手动运行一次，推送效果如下：

说明：防止手机信息骚扰。到达设置的涨跌幅时才推送消息，没有时则不发送通知，并且首次推送通知后，再次推送间隔须1小时。运行时间为开市时才运行。

最后附上脚本代码：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
import time
import os
import json
from typing import Dict, Optional
# 环境变量配置
STOCK_CODES = os.environ.get("STOCK_CODES", "sh000422").split(",")     # 股票代码列表（逗号分隔）
PUSHDEER_PUSHKEY = os.environ.get("PUSHDEER_PUSHKEY")                 # PushDeer密钥
THRESHOLD_PERCENT = float(os.environ.get("THRESHOLD_PERCENT", "1.5")) # 涨跌阈值
INTERVAL_SECONDS = int(os.environ.get("INTERVAL_SECONDS", "3600"))    # 推送间隔
# 固定配置
TENCENT_API_URL = "http://qt.gtimg.cn/q={codes}"
TIME_FILE = "stock_push_times.json"  # 改为JSON格式存储时间戳
def validate_config():
    """验证必要配置"""
    if not PUSHDEER_PUSHKEY:
        raise ValueError("PUSHDEER_PUSHKEY 环境变量未设置")
    if not STOCK_CODES:
        raise ValueError("STOCK_CODES 不能为空")
    if THRESHOLD_PERCENT <= 0:
        raise ValueError("THRESHOLD_PERCENT 必须大于0")
def load_push_times() -> Dict[str, float]:
    """加载推送时间记录"""
    try:
        with open(TIME_FILE, 'r') as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {}
def save_push_time(stock_code: str, push_time: float):
    """保存单个股票的推送时间"""
    push_times = load_push_times()
    push_times[stock_code] = push_time
    try:
        with open(TIME_FILE, 'w') as f:
            json.dump(push_times, f)
    except Exception as e:
        print(f"保存时间记录失败: {e}")
def get_stocks_data() -> Dict[str, dict]:
    """批量获取股票数据"""
    headers = {
        "Referer": "http://qt.gtimg.cn/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }

    try:
        resp = requests.get(TENCENT_API_URL.format(codes=",".join(STOCK_CODES)), 
                          headers=headers, timeout=15)
        resp.encoding = 'gbk'
        return parse_stock_data(resp.text)
    except Exception as e:
        print(f"获取股票数据失败: {e}")
        return {}
def parse_stock_data(raw_data: str) -> Dict[str, dict]:
    """解析腾讯财经返回数据"""
    stocks = {}
    for line in raw_data.splitlines():
        if not line:
            continue
        try:
            # 示例数据行：v_sh601006="1~大秦铁路~600006~6.74~6.72~6.73...
            code_part, data_part = line.split("=")
            stock_code = code_part.split("_")[1].lower()  # 统一转为小写
            data_str = data_part.strip('";')
            data = data_str.split("~")

            if len(data) < 5:
                continue

            stocks[stock_code] = {
                "name": data[1],
                "price": float(data[3]),
                "prev_close": float(data[4])
            }
        except Exception as e:
            print(f"解析数据异常: {e}")
    return stocks
def calculate_change(current: float, previous: float) -> float:
    """计算涨跌幅百分比"""
    if previous == 0:
        return 0.0
    return round((current - previous) / previous * 100, 2)
def check_cooldown(stock_code: str) -> bool:
    """检查单个股票的冷却时间"""
    push_times = load_push_times()
    last_push = push_times.get(stock_code, 0)
    return (time.time() - last_push) >= INTERVAL_SECONDS
def push_notification(text: str, desp: str) -> bool:
    """通过PushDeer发送通知"""
    push_url = "https://api2.pushdeer.com/message/push"
    params = {
        "pushkey": PUSHDEER_PUSHKEY,
        "text": text,
        "desp": desp
    }
    try:
        resp = requests.get(push_url, params=params, timeout=10)
        return resp.json().get("code") == 0
    except Exception as e:
        print(f"推送失败: {e}")
        return False
def process_stock(stock_code: str, data: dict):
    """处理单个股票的逻辑"""
    current = data["price"]
    prev_close = data["prev_close"]
    stock_name = data["name"]

    change = calculate_change(current, prev_close)
    if abs(change) < THRESHOLD_PERCENT:
        print(f"[{stock_code}] 涨跌幅未达阈值: {change}%")
        return

    if not check_cooldown(stock_code):
        print(f"[{stock_code}] 冷却时间内不推送")
        return

    # 准备推送内容
    direction = "涨" if change > 0 else "跌"
    text = f"{stock_name} 价格波动预警"
    desp = f"""股票代码: {stock_code}
当前价格: {current}元
昨日收盘: {prev_close}元
涨跌幅: {change}% ({direction}幅超过{THRESHOLD_PERCENT}%)"""

    if push_notification(text, desp):
        save_push_time(stock_code, time.time())
        print(f"[{stock_code}] 推送成功")
    else:
        print(f"[{stock_code}] 推送失败")
def main():
    try:
        validate_config()
        stocks_data = get_stocks_data()

        if not stocks_data:
            print("未获取到有效股票数据")
            return

        for code in STOCK_CODES:
            # 统一转为小写匹配
            code_lower = code.strip().lower()
            if stock_data := stocks_data.get(code_lower):
                process_stock(code_lower, stock_data)
            else:
                print(f"[{code}] 未找到相关数据")

    except Exception as e:
        print(f"程序异常: {e}")
if __name__ == "__main__":
    main()