用 Claude Code 给 Pionex 机器人加智能监控：边沿预警 + Telegram 推送

import os, json, urllib.request, time
from datetime import datetime

TG_TOKEN = os.environ["TG_TOKEN"]
TG_CHAT_ID = os.environ["TG_CHAT_ID"]
BOT_CONFIGS = json.loads(os.environ.get("BOT_CONFIGS", "[]"))

# 阈值：距离边沿 10% 触发预警
ALERT_THRESHOLD = 0.10

def get_price(symbol):
    """symbol 格式 'BTC/USDT' → Binance API 'BTCUSDT'"""
    s = symbol.replace("/", "")
    url = f"https://api.binance.com/api/v3/ticker/price?symbol={s}"
    with urllib.request.urlopen(url, timeout=10) as r:
        return float(json.loads(r.read())["price"])

def send_tg(text):
    """Telegram Bot 推送"""
    url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"
    data = json.dumps({"chat_id": TG_CHAT_ID, "text": text}).encode()
    req = urllib.request.Request(url, data=data,
                                 headers={"Content-Type": "application/json"})
    try: urllib.request.urlopen(req, timeout=5)
    except Exception as e: print(f"TG send failed: {e}")

def check_bot(bot):
    symbol = bot["symbol"]
    low, high = bot["low"], bot["high"]
    name = bot["name"]
    try: price = get_price(symbol)
    except Exception as e:
        print(f"{symbol}: get price fail: {e}")
        return
    
    # 距离上沿的百分比
    dist_to_high = (high - price) / price
    dist_to_low = (price - low) / price
    
    # 判断是否接近边沿
    if dist_to_high < ALERT_THRESHOLD and dist_to_high > 0:
        pct = dist_to_high * 100
        send_tg(f"🟡 {name} 接近上沿\n当前: ${price:,.2f}\n上沿: ${high:,.2f}\n距离: {pct:.2f}%")
    elif dist_to_low < ALERT_THRESHOLD and dist_to_low > 0:
        pct = dist_to_low * 100
        send_tg(f"🟡 {name} 接近下沿\n当前: ${price:,.2f}\n下沿: ${low:,.2f}\n距离: {pct:.2f}%")
    elif price >= high:
        send_tg(f"🔴 {name} 已穿越上沿!\n当前: ${price:,.2f}\n上沿: ${high:,.2f}\n机器人可能已停，建议检查")
    elif price <= low:
        send_tg(f"🔴 {name} 已跌破下沿!\n当前: ${price:,.2f}\n下沿: ${low:,.2f}\n建议手动平仓或扩大区间")

if __name__ == "__main__":
    for bot in BOT_CONFIGS:
        check_bot(bot)

<?xml version="1.0" encoding="UTF-8"?>
<plist version="1.0">
<dict>
    <key>Label</key><string>com.as.pionex-monitor</string>
    <key>ProgramArguments</key>
    <array>
        <string>/usr/bin/python3</string>
        <string>/Users/YOU/scripts/pionex_monitor.py</string>
    </array>
    <key>StartInterval</key><integer>300</integer>
    <key>EnvironmentVariables</key>
    <dict>
        <key>TG_TOKEN</key><string>YOUR_TOKEN</string>
        <key>TG_CHAT_ID</key><string>YOUR_CHAT_ID</string>
        <key>BOT_CONFIGS</key><string>[{"symbol":"BTC/USDT","low":60000,"high":70000,"name":"BTC 网格"}]</string>
    </dict>
    <key>StandardOutPath</key><string>/tmp/pionex_monitor.log</string>
    <key>StandardErrorPath</key><string>/tmp/pionex_monitor.log</string>
</dict>
</plist>

launchctl load ~/Library/LaunchAgents/com.as.pionex-monitor.plist

def should_alert(key, window_sec=7200):
    """用 /tmp 里的 marker 文件去重"""
    import os
    marker = f"/tmp/.pionex_alert_{key}"
    if os.path.isfile(marker) and time.time() - os.path.getmtime(marker) < window_sec:
        return False
    open(marker, "w").close()
    return True