exchange_monitor_sync/sync/manager.py

import asyncio
from loguru import logger
import signal
import sys
import time
from typing import Dict

from config.settings import SYNC_CONFIG
from .position_sync_batch import PositionSyncBatch
from .order_sync_batch import OrderSyncBatch  # 使用批量版本
from .account_sync_batch import AccountSyncBatch
from utils.batch_position_sync import BatchPositionSync
from utils.batch_order_sync import BatchOrderSync
from utils.batch_account_sync import BatchAccountSync
from utils.redis_batch_helper import RedisBatchHelper

class SyncManager:
    """同步管理器（完整批量版本）"""
    
    def __init__(self):
        self.is_running = True
        self.sync_interval = SYNC_CONFIG['interval']
        
        # 初始化批量同步工具
        self.batch_tools = {}
        self.redis_helper = None
        
        # 初始化同步器
        self.syncers = []
        
        if SYNC_CONFIG['enable_position_sync']:
            position_sync = PositionSyncBatch()
            self.syncers.append(position_sync)
            self.batch_tools['position'] = BatchPositionSync(position_sync.db_manager)
            logger.info("启用持仓批量同步")
        
        if SYNC_CONFIG['enable_order_sync']:
            order_sync = OrderSyncBatch()
            self.syncers.append(order_sync)
            self.batch_tools['order'] = BatchOrderSync(order_sync.db_manager)
            
            # 初始化Redis批量助手
            if order_sync.redis_client:
                self.redis_helper = RedisBatchHelper(order_sync.redis_client.client)
            
            logger.info("启用订单批量同步")
        
        if SYNC_CONFIG['enable_account_sync']:
            account_sync = AccountSyncBatch()
            self.syncers.append(account_sync)
            self.batch_tools['account'] = BatchAccountSync(account_sync.db_manager)
            logger.info("启用账户信息批量同步")
        
        # 性能统计
        self.stats = {
            'total_syncs': 0,
            'last_sync_time': 0,
            'avg_sync_time': 0,
            'position': {'accounts': 0, 'positions': 0, 'time': 0},
            'order': {'accounts': 0, 'orders': 0, 'time': 0},
            'account': {'accounts': 0, 'records': 0, 'time': 0}
        }
        
        # 注册信号处理器
        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)
    
    async def start(self):
        """启动同步服务"""
        logger.info(f"同步服务启动，间隔 {self.sync_interval} 秒")
        
        while self.is_running:
            try:
                self.stats['total_syncs'] += 1
                sync_start = time.time()
                
                # 获取所有账号（只获取一次）
                accounts = await self._get_all_accounts()
                
                if not accounts:
                    logger.warning("未获取到任何账号，等待下次同步")
                    await asyncio.sleep(self.sync_interval)
                    continue
                
                logger.info(f"第{self.stats['total_syncs']}次同步开始，共 {len(accounts)} 个账号")
                
                # 并发执行所有同步
                await self._execute_all_syncers_concurrent(accounts)
                
                # 更新统计
                sync_time = time.time() - sync_start
                self._update_stats(sync_time)
                
                logger.info(f"同步完成，总耗时 {sync_time:.2f} 秒，等待 {self.sync_interval} 秒")
                await asyncio.sleep(self.sync_interval)
                
            except asyncio.CancelledError:
                logger.info("同步任务被取消")
                break
            except Exception as e:
                logger.error(f"同步任务异常: {e}")
                await asyncio.sleep(30)
    
    async def _get_all_accounts(self) -> Dict[str, Dict]:
        """获取所有账号"""
        if not self.syncers:
            return {}
        
        # 使用第一个同步器获取账号
        return self.syncers[0].get_accounts_from_redis()
    
    async def _execute_all_syncers_concurrent(self, accounts: Dict[str, Dict]):
        """并发执行所有同步器"""
        tasks = []
        
        # 持仓批量同步
        if 'position' in self.batch_tools:
            task = self._sync_positions_batch(accounts)
            tasks.append(task)
        
        # 订单批量同步
        if 'order' in self.batch_tools:
            task = self._sync_orders_batch(accounts)
            tasks.append(task)
        
        # 账户信息批量同步
        if 'account' in self.batch_tools:
            task = self._sync_accounts_batch(accounts)
            tasks.append(task)
        
        # 并发执行所有任务
        if tasks:
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 检查结果
            for i, result in enumerate(results):
                if isinstance(result, Exception):
                    logger.error(f"同步任务 {i} 失败: {result}")
    
    async def _sync_positions_batch(self, accounts: Dict[str, Dict]):
        """批量同步持仓数据"""
        try:
            start_time = time.time()
            
            # 收集所有持仓数据
            position_sync = next((s for s in self.syncers if isinstance(s, PositionSyncBatch)), None)
            if not position_sync:
                return
            
            all_positions = await position_sync._collect_all_positions(accounts)
            
            if not all_positions:
                self.stats['position'] = {'accounts': 0, 'positions': 0, 'time': 0}
                return
            
            # 使用批量工具同步
            batch_tool = self.batch_tools['position']
            success, stats = batch_tool.sync_positions_batch(all_positions)
            
            if success:
                elapsed = time.time() - start_time
                self.stats['position'] = {
                    'accounts': len(accounts),
                    'positions': stats['total'],
                    'time': elapsed
                }
                
        except Exception as e:
            logger.error(f"批量同步持仓失败: {e}")
            self.stats['position'] = {'accounts': 0, 'positions': 0, 'time': 0}
    
    async def _sync_orders_batch(self, accounts: Dict[str, Dict]):
        """批量同步订单数据"""
        try:
            start_time = time.time()
            
            # 收集所有订单数据
            order_sync = next((s for s in self.syncers if isinstance(s, OrderSyncBatch)), None)
            if not order_sync:
                return
            
            all_orders = await order_sync._collect_all_orders(accounts)
            
            if not all_orders:
                self.stats['order'] = {'accounts': 0, 'orders': 0, 'time': 0}
                return
            
            # 使用批量工具同步
            batch_tool = self.batch_tools['order']
            success, processed_count = batch_tool.sync_orders_batch(all_orders)
            
            if success:
                elapsed = time.time() - start_time
                self.stats['order'] = {
                    'accounts': len(accounts),
                    'orders': processed_count,
                    'time': elapsed
                }
                
        except Exception as e:
            logger.error(f"批量同步订单失败: {e}")
            self.stats['order'] = {'accounts': 0, 'orders': 0, 'time': 0}
    
    async def _sync_accounts_batch(self, accounts: Dict[str, Dict]):
        """批量同步账户信息数据"""
        try:
            start_time = time.time()
            
            # 收集所有账户数据
            account_sync = next((s for s in self.syncers if isinstance(s, AccountSyncBatch)), None)
            if not account_sync:
                return
            
            all_account_data = await account_sync._collect_all_account_data(accounts)
            
            if not all_account_data:
                self.stats['account'] = {'accounts': 0, 'records': 0, 'time': 0}
                return
            
            # 使用批量工具同步
            batch_tool = self.batch_tools['account']
            updated, inserted = batch_tool.sync_accounts_batch(all_account_data)
            
            elapsed = time.time() - start_time
            self.stats['account'] = {
                'accounts': len(accounts),
                'records': len(all_account_data),
                'time': elapsed
            }
            
        except Exception as e:
            logger.error(f"批量同步账户信息失败: {e}")
            self.stats['account'] = {'accounts': 0, 'records': 0, 'time': 0}
    
    def _update_stats(self, sync_time: float):
        """更新统计信息"""
        self.stats['last_sync_time'] = sync_time
        self.stats['avg_sync_time'] = (self.stats['avg_sync_time'] * 0.9 + sync_time * 0.1)
        
        # 打印详细统计
        stats_lines = [
            f"=== 第{self.stats['total_syncs']}次同步统计 ===",
            f"总耗时: {sync_time:.2f}秒 | 平均耗时: {self.stats['avg_sync_time']:.2f}秒"
        ]
        
        if self.stats['position']['accounts'] > 0:
            stats_lines.append(
                f"持仓: {self.stats['position']['accounts']}账号/{self.stats['position']['positions']}条"
                f"/{self.stats['position']['time']:.2f}秒"
            )
        
        if self.stats['order']['accounts'] > 0:
            stats_lines.append(
                f"订单: {self.stats['order']['accounts']}账号/{self.stats['order']['orders']}条"
                f"/{self.stats['order']['time']:.2f}秒"
            )
        
        if self.stats['account']['accounts'] > 0:
            stats_lines.append(
                f"账户: {self.stats['account']['accounts']}账号/{self.stats['account']['records']}条"
                f"/{self.stats['account']['time']:.2f}秒"
            )
        
        logger.info("\n".join(stats_lines))
    
    def signal_handler(self, signum, frame):
        """信号处理器"""
        logger.info(f"接收到信号 {signum}，正在关闭...")
        self.is_running = False
    
    async def stop(self):
        """停止同步服务"""
        self.is_running = False
        
        # 关闭所有数据库连接
        for syncer in self.syncers:
            if hasattr(syncer, 'db_manager'):
                syncer.db_manager.close()
        
        logger.info("同步服务停止")