129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
OKX API 连接测试脚本
|
||
用于验证API密钥配置是否正确，以及网络连接是否正常
|
||
"""
|
||
|
||
import okx.api.market as Market
|
||
import okx.api.account as Account
|
||
from config import API_KEY, SECRET_KEY, PASSPHRASE, TRADING_CONFIG
|
||
|
||
def test_public_api():
|
||
"""测试公共API（无需密钥）"""
|
||
print("=== 测试公共API ===")
|
||
|
||
try:
|
||
# 创建市场数据API实例（公共API）
|
||
market_api = Market.Market()
|
||
|
||
# 获取BTC-USDT行情
|
||
result = market_api.get_ticker(instId="BTC-USDT")
|
||
|
||
if result['code'] == '0':
|
||
price = float(result['data'][0]['last'])
|
||
print(f"✅ 公共API测试成功")
|
||
print(f"BTC/USDT 当前价格: ${price:,.2f}")
|
||
return True
|
||
else:
|
||
print(f"❌ 公共API测试失败: {result}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 公共API测试异常: {e}")
|
||
return False
|
||
|
||
def test_private_api():
|
||
"""测试私有API（需要密钥）"""
|
||
print("\n=== 测试私有API ===")
|
||
|
||
try:
|
||
# 检查API密钥是否已配置
|
||
if API_KEY == "your_api_key_here":
|
||
print("⚠️ API密钥未配置，跳过私有API测试")
|
||
print("请在config.py中配置你的API密钥")
|
||
return False
|
||
|
||
# 创建账户API实例
|
||
flag = "0" if TRADING_CONFIG["sandbox"] else "1"
|
||
account_api = Account.Account(
|
||
API_KEY, SECRET_KEY, PASSPHRASE,
|
||
flag=flag, debug=False
|
||
)
|
||
|
||
# 获取账户余额
|
||
result = account_api.get_account_balance()
|
||
|
||
if result['code'] == '0':
|
||
print("✅ 私有API测试成功")
|
||
print("账户余额信息:")
|
||
for balance in result['data']:
|
||
if balance['ccy'] in ['USDT', 'BTC']:
|
||
print(f" {balance['ccy']}: {balance['bal']}")
|
||
return True
|
||
else:
|
||
print(f"❌ 私有API测试失败: {result}")
|
||
if result['code'] == '50001':
|
||
print("可能原因: API密钥错误或权限不足")
|
||
elif result['code'] == '50002':
|
||
print("可能原因: 签名错误")
|
||
elif result['code'] == '50004':
|
||
print("可能原因: 请求过于频繁")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 私有API测试异常: {e}")
|
||
return False
|
||
|
||
def test_environment():
|
||
"""测试环境配置"""
|
||
print("\n=== 测试环境配置 ===")
|
||
|
||
print(f"沙盒环境: {'是' if TRADING_CONFIG['sandbox'] else '否'}")
|
||
print(f"交易对: {TRADING_CONFIG['symbol']}")
|
||
print(f"交易数量: {TRADING_CONFIG['position_size']} BTC")
|
||
|
||
# 检查依赖包
|
||
try:
|
||
import pandas as pd
|
||
import numpy as np
|
||
print("✅ 依赖包检查通过")
|
||
except ImportError as e:
|
||
print(f"❌ 依赖包缺失: {e}")
|
||
print("请运行: pip install pandas numpy")
|
||
return False
|
||
|
||
return True
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
print("🚀 OKX API 连接测试")
|
||
print("=" * 50)
|
||
|
||
# 测试环境配置
|
||
env_ok = test_environment()
|
||
|
||
# 测试公共API
|
||
public_ok = test_public_api()
|
||
|
||
# 测试私有API
|
||
private_ok = test_private_api()
|
||
|
||
# 总结
|
||
print("\n" + "=" * 50)
|
||
print("📊 测试结果总结:")
|
||
print(f"环境配置: {'✅ 通过' if env_ok else '❌ 失败'}")
|
||
print(f"公共API: {'✅ 通过' if public_ok else '❌ 失败'}")
|
||
print(f"私有API: {'✅ 通过' if private_ok else '❌ 失败'}")
|
||
|
||
if env_ok and public_ok:
|
||
print("\n🎉 基础功能测试通过！可以运行主程序了。")
|
||
if private_ok:
|
||
print("🔐 API密钥配置正确，可以进行交易操作。")
|
||
else:
|
||
print("⚠️ API密钥需要配置或有问题，只能使用公共功能。")
|
||
else:
|
||
print("\n❌ 存在配置问题，请检查后重试。")
|
||
|
||
if __name__ == "__main__":
|
||
main() |