xm/ccxt_with_mt5
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
lz_db
2025-11-29 22:42:32 +08:00
parent b298c01743
commit efd59755d8
3 changed files with 64 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
test/test_mt5_tools.py
View File

@@ -62,39 +62,49 @@ async def quick_order_test():
logger.info("🚀 快速订单测试开始")
exchange = mt5({
'user': 62333850,
'password': 'tecimil4',
'host': '78.140.180.198',
'port': 443,
'sandbox': True,
'apiKey': '76888962',
'secret': 'LZ-trade666888',
'verbose': False, # 启用详细日志
'hostname': '43.133.206.185:5000',
'options': {
# 'server': '147.160.254.81:443', # 使用服务器名称
# 或者
'host': '18.163.85.196',
'port': 443,
},
})
try:
# 测试连接
balance = await exchange.fetch_balance()
logger.info(f"✅ 连接成功，余额: {balance['total'].get('USD', 'N/A')}")
# balance = await exchange.fetch_balance()
# logger.info(f"✅ 连接成功，余额: {balance}")
positions = await exchange.fetch_positions()
logger.info(f"✅ 连接成功，信息: {positions}")
# 获取市场信息
markets = await exchange.fetch_markets()
logger.info(f"✅ 获取到 {len(markets)} 个交易对")
# markets = await exchange.fetch_markets()
# logger.info(f"✅ 获取到 {len(markets)} 个交易对")
# 获取当前价格
ticker = await exchange.fetch_ticker('EUR/USD')
logger.info(f"✅ EUR/USD 当前价格: 买={ticker['bid']}, 卖={ticker['ask']}")
# ticker = await exchange.fetch_ticker('EUR/USD')
# logger.info(f"✅ EUR/USD 当前价格: 买={ticker['bid']}, 卖={ticker['ask']}")
# 获取订单簿
orderbook = await exchange.fetch_order_book('EUR/USD')
logger.info(f"✅ EUR/USD 订单簿深度: {len(orderbook['bids'])} 买单, {len(orderbook['asks'])} 卖单")
# orderbook = await exchange.fetch_order_book('EUR/USD')
# logger.info(f"✅ EUR/USD 订单簿深度: {len(orderbook['bids'])} 买单, {len(orderbook['asks'])} 卖单")
# 获取开单
open_orders = await exchange.fetch_open_orders()
logger.info(f"✅ 当前开单数量: {len(open_orders)}")
# open_orders = await exchange.fetch_open_orders()
# logger.info(f"✅ 当前开单数量: {len(open_orders)}")
for order in open_orders:
logger.info(f" 订单 {order['id']}: {order['symbol']} {order['side']} {order['type']} {order['status']}")
# for order in open_orders:
# logger.info(f" 订单 {order['id']}: {order['symbol']} {order['side']} {order['type']} {order['status']}")
except Exception as e:
logger.error(f"❌ 测试失败: {e}")
# 抛出异常，终止程序
raise e
finally:
await exchange.close()
@@ -107,10 +117,13 @@ async def websocket_quick_test():
'apiKey': '76888962',
'secret': 'LZ-trade666888',
'verbose': False, # 启用详细日志
# 'debug': True, # 启用详细调试信息
'hostname': '43.167.188.220:5000',
'host': '18.163.85.196',
'port': 443,
'hostname': '43.133.206.185:5000',
'options': {
# 'server': '147.160.254.81:443', # 使用服务器名称
# 或者
'host': '18.163.85.196',
'port': 443,
},
})
try:
@@ -149,4 +162,5 @@ async def websocket_quick_test():
if __name__ == "__main__":
# 运行快速测试
asyncio.run(websocket_quick_test())
asyncio.run(quick_order_test())
# asyncio.run(websocket_quick_test())