""" 交易部 → 最高统帅部 连接客户端 职责： 入站：接收战略部的交易命令（strategy.signal）→ 调整仓位 出站：推送交易状态（trading.status）→ 战略部/情报部 推送盈亏更新（trading.pnl）→ 情报部 接收风控告警（risk.alert）→ 触发暂停/平仓 注意：交易部不产生宏观判断，只执行战略部的交易命令。 风控部告警具有更高优先级——可覆盖战略部信号。 """ import json import logging import threading import time from datetime import datetime from dataclasses import dataclass, field, asdict from typing import Optional, Callable import httpx logger = logging.getLogger("trading.gateway_client") # ══════════════════════════════════════════ # 数据模型 # ══════════════════════════════════════════ @dataclass class StrategyCommand: """战略部发来的交易命令""" asset: str scenario: str probability: float direction: str # long / short / neutral conviction: str # high / medium / low reasoning: str = "" analyst: str = "" timestamp: str = "" msg_id: str = "" requires_action: bool = False # conviction=high + prob≥0.65 def should_adjust_position(self) -> bool: return self.requires_action and self.conviction == "high" def get_max_position_ratio(self) -> float: """根据信号确定仓位上限""" if self.conviction == "high" and self.probability >= 0.65: return 0.30 elif self.conviction == "medium": return 0.20 else: return 0.10 @dataclass class RiskCommand: """风控部发来的告警命令（优先级高于战略部信号）""" rule: str severity: str address: str chain: str details: dict = field(default_factory=dict) action_required: bool = False # 高危自动平仓 msg_id: str = "" def should_force_close(self) -> bool: return self.severity == "高危" and self.action_required @dataclass class InfrastructureCommand: """统帅部发来的基础设施指令""" action: str # pause / resume / shutdown reason: str = "" duration_minutes: int = 0 msg_id: str = "" # ══════════════════════════════════════════ # 网关客户端 # ══════════════════════════════════════════ class TradingGatewayClient: """交易部与统帅部网关的通信客户端""" def __init__(self, gateway_url: str = "http://127.0.0.1:8000", internal_token: str = "supreme-internal-token-2026", poll_interval: float = 5.0): self.gateway_url = gateway_url.rstrip("/") self._headers = { "X-Internal-Token": internal_token, "X-Source-Dept": "trading", "Content-Type": "application/json", } self.poll_interval = poll_interval self._running = False self._thread: Optional[threading.Thread] = None # 待处理的指令队列 self.pending_strategy_commands: list[StrategyCommand] = [] self.pending_risk_commands: list[RiskCommand] = [] self.pending_infra_commands: list[InfrastructureCommand] = [] self._lock = threading.Lock() # 回调（由 TradeManager 注册） self.on_strategy_signal: Optional[Callable] = None self.on_risk_alert: Optional[Callable] = None self.on_infra_command: Optional[Callable] = None logger.info(f"交易部客户端就绪 → 网关: {gateway_url}") # ══════════════════════════════════════ # 出站：推送交易状态 # ══════════════════════════════════════ def send_trading_status(self, equity: float, pnl_24h: float, positions: list, score: float, strategy_name: str = "alpha") -> dict: """推送交易状态到网关（→ 战略部 + 情报部）""" payload = { "source_dept": "trading", "strategy": strategy_name, "equity": round(equity, 4), "pnl_24h": round(pnl_24h, 4), "positions": positions, "score": round(score, 4), "timestamp": datetime.now().isoformat(), } return self._post("/api/command/relay/trading-status", payload) def send_execution_report(self, trade: dict) -> dict: """推送成交回报""" payload = { "source_dept": "trading", "trade": trade, "timestamp": datetime.now().isoformat(), } return self._post("/api/command/relay/trading-execution", payload) def send_pnl_update(self, pnl: float, pnl_pct: float, strategy_name: str = "alpha") -> dict: """推送盈亏更新（→ 情报部用于日报）""" payload = { "source_dept": "trading", "strategy": strategy_name, "pnl": round(pnl, 4), "pnl_pct": round(pnl_pct, 4), "timestamp": datetime.now().isoformat(), } return self._post("/api/command/relay/trading-pnl", payload) # ══════════════════════════════════════ # 出站：请求基础设施指令 # ══════════════════════════════════════ def request_self_pause(self, reason: str, minutes: int = 30) -> dict: """交易部自行请求暂停（遇到异常时）""" logger.warning(f"交易部请求自暂停: {reason}") return self._post("/api/command/pause", { "reason": f"[交易部自请求] {reason}", "duration_minutes": minutes, }) # ══════════════════════════════════════ # 入站轮询：接收指令 # ══════════════════════════════════════ def start_polling(self): """启动后台轮询，接收网关转发的指令""" if self._running: return self._running = True self._thread = threading.Thread(target=self._poll_loop, daemon=True) self._thread.start() logger.info("交易部指令轮询已启动") def stop_polling(self): self._running = False def _poll_loop(self): while self._running: try: messages = self._get_recent_messages() for msg in messages: self._dispatch(msg) except Exception as e: logger.debug(f"轮询错误: {e}") time.sleep(self.poll_interval) def _get_recent_messages(self) -> list: result = self._get("/api/status") return result.get("message_bus", {}).get("latest", []) def _dispatch(self, msg: dict): """分发消息到对应处理器""" msg_type = msg.get("msg_type", "") source = msg.get("source", "") payload = msg.get("payload", {}) priority = msg.get("priority", "") with self._lock: # 战略部 → 交易部：交易命令 if msg_type == "strategy.signal" and source == "strategy": cmd = StrategyCommand( asset=payload.get("asset", ""), scenario=payload.get("scenario", ""), probability=payload.get("probability", 0.5), direction=payload.get("direction", "neutral"), conviction=payload.get("conviction", "low"), reasoning=payload.get("reasoning", ""), analyst=payload.get("analyst", ""), msg_id=msg.get("msg_id", ""), requires_action=payload.get("requires_action", False), ) self.pending_strategy_commands.append(cmd) logger.info(f"🎯 收到交易命令: {cmd.asset} {cmd.direction} " f"conviction={cmd.conviction}") if self.on_strategy_signal: self.on_strategy_signal(cmd) # 风控部 → 交易部：告警 elif msg_type in ("risk.alert", "risk.sanction") and source == "risk": cmd = RiskCommand( rule=payload.get("alert_rule", ""), severity=payload.get("severity", ""), address=payload.get("address", ""), chain=payload.get("chain", ""), details=payload.get("details", {}), action_required=payload.get("action_required", False), msg_id=msg.get("msg_id", ""), ) self.pending_risk_commands.append(cmd) logger.warning(f"🚨 收到风控告警: [{cmd.severity}] {cmd.rule}") if self.on_risk_alert: self.on_risk_alert(cmd) # 统帅部 → 交易部：基础设施指令 elif msg_type in ("command.pause", "command.resume", "command.shutdown"): cmd = InfrastructureCommand( action=msg_type.replace("command.", ""), reason=payload.get("reason", ""), duration_minutes=payload.get("duration_minutes", 0), msg_id=msg.get("msg_id", ""), ) self.pending_infra_commands.append(cmd) logger.warning(f"⚠️ 收到基础设施指令: {cmd.action}") if self.on_infra_command: self.on_infra_command(cmd) def get_pending_commands(self) -> dict: """获取所有待处理指令""" with self._lock: result = { "strategy": [asdict(c) for c in self.pending_strategy_commands[-10:]], "risk": [asdict(c) for c in self.pending_risk_commands[-10:]], "infra": [asdict(c) for c in self.pending_infra_commands[-5:]], } return result # ══════════════════════════════════════ # HTTP 底层 # ══════════════════════════════════════ def _post(self, path: str, data: dict) -> dict: try: with httpx.Client(timeout=10.0) as client: r = client.post(f"{self.gateway_url}{path}", headers=self._headers, json=data) r.raise_for_status() return r.json() except Exception as e: logger.error(f"网关POST失败 {path}: {e}") return {"error": str(e)} def _get(self, path: str) -> dict: try: with httpx.Client(timeout=10.0) as client: r = client.get(f"{self.gateway_url}{path}", headers=self._headers) r.raise_for_status() return r.json() except Exception as e: logger.debug(f"网关GET失败 {path}: {e}") return {} def ping(self) -> bool: result = self._get("/api/health") return result.get("status") == "ok"