add

.env

@@ -0,0 +1,35 @@
+# Redis配置
+REDIS_HOST=localhost
+REDIS_PORT=6379
+REDIS_PASSWORD=
+REDIS_DB=0
+
+# MySQL配置
+DB_HOST=localhost
+DB_PORT=3306
+DB_USER=root
+DB_PASSWORD=your_password
+DB_DATABASE=exchange_monitor
+DB_POOL_SIZE=10
+DB_MAX_OVERFLOW=20
+DB_POOL_RECYCLE=3600
+
+# SQLAlchemy配置
+SQLALCHEMY_ECHO=false
+SQLALCHEMY_ECHO_POOL=false
+
+# 同步配置
+SYNC_INTERVAL=60
+RECENT_DAYS=3
+CHUNK_SIZE=1000
+ENABLE_POSITION_SYNC=true
+ENABLE_ORDER_SYNC=true
+ENABLE_ACCOUNT_SYNC=true
+
+# 日志配置
+LOG_LEVEL=INFO
+LOG_ROTATION=10 MB
+LOG_RETENTION=7 days
+
+# 计算机名（用于过滤账号）
+COMPUTER_NAME=lz_c01

config/database.py

@@ -0,0 +1,25 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# MySQL数据库配置
+DATABASE_CONFIG = {
+    'host': os.getenv('DB_HOST', 'localhost'),
+    'port': int(os.getenv('DB_PORT', 3306)),
+    'user': os.getenv('DB_USER', 'root'),
+    'password': os.getenv('DB_PASSWORD', ''),
+    'database': os.getenv('DB_DATABASE', 'exchange_monitor'),
+    'charset': 'utf8mb4',
+    'pool_size': int(os.getenv('DB_POOL_SIZE', 10)),
+    'max_overflow': int(os.getenv('DB_MAX_OVERFLOW', 20)),
+    'pool_recycle': int(os.getenv('DB_POOL_RECYCLE', 3600))
+}
+
+# SQLAlchemy配置
+SQLALCHEMY_CONFIG = {
+    'echo': os.getenv('SQLALCHEMY_ECHO', 'False').lower() == 'true',
+    'echo_pool': os.getenv('SQLALCHEMY_ECHO_POOL', 'False').lower() == 'true',
+    'pool_pre_ping': True,
+    'pool_recycle': 3600
+}

config/settings.py

@@ -0,0 +1,35 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Redis配置
+REDIS_CONFIG = {
+    'host': os.getenv('REDIS_HOST', 'localhost'),
+    'port': int(os.getenv('REDIS_PORT', 6379)),
+    'password': os.getenv('REDIS_PASSWORD', ''),
+    'db': int(os.getenv('REDIS_DB', 0)),
+    'decode_responses': True,
+    'max_connections': 30
+}
+
+# 同步配置
+SYNC_CONFIG = {
+    'interval': int(os.getenv('SYNC_INTERVAL', 60)),  # 同步间隔（秒）
+    'recent_days': int(os.getenv('RECENT_DAYS', 3)),  # 同步最近几天数据
+    'chunk_size': int(os.getenv('CHUNK_SIZE', 1000)),  # 批量处理大小
+    'enable_position_sync': os.getenv('ENABLE_POSITION_SYNC', 'true').lower() == 'true',
+    'enable_order_sync': os.getenv('ENABLE_ORDER_SYNC', 'true').lower() == 'true',
+    'enable_account_sync': os.getenv('ENABLE_ACCOUNT_SYNC', 'true').lower() == 'true'
+}
+
+# 日志配置
+LOG_CONFIG = {
+    'level': os.getenv('LOG_LEVEL', 'INFO'),
+    'rotation': os.getenv('LOG_ROTATION', '10 MB'),
+    'retention': os.getenv('LOG_RETENTION', '7 days'),
+    'format': '{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}:{function}:{line} - {message}'
+}
+
+# 计算机名配置（用于过滤账号）
+COMPUTER_NAME = os.getenv('COMPUTER_NAME', 'lz_c01')

main.py

