arbitrage/signal001/ok_chain_client.py

import requests
import hmac
import hashlib
import base64
import datetime
import urllib.parse
import json
import time # 只是为了在main的例子中演示循环

# 定义 API 凭证
api_config = {
  "api_key": 'a05643ab-fb17-402b-94a8-a886bd343301',  # 请替换为您的真实 API Key
  "secret_key": '9D59B53EB1E60B1B5F290D3698A8C9DA', # 请替换为您的真实 Secret Key
  "passphrase": 'Qwe123123.', # 请替换为您的真实 Passphrase
}
# api_config = {
#   "api_key": '3d34112b-6c78-4a37-8454-096df28bd5d0',  # 请替换为您的真实 API Key
#   "secret_key": '4E129F37B2836A2C341ECC7C22D1D706', # 请替换为您的真实 Secret Key
#   "passphrase": 'Qwe123123.', # 请替换为您的真实 Passphrase
# }

BASE_URL = "https://web3.okx.com"

def get_timestamp():
    """
    获取 ISO 8601 格式时间戳，精确到秒，以 'Z' 结尾。
    示例: 2025-05-13T03:55:34Z
    """
    return datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')

def pre_hash_string(timestamp, method, request_path, params_dict=None):
    """
    根据字符串和参数创建预签名字符串。
    """
    params_str = ""
    method_upper = method.upper()

    if method_upper == 'GET' and params_dict:
        # 对GET请求的参数进行URL编码
        # OKX V5 API文档通常要求参数按字母顺序排序后进行签名，但您的Node.js示例中querystring.stringify默认不排序。
        # 如果API要求排序，可以使用：sorted_params = sorted(params_dict.items())
        # params_str = '?' + urllib.parse.urlencode(sorted_params)
        # 为了与您的Node.js querystring.stringify行为一致（不排序），我们直接编码
        params_str = '?' + urllib.parse.urlencode(params_dict)
    elif method_upper == 'POST' and params_dict:
        # 对POST请求的body进行JSON字符串化
        params_str = json.dumps(params_dict)
    # 如果是POST请求但没有body (params_dict is None or empty), params_str 保持为 ""
    # 这是OKX API文档通常的要求: "requestBody... The Body of the request (options), "" if it is not a POST request or if there is no request body."

    return str(timestamp) + method_upper + request_path + params_str

def sign(message, secret_key):
    """
    使用 HMAC-SHA256 对预签名字符串进行签名。
    """
    mac = hmac.new(secret_key.encode('utf-8'), message.encode('utf-8'), hashlib.sha256)
    # digest() 返回 bytes, b64encode() 也返回 bytes, 最后 decode() 成 utf-8 字符串
    return base64.b64encode(mac.digest()).decode('utf-8')

def create_signature_headers(method, request_path, params_dict=None):
    """
    生成签名及请求头。
    """
    timestamp = get_timestamp()
    message_to_sign = pre_hash_string(timestamp, method, request_path, params_dict)
    signature = sign(message_to_sign, api_config['secret_key'])

    headers = {
        'OK-ACCESS-KEY': api_config['api_key'],
        'OK-ACCESS-SIGN': signature,
        'OK-ACCESS-TIMESTAMP': timestamp,
        'OK-ACCESS-PASSPHRASE': api_config['passphrase'],
        'Content-Type': 'application/json' # 几乎所有OKX V5 API都推荐或要求这个
    }
    return headers

def send_get_request(request_path, params_dict=None):
    """
    发送GET请求。
    """
    headers = create_signature_headers("GET", request_path, params_dict)
    full_url = BASE_URL + request_path

    try:
        response = requests.get(full_url, headers=headers, params=params_dict) # requests库会自动处理params的URL编码
        response.raise_for_status() # 如果HTTP请求返回了不成功的状态码 (4xx or 5xx)，则抛出HTTPError异常
        return response.json() # 假设返回的是JSON
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")
        print(f"Response content: {response.content.decode()}")
    except requests.exceptions.RequestException as req_err:
        print(f"Request exception occurred: {req_err}")
    except json.JSONDecodeError:
        print(f"Failed to decode JSON. Response content: {response.text}")
    return None

def send_post_request(request_path, body_params_dict=None):
    """
    发送POST请求。
    """
    # 对于POST请求，params_dict 用于签名，并且也作为请求体发送
    headers = create_signature_headers("POST", request_path, body_params_dict)
    full_url = BASE_URL + request_path

    try:
        # 如果 body_params_dict 为 None 或空字典，requests.post 的 json 参数会发送 "{}" 或不发送body
        # 这取决于API如何处理空POST体。通常如果签名时 body为空字符串，post的data也应为空。
        # 如果 body_params_dict 为 None, 且签名时也用空字符串表示 body，那么发送 data=""
        if body_params_dict:
            response = requests.post(full_url, headers=headers, json=body_params_dict)
        else:
            response = requests.post(full_url, headers=headers, data="") # 如果签名用了空body，这里也用空data

        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")
        print(f"Response content: {response.content.decode()}")
    except requests.exceptions.RequestException as req_err:
        print(f"Request exception occurred: {req_err}")
    except json.JSONDecodeError:
        print(f"Failed to decode JSON. Response content: {response.text}")
    return None

def swap(chain_id, amount, from_token_address, to_token_address, slippage, user_wallet_address, receiver_address=None, gas_level='average'):
    get_request_path = '/api/v5/dex/aggregator/swap'
    get_params = {
        'chainIndex': chain_id,
        'amount': amount,
        'fromTokenAddress': from_token_address,
        'toTokenAddress': to_token_address,
        'slippage': slippage,
        'userWalletAddress': user_wallet_address,
        'gasLevel': gas_level
    }

    if receiver_address is not None:
        get_params['swapReceiverAddress'] = receiver_address

    return send_get_request(get_request_path, get_params)

