""" 统一仓位 sizing 规则。 目标: - 单笔仓位按账户权益百分比控制保证金 - 总杠杆限制按“名义仓位空间”换算成“可加保证金” - 给模拟盘和执行决策层复用,避免多套逻辑漂移 """ from typing import Dict, Optional, Tuple from app.utils.logger import logger DEFAULT_POSITION_SIZE_MARGIN_PCTS: Dict[str, float] = { "micro": 0.01, "light": 0.08, "medium": 0.12, "heavy": 0.18, } DEFAULT_SIGNAL_POSITION_SIZE_BY_TIMEFRAME: Dict[str, str] = { "short_term": "light", "medium_term": "light", "long_term": "medium", } DEFAULT_TIMEFRAME_MARGIN_MULTIPLIERS: Dict[str, float] = { "short_term": 0.90, "medium_term": 1.00, "long_term": 1.10, } DEFAULT_GRADE_MARGIN_MULTIPLIERS: Dict[str, float] = { "A": 1.10, "B": 1.00, "C": 0.80, "D": 0.00, } def normalize_signal_type(signal_type: Optional[str]) -> str: normalized = (signal_type or "medium_term").strip().lower() if normalized not in {"short_term", "medium_term", "long_term"}: return "medium_term" return normalized def infer_signal_grade(confidence: Optional[float], explicit_grade: Optional[str] = None) -> str: if explicit_grade: grade = str(explicit_grade).strip().upper() if grade in {"A", "B", "C", "D"}: return grade confidence_value = float(confidence or 0) if confidence_value >= 80: return "A" if confidence_value >= 60: return "B" if confidence_value >= 40: return "C" return "D" def normalize_position_size( position_size: Optional[str], signal_type: Optional[str], defaults: Optional[Dict[str, str]] = None, ) -> str: normalized_type = normalize_signal_type(signal_type) normalized_size = (position_size or "").strip().lower() if normalized_size in DEFAULT_POSITION_SIZE_MARGIN_PCTS: return normalized_size fallback_defaults = defaults or DEFAULT_SIGNAL_POSITION_SIZE_BY_TIMEFRAME return fallback_defaults.get(normalized_type, "light") def resolve_target_margin_pct( position_size: Optional[str], signal_type: Optional[str], confidence: Optional[float] = None, grade: Optional[str] = None, size_margin_pcts: Optional[Dict[str, float]] = None, timeframe_multipliers: Optional[Dict[str, float]] = None, grade_multipliers: Optional[Dict[str, float]] = None, default_positions: Optional[Dict[str, str]] = None, ) -> Tuple[float, str, str, str]: normalized_type = normalize_signal_type(signal_type) normalized_size = normalize_position_size(position_size, normalized_type, default_positions) normalized_grade = infer_signal_grade(confidence, grade) base_margin_pcts = size_margin_pcts or DEFAULT_POSITION_SIZE_MARGIN_PCTS type_multipliers = timeframe_multipliers or DEFAULT_TIMEFRAME_MARGIN_MULTIPLIERS effective_grade_multipliers = grade_multipliers or DEFAULT_GRADE_MARGIN_MULTIPLIERS base_pct = base_margin_pcts.get(normalized_size, base_margin_pcts["light"]) timeframe_multiplier = type_multipliers.get(normalized_type, 1.0) grade_multiplier = effective_grade_multipliers.get(normalized_grade, 1.0) target_pct = base_pct * timeframe_multiplier * grade_multiplier reason = ( f"{normalized_type} {normalized_grade}级 {normalized_size}仓位" f" -> {target_pct * 100:.1f}%权益保证金" ) return target_pct, reason, normalized_size, normalized_grade def calculate_margin_and_position_value( *, balance: float, available_margin: float, current_total_leverage: float, max_total_leverage: float, order_leverage: float, target_margin_pct: float, max_margin_pct: float, min_margin: float = 0.0, min_position_value: float = 0.0, min_effective_leverage: float = 1.0, reserve_ratio: float = 0.05, ) -> Tuple[float, float, str]: if balance <= 0: return 0.0, 0.0, "账户余额无效" if available_margin <= 0: return 0.0, 0.0, "可用保证金不足" if order_leverage <= 0: return 0.0, 0.0, "订单杠杆无效" if min_effective_leverage <= 0: min_effective_leverage = 1.0 if order_leverage < min_effective_leverage: return 0.0, 0.0, ( f"订单杠杆 {order_leverage:.1f}x 低于最小有效杠杆 " f"{min_effective_leverage:.1f}x" ) if target_margin_pct <= 0: return 0.0, 0.0, "目标保证金比例无效" reserve_ratio = min(max(reserve_ratio, 0.0), 0.5) buffer_available_margin = max(0.0, available_margin * (1 - reserve_ratio)) if buffer_available_margin <= 0: return 0.0, 0.0, "可用保证金不足" max_margin_by_platform = balance * max_margin_pct if max_margin_pct > 0 else buffer_available_margin leverage_headroom = max(0.0, max_total_leverage - current_total_leverage) if leverage_headroom <= 0: return 0.0, 0.0, f"已达最大总杠杆 {current_total_leverage:.2f}x/{max_total_leverage:.2f}x" # 总杠杆限制是“还能开的名义仓位”,这里换算成“还能加的保证金”。 max_margin_by_total_leverage = (balance * leverage_headroom) / order_leverage hard_cap = min(buffer_available_margin, max_margin_by_platform, max_margin_by_total_leverage) if hard_cap <= 0: return 0.0, 0.0, "可开保证金额度不足" target_margin = balance * target_margin_pct margin = min(target_margin, hard_cap) adjustments = [] if min_margin > 0 and margin < min_margin: if min_margin <= hard_cap: margin = min_margin adjustments.append(f"按最小保证金抬升至 ${min_margin:.2f}") else: return 0.0, 0.0, ( f"最小保证金 ${min_margin:.2f} 超过当前可用额度 " f"${hard_cap:.2f}" ) if min_position_value > 0: required_margin_for_min_position = min_position_value / order_leverage if margin < required_margin_for_min_position: if required_margin_for_min_position <= hard_cap: margin = required_margin_for_min_position adjustments.append( f"按最低名义仓位 ${min_position_value:.2f} 抬升保证金至 ${margin:.2f}" ) else: adjustments.append( f"最低名义仓位 ${min_position_value:.2f} 受当前上限约束,实际最多 ${hard_cap * order_leverage:.2f}" ) position_value = margin * order_leverage if position_value < 50: return 0.0, 0.0, "可开仓位不足 $50" margin = round(margin, 2) position_value = round(position_value, 2) detail = ( f"目标保证金 ${target_margin:.2f}, 实际保证金 ${margin:.2f}, " f"名义仓位 ${position_value:.2f}, 单笔杠杆 {order_leverage:.1f}x" ) if adjustments: detail = f"{detail}, 调整: {'; '.join(adjustments)}" logger.info(f"仓位预算计算: {detail}") return margin, position_value, detail