@@ -0,0 +1,52 @@
+import asyncio
+import sys
+from loguru import logger
+import os
+
+# 配置日志
+from config.settings import LOG_CONFIG
+
+logger.remove()  # 移除默认处理器
+logger.add(
+    sys.stdout,
+    level=LOG_CONFIG['level'],
+    format=LOG_CONFIG['format']
+)
+logger.add(
+    "logs/sync_{time:YYYY-MM-DD}.log",
+    rotation=LOG_CONFIG['rotation'],
+    retention=LOG_CONFIG['retention'],
+    level=LOG_CONFIG['level'],
+    format=LOG_CONFIG['format']
+)
+
+async def main():
+    """主程序"""
+    from sync.manager import SyncManager
+    
+    logger.info("=== 交易所数据同步服务启动 ===")
+    logger.info(f"工作目录: {os.getcwd()}")
+    
+    try:
+        manager = SyncManager()
+        await manager.start()
+        
+    except KeyboardInterrupt:
+        logger.info("用户中断")
+    except Exception as e:
+        logger.error(f"服务运行异常: {e}")
+        import traceback
+        logger.error(traceback.format_exc())
+    finally:
+        logger.info("=== 交易所数据同步服务停止 ===")
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("程序被用户中断")
+    except Exception as e:
+        logger.error(f"程序异常: {e}")
+        import traceback
+        logger.error(traceback.format_exc())
+        sys.exit(1)

models/orm_models.py

@@ -0,0 +1,76 @@
+from sqlalchemy import create_engine, MetaData
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
+from sqlalchemy import Integer, String, Float, DateTime, DECIMAL, func
+from datetime import datetime
+from typing import Optional
+import json
+
+class Base(DeclarativeBase):
+    """SQLAlchemy基础类"""
+    pass
+
+class StrategyPosition(Base):
+    """持仓表模型"""
+    __tablename__ = 'deh_strategy_position_new'
+    
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment='自增id')
+    st_id: Mapped[int] = mapped_column(Integer, nullable=False, comment='策略id')
+    k_id: Mapped[int] = mapped_column(Integer, default=0, comment='对应strategy_key 的ID')
+    asset: Mapped[str] = mapped_column(String(32), default='USDT', comment='使用资产名称')
+    symbol: Mapped[str] = mapped_column(String(50), nullable=False, comment='交易对')
+    price: Mapped[Optional[float]] = mapped_column(Float, comment='持仓均价')
+    side: Mapped[str] = mapped_column(String(10), nullable=False, comment='方向')
+    
+    # 注意：这里属性名是 qty，但对应数据库字段是 sum
+    qty: Mapped[float] = mapped_column("sum", Float, nullable=False, comment='仓位(张数)')
+    
+    asset_num: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='资产数量')
+    asset_profit: Mapped[Optional[float]] = mapped_column(DECIMAL(20, 8), comment='利润数量')
+    leverage: Mapped[int] = mapped_column(Integer, default=0, comment='杠杆倍数')
+    uptime: Mapped[int] = mapped_column(Integer, nullable=False, comment='更新时间')
+    profit_price: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='止盈价格')
+    stop_price: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='止损价格')
+    liquidation_price: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='强平价格')
+    
+    def __repr__(self) -> str:
+        return f"StrategyPosition(id={self.id!r}, k_id={self.k_id!r}, symbol={self.symbol!r})"
+
+class StrategyOrder(Base):
+    """订单表模型"""
+    __tablename__ = 'deh_strategy_order_new'
+    
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    st_id: Mapped[Optional[int]] = mapped_column(Integer, comment='策略id')
+    k_id: Mapped[int] = mapped_column(Integer, default=0, comment='对应strategy_key 的ID')
+    asset: Mapped[Optional[str]] = mapped_column(String(32), default='USDT', comment='资产名称')
+    order_id: Mapped[Optional[str]] = mapped_column(String(765), comment='订单id')
+    symbol: Mapped[Optional[str]] = mapped_column(String(120), comment='交易对')
+    side: Mapped[Optional[str]] = mapped_column(String(120), comment='订单方向')
+    price: Mapped[Optional[float]] = mapped_column(Float, comment='订单价格')
+    time: Mapped[Optional[int]] = mapped_column(Integer, comment='订单时间')
+    order_qty: Mapped[Optional[float]] = mapped_column(Float, comment='订单挂单数量')
+    last_qty: Mapped[Optional[float]] = mapped_column(Float, comment='订单成交数量')
+    avg_price: Mapped[Optional[float]] = mapped_column(Float, comment='订单成交均价')
+    exchange_id: Mapped[Optional[int]] = mapped_column(Integer, comment='交易所id')
+    
+    def __repr__(self) -> str:
+        return f"StrategyOrder(id={self.id!r}, order_id={self.order_id!r})"
+
+class StrategyKX(Base):
+    """账户信息表模型"""
+    __tablename__ = 'deh_strategy_kx_new'
+    
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment='自增id')
+    st_id: Mapped[Optional[int]] = mapped_column(Integer, comment='策略id')
+    k_id: Mapped[int] = mapped_column(Integer, default=0, comment='对应strategy_key 的ID')
+    asset: Mapped[Optional[str]] = mapped_column(String(32), default='USDT', comment='资产名')
+    balance: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='当日账户金额')
+    withdrawal: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='当日提现')
+    deposit: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='当日充值')
+    other: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='当日其他')
+    profit: Mapped[float] = mapped_column(DECIMAL(20, 8), default=0.00000000, comment='当日利润')
+    time: Mapped[Optional[int]] = mapped_column(Integer, comment='时间')
+    up_time: Mapped[Optional[datetime]] = mapped_column(DateTime, default=func.now(), onupdate=func.now(), comment='最后更新时间')
+    
+    def __repr__(self) -> str:
+        return f"StrategyKX(id={self.id!r}, k_id={self.k_id!r}, time={self.time!r})"

