""" 交易部 → 情报部 非结构化市场数据推送 职责： 从 MarketAnalyzer 的分析结果中提取非结构化微观数据， 每60秒通过总线 Push 到情报部，供给日报分析使用。 推送数据字段： - orderbook_imbalance: 订单簿挂单失衡（买方vs卖方深度） - taker_volume_ratio: Taker主动买卖量比率 - oi_change_5m: OI未平仓持仓变化（5分钟窗口） - funding_rate: 持仓资金费率 - liquidation_events: 清算事件明细（多/空清、密集区） - factor_scores: 15因子完整评分快照 """ import json import logging import threading import time from datetime import datetime from typing import Optional import httpx logger = logging.getLogger("trading.data_pusher") class DataPusher: """非结构化市场数据推送器 在主循环中每60秒调用一次 push()，提取 analyzer 产出的 微观数据并 POST 到最高统帅部网关的 /api/command/relay/trading-data， 网关路由转发到情报部。 """ def __init__(self, gateway_url: str = "http://127.0.0.1:8000", internal_token: str = "supreme-internal-token-2026", poll_interval: float = 60.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._last_push_time: float = 0.0 self._lock = threading.Lock() # 缓存最近一次分析数据，避免推送空数据 self._latest_result: Optional[dict] = None logger.info(f"交易部数据推送器就绪 → 网关: {gateway_url} (间隔{self.poll_interval}s)") # ══════════════════════════════════════════ # 数据提取 # ══════════════════════════════════════════ @staticmethod def extract_trading_data(result: dict) -> list[dict]: """从 TradeManager.run_cycle() 的返回结果中提取非结构化数据。 result['signals'] 是每个标的的信号列表，每个 signal 包含: - symbol, price, score, direction - factors: 15+因子评分dict - raw: 原始数据（liq_detail, ob_skew, taker等） - funding_rate, max_pain, news_headlines Returns: 按标的组织的非结构化数据列表 """ signals = result.get("signals", []) if not signals: return [] extracted = [] for sig in signals: factors = sig.get("factors", {}) raw = sig.get("raw", {}) liq_detail = raw.get("liq_detail", {}) if isinstance(raw, dict) else {} # 清算事件摘要 liq_summary = {} if isinstance(liq_detail, dict): liq_summary = { "hist": liq_detail.get("hist", ""), "zones": liq_detail.get("zones", ""), "prob": liq_detail.get("prob", ""), "ls_ratio": liq_detail.get("ls_ratio"), "final": liq_detail.get("final"), "wall_direction": liq_detail.get("wall_direction", 0), "proximity": liq_detail.get("proximity", 0), } entry = { "symbol": sig.get("symbol", "?"), "price": sig.get("price", 0), "direction": sig.get("direction", "WAIT"), "total_score": sig.get("score", 0), # 核心非结构化指标 "orderbook_imbalance": round(factors.get("orderbook", 0), 4), "taker_volume_ratio": round(raw.get("taker", 0), 4) if isinstance(raw, dict) else round(factors.get("taker", 0), 4), "oi_change_5m": round(raw.get("oi", 0), 4) if isinstance(raw, dict) else round(factors.get("oi", 0), 4), "funding_rate": sig.get("funding_rate", 0), "liquidation_events": liq_summary, # 完整因子评分快照 "factor_scores": { "trend": factors.get("trend", 0), "orderbook": factors.get("orderbook", 0), "funding": factors.get("funding", 0), "taker": factors.get("taker", 0), "oi": factors.get("oi", 0), "maxpain": factors.get("maxpain", 0), "vol_delta": factors.get("vol_delta", 0), "btc_corr": factors.get("btc_corr", 0), "gamma": factors.get("gamma", 0), "iv": factors.get("iv", 0), "exhaust": factors.get("exhaust", 0), "liq_cool": factors.get("liq_cool", 0), "mean_revert": factors.get("mean_revert", 0), "news": factors.get("news", 0), "smart_money": factors.get("smart_money", 0), "copy_trade": factors.get("copy_trade", 0), "mtf": factors.get("mtf", 0), "ob_liq": factors.get("ob_liq", 0), "low_lev": factors.get("low_lev", 0), "liq_ex": factors.get("liq_ex", 0), }, # 原始微观数据（供情报部深度使用） "raw_micro": { "ob_skew": raw.get("ob_skew", 0) if isinstance(raw, dict) else 0, "ask_near": raw.get("ask_near", 0) if isinstance(raw, dict) else 0, "bid_near": raw.get("bid_near", 0) if isinstance(raw, dict) else 0, "fr_raw": raw.get("fr_raw", 0) if isinstance(raw, dict) else 0, "taker_avg": raw.get("taker_avg", 0) if isinstance(raw, dict) else 0, "oi_avg": raw.get("oi_avg", 0) if isinstance(raw, dict) else 0, "ob_avg": raw.get("ob_avg", 0) if isinstance(raw, dict) else 0, "fng": raw.get("fng", 0) if isinstance(raw, dict) else 0, }, } extracted.append(entry) return extracted # ══════════════════════════════════════════ # 推送逻辑 # ══════════════════════════════════════════ def update_result(self, result: dict): """更新最新的分析结果缓存（由主循环在每次 run_cycle 后调用）。""" with self._lock: self._latest_result = result def should_push(self) -> bool: """判断是否到达推送时机（每60秒一次）。""" with self._lock: now = time.time() if now - self._last_push_time >= self.poll_interval: return True return False def push(self) -> dict: """执行一次数据推送。 从缓存的最新分析结果中提取非结构化数据， POST 到网关的 /api/command/relay/trading-data 端点。 """ with self._lock: result = self._latest_result if result is None: return {"status": "skip", "reason": "no_data_yet"} try: trading_data = self.extract_trading_data(result) if not trading_data: return {"status": "skip", "reason": "empty_signals"} payload = { "source_dept": "trading", "data_type": "unstructured_market_micro", "symbols": [d["symbol"] for d in trading_data], "trading_data": trading_data, "timestamp": datetime.now().isoformat(), } resp = self._post("/api/command/relay/trading-data", payload) with self._lock: self._last_push_time = time.time() logger.info(f"[DataPusher] 已推送 {len(trading_data)} 个标的的非结构化数据到情报部") return {"status": "sent", "symbols": payload["symbols"], "response": resp} except Exception as e: logger.error(f"[DataPusher] 推送失败: {e}") return {"status": "error", "reason": str(e)} # ══════════════════════════════════════════ # 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"DataPusher POST失败 {path}: {e}") return {"error": str(e)} def ping(self) -> bool: """检查网关连通性。""" try: with httpx.Client(timeout=5.0) as client: r = client.get(f"{self.gateway_url}/api/health", headers=self._headers) return r.status_code == 200 except Exception: return False