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可用 |

这个股票监控 Skill 功能比较实用，能同时监控多只股票的价格涨跌并自动提醒，配置简单无需注册账号。界面交互和提醒去重机制做得不错，使用体验较好。但目前只有功能文档，缺少新手快速上手指南，首次配置需要一定学习成本，整体质量中规中矩，适合对股票监控有明确需求的用户使用。