requirements.txt

@@ -0,0 +1,14 @@
+# 基础依赖
+python-dotenv>=1.0.0
+loguru>=0.7.0
+
+# 数据库
+SQLAlchemy>=2.0.0
+pymysql>=1.1.0
+aiomysql>=0.2.0
+
+# Redis
+redis>=5.0.0
+
+# 异步
+asyncio>=3.4.3

sync/account_sync.py

@@ -0,0 +1,183 @@
+from .base_sync import BaseSync
+from loguru import logger
+from typing import List, Dict
+import json
+import time
+from datetime import datetime, timedelta
+
+class AccountSync(BaseSync):
+    """账户信息同步器"""
+    
+    async def sync(self):
+        """同步账户信息数据"""
+        try:
+            # 获取所有账号
+            accounts = self.get_accounts_from_redis()
+            
+            for k_id_str, account_info in accounts.items():
+                try:
+                    k_id = int(k_id_str)
+                    st_id = account_info.get('st_id', 0)
+                    exchange_id = account_info['exchange_id']
+                    
+                    if k_id <= 0 or st_id <= 0:
+                        continue
+                    
+                    # 从Redis获取账户信息数据
+                    account_data = await self._get_account_info_from_redis(k_id, st_id, exchange_id)
+                    
+                    # 同步到数据库
+                    if account_data:
+                        success = self._sync_account_info_to_db(account_data)
+                        if success:
+                            logger.debug(f"账户信息同步成功: k_id={k_id}")
+                    
+                except Exception as e:
+                    logger.error(f"同步账号 {k_id_str} 账户信息失败: {e}")
+                    continue
+            
+            logger.info("账户信息同步完成")
+            
+        except Exception as e:
+            logger.error(f"账户信息同步失败: {e}")
+    
+    async def _get_account_info_from_redis(self, k_id: int, st_id: int, exchange_id: str) -> List[Dict]:
+        """从Redis获取账户信息数据"""
+        try:
+            redis_key = f"{exchange_id}:balance:{k_id}"
+            redis_funds = self.redis_client.client.hgetall(redis_key)
+            
+            if not redis_funds:
+                return []
+            
+            # 按天统计数据
+            from config.settings import SYNC_CONFIG
+            recent_days = SYNC_CONFIG['recent_days']
+            
+            today = datetime.now()
+            date_stats = {}
+            
+            # 收集所有日期的数据
+            for fund_key, fund_json in redis_funds.items():
+                try:
+                    fund_data = json.loads(fund_json)
+                    date_str = fund_data.get('lz_time', '')
+                    lz_type = fund_data.get('lz_type', '')
+                    
+                    if not date_str or lz_type not in ['lz_balance', 'deposit', 'withdrawal']:
+                        continue
+                    
+                    # 只处理最近N天的数据
+                    date_obj = datetime.strptime(date_str, '%Y-%m-%d')
+                    if (today - date_obj).days > recent_days:
+                        continue
+                    
+                    if date_str not in date_stats:
+                        date_stats[date_str] = {
+                            'balance': 0.0,
+                            'deposit': 0.0,
+                            'withdrawal': 0.0,
+                            'has_balance': False
+                        }
+                    
+                    lz_amount = float(fund_data.get('lz_amount', 0))
+                    
+                    if lz_type == 'lz_balance':
+                        date_stats[date_str]['balance'] = lz_amount
+                        date_stats[date_str]['has_balance'] = True
+                    elif lz_type == 'deposit':
+                        date_stats[date_str]['deposit'] += lz_amount
+                    elif lz_type == 'withdrawal':
+                        date_stats[date_str]['withdrawal'] += lz_amount
+                        
+                except (json.JSONDecodeError, ValueError) as e:
+                    logger.debug(f"解析Redis数据失败: {fund_key}, error={e}")
+                    continue
+            
+            # 转换为账户信息数据
+            account_data_list = []
+            sorted_dates = sorted(date_stats.keys())
+            prev_balance = 0.0
+            
+            for date_str in sorted_dates:
+                stats = date_stats[date_str]
+                
+                # 如果没有余额数据但有充提数据，仍然处理
+                if not stats['has_balance'] and stats['deposit'] == 0 and stats['withdrawal'] == 0:
+                    continue
+                
+                balance = stats['balance']
+                deposit = stats['deposit']
+                withdrawal = stats['withdrawal']
+                
+                # 计算利润
+                profit = balance - deposit - withdrawal - prev_balance
+                
+                # 转换时间戳
+                date_obj = datetime.strptime(date_str, '%Y-%m-%d')
+                time_timestamp = int(date_obj.timestamp())
+                
+                account_data = {
+                    'st_id': st_id,
+                    'k_id': k_id,
+                    'balance': balance,
+                    'withdrawal': withdrawal,
+                    'deposit': deposit,
+                    'other': 0.0,  # 暂时为0
+                    'profit': profit,
+                    'time': time_timestamp
+                }
+                
+                account_data_list.append(account_data)
+                
+                # 更新前一天的余额
+                if stats['has_balance']:
+                    prev_balance = balance
+            
+            return account_data_list
+            
+        except Exception as e:
+            logger.error(f"获取Redis账户信息失败: k_id={k_id}, error={e}")
+            return []
+    
+    def _sync_account_info_to_db(self, account_data_list: List[Dict]) -> bool:
+        """同步账户信息到数据库"""
+        session = self.db_manager.get_session()
+        try:
+            with session.begin():
+                for account_data in account_data_list:
+                    try:
+                        # 查询是否已存在
+                        existing = session.execute(
+                            select(StrategyKX).where(
+                                and_(
+                                    StrategyKX.k_id == account_data['k_id'],
+                                    StrategyKX.st_id == account_data['st_id'],
+                                    StrategyKX.time == account_data['time']
+                                )
+                            )
+                        ).scalar_one_or_none()
+                        
+                        if existing:
+                            # 更新
+                            existing.balance = account_data['balance']
+                            existing.withdrawal = account_data['withdrawal']
+                            existing.deposit = account_data['deposit']
+                            existing.other = account_data['other']
+                            existing.profit = account_data['profit']
+                        else:
+                            # 插入
+                            new_account = StrategyKX(**account_data)
+                            session.add(new_account)
+                            
+                    except Exception as e:
+                        logger.error(f"处理账户数据失败: {account_data}, error={e}")
+                        continue
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"同步账户信息到数据库失败: error={e}")
+            return False
+        finally:
+            session.close()

