""" 执行监管器 从 CryptoAgent 主循环中拆分执行后监管职责，负责： - 挂单超时清理 - 持仓管理（止盈 / 超时退出 / 移动止损） - Bitget 挂单成交后的 TP/SL 补设 - Bitget 持仓保护单缺失补救 第一版先作为确定性协调器运行，不引入新的 LLM 决策。 """ from __future__ import annotations from datetime import datetime from typing import Any, Dict, List from app.crypto_agent.execution_targets import ExecutionTarget from app.utils.logger import logger class ExecutionGuardian: """执行监管协调器。""" def __init__(self, agent: Any): self.agent = agent self._state: Dict[str, Any] = { "last_run_at": None, "last_status": "idle", "last_error": "", "last_actions": [], } def get_status(self) -> Dict[str, Any]: return { "last_run_at": self._state.get("last_run_at"), "last_status": self._state.get("last_status", "idle"), "last_error": self._state.get("last_error", ""), "targets": [self._serialize_target(target) for target in self._iter_targets()], "last_actions": list(self._state.get("last_actions", []))[:20], } def _serialize_target(self, target: ExecutionTarget) -> Dict[str, Any]: return { "target_key": target.target_key, "platform": target.platform, "account_id": target.account_id, "supports_pending_timeout": target.supports_pending_timeout, "supports_position_management": target.supports_position_management, "supports_tpsl_repair": target.supports_tpsl_repair, } def _iter_targets(self) -> List[ExecutionTarget]: targets = self.agent.get_execution_targets() if not isinstance(targets, list): return [] return targets async def run_cycle(self): """执行一轮监管扫描。""" self._state["last_run_at"] = datetime.now().isoformat() self._state["last_status"] = "running" self._state["last_error"] = "" self._state["last_actions"] = [] try: for target in self._iter_targets(): if self.agent._is_platform_halted(target.target_key): continue if target.supports_pending_timeout: await self._check_pending_order_timeouts(target) if target.supports_position_management: await self._check_position_management(target) if target.supports_tpsl_repair: await self._check_and_set_pending_tp_sl(target) await self._check_missing_tp_sl(target) self._state["last_status"] = "completed" except Exception as e: self._state["last_status"] = "error" self._state["last_error"] = str(e) logger.error(f"ExecutionGuardian 运行异常: {e}") raise def _record_action(self, action_type: str, platform: str, symbol: str = "", detail: str = ""): self._state.setdefault("last_actions", []).insert(0, { "timestamp": datetime.now().isoformat(), "action_type": action_type, "platform": platform, "symbol": symbol, "detail": detail, }) self._state["last_actions"] = self._state["last_actions"][:20] async def _check_pending_order_timeouts(self, target: ExecutionTarget): """检查各平台挂单超时。""" pending_orders = [] if target.platform == 'PaperTrading': pending_orders = target.service.get_open_orders() elif target.platform == 'Bitget': pending_orders = target.service.get_open_orders() if target.service else [] if not pending_orders: return timeout_orders = target.executor.check_pending_order_timeout(pending_orders) for order_info in timeout_orders: order_id = order_info.get('order_id') symbol = order_info.get('symbol', '') reason = order_info.get('reason', '') logger.info(f" ⏰ [{target.target_key}] {symbol} {reason}") result = await target.executor.execute_cancel(order_id, symbol) if result.get('success'): self._record_action("cancel_timeout", target.target_key, symbol, reason) logger.info(f" ✅ 已取消超时挂单: {order_id}") message = ( f"⏰ 挂单超时自动取消\n\n" f"平台: {target.platform}\n" f"账户: {target.account_id}\n" f"交易对: {symbol}\n" f"订单ID: {order_id}\n" f"原因: {reason}" ) await self.agent._send_alert_notification(f"⏰ [{target.target_key}] 挂单超时", message) else: error = result.get('error', '未知错误') logger.error(f" ❌ 取消失败: {error}") async def _check_position_management(self, target: ExecutionTarget): """检查各平台持仓管理（止盈/止损/移动止损）。""" current_prices = {} volatility_data = {} for symbol in self.agent.symbols: try: data = self.agent.exchange.get_multi_timeframe_data(symbol) current_prices[symbol] = float(data['5m'].iloc[-1]['close']) if '1h' in data and 'atr' in data['1h'].columns: atr_value = data['1h']['atr'].iloc[-1] price_1h = data['1h']['close'].iloc[-1] if atr_value and price_1h > 0: volatility_data[symbol] = float(atr_value) / float(price_1h) except Exception: continue if target.platform == 'PaperTrading': positions = target.service.get_open_positions() elif target.platform == 'Bitget': positions = target.service.get_open_positions() if target.service else [] else: positions = [] if not positions: return actions = target.executor.check_position_management(positions, current_prices, volatility_data) for action_info in actions: symbol = action_info.get('symbol') action = action_info.get('action') reason = action_info.get('reason', '') logger.info(f" 📊 [{target.target_key}] {symbol} {reason}") if action in {'TAKE_PROFIT', 'TIME_EXIT'}: normalized_symbol = self.agent._normalize_symbol(symbol) close_order_ids = [ p.get('order_id') for p in positions if self.agent._normalize_symbol(p.get('symbol', '')) == normalized_symbol and p.get('order_id') ] decision = { 'decision': 'CLOSE', 'symbol': normalized_symbol, 'orders_to_close': close_order_ids, 'reason': reason, } result = await target.executor.execute_close(decision, current_prices.get(symbol, 0)) if result.get('success'): self._record_action(action.lower(), target.target_key, normalized_symbol, reason) title = "💰" if action == 'TAKE_PROFIT' else "⏰" text = "自动止盈" if action == 'TAKE_PROFIT' else "持仓超时平仓" await self.agent._send_alert_notification( f"{title} [{target.target_key}] {text}", f"交易对: {symbol}\n原因: {reason}" ) elif action == 'MOVE_SL': new_sl = action_info.get('new_sl') pnl_pct = action_info.get('pnl_pct', 0) if new_sl: move_result = await target.executor.move_stop_loss(symbol=symbol, new_stop_loss=new_sl) if move_result.get('success'): self._record_action("move_sl", target.target_key, symbol, f"new_sl={new_sl}") await self.agent._send_alert_notification( f"🔒 [{target.target_key}] 移动止损", f"交易对: {symbol}\n新止损: ${new_sl:.2f}\n原因: {reason}" ) await target.executor.send_execution_notification( operation='POSITION_MANAGEMENT', symbol=symbol, result={'success': True, 'action': 'MOVE_SL', 'reason': reason}, details={ 'new_sl': new_sl, 'pnl_percent': pnl_pct, 'account_id': target.account_id, 'target_key': target.target_key, } ) async def _check_and_set_pending_tp_sl(self, target: ExecutionTarget): """检查 Bitget 挂单是否已成交，若成交则补设止盈止损。""" if target.platform != 'Bitget': return pending_state = self.agent._get_pending_tp_sl_state(target.pending_tpsl_state_key or target.target_key) if not pending_state: return for order_id, info in list(pending_state.items()): symbol = self.agent._normalize_symbol(info['symbol']) coin = symbol.replace('USDT', '') open_orders = target.service.get_open_orders(symbol) still_open = any(str(o.get('order_id')) == order_id for o in open_orders) if still_open: continue position = target.service.get_position_for_symbol(coin) if not position: logger.info(f"[{target.target_key}] 挂单追踪 {order_id} 已结束：{symbol} 无持仓，移除待补设任务") self._record_action("cleanup_pending_tpsl", target.target_key, symbol, f"order_id={order_id}") del pending_state[order_id] continue tp_price = info.get('tp_price') sl_price = info.get('sl_price') logger.info(f"[{target.target_key}] 挂单 {order_id} ({symbol}) 已成交，补设 TP/SL...") tp_sl_result = target.service.set_tp_sl( symbol=coin, is_long=position.get('size', 0) > 0, size=abs(position.get('size', 0)), tp_price=tp_price, sl_price=sl_price, ) info['retry_count'] = int(info.get('retry_count', 0)) + 1 tp_set = tp_sl_result.get('tp_set', False) sl_set = tp_sl_result.get('sl_set', False) if tp_set and sl_set: self._record_action("repair_tpsl", target.target_key, symbol, f"order_id={order_id}") logger.info(f"[{target.target_key}] ✅ TP/SL 补设成功: {symbol} TP={tp_price} SL={sl_price}") del pending_state[order_id] continue if tp_set or sl_set: missing_tp = tp_price if not tp_set else None missing_sl = sl_price if not sl_set else None pending_state[order_id] = self.agent._build_pending_tp_sl_task( symbol=symbol, is_long=position.get('size', 0) > 0, size=abs(position.get('size', 0)), tp_price=missing_tp, sl_price=missing_sl, retry_count=info.get('retry_count', 0), first_seen_at=info.get('first_seen_at'), last_alert_at=info.get('last_alert_at'), ) set_text = "TP" if tp_set else "SL" fail_text = "TP" if not tp_set else "SL" await self.agent._maybe_alert_tp_sl_incomplete( target.target_key, order_id, pending_state[order_id], f"{set_text}已设，{fail_text}补设失败", ) continue await self.agent._maybe_alert_tp_sl_incomplete( target.target_key, order_id, info, str(tp_sl_result.get('errors') or 'TP/SL补设失败'), ) async def _check_missing_tp_sl(self, target: ExecutionTarget): """定时检查 Bitget 持仓是否缺少止盈止损，缺少则从信号补救。""" if target.platform != 'Bitget' or not target.service: return positions = target.service.get_open_positions() if not positions: return for pos in positions: symbol = pos.get('symbol', '') if not symbol: continue coin = symbol.replace('USDT', '') tp_sl = target.service.get_tp_sl_prices(coin) has_tp = tp_sl.get('take_profit') is not None has_sl = tp_sl.get('stop_loss') is not None if has_tp and has_sl: continue latest_signal = self.agent.signal_db.get_latest_signal('crypto', symbol) if not latest_signal: missing = ('止盈' if not has_tp else '') + ('/' if not has_tp and not has_sl else '') + ('止损' if not has_sl else '') logger.warning(f"[{target.target_key}] ⚠️ {symbol} 缺少{missing}，且无历史信号可补救") continue tp_price = latest_signal.get('take_profit') sl_price = latest_signal.get('stop_loss') if not tp_price and not sl_price: logger.warning(f"[{target.target_key}] ⚠️ {symbol} 缺少止盈止损，最近信号也无 TP/SL") continue set_tp = tp_price if not has_tp else None set_sl = sl_price if not has_sl else None missing_parts = [] if not has_tp: missing_parts.append(f"TP={set_tp}") if not has_sl: missing_parts.append(f"SL={set_sl}") missing_desc = ' & '.join(missing_parts) logger.warning(f"[{target.target_key}] 🔧 {symbol} 缺少 {missing_desc}，从信号补救...") size = abs(pos.get('size', 0)) if size <= 0: continue tp_sl_result = target.service.set_tp_sl( symbol=coin, is_long=pos.get('size', 0) > 0, size=size, tp_price=set_tp, sl_price=set_sl, ) tp_set = tp_sl_result.get('tp_set', False) sl_set = tp_sl_result.get('sl_set', False) if tp_set or sl_set: self._record_action("fallback_tpsl", target.target_key, symbol, missing_desc) set_parts = [] if tp_set: set_parts.append(f"TP={set_tp}") if sl_set: set_parts.append(f"SL={set_sl}") logger.info(f"[{target.target_key}] ✅ 补救成功: {symbol} {' & '.join(set_parts)}") else: await self.agent._maybe_alert_tp_sl_incomplete( target.target_key, f"{target.target_key}:fallback:{symbol}", self.agent._build_pending_tp_sl_task( symbol=coin, is_long=pos.get('size', 0) > 0, size=size, tp_price=set_tp, sl_price=set_sl, retry_count=self.agent.TP_SL_RETRY_ALERT_THRESHOLD, ), str(tp_sl_result.get('errors') or '兜底补设失败'), force=True, )