用 Claude 写 Binance API 自动风控脚本：爆仓前 3 小时预警

import os, time, hmac, hashlib, json, urllib.request, urllib.parse
from datetime import datetime

API_KEY = os.environ["BINANCE_API_KEY"]
API_SECRET = os.environ["BINANCE_API_SECRET"].encode()
TG_TOKEN = os.environ["TG_TOKEN"]
TG_CHAT_ID = os.environ["TG_CHAT_ID"]

BASE = "https://fapi.binance.com"

# 阈值
SAFE = 5.0    # 账户余额 / 维持保证金
WARN = 2.0
DANGER = 1.2  # 低于这个立刻警报


def sign(params):
    ts = int(time.time() * 1000)
    params["timestamp"] = ts
    query = urllib.parse.urlencode(params)
    sig = hmac.new(API_SECRET, query.encode(), hashlib.sha256).hexdigest()
    return query + "&signature=" + sig


def api_get(path, params=None):
    query = sign(params or {})
    req = urllib.request.Request(f"{BASE}{path}?{query}",
                                 headers={"X-MBX-APIKEY": API_KEY})
    with urllib.request.urlopen(req, timeout=10) as r:
        return json.loads(r.read())


def send_tg(text):
    url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"
    data = json.dumps({"chat_id": TG_CHAT_ID, "text": text, "parse_mode": "Markdown"}).encode()
    req = urllib.request.Request(url, data=data,
                                 headers={"Content-Type": "application/json"})
    try: urllib.request.urlopen(req, timeout=5)
    except: pass


def check():
    """查账户 + 仓位"""
    account = api_get("/fapi/v2/account")
    wallet = float(account["totalWalletBalance"])
    maint = float(account["totalMaintMargin"])
    pnl = float(account["totalUnrealizedProfit"])
    
    if maint == 0:
        return  # 没开仓
    
    ratio = (wallet + pnl) / maint
    status = "🟢 安全" if ratio > SAFE else ("🟡 警戒" if ratio > WARN else "🔴 危险")
    
    # 构造仓位明细
    positions = [p for p in account["positions"] if float(p.get("positionAmt", 0)) != 0]
    lines = []
    for p in positions:
        sym = p["symbol"]
        amt = float(p["positionAmt"])
        entry = float(p["entryPrice"])
        mark = float(p.get("markPrice", entry))
        unr = float(p.get("unrealizedProfit", 0))
        lev = p.get("leverage", "?")
        lines.append(f"`{sym}` x{lev} · {amt:+.4f} @ ${entry:,.2f} · 现价 ${mark:,.2f} · {unr:+.2f}")
    
    msg = f"""*{status}* · 合约账户健康度
账户余额: `${wallet:,.2f}` ({pnl:+,.2f} 未实现)
总维持保证金: `${maint:,.2f}`
健康比率: `{ratio:.2f}x` (> {SAFE:.0f} 安全)

*仓位* ({len(positions)} 个):
""" + "\n".join(lines)
    
    # 只在警戒/危险时发，或者每天一次常规汇报（8:00）
    now = datetime.now()
    is_report_time = now.hour == 8 and now.minute < 10
    
    if ratio < WARN:
        send_tg(msg)
    elif ratio < DANGER:
        send_tg("🚨 *紧急* 3 小时内可能爆仓!\n\n" + msg)
    elif is_report_time:
        send_tg("📊 日报\n\n" + msg)


if __name__ == "__main__":
    try: check()
    except Exception as e:
        send_tg(f"⚠️ 监控脚本报错: {e}")

<key>StartInterval</key><integer>600</integer>
<key>EnvironmentVariables</key>
<dict>
    <key>BINANCE_API_KEY</key><string>YOUR_KEY</string>
    <key>BINANCE_API_SECRET</key><string>YOUR_SECRET</string>
    <key>TG_TOKEN</key><string>...</string>
    <key>TG_CHAT_ID</key><string>...</string>
</dict>