sync/base_sync.py

@@ -0,0 +1,90 @@
+from abc import ABC, abstractmethod
+from loguru import logger
+from typing import List, Dict, Any
+import json
+
+from utils.redis_client import RedisClient
+from utils.database_manager import DatabaseManager
+
+class BaseSync(ABC):
+    """同步基类"""
+    
+    def __init__(self):
+        self.redis_client = RedisClient()
+        self.db_manager = DatabaseManager()
+        self.computer_name = None  # 从配置读取
+    
+    @abstractmethod
+    async def sync(self):
+        """执行同步"""
+        pass
+    
+    def get_accounts_from_redis(self) -> Dict[str, Dict]:
+        """从Redis获取账号配置"""
+        try:
+            if self.computer_name is None:
+                from config.settings import COMPUTER_NAME
+                self.computer_name = COMPUTER_NAME
+            
+            # 从Redis获取数据
+            result = self.redis_client.client.hgetall(f"{self.computer_name}_strategy_api")
+            if not result:
+                logger.warning(f"未找到 {self.computer_name} 的策略API配置")
+                return {}
+            
+            accounts_dict = {}
+            for exchange_name, accounts_json in result.items():
+                try:
+                    accounts = json.loads(accounts_json)
+                    if not accounts:
+                        continue
+                    
+                    # 格式化交易所ID
+                    exchange_id = self.format_exchange_id(exchange_name)
+                    
+                    for account_id, account_info in accounts.items():
+                        parsed_account = self.parse_account(exchange_id, account_id, account_info)
+                        if parsed_account:
+                            accounts_dict[account_id] = parsed_account
+                            
+                except json.JSONDecodeError as e:
+                    logger.error(f"解析交易所 {exchange_name} 的JSON数据失败: {e}")
+                    continue
+            
+            return accounts_dict
+            
+        except Exception as e:
+            logger.error(f"获取账户信息失败: {e}")
+            return {}
+    
+    def format_exchange_id(self, key: str) -> str:
+        """格式化交易所ID"""
+        key = key.lower().strip()
+        
+        # 交易所名称映射
+        exchange_mapping = {
+            'metatrader': 'mt5',
+            'binance_spot_test': 'binance',
+            'binance_spot': 'binance',
+            'binance': 'binance',
+            'gate_spot': 'gate',
+            'okex': 'okx'
+        }
+        
+        return exchange_mapping.get(key, key)
+    
+    def parse_account(self, exchange_id: str, account_id: str, account_info: str) -> Dict:
+        """解析账号信息"""
+        try:
+            source_account_info = json.loads(account_info)
+            account_data = {
+                'exchange_id': exchange_id,
+                'k_id': account_id,
+                'st_id': int(source_account_info.get('st_id', 0)),
+                'add_time': int(source_account_info.get('add_time', 0))
+            }
+            return {**source_account_info, **account_data}
+            
+        except json.JSONDecodeError as e:
+            logger.error(f"解析账号 {account_id} 数据失败: {e}")
+            return {}

