shield_lock
金大哥
← 返回 Skill 列表
extension
分类: 数据与分析无需 API Key

股票涨跌自动监控提醒

股票涨跌自动监控提醒 v1.0.1 — 修复quick_add TypeError、优化逐股查询(不再全市场拉取)、新增5分钟去重机制。设置价格/涨跌幅阈值，A股实时监控，突破时自动推送提醒。支持多股票同时监控、批量配置、历史提醒记录。

person作者: user_d9314a2ahubcommunity
download下载资源包

slug: stock-price-alert-cn name: stock-price-alert description: "股票涨跌自动监控提醒 v1.0.1 — 修复quick_add TypeError、优化逐股查询(不再全市场拉取)、新增5分钟去重机制。设置价格/涨跌幅阈值，A股实时监控，突破时自动推送提醒。支持多股票同时监控、批量配置、历史提醒记录。" version: 1.0.1 trigger:

  • 股票监控
  • 股价提醒
  • 涨跌提醒
  • 价格预警
  • 股票预警
  • 股价监控
  • 设置提醒
  • 股票通知
  • 突破提醒
  • 添加监控
  • 我的监控
  • 查看提醒 category:
  • 数据分析 tags:
  • 股票
  • 监控
  • 提醒
  • A股
  • 实时行情 displayName: "股票涨跌自动监控提醒" summary: "股票涨跌自动监控提醒 — 设置价格/涨跌幅阈值，A 股实时监控，突破时自动推送提醒。支持多股票同时监控、批量配置、历史提醒记录、5 分钟去重。"

股票涨跌自动监控提醒 v1.0.1

触发条件

当用户提到 股票监控/股价提醒/涨跌提醒/价格预警/股票预警/股价监控/设置提醒/股票通知/突破提醒/添加监控/查看提醒 时自动加载。

快速决策树

用户请求股票监控
  ├── 添加监控股票？→ [A] 设置监控 + 价格/涨跌幅阈值
  ├── 查看当前监控？→ [B] 列出所有监控 + 当前价格状态
  ├── 手动检查触发？→ [C] 轮询所有监控 → 输出触发清单
  ├── 删除/修改监控？→ [D] 管理监控列表
  └── 查看历史提醒？→ [E] 提醒记录查询

依赖安装

pip install akshare pandas -q

akshare 免费无需 API Key，数据来自东方财富等公开接口。

工作流 A：添加监控

A1: 添加价格阈值监控

import json, os

WATCHLIST_FILE = os.path.expanduser("~/.hermes/stock_watchlist.json")

def load_watchlist():
    if os.path.exists(WATCHLIST_FILE):
        with open(WATCHLIST_FILE, 'r', encoding='utf-8') as f:
            return json.load(f)
    return []

