136 lines
3.9 KiB
Python
136 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import asyncio
|
|
import logging
|
|
from ccxt import mt5 as mt5_sync
|
|
from ccxt.async_support import mt5 as mt5_async
|
|
|
|
# 简单日志配置
|
|
logging.basicConfig(level=logging.INFO, format='%(message)s')
|
|
logger = logging.getLogger('MT5-Quick-Test')
|
|
|
|
CONFIG = {
|
|
'apiKey': '76888962',
|
|
'secret': 'LZ-trade666888',
|
|
'verbose': False, # 启用详细日志
|
|
'hostname': '43.167.188.220:5000',
|
|
'host': '18.163.85.196',
|
|
'port': 443,
|
|
}
|
|
|
|
|
|
def quick_sync_test():
|
|
"""快速同步测试"""
|
|
logger.info("🔧 快速同步测试...")
|
|
|
|
try:
|
|
exchange = mt5_sync(CONFIG)
|
|
|
|
# 测试连接
|
|
logger.info("1. 测试连接...")
|
|
token = exchange.get_token()
|
|
logger.info(f" ✅ Token: {token}")
|
|
|
|
# # 测试交易对
|
|
logger.info("4. 测试交易对...")
|
|
markets = exchange.fetch_markets()
|
|
logger.info(f" ✅ 交易对数量: {len(markets)}")
|
|
|
|
# # 测试余额
|
|
# logger.info("2. 测试余额...")
|
|
# balance = exchange.fetch_balance()
|
|
# logger.info(f" ✅ 余额: {balance['total'].get('USDT', 'N/A')}")
|
|
|
|
# 测试行情
|
|
# logger.info("3. 测试行情...")
|
|
# ticker = exchange.fetch_ticker('EUR/USD')
|
|
# logger.info(f" ✅ EUR/USD: {ticker['bid']} / {ticker['ask']}")
|
|
|
|
# 测试订单
|
|
logger.info("3. 测试订单...")
|
|
ticker = exchange.fetch_closed_orders('EUR/USD')
|
|
logger.info(f" ✅ EUR/USD: {ticker['bid']} / {ticker['ask']}")
|
|
|
|
|
|
logger.info("🎉 快速同步测试完成！")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ 快速同步测试失败: {e}")
|
|
return False
|
|
|
|
|
|
async def quick_async_test():
|
|
"""快速异步测试"""
|
|
logger.info("\n🔧 快速异步测试...")
|
|
|
|
try:
|
|
exchange = mt5_async(CONFIG)
|
|
|
|
# 测试连接
|
|
logger.info("1. 测试异步连接...")
|
|
token = await exchange.get_token()
|
|
logger.info(f" ✅ Token: {token}")
|
|
|
|
# 测试余额
|
|
logger.info("2. 测试异步余额...")
|
|
balance = await exchange.fetch_balance()
|
|
logger.info(f" ✅ 余额: {balance['total'].get('USDT', 'N/A')}")
|
|
|
|
# 测试行情
|
|
logger.info("3. 测试异步行情...")
|
|
ticker = await exchange.fetch_ticker('EUR/USD')
|
|
logger.info(f" ✅ EUR/USD: {ticker['bid']} / {ticker['ask']}")
|
|
|
|
await exchange.close()
|
|
logger.info("🎉 快速异步测试完成！")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ 快速异步测试失败: {e}")
|
|
return False
|
|
|
|
|
|
async def main():
|
|
"""运行快速测试"""
|
|
print("🚀 MT5 快速测试")
|
|
print("="*40)
|
|
|
|
# 同步测试
|
|
sync_success = quick_sync_test()
|
|
|
|
# 异步测试
|
|
# async_success = await quick_async_test()
|
|
async_success = True
|
|
|
|
print("\n" + "="*40)
|
|
print("📊 快速测试结果")
|
|
print(f"同步测试: {'✅ 通过' if sync_success else '❌ 失败'}")
|
|
print(f"异步测试: {'✅ 通过' if async_success else '❌ 失败'}")
|
|
|
|
if sync_success and async_success:
|
|
print("🎉 所有快速测试通过！")
|
|
else:
|
|
print("⚠️ 部分测试失败，请运行完整测试套件查看详情")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# asyncio.run(main())
|
|
|
|
exchange = mt5_sync(CONFIG)
|
|
|
|
# 测试连接
|
|
logger.info("1. 测试连接...")
|
|
token = exchange.get_token()
|
|
logger.info(f" ✅ Token: {token}")
|
|
|
|
# # 测试交易对
|
|
logger.info("4. 测试交易对...")
|
|
markets = exchange.fetch_markets()
|
|
logger.info(f" ✅ 交易对数量: {len(markets)}")
|
|
|
|
|
|
logger.info("3. 测试订单...")
|
|
ticker = exchange.fetch_closed_orders('BTC/USD')
|
|
logger.info(f"{ticker}") |