sync/manager.py

@@ -0,0 +1,66 @@
+import asyncio
+from loguru import logger
+from typing import List, Dict
+import signal
+import sys
+
+from config.settings import SYNC_CONFIG
+from .position_sync import PositionSync
+from .order_sync import OrderSync
+from .account_sync import AccountSync
+
+class SyncManager:
+    """同步管理器"""
+    
+    def __init__(self):
+        self.is_running = True
+        self.sync_interval = SYNC_CONFIG['interval']
+        
+        # 初始化同步器
+        self.syncers = []
+        
+        if SYNC_CONFIG['enable_position_sync']:
+            self.syncers.append(PositionSync())
+            logger.info("启用持仓同步")
+        
+        if SYNC_CONFIG['enable_order_sync']:
+            self.syncers.append(OrderSync())
+            logger.info("启用订单同步")
+        
+        if SYNC_CONFIG['enable_account_sync']:
+            self.syncers.append(AccountSync())
+            logger.info("启用账户信息同步")
+        
+        # 注册信号处理器
+        signal.signal(signal.SIGINT, self.signal_handler)
+        signal.signal(signal.SIGTERM, self.signal_handler)
+    
+    def signal_handler(self, signum, frame):
+        """信号处理器"""
+        logger.info(f"接收到信号 {signum}，正在关闭...")
+        self.is_running = False
+    
+    async def start(self):
+        """启动同步服务"""
+        logger.info(f"同步服务启动，间隔 {self.sync_interval} 秒")
+        
+        while self.is_running:
+            try:
+                # 执行所有同步器
+                tasks = [syncer.sync() for syncer in self.syncers]
+                await asyncio.gather(*tasks, return_exceptions=True)
+                
+                logger.debug(f"同步完成，等待 {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)  # 出错后等待30秒
+    
+    async def stop(self):
+        """停止同步服务"""
+        self.is_running = False
+        logger.info("同步服务停止")

sync/order_sync.py

@@ -0,0 +1,166 @@
+from .base_sync import BaseSync
+from loguru import logger
+from typing import List, Dict
+import json
+import time
+from datetime import datetime, timedelta
+
+class OrderSync(BaseSync):
+    """订单数据同步器"""
+    
+    async def sync(self):
+        """同步订单数据"""
+        try:
+            # 获取所有账号
+            accounts = self.get_accounts_from_redis()
+            
+            for k_id_str, account_info in accounts.items():
+                try:
+                    k_id = int(k_id_str)
+                    st_id = account_info.get('st_id', 0)
+                    exchange_id = account_info['exchange_id']
+                    
+                    if k_id <= 0 or st_id <= 0:
+                        continue
+                    
+                    # 从Redis获取最近N天的订单数据
+                    orders = await self._get_recent_orders_from_redis(k_id, exchange_id)
+                    
+                    # 同步到数据库
+                    if orders:
+                        success = self._sync_orders_to_db(k_id, st_id, orders)
+                        if success:
+                            logger.debug(f"订单同步成功: k_id={k_id}, 订单数={len(orders)}")
+                    
+                except Exception as e:
+                    logger.error(f"同步账号 {k_id_str} 订单失败: {e}")
+                    continue
+            
+            logger.info("订单数据同步完成")
+            
+        except Exception as e:
+            logger.error(f"订单同步失败: {e}")
+    
+    async def _get_recent_orders_from_redis(self, k_id: int, exchange_id: str) -> List[Dict]:
+        """从Redis获取最近N天的订单数据"""
+        try:
+            redis_key = f"{exchange_id}:orders:{k_id}"
+            
+            # 计算最近N天的日期
+            from config.settings import SYNC_CONFIG
+            recent_days = SYNC_CONFIG['recent_days']
+            
+            today = datetime.now()
+            recent_dates = []
+            for i in range(recent_days):
+                date = today - timedelta(days=i)
+                date_format = date.strftime('%Y-%m-%d')
+                recent_dates.append(date_format)
+            
+            # 获取所有key
+            all_keys = self.redis_client.client.hkeys(redis_key)
+            
+            orders_list = []
+            for key in all_keys:
+                key_str = key.decode('utf-8') if isinstance(key, bytes) else key
+                
+                if key_str == 'positions':
+                    continue
+                
+                # 检查是否以最近N天的日期开头
+                for date_format in recent_dates:
+                    if key_str.startswith(date_format + '_'):
+                        try:
+                            order_json = self.redis_client.client.hget(redis_key, key_str)
+                            if order_json:
+                                order = json.loads(order_json)
+                                
+                                # 验证时间
+                                order_time = order.get('time', 0)
+                                if order_time >= int(time.time()) - recent_days * 24 * 3600:
+                                    orders_list.append(order)
+                                
+                                break
+                        except:
+                            break
+            
+            return orders_list
+            
+        except Exception as e:
+            logger.error(f"获取Redis订单数据失败: k_id={k_id}, error={e}")
+            return []
+    
+    def _sync_orders_to_db(self, k_id: int, st_id: int, orders_data: List[Dict]) -> bool:
+        """同步订单数据到数据库"""
+        session = self.db_manager.get_session()
+        try:
+            # 准备批量数据
+            insert_data = []
+            for order_data in orders_data:
+                try:
+                    order_dict = self._convert_order_data(order_data)
+                    
+                    # 检查完整性
+                    required_fields = ['order_id', 'symbol', 'side', 'time']
+                    if not all(order_dict.get(field) for field in required_fields):
+                        continue
+                    
+                    insert_data.append(order_dict)
+                    
+                except Exception as e:
+                    logger.error(f"转换订单数据失败: {order_data}, error={e}")
+                    continue
+            
+            if not insert_data:
+                return True
+            
+            with session.begin():
+                # 使用参数化批量插入
+                sql = """
+                INSERT INTO deh_strategy_order_new 
+                (st_id, k_id, asset, order_id, symbol, side, price, time, 
+                 order_qty, last_qty, avg_price, exchange_id)
+                VALUES 
+                (:st_id, :k_id, :asset, :order_id, :symbol, :side, :price, :time,
+                 :order_qty, :last_qty, :avg_price, :exchange_id)
+                ON DUPLICATE KEY UPDATE
+                    side = VALUES(side),
+                    price = VALUES(price),
+                    time = VALUES(time),
+                    order_qty = VALUES(order_qty),
+                    last_qty = VALUES(last_qty),
+                    avg_price = VALUES(avg_price)
+                """
+                
+                # 分块执行
+                from config.settings import SYNC_CONFIG
+                chunk_size = SYNC_CONFIG['chunk_size']
+                
+                for i in range(0, len(insert_data), chunk_size):
+                    chunk = insert_data[i:i + chunk_size]
+                    session.execute(text(sql), chunk)
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"同步订单到数据库失败: k_id={k_id}, error={e}")
+            return False
+        finally:
+            session.close()
+    
+    def _convert_order_data(self, data: Dict) -> Dict:
+        """转换订单数据格式"""
+        return {
+            'st_id': int(data.get('st_id', 0)),
+            'k_id': int(data.get('k_id', 0)),
+            'asset': 'USDT',
+            'order_id': str(data.get('order_id', '')),
+            'symbol': data.get('symbol', ''),
+            'side': data.get('side', ''),
+            'price': float(data.get('price', 0)) if data.get('price') is not None else None,
+            'time': int(data.get('time', 0)) if data.get('time') is not None else None,
+            'order_qty': float(data.get('order_qty', 0)) if data.get('order_qty') is not None else None,
+            'last_qty': float(data.get('last_qty', 0)) if data.get('last_qty') is not None else None,
+            'avg_price': float(data.get('avg_price', 0)) if data.get('avg_price') is not None else None,
+            'exchange_id': None  # 忽略该字段
+        }