def save_watchlist(data):
    with open(WATCHLIST_FILE, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def add_price_alert(stock_code, stock_name="", upper_price=None, lower_price=None,
                    upper_pct=None, lower_pct=None, note=""):
    """
    添加股票价格监控
    stock_code: 股票代码，如 '000001' 或 '600519'
    upper_price: 上涨到该价格时提醒（元）
    lower_price: 下跌到该价格时提醒（元）
    upper_pct: 涨幅超过该百分比时提醒（如 5 表示 5%）
    lower_pct: 跌幅超过该百分比时提醒（如 -3 表示跌 3%）
    """
    watchlist = load_watchlist()

    # 检查是否已存在
    code_clean = stock_code.replace('sh', '').replace('sz', '').replace('.', '')
    for item in watchlist:
        if item['code'] == code_clean:
            # 更新已有监控
            if upper_price: item['upper_price'] = upper_price
            if lower_price: item['lower_price'] = lower_price
            if upper_pct is not None: item['upper_pct'] = upper_pct
            if lower_pct is not None: item['lower_pct'] = lower_pct
            if note: item['note'] = note
            save_watchlist(watchlist)
            return {"action": "updated", "code": code_clean, "name": item.get('name', '')}

    # 新建监控
    entry = {
        "code": code_clean,
        "name": stock_name or code_clean,
        "upper_price": upper_price,
        "lower_price": lower_price,
        "upper_pct": upper_pct,
        "lower_pct": lower_pct,
        "note": note,
        "created_at": __import__('datetime').datetime.now().isoformat(),
        "last_price": None,
        "last_check": None,
    }
    watchlist.append(entry)
    save_watchlist(watchlist)
    return {"action": "added", "code": code_clean, "name": stock_name or code_clean}

# 示例用法
# add_price_alert("000001", "平安银行", upper_price=12.50, lower_price=10.00, note="测试")
# add_price_alert("600519", "贵州茅台", upper_pct=5, lower_pct=-3)

A2: 快速添加（交互式）

def quick_add(stock_code, threshold_str):
    """
    快速添加监控
    threshold_str 格式：
      ">15.5" — 突破15.5元提醒
      "<10.0" — 跌破10元提醒  
      "12~15" — 12-15元区间监控
      "+5%" — 涨5%提醒
      "-3%" — 跌3%提醒
      ">15.5,-5%" — 组合条件
    """
    import re

    upper_price, lower_price = None, None
    upper_pct, lower_pct = None, None

    parts = threshold_str.replace('，', ',').split(',')

    for part in parts:
        part = part.strip()
        # 价格阈值
        m = re.match(r'>(\d+\.?\d*)', part)
        if m: upper_price = float(m.group(1))

        m = re.match(r'<(\d+\.?\d*)', part)
        if m: lower_price = float(m.group(1))

        m = re.match(r'(\d+\.?\d*)~(\d+\.?\d*)', part)
        if m:
            lower_price = float(m.group(1))
            upper_price = float(m.group(2))

        # 百分比阈值
        m = re.match(r'\+(\d+\.?\d*)%', part)
        if m: upper_pct = float(m.group(1))

        m = re.match(r'\-(\d+\.?\d*)%', part)
        if m: lower_pct = -float(m.group(1))

    return add_price_alert(stock_code,
                           upper_price=upper_price, lower_price=lower_price,
                           upper_pct=upper_pct, lower_pct=lower_pct)

# 示例
# quick_add("000001", ">12.5,<10.0")
# quick_add("600519", "+5%,-3%")

工作流 B：查看监控列表

def list_watchlist():
    """列出所有监控及当前价格状态"""
    watchlist = load_watchlist()
    if not watchlist:
        return "当前没有监控中的股票。使用「添加监控」开始。\n\n示例：监控 000001 12~15元"

    # 获取实时价格
    prices = get_realtime_prices([w['code'] for w in watchlist])

    lines = ["## 📊 股票监控列表\n"]
    lines.append(f"共 {len(watchlist)} 只股票\n")

    for w in watchlist:
        code = w['code']
        name = w.get('name', code)
        current = prices.get(code, {}).get('price')
        change_pct = prices.get(code, {}).get('change_pct')

        lines.append(f"### {name}（{code}）")
        if current:
            arrow = "🔺" if (change_pct or 0) > 0 else ("🔻" if (change_pct or 0) < 0 else "➖")
            lines.append(f"  现价: {arrow} {current:.2f}元 ({change_pct:+.2f}%)")
        else:
            lines.append(f"  现价: 获取中...")

        conditions = []
        if w.get('upper_price'):
            conditions.append(f"涨到 {w['upper_price']}元")
            if current and current >= w['upper_price']:
                conditions[-1] += " 🚨已触发"
        if w.get('lower_price'):
            conditions.append(f"跌到 {w['lower_price']}元")
            if current and current <= w['lower_price']:
                conditions[-1] += " 🚨已触发"
        if w.get('upper_pct'):
            conditions.append(f"涨幅超 {w['upper_pct']}%")
        if w.get('lower_pct'):
            conditions.append(f"跌幅超 {abs(w['lower_pct'])}%")

        if w.get('note'):
            conditions.append(f"备注: {w['note']}")

        for c in conditions:
            lines.append(f"  - {c}")
        lines.append("")

    return '\n'.join(lines)

工作流 C：检查并触发提醒

C1: 获取实时价格

def get_realtime_prices(codes):
    """批量获取股票实时价格（逐股查询东方财富API，避免全市场拉取）"""
    import urllib.request, json
    result = {}
    for code in codes:
        try:
            market = "1" if code.startswith('6') else "0"
            url = f"https://push2.eastmoney.com/api/qt/stock/get?secid={market}.{code}&fields=f43,f44,f45,f46,f47,f48,f50,f51,f52,f57,f58,f60,f169,f170"
            req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
            resp = urllib.request.urlopen(req, timeout=10)
            data = json.loads(resp.read())
            d = data.get('data', {})
            if d:
                result[code] = {
                    'price': d.get('f43', 0) / 100 if d.get('f43') else None,
                    'change_pct': d.get('f170', 0) / 100 if d.get('f170') else None,
                    'high': d.get('f44', 0) / 100 if d.get('f44') else None,
                    'low': d.get('f45', 0) / 100 if d.get('f45') else None,
                    'volume': d.get('f47', 0),
                    'name': d.get('f58', code),
                }
        except Exception as e:
            result[code] = {'price': None, 'error': str(e)}
    return result

def get_realtime_prices_bulk(codes):
    """备选：通过 akshare 全市场拉取（适合监控5+只股票时减少请求数）"""
    try:
        import akshare as ak
        df = ak.stock_zh_a_spot_em()
        result = {}
        for code in codes:
            row = df[df['代码'] == code]
            if not row.empty:
                r = row.iloc[0]
                result[code] = {
                    'price': float(r['最新价']),
                    'change_pct': float(r['涨跌幅']),
                    'high': float(r['最高']),
                    'low': float(r['最低']),
                    'volume': float(r['成交量']),
                    'name': r['名称'],
                }
        return result
    except Exception as e:
        print(f"akshare批量获取失败: {e}，回退到逐股查询...")
        return get_realtime_prices(codes)

C2: 检查提醒触发

import os, json
from datetime import datetime

ALERT_LOG_FILE = os.path.expanduser("~/.hermes/stock_alerts.json")

def load_alerts_log():
    if os.path.exists(ALERT_LOG_FILE):
        with open(ALERT_LOG_FILE, 'r', encoding='utf-8') as f:
            return json.load(f)
    return []

def save_alerts_log(data):
    with open(ALERT_LOG_FILE, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def check_alerts():
    """检查所有监控，返回触发的提醒列表（v1.0.1 新增去重：同股票同类型5分钟内不重复）"""
    watchlist = load_watchlist()
    if not watchlist:
        return []

    codes = [w['code'] for w in watchlist]
    prices = get_realtime_prices(codes)
    alerts_log = load_alerts_log()
    triggered = []
    
    # 去重窗口：5分钟内同股票同类型不重复
    DEDUP_WINDOW = 300  # 秒
    now_ts = datetime.now().timestamp()

    for w in watchlist:
        code = w['code']
        current = prices.get(code, {})
        price = current.get('price')
        change_pct = current.get('change_pct')

        if price is None:
            continue

        alerts = []
        # 价格突破检查
        if w.get('upper_price') and price >= w['upper_price']:
            alerts.append(("upper_price", f"🔺 突破上限 {w['upper_price']}元（现价 {price:.2f}）"))
        if w.get('lower_price') and price <= w['lower_price']:
            alerts.append(("lower_price", f"🔻 跌破下限 {w['lower_price']}元（现价 {price:.2f}）"))
        if w.get('upper_pct') and change_pct and change_pct >= w['upper_pct']:
            alerts.append(("upper_pct", f"📈 涨幅 {change_pct:+.2f}% 超阈值 {w['upper_pct']}%"))
        if w.get('lower_pct') and change_pct and change_pct <= w['lower_pct']:
            alerts.append(("lower_pct", f"📉 跌幅 {change_pct:+.2f}% 超阈值 {w['lower_pct']}%"))

        if alerts:
            # 去重：检查最近记录中是否有相同股票+类型
            for alert_type, alert_msg in alerts:
                is_dup = False
                for past in alerts_log:
                    past_ts = datetime.fromisoformat(past['time']).timestamp()
                    if (now_ts - past_ts) < DEDUP_WINDOW and past['code'] == code:
                        # 检查同一类型是否已在窗口内触发
                        for past_alert in past.get('alerts', []):
                            if alert_type in past_alert:
                                is_dup = True
                                break
                    if is_dup:
                        break
                
                if is_dup:
                    continue  # 跳过重复提醒
                
                # 非重复 → 记录
                record = {
                    "code": code,
                    "name": w.get('name', code),
                    "price": price,
                    "change_pct": change_pct,
                    "alerts": [alert_msg],
                    "time": datetime.now().isoformat(),
                }
                triggered.append(record)
                alerts_log.append(record)

        # 更新最后价格
        w['last_price'] = price
        w['last_check'] = datetime.now().isoformat()

    save_watchlist(watchlist)
    if triggered:
        save_alerts_log(alerts_log)

    return triggered

def format_alerts(triggered):
    """格式化提醒输出"""
    if not triggered:
        return "✅ 所有监控正常，无触发提醒。"

    lines = ["## 🚨 股票提醒触发！\n"]
    for r in triggered:
        name = r['name']
        code = r['code']
        price = r['price']
        change = r.get('change_pct', 0)
        arrow = "🔺" if (change or 0) > 0 else "🔻"
        lines.append(f"### {name}（{code}）{arrow} {price:.2f}元 ({change:+.2f}%)")
        for a in r['alerts']:
            lines.append(f"  - {a}")
        lines.append(f"  *触发时间: {r['time']}*")
        lines.append("")

    return '\n'.join(lines)

C3: 一键检查

def run_check():
    """执行一次完整检查并输出结果"""
    triggered = check_alerts()
    return format_alerts(triggered)

# 使用：
# print(run_check())

工作流 D：管理监控

def remove_alert(stock_code):
    """删除某只股票的监控"""
    code_clean = stock_code.replace('sh', '').replace('sz', '').replace('.', '')
    watchlist = load_watchlist()
    new_list = [w for w in watchlist if w['code'] != code_clean]
    if len(new_list) == len(watchlist):
        return f"未找到 {code_clean} 的监控"
    save_watchlist(new_list)
    return f"已删除 {code_clean} 的监控（剩余 {len(new_list)} 只）"

def clear_all():
    """清空所有监控"""
    save_watchlist([])
    return "已清空所有监控"

def modify_alert(stock_code, **kwargs):
    """修改监控 — 使用方法同 add_price_alert，会更新已有条目"""
    return add_price_alert(stock_code, **kwargs)

工作流 E：历史提醒

def alert_history(days=7):
    """查看最近的提醒记录"""
    alerts = load_alerts_log()
    if not alerts:
        return "暂无提醒记录。"

    cutoff = datetime.now().timestamp() - days * 86400
    recent = [a for a in alerts if datetime.fromisoformat(a['time']).timestamp() > cutoff]

    if not recent:
        return f"最近 {days} 天无提醒记录。"

    lines = [f"## 📋 最近 {days} 天提醒记录\n"]
    lines.append(f"共 {len(recent)} 条触发记录\n")

    for r in reversed(recent):
        name = r['name']
        code = r['code']
        lines.append(f"- **{r['time'][:16]}** | {name}({code}) | {r['price']:.2f}元")
        for a in r['alerts']:
            lines.append(f"    {a}")

    return '\n'.join(lines)

定时自动检查（推荐设置）

设置定时任务，每 5 分钟检查一次（仅交易时段 9:30-15:00）：

# 创建检查脚本
cat > ~/.hermes/scripts/stock_check.py << 'EOF'
"""股票监控定时检查"""
import json, os, sys
sys.path.insert(0, os.path.dirname(__file__))
from datetime import datetime

# 判断是否交易时段
now = datetime.now()
weekday = now.weekday()  # 0=Mon, 6=Sun
hour, minute = now.hour, now.minute
is_trading = (weekday < 5 and (
    (hour == 9 and minute >= 30) or
    (hour == 10) or (hour == 11 and minute <= 30) or
    (hour >= 13 and hour < 15)
))

if not is_trading:
    # 非交易时段静默
    sys.exit(0)

# 执行检查
WATCHLIST_FILE = os.path.expanduser("~/.hermes/stock_watchlist.json")
if not os.path.exists(WATCHLIST_FILE):
    sys.exit(0)

# 简化的价格获取
import urllib.request

def get_price(code):
    market = "1" if code.startswith('6') else "0"
    url = f"https://push2.eastmoney.com/api/qt/stock/get?secid={market}.{code}&fields=f43,f170"
    try:
        req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
        resp = urllib.request.urlopen(req, timeout=5)
        data = json.loads(resp.read()).get('data', {})
        return data.get('f43', 0) / 100, data.get('f170', 0) / 100
    except:
        return None, None

with open(WATCHLIST_FILE, 'r', encoding='utf-8') as f:
    watchlist = json.load(f)

triggered = []
for w in watchlist:
    price, change = get_price(w['code'])
    if price is None:
        continue
    alerts = []
    if w.get('upper_price') and price >= w['upper_price']:
        alerts.append(f"🔺突破{w['upper_price']}元→现价{price:.2f}")
    if w.get('lower_price') and price <= w['lower_price']:
        alerts.append(f"🔻跌破{w['lower_price']}元→现价{price:.2f}")
    if w.get('upper_pct') and change and change >= w['upper_pct']:
        alerts.append(f"📈涨幅{change:+.2f}%超阈值{w['upper_pct']}%")
    if w.get('lower_pct') and change and change <= w['lower_pct']:
        alerts.append(f"📉跌幅{change:+.2f}%超阈值{w['lower_pct']}%")
    if alerts:
        triggered.append(f"{w.get('name',w['code'])}({w['code']}) {price:.2f}元 {change:+.2f}%\n" + "\n".join(alerts))

if triggered:
    print("🚨 股票提醒触发！\n\n" + "\n\n".join(triggered))
EOF

# 创建 cron 任务（交易时段每5分钟）
hermes cron create "*/5 9-11,13-14 * * 1-5" --script stock_check.py --no-agent

常见使用场景

场景1：设置价格区间监控

用户：帮我监控平安银行，12到15元区间

→ 调用 quick_add("000001", "12~15")

场景2：涨跌幅提醒

用户：贵州茅台涨3%或跌2%提醒我

→ 调用 quick_add("600519", "+3%,-2%")

场景3：查看所有监控

用户：我有哪些股票在监控？

→ 调用 list_watchlist()

场景4：手动检查

用户：现在检查一下股票有没有触发提醒

→ 调用 run_check()

数据文件

| 文件 | 用途 | |------|------| | ~/.hermes/stock_watchlist.json | 监控列表配置 | | ~/.hermes/stock_alerts.json | 历史提醒记录 |

错误速查

| 错误 | 原因 | 解决 | |------|------|------| | 获取行情失败 | 网络问题或非交易时段 | 用备用API或稍后重试 | | akshare未安装 | 缺少依赖 | pip install akshare pandas | | 代码无效 | 股票代码格式错误 | 使用6位纯数字代码 | | 无数据返回 | 非交易时段 | 交易日9:30-15:00可用 |