OKX交易所API认证签名Python实现：从小白到生产级代码的血泪史

记得三年前凌晨3点，我的量化交易机器人突然停止工作。日志里只有一行冷冰冰的 401 Unauthorized - signature verification failed。那天晚上我在OKX API文档和Stack Overflow之间来回切换了整整4个小时，头发掉了十几根——今天我把踩过的坑全部整理成这篇教程，让你避免重蹈覆辙。

为什么OKX的签名这么难搞？

OKX采用了基于HMAC SHA256的消息认证机制，和币安、Binance、火币等主流交易所类似，但参数格式和签名算法细节各不相同。最坑的是：timestamp + method + requestPath + body 的拼接顺序错一位就是401错误。

前置准备：获取API Key

登录 OKX官网 → 账户中心 → API Key
创建API Key，选择权限（只读/交易/提币）
保存 apiKey, secretKey, passphrase 三个值
重要：secretKey只在创建时显示一次，之后无法查看！

核心签名算法实现

import hmac
import base64
import time
import json
from typing import Dict, Optional
import requests

class OKXSigner:
    """OKX API 签名生成器"""
    
    def __init__(self, api_key: str, secret_key: str, passphrase: str, passphrase_type: str = "plain"):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase
        self.passphrase_type = passphrase_type  # 可选: plain, T-OTP
    
    def _sign(self, message: str) -> str:
        """生成HMAC SHA256签名，Base64编码"""
        mac = hmac.new(
            self.secret_key.encode('utf-8'),
            message.encode('utf-8'),
            digestmod='sha256'
        )
        return base64.b64encode(mac.digest()).decode('utf-8')
    
    def sign(self, timestamp: str, method: str, request_path: str, body: str = "") -> Dict[str, str]:
        """
        构建带签名的请求头
        :param timestamp: ISO 8601格式，如 '2024-01-15T10:30:00.000Z'
        :param method: GET, POST, DELETE等
        :param request_path: /api/v5/account/balance
        :param body: JSON字符串，如 '{}' 或 ''
        :return: 包含所有认证头的字典
        """
        # ⚠️ 关键：签名消息 = timestamp + method + requestPath + body
        message = timestamp + method + request_path + body
        
        signature = self._sign(message)
        
        return {
            'OK-ACCESS-KEY': self.api_key,
            'OK-ACCESS-SIGN': signature,
            'OK-ACCESS-TIMESTAMP': timestamp,
            'OK-ACCESS-PASSPHRASE': self.passphrase,
            'Content-Type': 'application/json',
        }


def get_timestamp() -> str:
    """获取OKX要求的ISO 8601格式时间戳"""
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'


class OKXClient:
    """OKX API客户端"""
    
    BASE_URL = "https://www.okx.com"
    
    def __init__(self, api_key: str, secret_key: str, passphrase: str):
        self.signer = OKXSigner(api_key, secret_key, passphrase)
        self.session = requests.Session()
    
    def _request(self, method: str, endpoint: str, params: Optional[Dict] = None, body: Optional[Dict] = None) -> Dict:
        """
        发送已签名的API请求
        """
        timestamp = get_timestamp()
        url = self.BASE_URL + endpoint
        
        # 构建body字符串
        body_str = json.dumps(body) if body else ""
        
        # 获取签名头
        headers = self.signer.sign(timestamp, method, endpoint, body_str)
        
        # 发送请求
        if method == "GET" and params:
            response = self.session.get(url, headers=headers, params=params)
        elif method == "POST":
            response = self.session.post(url, headers=headers, data=body_str)
        elif method == "DELETE":
            response = self.session.delete(url, headers=headers)
        else:
            response = self.session.request(method, url, headers=headers)
        
        # 错误处理
        if response.status_code != 200:
            print(f"❌ HTTP {response.status_code}: {response.text}")
            response.raise_for_status()
        
        result = response.json()
        
        # 检查业务错误码
        if result.get('code') != '0':
            error_msg = result.get('msg', 'Unknown error')
            print(f"❌ API Error [{result.get('code')}]: {error_msg}")
            raise Exception(f"API Error: {error_msg}")
        
        return result['data']
    
    def get_balance(self) -> list:
        """获取账户余额"""
        return self._request("GET", "/api/v5/account/balance")
    
    def get_positions(self) -> list:
        """获取持仓信息"""
        return self._request("GET", "/api/v5/account/positions")
    
    def place_order(self, inst_id: str, td_mode: str, side: str, ord_type: str, sz: str, px: Optional[str] = None) -> list:
        """下单"""
        body = {
            "instId": inst_id,
            "tdMode": td_mode,  # cross, isolated, cash
            "side": side,       # buy, sell
            "ordType": ord_type, # market, limit, stop_loss等
            "sz": sz,
        }
        if px:
            body["px"] = px
        
        return self._request("POST", "/api/v5/trade/order", body=body)