sync/position_sync.py

@@ -0,0 +1,174 @@
+from .base_sync import BaseSync
+from loguru import logger
+from typing import List, Dict
+import json
+from datetime import datetime, timedelta
+
+class PositionSync(BaseSync):
+    """持仓数据同步器"""
+    
+    async def sync(self):
+        """同步持仓数据"""
+        try:
+            # 获取所有账号
+            accounts = self.get_accounts_from_redis()
+            
+            for k_id_str, account_info in accounts.items():
+                try:
+                    k_id = int(k_id_str)
+                    st_id = account_info.get('st_id', 0)
+                    exchange_id = account_info['exchange_id']
+                    
+                    if k_id <= 0 or st_id <= 0:
+                        continue
+                    
+                    # 从Redis获取持仓数据
+                    positions = await self._get_positions_from_redis(k_id, exchange_id)
+                    
+                    # 同步到数据库
+                    if positions:
+                        success = self._sync_positions_to_db(k_id, st_id, positions)
+                        if success:
+                            logger.debug(f"持仓同步成功: k_id={k_id}, 持仓数={len(positions)}")
+                    
+                except Exception as e:
+                    logger.error(f"同步账号 {k_id_str} 持仓失败: {e}")
+                    continue
+            
+            logger.info("持仓数据同步完成")
+            
+        except Exception as e:
+            logger.error(f"持仓同步失败: {e}")
+    
+    async def _get_positions_from_redis(self, k_id: int, exchange_id: str) -> List[Dict]:
+        """从Redis获取持仓数据"""
+        try:
+            redis_key = f"{exchange_id}:positions:{k_id}"
+            redis_data = self.redis_client.client.hget(redis_key, 'positions')
+            
+            if not redis_data:
+                return []
+            
+            positions = json.loads(redis_data)
+            
+            # 添加账号信息
+            for position in positions:
+                position['k_id'] = k_id
+            
+            return positions
+            
+        except Exception as e:
+            logger.error(f"获取Redis持仓数据失败: k_id={k_id}, error={e}")
+            return []
+    
+    def _sync_positions_to_db(self, k_id: int, st_id: int, positions_data: List[Dict]) -> bool:
+        """同步持仓数据到数据库"""
+        session = self.db_manager.get_session()
+        try:
+            # 使用批量优化方案
+            from sqlalchemy.dialects.mysql import insert
+            
+            # 准备数据
+            insert_data = []
+            keep_keys = set()  # 需要保留的(symbol, side)组合
+            
+            for pos_data in positions_data:
+                try:
+                    # 转换数据（这里需要实现转换逻辑）
+                    pos_dict = self._convert_position_data(pos_data)
+                    if not all([pos_dict.get('symbol'), pos_dict.get('side')]):
+                        continue
+                    
+                    # 重命名qty为sum
+                    if 'qty' in pos_dict:
+                        pos_dict['sum'] = pos_dict.pop('qty')
+                    
+                    insert_data.append(pos_dict)
+                    keep_keys.add((pos_dict['symbol'], pos_dict['side']))
+                    
+                except Exception as e:
+                    logger.error(f"转换持仓数据失败: {pos_data}, error={e}")
+                    continue
+            
+            with session.begin():
+                if not insert_data:
+                    # 清空该账号持仓
+                    session.execute(
+                        delete(StrategyPosition).where(
+                            and_(
+                                StrategyPosition.k_id == k_id,
+                                StrategyPosition.st_id == st_id
+                            )
+                        )
+                    )
+                    return True
+                
+                # 批量插入/更新
+                stmt = insert(StrategyPosition.__table__).values(insert_data)
+                
+                update_dict = {
+                    'price': stmt.inserted.price,
+                    'sum': stmt.inserted.sum,
+                    'asset_num': stmt.inserted.asset_num,
+                    'asset_profit': stmt.inserted.asset_profit,
+                    'leverage': stmt.inserted.leverage,
+                    'uptime': stmt.inserted.uptime,
+                    'profit_price': stmt.inserted.profit_price,
+                    'stop_price': stmt.inserted.stop_price,
+                    'liquidation_price': stmt.inserted.liquidation_price
+                }
+                
+                stmt = stmt.on_duplicate_key_update(**update_dict)
+                session.execute(stmt)
+                
+                # 删除多余持仓
+                if keep_keys:
+                    existing_positions = session.execute(
+                        select(StrategyPosition).where(
+                            and_(
+                                StrategyPosition.k_id == k_id,
+                                StrategyPosition.st_id == st_id
+                            )
+                        )
+                    ).scalars().all()
+                    
+                    to_delete_ids = []
+                    for existing in existing_positions:
+                        key = (existing.symbol, existing.side)
+                        if key not in keep_keys:
+                            to_delete_ids.append(existing.id)
+                    
+                    if to_delete_ids:
+                        session.execute(
+                            delete(StrategyPosition).where(
+                                StrategyPosition.id.in_(to_delete_ids)
+                            )
+                        )
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"同步持仓到数据库失败: k_id={k_id}, error={e}")
+            return False
+        finally:
+            session.close()
+    
+    def _convert_position_data(self, data: Dict) -> Dict:
+        """转换持仓数据格式"""
+        # 这里实现具体的转换逻辑
+        return {
+            'st_id': int(data.get('st_id', 0)),
+            'k_id': int(data.get('k_id', 0)),
+            'asset': 'USDT',
+            'symbol': data.get('symbol', ''),
+            'side': data.get('side', ''),
+            'price': float(data.get('price', 0)) if data.get('price') is not None else None,
+            'qty': float(data.get('qty', 0)) if data.get('qty') is not None else None,
+            'asset_num': float(data.get('asset_num', 0)) if data.get('asset_num') is not None else None,
+            'asset_profit': float(data.get('asset_profit', 0)) if data.get('asset_profit') is not None else None,
+            'leverage': int(data.get('leverage', 0)) if data.get('leverage') is not None else None,
+            'uptime': int(data.get('uptime', 0)) if data.get('uptime') is not None else None,
+            'profit_price': float(data.get('profit_price', 0)) if data.get('profit_price') is not None else None,
+            'stop_price': float(data.get('stop_price', 0)) if data.get('stop_price') is not None else None,
+            'liquidation_price': float(data.get('liquidation_price', 0)) if data.get('liquidation_price') is not None else None
+        }