def approve(chain_id, token_contract_address, approve_amount):
    get_request_path = '/api/v5/dex/aggregator/approve-transaction'
    get_params = {
        'chainIndex': chain_id,
        'tokenContractAddress': token_contract_address,
        'approveAmount': approve_amount,
    }

    return send_get_request(get_request_path, get_params)

if __name__ == "__main__":
    import decimal
    import pprint

    pprint.pprint(swap(1, decimal.Decimal('1000') * (10 ** 6), '0xdAC17F958D2ee523a2206206994597C13D831ec7', '0xf816507E690f5Aa4E29d164885EB5fa7a5627860', 1, '0xf816507E690f5Aa4E29d164885EB5fa7a5627860'))
    # pprint.pprint(approve(1, '0xdAC17F958D2ee523a2206206994597C13D831ec7', '1000000'))

    # print("测试时间戳格式:", get_timestamp())
    # print("-" * 30)

    # GET 请求示例 (OKX Web3 API - DEX 聚合器询价)
    # 参考文档: https://www.okx.com/web3/build/docs/sdks-and-apis/dex-api/aggregate-and-swap/get-quote
    # 注意：这是 Web3 API，其 Base URL 和普通 CEX API 可能不同。
    # 确保 BASE_URL 设置为 https://web3.okx.com
    # 并且您的 API Key 拥有 Web3 API 的权限。
    # 如果是普通 CEX API，BASE_URL 通常是 https://www.okx.com
    
    # print("发送 GET 请求示例:")
    # get_request_path = '/api/v5/dex/aggregator/quote'
    # get_params = {
    #     'chainId': '42161', # Arbitrum One
    #     'amount': '1000000000000', # 0.001 WETH (假设WETH是18位小数)
    #     'toTokenAddress': '0xff970a61a04b1ca14834a43f5de4533ebddb5cc8', # USDC on Arbitrum
    #     'fromTokenAddress': '0x82aF49447D8a07e3bd95BD0d56f35241523fBab1' # WETH on Arbitrum
    # }
    # # 确保 API key, secret, passphrase 是针对 web3.okx.com (如果这是你测试的 endpoint)
    # # 或者，如果你测试的是 CEX API, 确保 endpoint 和 API key 对应
    # print(f"请求路径: {get_request_path}")
    # print(f"请求参数: {get_params}")
    # response_data_get = send_get_request(get_request_path, get_params)
    # if response_data_get:
    #     print("GET 请求成功，响应:")
    #     print(json.dumps(response_data_get, indent=2))
    # else:
    #     print("GET 请求失败。")

    # print("-" * 30)

    # POST 请求示例 (这里用一个CEX API的例子，因为Web3 API 大部分是GET)
    # 例如，下单 (Trade API: /api/v5/trade/order)
    # BASE_URL 应为 "https://www.okx.com"
    # 注意：以下POST示例需要 CEX API 权限，并且请求体结构特定于此接口。仅作演示。
    # print("\n发送 POST 请求示例 (CEX Trade API - 下单):")
    # BASE_URL_CEX = "https://www.okx.com" # 切换到CEX
    # # 重新设置 BASE_URL 以便 send_post_request 使用正确的域
    # original_base_url = BASE_URL
    # BASE_URL = BASE_URL_CEX
    #
    # post_request_path_trade = '/api/v5/trade/order'
    # post_params_trade = {
    #     "instId": "BTC-USDT",
    #     "tdMode": "cash", # "cash" 现货, "cross" 全仓杠杆, "isolated" 逐仓杠杆
    #     "side": "buy",
    #     "ordType": "limit",
    #     "sz": "0.0001", # 数量
    #     "px": "20000"   # 价格
    # }
    # print(f"请求路径: {post_request_path_trade}")
    # print(f"请求体: {post_params_trade}")
    # # 确保 API key 对 www.okx.com 有交易权限
    # print("--- 警告: 此POST请求会实际下单，请谨慎测试 ---")
    # # response_data_post = send_post_request(post_request_path_trade, post_params_trade)
    # # if response_data_post:
    # #     print("POST 请求成功，响应:")
    # #     print(json.dumps(response_data_post, indent=2))
    # # else:
    # #     print("POST 请求失败。")
    #
    # BASE_URL = original_base_url # 恢复原始 BASE_URL
    # print("-" * 30)

    # 如果您想测试 Node.js 代码中注释掉的 POST 示例:
    # /api/v5/mktplace/nft/ordinals/listings (这是一个 NFT 市场的 API，也属于 Web3 API)
    # BASE_URL 应为 "https://web3.okx.com"
    # print("\n发送 POST 请求示例 (Web3 Marketplace Ordinals):")
    # post_request_path_ordinals = '/api/v5/mktplace/nft/ordinals/listings'
    # post_params_ordinals = {
    #     'slug': 'sats' # 注意：此参数可能不完整或不正确，请参照最新API文档
    #                    # 通常获取listings需要更多参数，比如分页，排序等
    # }
    # print(f"请求路径: {post_request_path_ordinals}")
    # print(f"请求体: {post_params_ordinals}")
    # response_data_post_ordinals = send_post_request(post_request_path_ordinals, post_params_ordinals)
    # if response_data_post_ordinals:
    #     print("POST 请求成功，响应:")
    #     print(json.dumps(response_data_post_ordinals, indent=2))
    # else:
    #     print("POST 请求失败。")