ccxt_with_mt5/test/test_mt5_tools.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import asyncio
import json
import logging
from ccxt.pro import mt5

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('MT5-Tools')


class MT5OrderAnalyzer:
    """MT5 订单分析工具"""
    
    def __init__(self, exchange):
        self.exchange = exchange
        self.order_history = []
        
    async def analyze_orders(self, symbol=None, days=7):
        """分析订单数据"""
        try:
            # 获取订单历史
            orders = await self.exchange.fetch_orders(symbol)
            
            # 分析订单统计
            total_orders = len(orders)
            open_orders = len([o for o in orders if o['status'] == 'open'])
            closed_orders = len([o for o in orders if o['status'] == 'closed'])
            canceled_orders = len([o for o in orders if o['status'] == 'canceled'])
            
            # 分析盈亏
            profitable_trades = []
            losing_trades = []
            
            for order in orders:
                if order['status'] == 'closed' and order.get('cost'):
                    if order.get('filled', 0) > 0:
                        # 这里需要根据具体订单数据计算盈亏
                        pass
            
            logger.info(f"📈 订单分析结果:")
            logger.info(f"   总订单数: {total_orders}")
            logger.info(f"   开单数: {open_orders}")
            logger.info(f"   平单数: {closed_orders}")
            logger.info(f"   取消单数: {canceled_orders}")
            
            return {
                'total_orders': total_orders,
                'open_orders': open_orders,
                'closed_orders': closed_orders,
                'canceled_orders': canceled_orders
            }
            
        except Exception as e:
            logger.error(f"分析订单失败: {e}")
            return {}


async def quick_order_test():
    """快速订单测试"""
    logger.info("🚀 快速订单测试开始")
    
    exchange = mt5({
        '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}")

        positions = await exchange.fetch_positions()
        logger.info(f"✅ 连接成功，信息: {positions}")
        
        # 获取市场信息
        # 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']}")
        
        # 获取订单簿
        # 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)}")

        # 获取订单
        # closed_orders = await exchange.fetch_closed_orders()
        # for order in closed_orders:
        #     # del order['info']
        #     logger.info(f"✅ 历史订单: {order}")
        # logger.info(f"✅ 当前订单: {closed_orders}")
        
        # 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()


async def websocket_quick_test():
    """WebSocket 快速测试"""
    logger.info("🔌 WebSocket 快速测试开始")
    
    exchange = mt5({
        '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:
        # 监听订单更新
        async def order_listener():
            while True:
                # print("111111111")
                res = await exchange.watch_orders()
                print("===========================收到信息")
                for order in res:
                    del order['info']
                    logger.info(f"📦 订单更新: {order}")
        
        # 监听持仓更新
        async def positions_listener():
            while True:
                res = await exchange.watch_positions()
                print("===========================收到信息")
                print(len(res))
                for position in res:
                    # del position['info']
                    logger.info(f"📦 持仓更新: {position}")
        # 监听余额更新  
        # async def balance_listener():
        #     balance = await exchange.watch_balance()
        #     total = sum([v for v in balance['total'].values() if v is not None])
        #     logger.info(f"💰 余额更新: 总余额 {total:.2f}")
        
        # 运行监听器
        await asyncio.gather(
            order_listener(),
            # positions_listener(),
            # balance_listener(),
            return_exceptions=True
        )
        
    except Exception as e:
        # logger.error(f"WebSocket 测试错误: {e}")
        logger.exception("WebSocket 测试错误:")
    finally:
        await exchange.close()


if __name__ == "__main__":
    # 运行快速测试
    asyncio.run(quick_order_test())
    # asyncio.run(websocket_quick_test())