实战示例：获取账户余额

# 使用示例
if __name__ == "__main__":
    # ⚠️ 请替换为你自己的API密钥
    API_KEY = "your_api_key_here"
    SECRET_KEY = "your_secret_key_here"
    PASSPHRASE = "your_passphrase_here"
    
    client = OKXClient(API_KEY, SECRET_KEY, PASSPHRASE)
    
    try:
        # 获取账户余额
        print("📊 获取账户余额...")
        balances = client.get_balance()
        
        for acc in balances:
            print(f"\n账户: {acc.get('Details', [{}])[0].get('ccy', 'N/A')}")
            print(f"  余额: {acc.get('Details', [{}])[0].get('cashBal', 'N/A')}")
            print(f"  可用: {acc.get('Details', [{}])[0].get('availBal', 'N/A')}")
        
        # 获取BTC持仓
        print("\n📈 获取BTC-USDT持仓...")
        positions = client.get_positions()
        for pos in positions:
            if pos.get('instId') == 'BTC-USDT-SWAP':
                print(f"  数量: {pos.get('Pos')} 张")
                print(f"  收益: {pos.get('Pnl')} USDT")
                print(f"  保证金: {pos.get('Margin')} USDT")
                
    except Exception as e:
        print(f"❌ 请求失败: {e}")

异步版本：适合高频交易

import asyncio
import aiohttp
import hmac
import base64
import json
from datetime import datetime, timezone

class OKXAsyncSigner:
    """异步版本的OKX签名器"""
    
    def __init__(self, api_key: str, secret_key: str, passphrase: str):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase
    
    def _sign(self, message: str) -> str:
        mac = hmac.new(
            self.secret_key.encode('utf-8'),
            message.encode('utf-8'),
            digestmod='sha256'
        )
        return base64.b64encode(mac.digest()).decode('utf-8')
    
    def get_headers(self, method: str, request_path: str, body: str = "") -> Dict[str, str]:
        timestamp = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
        message = timestamp + method + request_path + body
        signature = self._sign(message)
        
        return {
            'OK-ACCESS-KEY': self.api_key,
            'OK-ACCESS-SIGN': signature,
            'OK-ACCESS-TIMESTAMP': timestamp,
            'OK-ACCESS-PASSPHRASE': self.passphrase,
            'Content-Type': 'application/json',
        }


class OKXAsyncClient:
    """异步OKX API客户端 - 适合高频交易场景"""
    
    BASE_URL = "https://www.okx.com"
    
    def __init__(self, api_key: str, secret_key: str, passphrase: str):
        self.signer = OKXAsyncSigner(api_key, secret_key, passphrase)
        self.session: Optional[aiohttp.ClientSession] = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
    
    async def request(self, method: str, endpoint: str, body: Optional[Dict] = None) -> Dict:
        body_str = json.dumps(body) if body else ""
        headers = self.signer.get_headers(method, endpoint, body_str)
        url = self.BASE_URL + endpoint
        
        async with self.session.request(method, url, headers=headers, data=body_str) as response:
            result = await response.json()
            
            if response.status != 200:
                raise Exception(f"HTTP {response.status}: {await response.text()}")
            
            if result.get('code') != '0':
                raise Exception(f"API Error: {result.get('msg')}")
            
            return result['data']
    
    async def get_ticker(self, inst_id: str) -> Dict:
        """获取行情"""
        return await self.request("GET", f"/api/v5/market/ticker?instId={inst_id}")
    
    async def batch_place_orders(self, orders: list) -> list:
        """批量下单 - 单次最多50个"""
        return await self.request("POST", "/api/v5/trade/batch-orders", {"orders": orders})


