okx交易策略代码时间戳错误Timestamp request expired

Question

import asyncio
import hmac
import hashlib
import time
import json
import websockets
import datetime
import requests

API_KEY = ""
SECRET_KEY = ""
PASSPHRASE = ""
WS_URL = 'wss://ws.okx.com:8443/ws/v5/private'

def get_okx_server_timestamp():
    """ 获取 OKX 服务器时间（毫秒级） """
    try:
        response = requests.get('https://www.okx.com/api/v5/public/time', timeout=3)
        response.raise_for_status()
        server_time = int(response.json()['data'][0]['ts'])  # 确保是整数
        return str(server_time)
    except Exception as e:
        print(f"获取服务器时间失败: {str(e)}")
        return str(int(time.time() * 1000))  # 仅限调试时使用

def generate_signature(secret, message):
    """ 生成 HMAC-SHA256 签名 """
    return hmac.new(secret.encode('utf-8'), message.encode('utf-8'), hashlib.sha256).hexdigest()

def get_login_message():
    """ 生成登录请求数据 """
    server_ts = get_okx_server_timestamp()  # 获取 OKX 服务器时间
    print(f"[时间校验] 服务器时间戳: {server_ts} ({datetime.datetime.fromtimestamp(int(server_ts) / 1000, datetime.UTC)})")

    # 生成签名
    message = server_ts + "GET" + "/users/self/verify"
    signature = generate_signature(SECRET_KEY, message)

    return {
        "op": "login",
        "args": [{
            "apiKey": API_KEY,
            "passphrase": PASSPHRASE,
            "timestamp": server_ts,
            "sign": signature
        }]
    }

async def websocket_client():
    """ 连接 WebSocket 并进行登录 """
    try:
        async with websockets.connect(WS_URL) as ws:
            login_msg = json.dumps(get_login_message())

            # 发送前稍作延迟，确保时间戳不过期
            time.sleep(0.1)

            await ws.send(login_msg)
            print(f"已发送登录请求: {login_msg}")

            # 获取并解析服务器响应
            response = json.loads(await ws.recv())
            print(f"登录响应: {response}")

            if response.get('code') != '0':
                print(f"登录失败: {response}")
                return
            print("✅ 登录成功！开始监听数据...")

            # 监听 WebSocket 消息
            while True:
                msg = await ws.recv()
                print(f"📩 收到消息: {msg}")

    except websockets.exceptions.ConnectionClosed as e:
        print(f"🔴 连接断开: code={e.code}, reason={e.reason}")
    except Exception as e:
        print(f"⚠️ 错误: {str(e)}")

if __name__ == "__main__":
    print("=" * 40)
    print("🔍 首次运行时间校验：")
    print(f"⏰ 本地系统时间: {datetime.datetime.now()}")
    print(f"🌍 UTC 时间: {datetime.datetime.now(datetime.UTC)}")
    print("=" * 40)

    asyncio.run(websocket_client())
    server_time = get_okx_server_timestamp()
    print(f"✅ OKX 服务器时间: {server_time} (毫秒)")
    print(f"✅ 当前 UTC 时间: {int(time.time() * 1000)} (毫秒)")

我试了很多方法都是时间戳错误，甚至把电脑系统改为UTC时间，也不行

Answer 1

阿里嘎多学长整理AIGC生成，因移动端显示问题导致当前答案未能完全显示，请使用PC端查看更加详细的解答过程

问题解答

你遇到了 OKX 交易策略代码中的时间戳错误，错误信息是 "Timestamp request expired"，这意味着你的代码中使用的时间戳已经过期。

在 OKX 的 API 中，时间戳是用来验证请求的有效性的一种机制。时间戳是使用当前时间戳和 API 密钥进行加密生成的，然后在请求头中传递给服务器。

在你的代码中，你可能没有正确地生成时间戳或没有正确地传递时间戳到请求头中。

以下是一些可能的解决方案：

1. 检查时间戳生成代码是否正确。OKX 的时间戳生成代码通常是使用当前时间戳和 API 密钥进行加密生成的。例如：

import datetime
import hashlib

timestamp = int(datetime.datetime.now().timestamp() * 1000)
signature = hmac.new(api_secret.encode(), f"{api_key}:{timestamp}".encode(), hashlib.sha256).hexdigest()

1. 检查请求头中是否正确地传递时间戳。OKX 的 API 请求头中需要包含一个名为 X-Stamp 的参数，用于传递时间戳。例如：

headers = {
    'X-Stamp': str(timestamp),
    'X-API-KEY': api_key,
    'X-SIGN': signature
}

1. 检查当前时间戳是否已经过期。如果当前时间戳已经过期，需要重新生成时间戳。

如果你已经检查了这些问题并且仍然无法解决问题，可以提供更多的代码和错误信息，以便更好地帮助你解决问题。