exchange_monitor_sync/sync/position_sync.py

from .base_sync import BaseSync
from loguru import logger
from typing import List, Dict, Any, Set, Tuple
from sqlalchemy import text, and_, select, delete
from models.orm_models import StrategyPosition
import utils.helpers as helpers
import time

class PositionSyncBatch(BaseSync):
    """持仓数据批量同步器"""
    
    def __init__(self):
        super().__init__()
        self.batch_size = 500  # 每批处理数量
    
    async def sync_batch(self, accounts: Dict[str, Dict]):
        """批量同步所有账号的持仓数据"""
        try:
            logger.info(f"开始批量同步持仓数据，共 {len(accounts)} 个账号")
            
            start_time = time.time()
            
            # 1. 收集所有账号的持仓数据
            all_positions = await self.redis_client._collect_all_positions(accounts)
            
            
            if not all_positions:
                logger.info("无持仓数据需要同步")
                return
            
            logger.info(f"收集到 {len(all_positions)} 条持仓数据")
            
            # 2. 批量同步到数据库
            success, stats = await self._sync_positions_batch_to_db_optimized_v3(all_positions)
            
            elapsed = time.time() - start_time
            if success:
                logger.info(f"持仓批量同步完成: 处理 {stats['total']} 条，受影响 {stats['affected']} 条，"
                          f"删除 {stats['deleted']} 条，耗时 {elapsed:.2f}秒")
            else:
                logger.error("持仓批量同步失败")
            
        except Exception as e:
            logger.error(f"持仓批量同步失败: {e}")# 获取完整的错误信息
            import traceback
            error_details = {
                'error_type': type(e).__name__,
                'error_message': str(e),
                'traceback': traceback.format_exc()
            }
            logger.error("完整堆栈跟踪:\n{traceback}", traceback=error_details['traceback'])
        
    async def _sync_positions_batch_to_db_optimized(self, all_positions: List[Dict]) -> Tuple[bool, Dict]:
        """
        批量同步持仓数据（不使用临时表）
        
        Args:
            all_positions: 所有持仓数据列表，每个持仓包含k_id（账号ID）等字段
        
        Returns:
            Tuple[bool, Dict]: (是否成功, 结果统计)
        """
        if not all_positions:
            return True, {'total': 0, 'affected': 0, 'deleted': 0, 'errors': []}
        

        session = self.db_manager.get_session()

        results = {
            'total': 0,
            'affected': 0,
            'deleted': 0,
            'errors': []
        }

        # 按账号分组
        positions_by_account = {}
        for position in all_positions:
            # print(position['symbol'])
            k_id = position['k_id']
            if k_id not in positions_by_account:
                positions_by_account[k_id] = []
            positions_by_account[k_id].append(position)
        
        logger.info(f"开始批量处理 {len(positions_by_account)} 个账号的持仓数据")
        
        try:
            # 按分组处理（10个账号一组）
            account_ids = list(positions_by_account.keys())
            
            for group_idx in range(0, len(account_ids), 10):
                group_account_ids = account_ids[group_idx:group_idx + 10]
                logger.info(f"处理第 {group_idx//10 + 1} 组账号: {group_account_ids}")
                
                # 收集本组所有持仓数据
                group_positions = []
                for k_id in group_account_ids:
                    group_positions.extend(positions_by_account[k_id])
                
                if not group_positions:
                    continue
                
                # 处理持仓数据
                processed_positions = []
                account_position_keys = {}  # 记录每个账号的持仓标识
                
                for raw_position in group_positions:
                    try:
                        k_id = raw_position['k_id']
                        processed = self._convert_position_data(raw_position)
                        
                        # 检查必要字段
                        if not all([processed.get('symbol'), processed.get('side')]):
                            continue
                        
                        # 确保st_id存在
                        if 'st_id' not in processed:
                            processed['st_id'] = raw_position.get('st_id', 0)
                        
                        # 确保k_id存在
                        if 'k_id' not in processed:
                            processed['k_id'] = k_id
                        
                        # 重命名qty为sum（如果存在）
                        if 'qty' in processed:
                            processed['sum'] = processed.pop('qty')
                        
                        
                        processed_positions.append(processed)
                        
                        # 记录持仓唯一标识
                        if k_id not in account_position_keys:
                            account_position_keys[k_id] = set()
                        
                        position_key = f"{processed['st_id']}&{processed['symbol']}&{processed['side']}"
                        # print(position_key)
                        account_position_keys[k_id].add(position_key)
                        
                    except Exception as e:
                        logger.error(f"处理持仓数据失败: {raw_position}, error={e}")
                        continue
                
                # 批量插入或更新
                if processed_positions:
                    try:
                        # 使用ON DUPLICATE KEY UPDATE批量处理
                        upsert_sql = text("""
                            INSERT INTO deh_strategy_position_new 
                            (st_id, k_id, asset, symbol, side, price, `sum`, 
                            asset_num, asset_profit, leverage, uptime, 
                            profit_price, stop_price, liquidation_price)
                            VALUES 
                            (:st_id, :k_id, :asset, :symbol, :side, :price, :sum,
                            :asset_num, :asset_profit, :leverage, :uptime,
                            :profit_price, :stop_price, :liquidation_price)
                            ON DUPLICATE KEY UPDATE
                                price = VALUES(price),
                                `sum` = VALUES(`sum`),
                                asset_num = VALUES(asset_num),
                                asset_profit = VALUES(asset_profit),
                                leverage = VALUES(leverage),
                                uptime = VALUES(uptime),
                                profit_price = VALUES(profit_price),
                                stop_price = VALUES(stop_price),
                                liquidation_price = VALUES(liquidation_price)
                        """)
                        
                        result = session.execute(upsert_sql, processed_positions)
                        
                        # 正确计算插入和更新的数量
                        total_affected = result.rowcount  # 受影响的总行数
                        batch_size = len(processed_positions)  # 本次尝试插入的数量
                        
                        # 累加到总结果
                        results['total'] += batch_size
                        results['affected'] += total_affected
                        
                        logger.debug(f"第 {group_idx//10 + 1} 组: "
                                f"处理 {batch_size} 条, "
                                f"受影响 {total_affected} 条")
                        
                    except Exception as e:
                        logger.error(f"批量插入/更新失败: {e}", exc_info=True)
                        session.rollback()
                        results['errors'].append(f"批量插入/更新失败: {str(e)}")
                        # 继续处理下一组
                        continue
                
                # 删除本组每个账号中已不存在的持仓
                for k_id in group_account_ids:
                    try:
                        if k_id not in account_position_keys or not account_position_keys[k_id]:
                            # 如果该账号没有任何持仓，删除所有
                            delete_sql = text("""
                                DELETE FROM deh_strategy_position_new 
                                WHERE k_id = :k_id
                            """)
                            
                            result = session.execute(delete_sql, {'k_id': k_id})
                            deleted_count = result.rowcount
                            results['deleted'] += deleted_count
                            
                            if deleted_count > 0:
                                logger.debug(f"账号 {k_id}: 删除所有旧持仓，共 {deleted_count} 条")
                            
                        else:
                            # 构建当前持仓的条件
                            current_keys = account_position_keys[k_id]
                            
                            # 使用多个OR条件来处理IN子句的限制
                            conditions = []
                            params = {'k_id': k_id}
                            
                            for idx, key in enumerate(current_keys):
                                parts = key.split('&')
                                if len(parts) >= 3:  # 确保有st_id, symbol, side三部分
                                    st_id_val = parts[0]
                                    symbol_val = parts[1]
                                    side_val = parts[2]
                                    
                                    conditions.append(f"(st_id = :st_id_{idx} AND symbol = :symbol_{idx} AND side = :side_{idx})")
                                    params[f'st_id_{idx}'] = int(st_id_val) if st_id_val.isdigit() else st_id_val
                                    params[f'symbol_{idx}'] = symbol_val
                                    params[f'side_{idx}'] = side_val
                            
                            if conditions:
                                conditions_str = " OR ".join(conditions)
                                
                                # 删除不在当前持仓列表中的记录
                                delete_sql = text(f"""
                                    DELETE FROM deh_strategy_position_new 
                                    WHERE k_id = :k_id 
                                    AND NOT ({conditions_str})
                                """)
                                
                                result = session.execute(delete_sql, params)
                                deleted_count = result.rowcount
                                results['deleted'] += deleted_count
                                
                                if deleted_count > 0:
                                    logger.debug(f"账号 {k_id}: 删除 {deleted_count} 条过期持仓")
                                    
                    except Exception as e:
                        logger.error(f"删除账号 {k_id} 旧持仓失败: {e}")
                        # 记录错误但继续处理其他账号
                        results['errors'].append(f"删除账号 {k_id} 旧持仓失败: {str(e)}")
                
                # 每组结束后提交
                try:
                    session.commit()
                    logger.debug(f"第 {group_idx//10 + 1} 组处理完成并提交")
                except Exception as e:
                    session.rollback()
                    logger.error(f"第 {group_idx//10 + 1} 组提交失败: {e}")
                    results['errors'].append(f"第 {group_idx//10 + 1} 组提交失败: {str(e)}")
            
            logger.info(f"批量同步完成: "
                    f"总数={results['total']}, "
                    f"受影响={results['affected']}, "
                    f"删除={results['deleted']}, "
                    f"错误数={len(results['errors'])}")
            
            
            success = len(results['errors']) == 0
            return success, results
            
        except Exception as e:
            session.rollback()
            logger.error(f"批量同步过程中发生错误: {e}", exc_info=True)
            results['errors'].append(f"同步过程错误: {str(e)}")
            return False, results
        
        finally:
            session.close()

    async def _sync_positions_batch_to_db_optimized_v3(self, all_positions: List[Dict]) -> Tuple[bool, Dict]:
        """
        最优化的批量同步（兼容所有MySQL版本）
        
        使用策略：
        1. 一次性UPSERT所有持仓数据
        2. 使用UNION ALL构造虚拟表进行JOIN删除
        
        Args:
            all_positions: 所有持仓数据列表
        
        Returns:
            Tuple[bool, Dict]: (是否成功, 结果统计)
        """
        if not all_positions:
            return True, {'total': 0, 'affected': 0, 'deleted': 0, 'errors': []}
        
        session = self.db_manager.get_session()
        
        results = {
            'total': 0,
            'affected': 0,
            'deleted': 0,
            'errors': []
        }
        
        try:
            session.begin()
            
            # 准备数据
            processed_positions = []
            current_position_records = set()  # 使用set去重，避免重复
            
            for raw_position in all_positions:
                try:
                    processed = self._convert_position_data(raw_position)
                    
                    if not all([processed.get('symbol'), processed.get('side')]):
                        continue
                    
                    if 'qty' in processed:
                        processed['sum'] = processed.pop('qty')
                    
                    k_id = processed.get('k_id', raw_position['k_id'])
                    st_id = processed.get('st_id', raw_position.get('st_id', 0))
                    symbol = processed.get('symbol')
                    side = processed.get('side')
                    
                    
                    processed_positions.append(processed)
                    
                    # 去重记录当前持仓
                    record_key = (k_id, st_id, symbol, side)
                    current_position_records.add(record_key)
                    
                except Exception as e:
                    logger.error(f"处理持仓数据失败: {raw_position}, error={e}")
                    continue
            
            if not processed_positions:
                session.commit()
                return True, results
            
            results['total'] = len(processed_positions)
            logger.info(f"准备同步 {results['total']} 条持仓数据，去重后 {len(current_position_records)} 条唯一记录")
            
            # 批量UPSERT
            upsert_sql = text("""
                INSERT INTO deh_strategy_position_new 
                (st_id, k_id, asset, symbol, side, price, `sum`, 
                asset_num, asset_profit, leverage, uptime, 
                profit_price, stop_price, liquidation_price)
                VALUES 
                (:st_id, :k_id, :asset, :symbol, :side, :price, :sum,
                :asset_num, :asset_profit, :leverage, :uptime,
                :profit_price, :stop_price, :liquidation_price)
                ON DUPLICATE KEY UPDATE
                    price = VALUES(price),
                    `sum` = VALUES(`sum`),
                    asset_num = VALUES(asset_num),
                    asset_profit = VALUES(asset_profit),
                    leverage = VALUES(leverage),
                    uptime = VALUES(uptime),
                    profit_price = VALUES(profit_price),
                    stop_price = VALUES(stop_price),
                    liquidation_price = VALUES(liquidation_price)
            """)
            
            result = session.execute(upsert_sql, processed_positions)
            
            total_affected = result.rowcount
            
            results['affected'] =total_affected
            
            logger.info(f"UPSERT完成: 总数 {results['total']} 条, 受影响 {results['affected']} 条")
            
            # 批量删除（使用UNION ALL构造虚拟表）
            if current_position_records:
                # 构建UNION ALL查询
                union_parts = []
                for record in current_position_records:
                    k_id, st_id, symbol, side = record
                    # 转义单引号
                    symbol_escaped = symbol.replace("'", "''")
                    side_escaped = side.replace("'", "''")
                    union_parts.append(f"SELECT {k_id} as k_id, {st_id} as st_id, '{symbol_escaped}' as symbol, '{side_escaped}' as side")
                
                if union_parts:
                    union_sql = " UNION ALL ".join(union_parts)
                    
                    # 或者使用LEFT JOIN方式
                    delete_sql_join = text(f"""
                        DELETE p FROM deh_strategy_position_new p
                        LEFT JOIN (
                            {union_sql}
                        ) AS current_pos ON 
                            p.k_id = current_pos.k_id
                            AND p.st_id = current_pos.st_id
                            AND p.symbol = current_pos.symbol
                            AND p.side = current_pos.side
                        WHERE current_pos.k_id IS NULL
                    """)
                    
                    result = session.execute(delete_sql_join)
                    deleted_count = result.rowcount
                    results['deleted'] = deleted_count
                    
                    logger.info(f"删除 {deleted_count} 条过期持仓")
            
            session.commit()
            
            logger.info(f"批量同步V3完成: 总数={results['total']}, "
                    f"受影响={results['affected']}, "
                    f"删除={results['deleted']}")
            
            return True, results
            
        except Exception as e:
            session.rollback()
            logger.error(f"批量同步V3失败: {e}", exc_info=True)
            results['errors'].append(f"同步失败: {str(e)}")
            return False, results
        
        finally:
            session.close()

    def _convert_position_data(self, data: Dict) -> Dict:
        """转换持仓数据格式"""
        try:
            
            return {
                'st_id': helpers.safe_int(data.get('st_id'), 0),
                'k_id': helpers.safe_int(data.get('k_id'), 0),
                'asset': data.get('asset', 'USDT'),
                'symbol': data.get('symbol', ''),
                'side': data.get('side', ''),
                'price': helpers.safe_float(data.get('price')),
                'qty': helpers.safe_float(data.get('qty')),  # 后面会重命名为sum
                'asset_num': helpers.safe_float(data.get('asset_num')),
                'asset_profit': helpers.safe_float(data.get('asset_profit')),
                'leverage': helpers.safe_int(data.get('leverage')),
                'uptime': helpers.safe_int(data.get('uptime')),
                'profit_price': helpers.safe_float(data.get('profit_price')),
                'stop_price': helpers.safe_float(data.get('stop_price')),
                'liquidation_price': helpers.safe_float(data.get('liquidation_price'))
            }
            
        except Exception as e:
            logger.error(f"转换持仓数据异常: {data}, error={e}")
            return {}