async def main():
    """异步示例"""
    async with OKXAsyncClient("api_key", "secret_key", "passphrase") as client:
        # 并行获取多个交易对行情
        tasks = [
            client.get_ticker("BTC-USDT"),
            client.get_ticker("ETH-USDT"),
            client.get_ticker("SOL-USDT"),
        ]
        
        tickers = await asyncio.gather(*tasks)
        
        for ticker in tickers:
            inst_id = ticker[0]['instId']
            last_price = ticker[0]['last']
            print(f"{inst_id}: ${last_price}")


if __name__ == "__main__":
    asyncio.run(main())

避免踩坑：签名细节清单

时间戳格式：必须是 ISO 8601，且必须是UTC时区，毫秒精度，结尾带Z
签名消息拼接顺序：timestamp + method + requestPath + body，不能有空格或换行
requestPath：必须包含查询参数（如有），格式为 /api/v5/endpoint?param=value
GET请求body：传空字符串 ""，而不是 "{}"
Content-Type：POST请求必须设为 application/json
passphrase：如果创建时选择了TOTP验证，这里要传加密后的密码

Lỗi thường gặp và cách khắc phục

Lỗi	Nguyên nhân	Cách khắc phục
401 Unauthorized - signature verification failed	签名消息拼接顺序错误，通常是body格式不对（GET请求传了"{}"）	`# GET请求body必须为空字符串 body_str = "" # ✅ 正确 body_str = "{}" # ❌ 会导致401 POST请求body body_str = json.dumps(body) if body else ""`
401 Invalid sign	时间戳与服务器时间偏差超过30秒，或使用了错误的secretKey	`# 检查服务器时间同步 import ntplib client = ntplib.NTPClient() response = client.request('pool.ntp.org') print(response.offset) # 偏差秒数或手动校准时间戳格式 timestamp = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'`
50125 - Incorrect trade password	passphrase填写错误，或API Key权限不足	`# 确认passphrase类型如果API Key是TOTP类型，passphrase需要是6位数字动态码如果是普通类型，直接填创建时设置的密码检查API Key权限 signer = OKXSigner( api_key, secret_key, passphrase, passphrase_type="T-OTP" # 动态密码 )`
Connection timeout / Read timeout	网络问题或OKX服务器负载高	`# 添加重试机制和超时设置 from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry session = requests.Session() retry = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) adapter = HTTPAdapter(max_retries=retry) session.mount('https://', adapter) response = session.get(url, headers=headers, timeout=(5, 10)) # (连接超时, 读取超时)`

生产环境建议

使用专用服务器：OKX对IP有限制，建议使用香港或新加坡服务器，延迟 < 50ms
签名缓存：对于只读API，签名有效期30秒，可以缓存复用
限流处理：公开数据 20次/2秒，私有数据 30次/2秒，超限会触发 429
日志记录：记录所有API调用，便于排查问题
资金安全：只读权限的API Key不要开启提币权限

如果需要AI API服务

如果你在开发量化策略时需要接入AI能力（如市场分析、自然语言处理等），可以考虑 HolySheep AI——支持 GPT-4、Claude、Gemini、DeepSeek 等主流模型，延迟低于50ms，价格比官方渠道低85%以上，支持微信/支付宝充值，对国内用户非常友好。

Giá tham khảo 2026:

GPT-4.1: $8/MTok
Claude Sonnet 4.5: $15/MTok
Gemini 2.5 Flash: $2.50/MTok
DeepSeek V3.2: $0.42/MTok

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký

OKX交易所API认证签名Python实现：从小白到生产级代码的血泪史

为什么OKX的签名这么难搞？

前置准备：获取API Key

核心签名算法实现

实战示例：获取账户余额

异步版本：适合高频交易

避免踩坑：签名细节清单

Lỗi thường gặp và cách khắc phục

POST请求body

或手动校准时间戳格式

如果API Key是TOTP类型，passphrase需要是6位数字动态码

如果是普通类型，直接填创建时设置的密码

检查API Key权限

生产环境建议

如果需要AI API服务

Tài nguyên liên quan

Bài viết liên quan

为什么OKX的签名这么难搞？

前置准备：获取API Key

核心签名算法实现

实战示例：获取账户余额

异步版本：适合高频交易

避免踩坑：签名细节清单

Lỗi thường gặp và cách khắc phục

POST请求body

或手动校准时间戳格式

如果API Key是TOTP类型，passphrase需要是6位数字动态码

如果是普通类型，直接填创建时设置的密码

检查API Key权限

生产环境建议

如果需要AI API服务

Tài nguyên liên quan

Bài viết liên quan

🔥 Thử HolySheep AI