""" 仓位管理服务 - 供模拟盘和实盘共用 基于 LLM 建议动态计算仓位大小 """ from typing import Dict, Any, Optional, Tuple from abc import ABC, abstractmethod from app.utils.logger import logger class PositionCalculator(ABC): """仓位计算器抽象基类""" @abstractmethod def get_account_status(self) -> Dict[str, Any]: """获取账户状态""" pass @abstractmethod def get_max_leverage(self) -> int: """获取最大杠杆""" pass class PositionManager: """ 仓位管理器 根据 LLM 建议的仓位大小(heavy/medium/light)动态计算实际保证金和持仓价值 模拟盘和实盘共用同一套仓位管理逻辑 """ def __init__(self, calculator: PositionCalculator): """ 初始化仓位管理器 Args: calculator: 仓位计算器(模拟盘或实盘) """ self.calculator = calculator def calculate_position( self, position_size: str, symbol: str, custom_ratios: Optional[Dict[str, float]] = None ) -> Tuple[float, float]: """ 根据 LLM 建议计算仓位 Args: position_size: LLM 建议的仓位大小 ('heavy', 'medium', 'light') symbol: 交易对 custom_ratios: 自定义仓位比例,可选 Returns: (margin, position_value) 元组 """ account = self.calculator.get_account_status() max_leverage = self.calculator.get_max_leverage() balance = account['current_balance'] current_position_value = account.get('total_position_value', 0) # 计算可用保证金空间 max_position_value = balance * max_leverage available_position_value = max_position_value - current_position_value if available_position_value <= 0: logger.warning(f"已达最大杠杆限制,无法开仓") return 0, 0 # 使用自定义比例或默认比例 ratios = custom_ratios or { 'heavy': 0.30, 'medium': 0.15, 'light': 0.05 } size_ratio = ratios.get(position_size, 0.05) # 计算目标持仓价值 target_position_value = available_position_value * size_ratio # 设置最小和最大限制 min_position_value = 1000 # 最小持仓价值 max_single_position = balance * 5 # 单笔最大不超过 5x 杠杆 position_value = max(min_position_value, min(target_position_value, max_single_position)) position_value = min(position_value, available_position_value) position_value = round(position_value, 2) # 计算对应的保证金 margin = round(position_value / max_leverage, 2) logger.info(f"动态仓位计算: {position_size} | 可用空间: ${available_position_value:,.0f} | " f"目标仓位: ${position_value:,.0f} | 保证金: ${margin:,.0f}") return margin, position_value class PaperPositionCalculator(PositionCalculator): """模拟盘仓位计算器""" def __init__(self, account_status_getter, max_leverage: int = 20): """ 初始化模拟盘计算器 Args: account_status_getter: 获取账户状态的函数 max_leverage: 最大杠杆倍数 """ self.account_status_getter = account_status_getter self._max_leverage = max_leverage def get_account_status(self) -> Dict[str, Any]: return self.account_status_getter() def get_max_leverage(self) -> int: return self._max_leverage class RealPositionCalculator(PositionCalculator): """实盘仓位计算器""" def __init__(self, balance: float, used_margin: float, total_position_value: float, max_leverage: int = 10): """ 初始化实盘计算器 Args: balance: 账户余额 used_margin: 已用保证金 total_position_value: 总持仓价值 max_leverage: 最大杠杆倍数 """ self._account_status = { 'current_balance': balance, 'used_margin': used_margin, 'total_position_value': total_position_value } self._max_leverage = max_leverage def get_account_status(self) -> Dict[str, Any]: return self._account_status def get_max_leverage(self) -> int: return self._max_leverage def calculate_paper_position( account_status_getter, position_size: str, symbol: str, max_leverage: int = 20 ) -> Tuple[float, float]: """ 计算模拟盘仓位(快捷方法) Args: account_status_getter: 获取账户状态的函数 position_size: LLM 建议的仓位大小 symbol: 交易对 max_leverage: 最大杠杆倍数 Returns: (margin, position_value) 元组 """ calculator = PaperPositionCalculator(account_status_getter, max_leverage) manager = PositionManager(calculator) return manager.calculate_position(position_size, symbol) def calculate_real_position( balance: float, used_margin: float, total_position_value: float, position_size: str, symbol: str, max_leverage: int = 10, custom_ratios: Optional[Dict[str, float]] = None ) -> Tuple[float, float]: """ 计算实盘仓位(快捷方法) Args: balance: 账户余额 used_margin: 已用保证金 total_position_value: 总持仓价值 position_size: LLM 建议的仓位大小 symbol: 交易对 max_leverage: 最大杠杆倍数 custom_ratios: 自定义仓位比例(实盘可用更保守的配置) Returns: (margin, position_value) 元组 """ calculator = RealPositionCalculator(balance, used_margin, total_position_value, max_leverage) manager = PositionManager(calculator) return manager.calculate_position(position_size, symbol, custom_ratios)