utils/database_manager.py

@@ -0,0 +1,75 @@
+from sqlalchemy import create_engine, select, update, insert, delete, and_, or_, text
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.exc import SQLAlchemyError
+from loguru import logger
+from typing import List, Dict, Optional, Any
+import json
+
+from config.database import DATABASE_CONFIG, SQLALCHEMY_CONFIG
+from models.orm_models import Base, StrategyPosition, StrategyOrder, StrategyKX
+
+class DatabaseManager:
+    """数据库管理器"""
+    
+    _instance = None
+    
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+    
+    def __init__(self):
+        if not self._initialized:
+            self._engine = None
+            self._session_factory = None
+            self._initialized = True
+            self._init_engine()
+    
+    def _init_engine(self):
+        """初始化数据库引擎"""
+        try:
+            # 构建数据库URL
+            db_url = (
+                f"mysql+pymysql://{DATABASE_CONFIG['user']}:{DATABASE_CONFIG['password']}"
+                f"@{DATABASE_CONFIG['host']}:{DATABASE_CONFIG['port']}"
+                f"/{DATABASE_CONFIG['database']}?charset={DATABASE_CONFIG['charset']}"
+            )
+            
+            # 创建引擎
+            self._engine = create_engine(
+                db_url,
+                echo=SQLALCHEMY_CONFIG['echo'],
+                echo_pool=SQLALCHEMY_CONFIG['echo_pool'],
+                pool_size=DATABASE_CONFIG['pool_size'],
+                max_overflow=DATABASE_CONFIG['max_overflow'],
+                pool_recycle=DATABASE_CONFIG['pool_recycle'],
+                pool_pre_ping=SQLALCHEMY_CONFIG['pool_pre_ping']
+            )
+            
+            # 创建会话工厂
+            self._session_factory = sessionmaker(
+                bind=self._engine,
+                expire_on_commit=False
+            )
+            
+            # 创建表（如果不存在）
+            Base.metadata.create_all(self._engine)
+            
+            logger.info("SQLAlchemy数据库引擎初始化成功")
+            
+        except Exception as e:
+            logger.error(f"数据库引擎初始化失败: {e}")
+            raise
+    
+    def get_session(self) -> Session:
+        """获取数据库会话"""
+        if self._session_factory is None:
+            self._init_engine()
+        return self._session_factory()
+    
+    def close(self):
+        """关闭数据库连接"""
+        if self._engine:
+            self._engine.dispose()
+            logger.info("数据库连接已关闭")

