外部 API 与接口文档

METHOD\nPATH_WITH_QUERY\nX-PM-Timestamp\nX-PM-Nonce\nSHA256_HEX(body)

import hashlib
import hmac
import os
import time
import uuid

import requests

BASE_URL = "https://api.phoenix-mountain.example"
# 凭证一律从环境变量读取，严禁写进代码或提交进仓库
API_KEY = os.environ["PM_API_KEY"]        # 密钥前缀，形如 pmk_live_7f3a2c
API_SECRET = os.environ["PM_API_SECRET"]  # 创建密钥时一次性返回的明文 Secret


def signed_headers(method: str, path_with_query: str, body: bytes = b"") -> dict:
    timestamp = str(int(time.time()))
    nonce = uuid.uuid4().hex  # 32 位随机串，5 分钟内不可重复
    body_hash = hashlib.sha256(body).hexdigest()
    payload = "\n".join([method.upper(), path_with_query, timestamp, nonce, body_hash])
    signature = hmac.new(API_SECRET.encode(), payload.encode(), hashlib.sha256).hexdigest()
    return {
        "X-PM-Key": API_KEY,
        "X-PM-Timestamp": timestamp,
        "X-PM-Nonce": nonce,
        "X-PM-Signature": signature,
        "Content-Type": "application/json",
    }


path = "/v1/signals?market=HK&status=ACTIVE"
resp = requests.get(BASE_URL + path, headers=signed_headers("GET", path), timeout=10)
if resp.status_code == 429:
    raise RuntimeError("触发限流，请按 Retry-After 指数退避后重试")
resp.raise_for_status()

# 过期信号必须视为无效，不得继续按有效信号处理
now = time.time()
usable = [
    s for s in resp.json()["items"]
    if s["status"] == "ACTIVE" and time.mktime(time.strptime(s["expiresAt"][:19], "%Y-%m-%dT%H:%M:%S")) > now
]