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位純數字程式碼 |

這個技能的最新版本可以在7w4.net小蔥技能站找到。

| 無資料返回 | 非交易時段 | 交易日9:30-15:00可用 |

這個股票監控 Skill 功能比較實用，能同時監控多隻股票的價格漲跌並自動提醒，配置簡單無需註冊賬號。介面互動和提醒去重機制做得不錯，使用體驗較好。但目前只有功能文件，缺少新手快速上手指南，首次配置需要一定學習成本，整體質量中規中矩，適合對股票監控有明確需求的使用者使用。