""" 风险管理模块 - 严格控制仓位和风险 """ import logging import os from datetime import datetime, timezone, timedelta from typing import Dict, List, Optional try: from .binance_client import BinanceClient from . import config from .atr_strategy import ATRStrategy except ImportError: from binance_client import BinanceClient import config from atr_strategy import ATRStrategy logger = logging.getLogger(__name__) class RiskManager: """风险管理类""" def __init__(self, client: BinanceClient): """ 初始化风险管理器 Args: client: 币安客户端 """ self.client = client # 不保存引用，每次都从 config.TRADING_CONFIG 读取最新配置 # self.config = config.TRADING_CONFIG # 移除，避免使用旧配置 # 初始化ATR策略 self.atr_strategy = ATRStrategy() async def check_position_size(self, symbol: str, quantity: float, leverage: Optional[int] = None) -> bool: """ 检查单笔仓位大小是否符合要求 Args: symbol: 交易对 quantity: 下单数量 leverage: 杠杆倍数（用于换算保证金）；若不传则使用配置的基础杠杆 Returns: 是否通过检查 """ try: logger.info(f"检查 {symbol} 单笔仓位大小...") # 获取账户余额 balance = await self.client.get_account_balance() available_balance = balance.get('available', 0) if available_balance <= 0: logger.warning(f"❌ {symbol} 账户可用余额不足: {available_balance:.2f} USDT") return False # 计算名义价值与保证金（名义价值/杠杆） ticker = await self.client.get_ticker_24h(symbol) if not ticker: logger.warning(f"❌ {symbol} 无法获取价格数据") return False current_price = ticker['price'] notional_value = quantity * current_price actual_leverage = leverage if leverage is not None else config.TRADING_CONFIG.get('LEVERAGE', 10) if not actual_leverage or actual_leverage <= 0: actual_leverage = 10 margin_value = notional_value / actual_leverage # 重要语义：POSITION_PERCENT 均按“保证金占用比例”计算（更符合 stop_loss/take_profit 的 margin 逻辑） max_margin_value = available_balance * config.TRADING_CONFIG['MAX_POSITION_PERCENT'] min_margin_value = available_balance * config.TRADING_CONFIG['MIN_POSITION_PERCENT'] max_margin_pct = config.TRADING_CONFIG['MAX_POSITION_PERCENT'] * 100 min_margin_pct = config.TRADING_CONFIG['MIN_POSITION_PERCENT'] * 100 logger.info(f" 数量: {quantity:.4f}") logger.info(f" 价格: {current_price:.4f} USDT") logger.info(f" 名义价值: {notional_value:.2f} USDT") logger.info(f" 杠杆: {actual_leverage}x") logger.info(f" 保证金: {margin_value:.4f} USDT") logger.info(f" 单笔最大保证金: {max_margin_value:.2f} USDT ({max_margin_pct:.1f}%)") logger.info(f" 单笔最小保证金: {min_margin_value:.2f} USDT ({min_margin_pct:.1f}%)") # 使用小的容差来处理浮点数精度问题（0.01 USDT） tolerance = 0.01 if margin_value > max_margin_value + tolerance: logger.warning( f"❌ {symbol} 单笔保证金过大: {margin_value:.4f} USDT > " f"最大限制: {max_margin_value:.2f} USDT " f"(超出: {margin_value - max_margin_value:.4f} USDT)" ) return False elif margin_value > max_margin_value: # 在容差范围内，允许通过（浮点数精度问题） logger.info( f"⚠ {symbol} 保证金略超限制但 within 容差: " f"{margin_value:.4f} USDT vs {max_margin_value:.2f} USDT " f"(差异: {margin_value - max_margin_value:.4f} USDT)" ) if margin_value < min_margin_value: logger.warning( f"❌ {symbol} 单笔保证金过小: {margin_value:.4f} USDT < " f"最小限制: {min_margin_value:.2f} USDT" ) return False logger.info(f"✓ {symbol} 单笔仓位大小检查通过") # 检查总仓位是否超过限制 logger.info(f"检查 {symbol} 总仓位限制...") if not await self.check_total_position(margin_value): return False logger.info( f"✓ {symbol} 所有仓位检查通过: 保证金 {margin_value:.4f} USDT " f"(账户可用余额: {available_balance:.2f} USDT)" ) return True except Exception as e: logger.error(f"检查仓位大小失败 {symbol}: {e}", exc_info=True) return False async def check_total_position(self, new_position_margin: float) -> bool: """ 检查总仓位是否超过限制 Args: new_position_margin: 新仓位保证金占用（USDT） Returns: 是否通过检查 """ try: # 获取当前持仓 positions = await self.client.get_open_positions() # 计算当前总保证金占用 current_position_values = [] total_margin_value = 0 for pos in positions: notional_value = abs(pos['positionAmt'] * pos['entryPrice']) lv = pos.get('leverage', None) try: lv = int(lv) if lv is not None else None except Exception: lv = None if not lv or lv <= 0: lv = config.TRADING_CONFIG.get('LEVERAGE', 10) or 10 margin_value = notional_value / lv current_position_values.append({ 'symbol': pos['symbol'], 'notional': notional_value, 'margin': margin_value, 'leverage': lv, 'amount': pos['positionAmt'], 'entryPrice': pos['entryPrice'] }) total_margin_value += margin_value # 加上新仓位 total_with_new = total_margin_value + new_position_margin # 获取账户余额 balance = await self.client.get_account_balance() total_balance = balance.get('total', 0) available_balance = balance.get('available', 0) if total_balance <= 0: logger.warning("账户总余额为0，无法开仓") return False max_total_margin = total_balance * config.TRADING_CONFIG['MAX_TOTAL_POSITION_PERCENT'] max_total_margin_pct = config.TRADING_CONFIG['MAX_TOTAL_POSITION_PERCENT'] * 100 # 详细日志 logger.info("=" * 60) logger.info("总仓位检查详情:") logger.info(f" 账户总余额: {total_balance:.2f} USDT") logger.info(f" 账户可用余额: {available_balance:.2f} USDT") logger.info(f" 总保证金上限: {max_total_margin:.2f} USDT ({max_total_margin_pct:.1f}%)") logger.info(f" 当前持仓数量: {len(positions)} 个") if current_position_values: logger.info(" 当前持仓明细:") for pos_info in current_position_values: logger.info( f" - {pos_info['symbol']}: " f"保证金 {pos_info['margin']:.4f} USDT " f"(名义 {pos_info['notional']:.2f} USDT, {pos_info['leverage']}x, " f"数量: {pos_info['amount']:.4f}, 入场价: {pos_info['entryPrice']:.4f})" ) logger.info(f" 当前总保证金: {total_margin_value:.4f} USDT") logger.info(f" 新仓位保证金: {new_position_margin:.4f} USDT") logger.info(f" 开仓后总保证金: {total_with_new:.4f} USDT") logger.info(f" 剩余可用保证金: {max_total_margin - total_margin_value:.4f} USDT") if total_with_new > max_total_margin: logger.warning("=" * 60) logger.warning( f"❌ 总保证金超限: {total_with_new:.4f} USDT > " f"最大限制: {max_total_margin:.2f} USDT" ) logger.warning( f" 超出: {total_with_new - max_total_margin:.4f} USDT " f"({((total_with_new - max_total_margin) / max_total_margin * 100):.1f}%)" ) logger.warning(" 建议: 平掉部分持仓或等待现有持仓平仓后再开新仓") logger.warning("=" * 60) return False logger.info( f"✓ 总保证金检查通过: {total_with_new:.4f} USDT / " f"最大限制: {max_total_margin:.2f} USDT " f"({(total_with_new / max_total_margin * 100):.1f}%)" ) logger.info("=" * 60) return True except Exception as e: logger.error(f"检查总仓位失败: {e}", exc_info=True) return False async def calculate_position_size( self, symbol: str, change_percent: float, leverage: Optional[int] = None, entry_price: Optional[float] = None, stop_loss_price: Optional[float] = None, side: Optional[str] = None, atr: Optional[float] = None, signal_strength: Optional[int] = None ) -> Optional[float]: """ 根据涨跌幅和风险参数计算合适的仓位大小 ⚠️ 优化：支持固定风险百分比计算（凯利公式）和信号强度分级 Args: symbol: 交易对 change_percent: 涨跌幅百分比 leverage: 杠杆倍数（可选） entry_price: 入场价格（可选，如果提供则用于固定风险计算） stop_loss_price: 止损价格（可选，如果提供则用于固定风险计算） side: 交易方向 'BUY' 或 'SELL'（可选，用于固定风险计算） atr: ATR值（可选，用于估算止损） signal_strength: 信号强度（可选，用于仓位分级） Returns: 建议的仓位数量，如果不符合条件则返回None """ try: logger.info(f"开始计算 {symbol} 的仓位大小...") # 获取账户余额 balance = await self.client.get_account_balance() available_balance = balance.get('available', 0) total_balance = balance.get('total', 0) logger.info(f" 账户可用余额: {available_balance:.2f} USDT") logger.info(f" 账户总余额: {total_balance:.2f} USDT") if available_balance <= 0: logger.warning(f"❌ {symbol} 账户可用余额不足: {available_balance:.2f} USDT") return None # 获取当前价格 ticker = await self.client.get_ticker_24h(symbol) if not ticker: logger.warning(f"❌ {symbol} 无法获取价格数据") return None current_price = ticker['price'] logger.info(f" 当前价格: {current_price:.4f} USDT") # 重要语义：MAX_POSITION_PERCENT 表示“单笔保证金占用比例” # 先确定实际杠杆（用于从保证金换算名义价值） actual_leverage = leverage if leverage is not None else config.TRADING_CONFIG.get('LEVERAGE', 10) if not actual_leverage or actual_leverage <= 0: actual_leverage = 10 # ⚠️ 优化3：固定风险百分比仓位计算（凯利公式） # 公式：仓位大小 = (总资金 * 每笔单子承受的风险%) / (入场价 - 止损价) use_fixed_risk = config.TRADING_CONFIG.get('USE_FIXED_RISK_SIZING', True) fixed_risk_percent = config.TRADING_CONFIG.get('FIXED_RISK_PERCENT', 0.02) # 默认2% quantity = None if use_fixed_risk and entry_price and side: # 如果未提供止损价格，先估算 if stop_loss_price is None: # 尝试使用ATR估算止损距离 if atr and atr > 0: atr_multiplier = config.TRADING_CONFIG.get('ATR_STOP_LOSS_MULTIPLIER', 1.8) if side == 'BUY': estimated_stop_loss = entry_price - (atr * atr_multiplier) else: # SELL estimated_stop_loss = entry_price + (atr * atr_multiplier) stop_loss_price = estimated_stop_loss else: # 使用固定百分比估算（基于保证金） stop_loss_pct = config.TRADING_CONFIG.get('STOP_LOSS_PERCENT', 0.10) # 先估算一个临时仓位来计算止损距离 temp_margin = total_balance * 0.05 # 临时使用5%保证金 temp_notional = temp_margin * actual_leverage temp_quantity = temp_notional / entry_price stop_loss_amount = temp_margin * stop_loss_pct if side == 'BUY': estimated_stop_loss = entry_price - (stop_loss_amount / temp_quantity) else: # SELL estimated_stop_loss = entry_price + (stop_loss_amount / temp_quantity) stop_loss_price = estimated_stop_loss # 计算止损距离 if side == 'BUY': stop_distance = entry_price - stop_loss_price else: # SELL stop_distance = stop_loss_price - entry_price if stop_distance > 0: # 固定风险金额 risk_amount = total_balance * fixed_risk_percent # 根据止损距离反算仓位 # 风险金额 = (入场价 - 止损价) × 数量 # 所以：数量 = 风险金额 / (入场价 - 止损价) quantity = risk_amount / stop_distance # 计算对应的保证金和名义价值 notional_value = quantity * entry_price margin_value = notional_value / actual_leverage logger.info(f" ⚠️ 使用固定风险百分比计算仓位:") logger.info(f" 固定风险: {fixed_risk_percent*100:.2f}% = {risk_amount:.4f} USDT") logger.info(f" 止损距离: {stop_distance:.4f} USDT ({stop_distance/entry_price*100:.2f}%)") logger.info(f" 计算数量: {quantity:.4f}") logger.info(f" 名义价值: {notional_value:.2f} USDT") logger.info(f" 保证金: {margin_value:.4f} USDT ({margin_value/total_balance*100:.2f}%)") # 检查是否超过最大仓位限制 max_position_percent = config.TRADING_CONFIG['MAX_POSITION_PERCENT'] max_margin_value = available_balance * max_position_percent if margin_value > max_margin_value: # 如果超过最大仓位，使用最大仓位 logger.warning(f" ⚠️ 固定风险计算的保证金 {margin_value:.4f} USDT > 最大限制 {max_margin_value:.2f} USDT") logger.info(f" ✓ 调整为最大仓位限制: {max_margin_value:.2f} USDT") margin_value = max_margin_value notional_value = margin_value * actual_leverage quantity = notional_value / entry_price else: # 使用固定风险计算的仓位 pass # quantity已经计算好了 # 如果未使用固定风险计算，使用原来的方法 if quantity is None: # ⚠️ 优化3：信号强度分级 - 9-10分高权重，8分轻仓 signal_multiplier = 1.0 if signal_strength is not None: signal_multipliers = config.TRADING_CONFIG.get('SIGNAL_STRENGTH_POSITION_MULTIPLIER', {8: 0.5, 9: 1.0, 10: 1.0}) signal_multiplier = signal_multipliers.get(signal_strength, 1.0) if signal_strength == 8: logger.info(f" ⚠️ 信号强度8分，使用50%仓位（轻仓试探）") elif signal_strength >= 9: logger.info(f" ✓ 信号强度{signal_strength}分，使用100%仓位（高质量信号）") # 根据涨跌幅调整仓位大小（涨跌幅越大，保证金占比可以适当增加） base_position_percent = config.TRADING_CONFIG['MAX_POSITION_PERCENT'] * signal_multiplier max_position_percent = config.TRADING_CONFIG['MAX_POSITION_PERCENT'] * signal_multiplier min_position_percent = config.TRADING_CONFIG['MIN_POSITION_PERCENT'] # 涨跌幅超过5%时，可以适当增加保证金占比，但必须遵守 MAX_POSITION_PERCENT 上限 if abs(change_percent) > 5: position_percent = min( base_position_percent * 1.5, max_position_percent ) logger.info(f" 涨跌幅 {change_percent:.2f}% > 5%，使用增强仓位比例: {position_percent*100:.1f}%") else: position_percent = base_position_percent logger.info(f" 涨跌幅 {change_percent:.2f}%，使用标准仓位比例: {position_percent*100:.1f}%") # 计算保证金与名义价值 margin_value = available_balance * position_percent notional_value = margin_value * actual_leverage logger.info(f" 计算保证金: {margin_value:.4f} USDT ({position_percent*100:.1f}% of {available_balance:.2f})") logger.info(f" 计算名义价值: {notional_value:.2f} USDT (保证金 {margin_value:.4f} × 杠杆 {actual_leverage}x)") # 计算数量 quantity = notional_value / current_price logger.info(f" 计算数量: {quantity:.4f}") # 计算名义价值和保证金（如果还未计算） if 'notional_value' not in locals() or 'margin_value' not in locals(): notional_value = quantity * current_price margin_value = notional_value / actual_leverage # 确保仓位价值满足最小名义价值要求（币安要求至少5 USDT） min_notional = 5.0 # 币安合约最小名义价值 calculated_notional = quantity * current_price if calculated_notional < min_notional: # 如果计算出的名义价值仍然不足，增加数量 required_quantity = min_notional / current_price logger.warning(f" ⚠ 计算出的名义价值 {calculated_notional:.2f} USDT < {min_notional:.2f} USDT") logger.info(f" ✓ 调整数量从 {quantity:.4f} 到 {required_quantity:.4f}") quantity = required_quantity notional_value = required_quantity * current_price margin_value = notional_value / actual_leverage # 计算名义价值和保证金（如果还未计算） if 'notional_value' not in locals() or 'margin_value' not in locals(): notional_value = quantity * current_price margin_value = notional_value / actual_leverage # 确保仓位价值满足最小名义价值要求（币安要求至少5 USDT） min_notional = 5.0 # 币安合约最小名义价值 calculated_notional = quantity * current_price if calculated_notional < min_notional: # 如果计算出的名义价值仍然不足，增加数量 required_quantity = min_notional / current_price logger.warning(f" ⚠ 计算出的名义价值 {calculated_notional:.2f} USDT < {min_notional:.2f} USDT") logger.info(f" ✓ 调整数量从 {quantity:.4f} 到 {required_quantity:.4f}") quantity = required_quantity notional_value = required_quantity * current_price margin_value = notional_value / actual_leverage # 检查最小保证金要求（margin 语义：MIN_MARGIN_USDT 本身就是保证金下限） min_margin_usdt = config.TRADING_CONFIG.get('MIN_MARGIN_USDT', 0.5) # 默认0.5 USDT logger.info(f" 当前保证金: {margin_value:.4f} USDT (杠杆: {actual_leverage}x)") if margin_value < min_margin_usdt: # 保证金不足，需要增加保证金 logger.warning( f" ⚠ 保证金 {margin_value:.4f} USDT < 最小保证金要求 {min_margin_usdt:.2f} USDT" ) # 检查是否可以使用更大的仓位价值（但不超过最大仓位限制） max_position_percent = config.TRADING_CONFIG['MAX_POSITION_PERCENT'] max_margin_value = available_balance * max_position_percent if min_margin_usdt <= max_margin_value: margin_value = min_margin_usdt notional_value = margin_value * actual_leverage quantity = notional_value / current_price logger.info( f" ✓ 调整保证金到 {margin_value:.2f} USDT " f"(名义 {notional_value:.2f} USDT) 以满足最小保证金要求" ) else: # 即使使用最大仓位也无法满足最小保证金要求 max_margin = max_margin_value logger.warning( f" ❌ 无法满足最小保证金要求: " f"需要 {min_margin_usdt:.2f} USDT 保证金，" f"但最大允许 {max_margin:.2f} USDT 保证金 (MAX_POSITION_PERCENT={max_position_percent*100:.2f}%)" ) logger.warning( f" 💡 建议: 增加账户余额到至少 " f"{min_margin_usdt / max_position_percent:.2f} USDT " f"才能满足最小保证金要求" ) return None # 检查是否通过风险控制 logger.info(f" 检查仓位大小是否符合风险控制要求...") # 计算最终的名义价值与保证金 final_notional_value = quantity * current_price final_margin = final_notional_value / actual_leverage if actual_leverage > 0 else final_notional_value # 添加最小名义价值检查（0.2 USDT），避免下无意义的小单子 MIN_NOTIONAL_VALUE = 0.2 # 最小名义价值0.2 USDT if final_notional_value < MIN_NOTIONAL_VALUE: logger.warning( f" ❌ {symbol} 名义价值 {final_notional_value:.4f} USDT < 最小要求 {MIN_NOTIONAL_VALUE:.2f} USDT" ) logger.warning(f" 💡 此类小单子意义不大，拒绝开仓") return None if await self.check_position_size(symbol, quantity, leverage=actual_leverage): logger.info( f"✓ {symbol} 仓位计算成功: {quantity:.4f} " f"(保证金: {final_margin:.4f} USDT, " f"名义价值: {final_notional_value:.2f} USDT, " f"保证金: {final_margin:.4f} USDT, 杠杆: {actual_leverage}x)" ) return quantity else: logger.warning(f"❌ {symbol} 仓位检查未通过，无法开仓") return None except Exception as e: logger.error(f"计算仓位大小失败 {symbol}: {e}", exc_info=True) return None async def should_trade(self, symbol: str, change_percent: float) -> bool: """ 判断是否应该交易 Args: symbol: 交易对 change_percent: 涨跌幅百分比 Returns: 是否应该交易 """ # 用户风险旋钮：自动交易总开关 if not bool(config.TRADING_CONFIG.get("AUTO_TRADE_ENABLED", True)): logger.info(f"{symbol} 自动交易已关闭（AUTO_TRADE_ENABLED=false），跳过") return False # 检查最小涨跌幅阈值 if abs(change_percent) < config.TRADING_CONFIG['MIN_CHANGE_PERCENT']: logger.debug(f"{symbol} 涨跌幅 {change_percent:.2f}% 小于阈值") return False # 检查是否已有持仓 / 总持仓数量限制 positions = await self.client.get_open_positions() try: max_open = int(config.TRADING_CONFIG.get("MAX_OPEN_POSITIONS", 0) or 0) except Exception: max_open = 0 if max_open > 0 and len(positions) >= max_open: logger.info(f"{symbol} 持仓数量已达上限：{len(positions)}/{max_open}，跳过开仓") return False existing_position = next( (p for p in positions if p['symbol'] == symbol), None ) if existing_position: logger.info(f"{symbol} 已有持仓，跳过") return False # 每日开仓次数限制（Redis 计数；无 Redis 时降级为内存计数） try: max_daily = int(config.TRADING_CONFIG.get("MAX_DAILY_ENTRIES", 0) or 0) except Exception: max_daily = 0 if max_daily > 0: c = await self._get_daily_entries_count() if c >= max_daily: logger.info(f"{symbol} 今日开仓次数已达上限：{c}/{max_daily}，跳过") return False return True def _daily_entries_key(self) -> str: try: aid = int(os.getenv("ATS_ACCOUNT_ID") or os.getenv("ACCOUNT_ID") or 1) except Exception: aid = 1 bj = timezone(timedelta(hours=8)) d = datetime.now(bj).strftime("%Y%m%d") return f"ats:acc:{aid}:daily_entries:{d}" def _seconds_until_beijing_day_end(self) -> int: bj = timezone(timedelta(hours=8)) now = datetime.now(bj) end = (now.replace(hour=23, minute=59, second=59, microsecond=0)) return max(60, int((end - now).total_seconds()) + 1) async def _get_daily_entries_count(self) -> int: key = self._daily_entries_key() try: # redis_cache 已有内存降级逻辑 return int(await self.client.redis_cache.get_int(key, 0)) except Exception: return 0 async def record_entry(self, symbol: str = "") -> None: """在“开仓真正成功”后调用，用于累计每日开仓次数。""" try: max_daily = int(config.TRADING_CONFIG.get("MAX_DAILY_ENTRIES", 0) or 0) except Exception: max_daily = 0 if max_daily <= 0: return key = self._daily_entries_key() ttl = self._seconds_until_beijing_day_end() try: n = await self.client.redis_cache.incr(key, 1, ttl=ttl) logger.info(f"{symbol} 今日开仓计数 +1：{n}/{max_daily}") except Exception: return def get_stop_loss_price( self, entry_price: float, side: str, quantity: float, leverage: int, stop_loss_pct: Optional[float] = None, klines: Optional[List] = None, bollinger: Optional[Dict] = None, atr: Optional[float] = None ) -> float: """ 计算止损价格（基于保证金的盈亏金额） Args: entry_price: 入场价格 side: 方向 'BUY' 或 'SELL' quantity: 持仓数量 leverage: 杠杆倍数 stop_loss_pct: 止损百分比（相对于保证金），如果为None则使用配置值 klines: K线数据，用于计算支撑/阻力位（作为辅助参考） bollinger: 布林带数据，用于计算动态止损（作为辅助参考） atr: 平均真实波幅，用于计算动态止损（作为辅助参考） Returns: 止损价格 """ # 计算保证金和仓位价值 position_value = entry_price * quantity margin = position_value / leverage if leverage > 0 else position_value # 优先使用ATR动态止损（如果启用且ATR可用） # 计算ATR百分比（如果提供了ATR绝对值） atr_percent = None if atr is not None and atr > 0 and entry_price > 0: atr_percent = atr / entry_price # 获取市场波动率（如果可用） volatility = None # 可以从symbol_info中获取，这里暂时为None # 使用ATR策略计算止损 stop_loss_price_atr, stop_distance_atr, atr_details = self.atr_strategy.calculate_stop_loss( entry_price, side, atr, atr_percent, volatility ) if stop_loss_price_atr is None: logger.debug(f"ATR不可用，使用固定百分比止损") # 获取止损百分比（相对于保证金） stop_loss_percent = stop_loss_pct or config.TRADING_CONFIG['STOP_LOSS_PERCENT'] # 计算止损金额（相对于保证金） stop_loss_amount = margin * stop_loss_percent # 计算基于保证金的止损价 # 止损金额 = (开仓价 - 止损价) × 数量 # 所以：止损价 = 开仓价 - (止损金额 / 数量) if side == 'BUY': # 做多，止损价低于入场价 stop_loss_price_margin = entry_price - (stop_loss_amount / quantity) else: # 做空，止损价高于入场价 stop_loss_price_margin = entry_price + (stop_loss_amount / quantity) # 同时计算基于价格百分比的止损价（作为最小值保护） # 获取最小价格变动百分比（如果配置了） min_price_change_pct = config.TRADING_CONFIG.get('MIN_STOP_LOSS_PRICE_PCT', None) if min_price_change_pct is not None: # 基于价格百分比的止损价 if side == 'BUY': stop_loss_price_price = entry_price * (1 - min_price_change_pct) else: stop_loss_price_price = entry_price * (1 + min_price_change_pct) else: stop_loss_price_price = None # 选择最终的止损价：优先ATR，其次保证金，最后价格百分比（取更宽松的） candidate_prices = [] if stop_loss_price_atr is not None: candidate_prices.append(('ATR', stop_loss_price_atr)) candidate_prices.append(('保证金', stop_loss_price_margin)) if stop_loss_price_price is not None: candidate_prices.append(('价格百分比', stop_loss_price_price)) # 选择“更宽松/更远”的止损： # - 做多(BUY)：止损越低越宽松 → 取最小值 # - 做空(SELL)：止损越高越宽松 → 取最大值 if side == 'BUY': stop_loss_price = min(p[1] for p in candidate_prices) selected_method = [p[0] for p in candidate_prices if p[1] == stop_loss_price][0] else: stop_loss_price = max(p[1] for p in candidate_prices) selected_method = [p[0] for p in candidate_prices if p[1] == stop_loss_price][0] # 如果提供了技术分析数据，计算技术止损（允许更紧的止损，但需要在保证金止损范围内） technical_stop = None if klines and len(klines) >= 10: # 计算支撑/阻力位 low_prices = [float(k[3]) for k in klines[-20:]] # 最近20根K线的最低价 high_prices = [float(k[2]) for k in klines[-20:]] # 最近20根K线的最高价 if side == 'BUY': # 做多，止损放在支撑位下方 # 找到近期波段低点 recent_low = min(low_prices) # 止损放在低点下方0.5% buffer = entry_price * 0.005 # 0.5%缓冲 technical_stop = recent_low - buffer # 如果布林带可用，也可以考虑布林带下轨 if bollinger and bollinger.get('lower'): bollinger_stop = bollinger['lower'] * 0.995 # 布林带下轨下方0.5% technical_stop = max(technical_stop, bollinger_stop) # 技术止损更紧，但需要确保在保证金止损范围内（不能超过保证金止损） if technical_stop < stop_loss_price and technical_stop >= stop_loss_price_margin: # 技术止损在合理范围内，可以考虑使用 candidate_prices.append(('技术分析', technical_stop)) logger.debug( f"技术止损 (BUY): {technical_stop:.4f} " f"(在保证金止损范围内)" ) else: # 做空，止损放在阻力位上方 # 找到近期波段高点 recent_high = max(high_prices) # 止损放在高点上方0.5% buffer = entry_price * 0.005 # 0.5%缓冲 technical_stop = recent_high + buffer # 如果布林带可用，也可以考虑布林带上轨 if bollinger and bollinger.get('upper'): bollinger_stop = bollinger['upper'] * 1.005 # 布林带上轨上方0.5% technical_stop = min(technical_stop, bollinger_stop) # 技术止损更紧，但需要确保在保证金止损范围内（不能超过保证金止损） if technical_stop > stop_loss_price and technical_stop <= stop_loss_price_margin: # 技术止损在合理范围内，可以考虑使用 candidate_prices.append(('技术分析', technical_stop)) logger.debug( f"技术止损 (SELL): {technical_stop:.4f} " f"(在保证金止损范围内)" ) # 重新选择最终的止损价（包括技术止损） # 仍保持“更宽松/更远”的选择规则 if side == 'BUY': final_stop_loss = min(p[1] for p in candidate_prices) selected_method = [p[0] for p in candidate_prices if p[1] == final_stop_loss][0] else: final_stop_loss = max(p[1] for p in candidate_prices) selected_method = [p[0] for p in candidate_prices if p[1] == final_stop_loss][0] logger.info( f"最终止损 ({side}): {final_stop_loss:.4f} (使用{selected_method}), " + (f"ATR={stop_loss_price_atr:.4f}, " if stop_loss_price_atr else "") + f"保证金={stop_loss_price_margin:.4f}, " + (f"价格={stop_loss_price_price:.4f}, " if stop_loss_price_price else "") + (f"技术={technical_stop:.4f}, " if technical_stop else "") + f"止损金额={stop_loss_amount:.2f} USDT ({stop_loss_percent*100:.1f}% of margin)" ) return final_stop_loss def get_take_profit_price( self, entry_price: float, side: str, quantity: float, leverage: int, take_profit_pct: Optional[float] = None, atr: Optional[float] = None, stop_distance: Optional[float] = None ) -> float: """ 计算止盈价格（基于保证金的盈亏金额，支持ATR动态止盈） Args: entry_price: 入场价格 side: 方向 'BUY' 或 'SELL' quantity: 持仓数量 leverage: 杠杆倍数 take_profit_pct: 止盈百分比（相对于保证金），如果为None则使用配置值 atr: 平均真实波幅，用于计算动态止盈（可选） Returns: 止盈价格 """ # 计算保证金和仓位价值 position_value = entry_price * quantity margin = position_value / leverage if leverage > 0 else position_value # 优先使用ATR动态止盈（如果启用且ATR可用） # 计算ATR百分比（如果提供了ATR绝对值） atr_percent = None if atr is not None and atr > 0 and entry_price > 0: atr_percent = atr / entry_price # 尝试从止损计算中获取止损距离（用于盈亏比计算） # 如果止损已经计算过，可以使用止损距离来计算止盈 stop_distance_for_rr = None # 注意：这里无法直接获取止损距离，需要调用方传递，或者使用ATR倍数计算 # 使用ATR策略计算止盈 # 优先使用盈亏比方法（基于止损距离），如果没有止损距离则使用ATR倍数 take_profit_price_atr, take_profit_distance_atr, atr_tp_details = self.atr_strategy.calculate_take_profit( entry_price, side, stop_distance, atr, atr_percent, use_risk_reward_ratio=(stop_distance is not None) ) if take_profit_price_atr is None: logger.debug(f"ATR不可用，使用固定百分比止盈") # 获取止盈百分比（相对于保证金） take_profit_percent = take_profit_pct or config.TRADING_CONFIG['TAKE_PROFIT_PERCENT'] # 计算止盈金额（相对于保证金） take_profit_amount = margin * take_profit_percent # 计算基于保证金的止盈价 # 止盈金额 = (止盈价 - 开仓价) × 数量 # 所以：止盈价 = 开仓价 + (止盈金额 / 数量) if side == 'BUY': # 做多，止盈价高于入场价 take_profit_price_margin = entry_price + (take_profit_amount / quantity) else: # 做空，止盈价低于入场价 take_profit_price_margin = entry_price - (take_profit_amount / quantity) # 同时计算基于价格百分比的止盈价（作为最小值保护） # 获取最小价格变动百分比（如果配置了） min_price_change_pct = config.TRADING_CONFIG.get('MIN_TAKE_PROFIT_PRICE_PCT', None) if min_price_change_pct is not None: # 基于价格百分比的止盈价 if side == 'BUY': take_profit_price_price = entry_price * (1 + min_price_change_pct) else: take_profit_price_price = entry_price * (1 - min_price_change_pct) else: take_profit_price_price = None # 选择最终的止盈价：优先ATR，其次保证金，最后价格百分比（取更宽松的） candidate_prices = [] if take_profit_price_atr is not None: candidate_prices.append(('ATR', take_profit_price_atr)) candidate_prices.append(('保证金', take_profit_price_margin)) if take_profit_price_price is not None: candidate_prices.append(('价格百分比', take_profit_price_price)) # 对做多取最大的值（更宽松），对做空取最小的值（更宽松） if side == 'BUY': take_profit_price = max(p[1] for p in candidate_prices) selected_method = [p[0] for p in candidate_prices if p[1] == take_profit_price][0] else: take_profit_price = min(p[1] for p in candidate_prices) selected_method = [p[0] for p in candidate_prices if p[1] == take_profit_price][0] logger.info( f"止盈计算 ({side}): " + (f"ATR={take_profit_price_atr:.4f}, " if take_profit_price_atr else "") + f"基于保证金={take_profit_price_margin:.4f}, " + (f"基于价格={take_profit_price_price:.4f}, " if take_profit_price_price else "") + f"最终止盈={take_profit_price:.4f} (使用{selected_method}, 取更宽松), " + f"止盈金额={take_profit_amount:.4f} USDT ({take_profit_percent*100:.1f}% of margin)" ) return take_profit_price async def calculate_dynamic_leverage(self, signal_strength: int, symbol: str = None, atr: Optional[float] = None, entry_price: Optional[float] = None) -> int: """ 根据信号强度计算动态杠杆倍数 信号强度越高，杠杆倍数越高，以最大化收益 同时检查交易对支持的最大杠杆限制 Args: signal_strength: 信号强度 (0-10) symbol: 交易对符号（可选，用于检查交易对的最大杠杆限制） Returns: 杠杆倍数 """ # 获取配置参数 use_dynamic_leverage = config.TRADING_CONFIG.get('USE_DYNAMIC_LEVERAGE', True) base_leverage = config.TRADING_CONFIG.get('LEVERAGE', 10) max_leverage = config.TRADING_CONFIG.get('MAX_LEVERAGE', 20) min_signal_strength = config.TRADING_CONFIG.get('MIN_SIGNAL_STRENGTH', 7) # ⚠️ 优化7：阶梯杠杆 - 小众币限制最高杠杆 max_leverage_small_cap = config.TRADING_CONFIG.get('MAX_LEVERAGE_SMALL_CAP', 5) atr_leverage_reduction_threshold = config.TRADING_CONFIG.get('ATR_LEVERAGE_REDUCTION_THRESHOLD', 0.05) # 5% # 检查是否为小众币（高波动率） is_small_cap = False if atr and entry_price and entry_price > 0: atr_percent = atr / entry_price if atr_percent >= atr_leverage_reduction_threshold: is_small_cap = True logger.info(f" ⚠️ {symbol} ATR波动率 {atr_percent*100:.2f}% >= {atr_leverage_reduction_threshold*100:.0f}%，识别为小众币，限制最大杠杆为{max_leverage_small_cap}x") max_leverage = min(max_leverage, max_leverage_small_cap) # 如果未启用动态杠杆，返回基础杠杆 if not use_dynamic_leverage: final_leverage = int(base_leverage) else: # 如果信号强度低于最小要求，使用基础杠杆 if signal_strength < min_signal_strength: final_leverage = int(base_leverage) else: # 计算动态杠杆：信号强度越高，杠杆越高 # 公式：杠杆 = 基础杠杆 + (信号强度 - 最小信号强度) * (最大杠杆 - 基础杠杆) / (10 - 最小信号强度) signal_range = 10 - min_signal_strength # 信号强度范围 leverage_range = max_leverage - base_leverage # 杠杆范围 if signal_range > 0: # 计算信号强度超出最小值的比例 strength_above_min = signal_strength - min_signal_strength leverage_increase = (strength_above_min / signal_range) * leverage_range dynamic_leverage = base_leverage + leverage_increase else: dynamic_leverage = base_leverage # 确保杠杆在合理范围内（不超过配置的最大杠杆） final_leverage = max(int(base_leverage), min(int(dynamic_leverage), int(max_leverage))) # 如果提供了交易对符号，检查交易对支持的最大杠杆限制 if symbol: try: symbol_info = await self.client.get_symbol_info(symbol) if symbol_info and 'maxLeverage' in symbol_info: symbol_max_leverage = symbol_info['maxLeverage'] if final_leverage > symbol_max_leverage: logger.warning( f"{symbol} 交易对最大杠杆限制为 {symbol_max_leverage}x, " f"计算杠杆 {final_leverage}x 超过限制，调整为 {symbol_max_leverage}x" ) final_leverage = symbol_max_leverage except Exception as e: logger.warning(f"获取 {symbol} 交易对杠杆限制失败: {e}，使用计算值 {final_leverage}x") logger.info( f"动态杠杆计算: 信号强度={signal_strength}/10, " f"基础杠杆={base_leverage}x, 计算杠杆={final_leverage}x" + (f", 交易对={symbol}" if symbol else "") ) return final_leverage