utils/redis_client.py

@@ -0,0 +1,54 @@
+import redis
+from loguru import logger
+from config.settings import REDIS_CONFIG
+
+class RedisClient:
+    """Redis客户端管理器"""
+    
+    _instance = None
+    
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+    
+    def __init__(self):
+        if not self._initialized:
+            self._pool = None
+            self._client = None
+            self._initialized = True
+    
+    @property
+    def client(self):
+        """获取Redis客户端（懒加载）"""
+        if self._client is None:
+            self._init_connection_pool()
+        return self._client
+    
+    def _init_connection_pool(self):
+        """初始化连接池"""
+        try:
+            self._pool = redis.ConnectionPool(
+                host=REDIS_CONFIG['host'],
+                port=REDIS_CONFIG['port'],
+                db=REDIS_CONFIG['db'],
+                decode_responses=REDIS_CONFIG['decode_responses'],
+                max_connections=REDIS_CONFIG['max_connections'],
+                socket_keepalive=True
+            )
+            self._client = redis.Redis(connection_pool=self._pool)
+            
+            # 测试连接
+            self._client.ping()
+            logger.info("Redis连接池初始化成功")
+            
+        except Exception as e:
+            logger.error(f"Redis连接失败: {e}")
+            raise
+    
+    def close(self):
+        """关闭连接池"""
+        if self._pool:
+            self._pool.disconnect()
+            logger.info("Redis连接池已关闭")