Trade/okx_monitor.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OKX 合约账户 只读监控 + 实时分析提醒
- 仅使用 GET 只读接口(账户/持仓/未成交/行情)，绝不写入/下单
- 密钥从同目录 okx_keys.json 读取(不要硬编码)
- 每5分钟运行, 触发条件弹 macOS 通知
"""
import subprocess, json, hmac, hashlib, base64, datetime, sys, os

HERE = os.path.dirname(os.path.abspath(__file__))
KEYS_FILE = os.path.join(HERE, "okx_keys.json")

# ===== 读取密钥 =====
def load_keys():
    with open(KEYS_FILE, "r", encoding="utf-8") as f:
        return json.load(f)

API_KEY = ""
SECRET = ""
PASSPHRASE = ""

try:
    k = load_keys()
    API_KEY = k["api_key"]
    SECRET = k["secret"]
    PASSPHRASE = k["passphrase"]
except Exception as e:
    print("读取密钥失败:", e)
    sys.exit(1)


def sign(timestamp, method, path, body=""):
    pre = timestamp + method.upper() + path + body
    mac = hmac.new(SECRET.encode("utf-8"), pre.encode("utf-8"), hashlib.sha256)
    return base64.b64encode(mac.digest()).decode()


def okx_get(path):
    ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    sig = sign(ts, "GET", path, "")
    curl = [
        "curl", "-s", "--max-time", "10",
        "-H", f"OK-ACCESS-KEY:{API_KEY}",
        "-H", f"OK-ACCESS-SIGN:{sig}",
        "-H", f"OK-ACCESS-TIMESTAMP:{ts}",
        "-H", f"OK-ACCESS-PASSPHRASE:{PASSPHRASE}",
        "-H", "Content-Type:application/json",
        f"https://www.okx.com{path}",
    ]
    out = subprocess.run(curl, capture_output=True, text=True, timeout=15).stdout
    try:
        return json.loads(out)
    except Exception:
        return {"code": "-1", "msg": out[:200]}


ALERTED = set()


def notify(title, msg):
    subprocess.run(["osascript", "-e",
                    f'display notification "{msg}" with title "{title}" sound name "Glass"'])


def mark_price(inst_id):
    """拉实时标记价(只读公共接口)"""
    try:
        out = subprocess.run(
            ["curl", "-s", "--max-time", "8",
             f"https://www.okx.com/api/v5/market/ticker?instId={inst_id}"],
            capture_output=True, text=True, timeout=12).stdout
        d = json.loads(out)
        if d.get("code") == "0":
            return float(d["data"][0]["last"])
    except Exception:
        pass
    return None


# 合约面值缓存(每张对应多少基础币)
_CTVAL = {}


def ctval(inst_id):
    if inst_id in _CTVAL:
        return _CTVAL[inst_id]
    try:
        out = subprocess.run(
            ["curl", "-s", "--max-time", "8",
             f"https://www.okx.com/api/v5/public/instruments?instType=SWAP"],
            capture_output=True, text=True, timeout=12).stdout
        for x in json.loads(out).get("data", []):
            if x["instId"] == inst_id:
                v = float(x.get("ctVal", 1))
                _CTVAL[inst_id] = v
                return v
    except Exception:
        pass
    _CTVAL[inst_id] = 1.0
    return 1.0


# 手续费档(普通用户Taker); 开平各一次
TAKER_FEE = 0.0005
# 资金费保守估算: 每8h最多按0.01%偏多, 默认持仓1天=3期(无持仓时长则按0算)
FUNDING_PER_PERIOD = 0.0001
FUNDING_PERIODS = 3


def be_price(inst, sz, entry, upl):
    """算覆盖所有费用后的回本价(多/空通用)"""
    cv = ctval(inst)
    notional = sz * cv * entry                 # 名义价值(开仓)
    fee = notional * TAKER_FEE * 2             # 开+平手续费
    funding = notional * FUNDING_PER_PERIOD * FUNDING_PERIODS
    total_cost = fee + funding                 # 需覆盖的总费用
    per_point = sz * cv                        # 每点价格变动盈亏
    need_move = total_cost / per_point         # 需涨/跌点数
    # 多单需涨, 空单需跌
    side = "long" if upl >= 0 else "long"
    return entry + need_move, total_cost, need_move


def analyze_and_alert(bal, pos):
    """基于持仓做风险提示: 浮亏预警 + 强平价关注 + 距强平<=3%紧急预警 + 费用回本价"""
    today = datetime.date.today().isoformat()
    if ALERTED and not any(x.endswith(today) for x in ALERTED):
        ALERTED.clear()

    details = pos.get("data", [])
    for p in details:
        inst = p.get("instId", "")
        pos_side = p.get("posSide", "")
        sz = float(p.get("pos", "0")) if p.get("pos") else 0.0
        lev = p.get("lever", "1")
        liq = p.get("liqPx", "")
        upl = float(p.get("upl", "0")) if p.get("upl") else 0.0
        ccy = p.get("ccy", "")
        if sz == 0 or not liq:
            continue
        liq = float(liq)
        tip = f"{inst} {pos_side} 量{sz} 杠杆{lev} 强平@{liq:.2f} 浮动{upl:+.2f}{ccy}"

        # 费用回本价
        be, cost, move = be_price(inst, sz, float(p.get("avgPx", 0)) or 0, upl)
        if float(p.get("avgPx", 0) or 0) > 0:
            print(f"  费用回本价: {be:.2f} (需覆盖费用{cost:.2f}USDT, 即再涨{move:.2f}点)")

        # 浮亏超阈值提醒
        if upl < -50:
            key = f"loss_{inst}_{today}"
            if key not in ALERTED:
                ALERTED.add(key)
                notify(f"[OKX 浮亏预警] {inst}", tip)

        # 强平价关注(每日一次)
        key0 = f"liq_{inst}_{today}"
        if key0 not in ALERTED:
            ALERTED.add(key0)
            notify(f"[OKX 持仓监控] {inst}", f"强平价 {liq:.2f} | 回本价{be:.2f} | {tip}")

        # 距强平 <=3% 紧急预警(每小时一次, 用分钟级key避免刷屏)
        mp = mark_price(inst)
        if mp:
            dist = abs(mp - liq) / mp * 100
            if dist <= 3.0:
                hm = datetime.datetime.now().strftime("%Y%m%d%H")
                key1 = f"risk_{inst}_{hm}"
                if key1 not in ALERTED:
                    ALERTED.add(key1)
                    notify(f"[OKX 紧急·近强平] {inst}",
                           f"现价{mp:.2f} 距强平{liq:.2f}仅{dist:.1f}%! 回本价{be:.2f} | 建议降杠杆/减仓 | {tip}")
            print(f"  {inst} 现价{mp:.2f} 强平{liq:.2f} 距强平{dist:.2f}%")


def main():
    if not API_KEY:
        sys.exit(0)
    # 只读接口
    bal = okx_get("/api/v5/account/balance")
    if bal.get("code") != "0":
        err = bal.get("msg", "未知错误")
        print("余额查询失败:", err)
        # 401/签名错也通知, 方便排查
        notify("[OKX 监控] 查询失败", err[:80])
        return
    pos = okx_get("/api/v5/account/positions?instType=SWAP")
    # 打印摘要
    try:
        total = float(bal["data"][0]["totalEq"])
        print(f"[{datetime.datetime.now():%H:%M}] 总权益: {total:.2f} USDT")
    except Exception:
        pass
    analyze_and_alert(bal, pos if pos.get("code") == "0" else {"data": []})


if __name__ == "__main__":
    main()