# -*- coding: utf-8 -*- """ Hunting Reversal Signal Engine — 猎杀行情反转系统 核心模块： - PreFilterGate : 市场环境前置过滤器（恐贪背离 + 费率极端 + OI高位） - MultiTimeframeResonance : 1m/5m/15m 多时间框架共振检测 - OrderBookLiqAnalyzer : 订单簿清算挂单 vs 真实挂单分析 - LiquidationExhaustion : 清算衰竭信号检测 - LowLeverageDetector : 低杠杆账户出现检测 - BookmapFakeWallDetector : Bookmap 挂单历史分布真假墙识别 """ import time import logging from collections import deque logger = logging.getLogger("MyTrader") # ═══════════════════════════════════════════════════════════════ # 1. 市场环境前置过滤器 # ═══════════════════════════════════════════════════════════════ class PreFilterGate: """ 前置过滤：恐贪指数背离 + 资金费率极端 + 未平仓高位 只有通过前置过滤，才进入信号检测阶段 用户规则： - 恐贪指数上升但价格未同步上涨 → 潜在反转契机 - 永续资金费率极正 → 过热回调风险；极负 → 轧空可能 - OI高位+价格停滞 → 多空对决，变盘前兆 """ def __init__(self): # 历史记录 self.fng_history = deque(maxlen=10) # (ts, fng_val, price) self.fr_history = deque(maxlen=20) # (ts, funding_rate) self.oi_history = deque(maxlen=20) # (ts, oi_value, price) def check(self, price: float, funding_rate: float, fng_val: int, oi_value: float) -> dict: """ 返回: {"passed": bool, "signals": list[str], "score": float} score: 0~1, 越高越满足猎杀条件 """ now = time.time() signals = [] score = 0.0 weight_total = 0.0 # ── 1. 恐贪背离检测 ── self.fng_history.append((now, fng_val, price)) fng_divergence = self._check_fng_divergence() if fng_divergence: signals.append(f"恐贪背离:{fng_divergence}") score += 0.35 weight_total += 0.35 # ── 2. 资金费率极端 ── self.fr_history.append((now, funding_rate)) fr_extreme = self._check_fr_extreme(funding_rate) if fr_extreme: signals.append(f"费率极端:{fr_extreme}") score += 0.30 weight_total += 0.30 # ── 3. OI高位+价格停滞 ── self.oi_history.append((now, oi_value, price)) oi_signal = self._check_oi_stagnation() if oi_signal: signals.append(f"OI:{oi_signal}") score += 0.35 weight_total += 0.35 passed = score >= 0.40 # 至少两个因子触发才算通过 return { "passed": passed, "signals": signals, "score": round(score, 3), } def _check_fng_divergence(self) -> str: """恐贪指数与价格背离""" if len(self.fng_history) < 3: return "" recent = list(self.fng_history)[-3:] fng_trend = recent[-1][1] - recent[0][1] # 恐贪上升/下降 price_chg = (recent[-1][2] - recent[0][2]) / recent[0][2] if recent[0][2] > 0 else 0 # 恐贪上升但价格下跌 → 看多反转契机（恐惧中酝酿上涨） if fng_trend > 3 and price_chg < -0.005: return "恐贪↑但价跌(看多反转)" # 恐贪下降但价格上涨 → 看空反转契机（贪婪中酝酿下跌） if fng_trend < -3 and price_chg > 0.005: return "恐贪↓但价涨(看空反转)" # 恐贪极端值 if recent[-1][1] <= 25: return "极度恐惧(底部区域)" if recent[-1][1] >= 75: return "极度贪婪(顶部区域)" return "" def _check_fr_extreme(self, fr: float) -> str: """资金费率极端检测""" if fr > 0.0008: # >0.08% 多头拥挤 return f"多头拥挤({fr*100:.3f}%)" if fr > 0.0005: # >0.05% 偏多 return f"偏多({fr*100:.3f}%)" if fr < -0.0005: # <-0.05% 空头拥挤 return f"空头拥挤({fr*100:.3f}%)" if fr < -0.0003: # <-0.03% 偏空 return f"偏空({fr*100:.3f}%)" return "" def _check_oi_stagnation(self) -> str: """OI高位 + 价格停滞""" if len(self.oi_history) < 6: return "" items = list(self.oi_history) recent_oi = [i[1] for i in items[-6:]] recent_px = [i[2] for i in items[-6:]] oi_max = max(recent_oi) oi_min = min(recent_oi) if oi_min <= 0: return "" oi_chg = (recent_oi[-1] - oi_min) / oi_min # OI 相对低点涨幅 px_range = (max(recent_px) - min(recent_px)) / min(recent_px) # 价格振幅 # OI高位（相对低点涨超5%）+ 价格窄幅震荡（振幅<1%） if oi_chg > 0.05 and px_range < 0.01: return "OI高位+价格停滞(变盘前兆)" # OI高位震荡 if oi_chg > 0.03 and px_range < 0.015: return "OI高位+窄幅震荡" return "" # ═══════════════════════════════════════════════════════════════ # 2. 多时间框架共振检测 # ═══════════════════════════════════════════════════════════════ class MultiTimeframeResonance: """ 用户规则：「只有 1m、5m、15m 同时出现时，才可以入」 - 1m: 价格突破关键结构（EMA20/布林带边界），伴随主动成交量放大 - 5m: RSI(14) 背离或进入极端区（<30 或 >70） - 15m: ATR(14) 扩张 + 方向与 1m/5m 一致 """ def __init__(self): self.rsi_history = deque(maxlen=30) # (ts, rsi_5m) self.atr_history = deque(maxlen=30) # (ts, atr_15m, atr_15m_prev) def check(self, candles_1m: list, candles_5m: list, candles_15m: list, price: float) -> dict: """ candles: [[ts, open, high, low, close, vol, ...], ...] (reversed, latest first) 返回: {"resonance": "LONG"|"SHORT"|"NONE", "score": float, "details": dict} """ result = {"resonance": "NONE", "score": 0.0, "details": {}} # ── 1分钟: 突破检测 ── sig_1m = self._check_1m(candles_1m) result["details"]["1m"] = sig_1m # ── 5分钟: RSI检测 ── sig_5m = self._check_5m(candles_5m) result["details"]["5m"] = sig_5m # ── 15分钟: ATR检测 ── sig_15m = self._check_15m(candles_15m) result["details"]["15m"] = sig_15m # ── 共振判定 ── def _dir(sig): return sig.get("direction", 0) # 1=LONG, -1=SHORT, 0=neutral d1, d5, d15 = _dir(sig_1m), _dir(sig_5m), _dir(sig_15m) s1, s5, s15 = sig_1m.get("score", 0), sig_5m.get("score", 0), sig_15m.get("score", 0) if d1 == d5 == d15 and d1 != 0: total = (s1 + s5 + s15) / 3 result["resonance"] = "LONG" if d1 == 1 else "SHORT" result["score"] = round(total, 3) elif d1 == d5 and d1 != 0 and s1 >= 0.6 and s5 >= 0.6: # 1m+5m强共振，15m中性 → 弱共振 total = (s1 + s5) / 2 * 0.7 result["resonance"] = "LONG" if d1 == 1 else "SHORT" result["score"] = round(total, 3) result["details"]["_note"] = "1m+5m共振(弱)" return result def _check_1m(self, candles: list) -> dict: """ 1分钟K线：价格突破EMA20/布林带上轨（做多）或下轨（做空）+ 成交量放大 """ if len(candles) < 20: return {"direction": 0, "score": 0, "desc": "数据不足"} closes = [float(c[4]) for c in reversed(candles[:20])] vols = [float(c[5]) for c in reversed(candles[:20])] current = closes[-1] # EMA20 ema20 = sum(closes) / 20 # 布林带（20周期, 2标准差） avg = sum(closes) / len(closes) variance = sum((c - avg) ** 2 for c in closes) / len(closes) std = variance ** 0.5 bb_upper = avg + 2 * std bb_lower = avg - 2 * std # 成交量放大（当前量 vs 近20根均量） vol_avg = sum(vols[:-1]) / max(len(vols) - 1, 1) vol_surge = vols[-1] / vol_avg if vol_avg > 0 else 1.0 score = 0.0 direction = 0 desc = "" # 突破上轨 + 放量 → 做多信号（但用户系统是反转交易，突破是条件确认） if current > bb_upper and vol_surge > 1.5: direction = 1 score = min(1.0, (current - bb_upper) / (bb_upper * 0.002) * 0.5 + min(vol_surge / 3, 1.0) * 0.5) desc = f"突破BB上轨 vol×{vol_surge:.1f}" elif current < bb_lower and vol_surge > 1.5: direction = -1 score = min(1.0, (bb_lower - current) / (bb_lower * 0.002) * 0.5 + min(vol_surge / 3, 1.0) * 0.5) desc = f"突破BB下轨 vol×{vol_surge:.1f}" # EMA20 突破 + 放量 elif current > ema20 and current > ema20 * 1.002 and vol_surge > 1.3: direction = 1 score = min(1.0, (current - ema20) / (ema20 * 0.002) * 0.4 + min(vol_surge / 2.5, 1.0) * 0.6) desc = f"突破EMA20 vol×{vol_surge:.1f}" elif current < ema20 and current < ema20 * 0.998 and vol_surge > 1.3: direction = -1 score = min(1.0, (ema20 - current) / (ema20 * 0.002) * 0.4 + min(vol_surge / 2.5, 1.0) * 0.6) desc = f"跌破EMA20 vol×{vol_surge:.1f}" return {"direction": direction, "score": round(score, 3), "desc": desc, "ema20": round(ema20, 2), "bb_upper": round(bb_upper, 2), "bb_lower": round(bb_lower, 2), "vol_surge": round(vol_surge, 2)} def _check_5m(self, candles: list) -> dict: """ 5分钟K线：RSI(14) 背离或极端区 """ if len(candles) < 14: return {"direction": 0, "score": 0, "desc": "数据不足"} closes = [float(c[4]) for c in reversed(candles[:20])] # RSI(14) gains = [max(closes[i] - closes[i-1], 0) for i in range(1, min(15, len(closes)))] losses = [max(closes[i-1] - closes[i], 0) for i in range(1, min(15, len(closes)))] avg_gain = sum(gains) / len(gains) if gains else 0 avg_loss = sum(losses) / len(losses) if losses else 0.0001 rsi = 100 - 100 / (1 + avg_gain / avg_loss) if avg_loss > 0 else 50 # 价格与RSI背离检测（用近6根5m = 30分钟） if len(closes) >= 10: price_chg = (closes[-1] - closes[-6]) / closes[-6] # 简化RSI历史：用当前RSI vs 6根前近似RSI old_gains = [max(closes[i] - closes[i-1], 0) for i in range(2, min(8, len(closes)))] old_losses = [max(closes[i-1] - closes[i], 0) for i in range(2, min(8, len(closes)))] old_avg_gain = sum(old_gains) / len(old_gains) if old_gains else 0 old_avg_loss = sum(old_losses) / len(old_losses) if old_losses else 0.0001 old_rsi = 100 - 100 / (1 + old_avg_gain / old_avg_loss) if old_avg_loss > 0 else 50 rsi_chg = rsi - old_rsi else: price_chg = 0 rsi_chg = 0 score = 0.0 direction = 0 desc = "" # RSI背离：价格新低但RSI上升 → 看多反转 if price_chg < -0.003 and rsi_chg > 3: direction = 1 score = min(1.0, abs(rsi_chg) / 10 + abs(price_chg) / 0.01 * 0.5) desc = f"RSI底背离(RSI:{rsi:.0f}↑)" # RSI背离：价格新高但RSI下降 → 看空反转 elif price_chg > 0.003 and rsi_chg < -3: direction = -1 score = min(1.0, abs(rsi_chg) / 10 + abs(price_chg) / 0.01 * 0.5) desc = f"RSI顶背离(RSI:{rsi:.0f}↓)" # RSI极端超卖 elif rsi < 30: direction = 1 score = min(1.0, (30 - rsi) / 20) desc = f"RSI超卖({rsi:.0f})" # RSI极端超买 elif rsi > 70: direction = -1 score = min(1.0, (rsi - 70) / 20) desc = f"RSI超买({rsi:.0f})" self.rsi_history.append((time.time(), rsi)) return {"direction": direction, "score": round(score, 3), "desc": desc, "rsi": round(rsi, 1)} def _check_15m(self, candles: list) -> dict: """ 15分钟K线：ATR(14) 扩张 + 方向确认 """ if len(candles) < 16: return {"direction": 0, "score": 0, "desc": "数据不足"} # ATR(14) 当前 tr_list = [] for i in range(min(14, len(candles) - 1)): high = float(candles[i][2]) low = float(candles[i][3]) prev_close = float(candles[i+1][4]) tr = max(high - low, abs(high - prev_close), abs(low - prev_close)) tr_list.append(tr) atr_current = sum(tr_list) / len(tr_list) if tr_list else 0 # ATR(14) 前期对比（前6根 vs 当前6根） tr_prev_list = [] for i in range(14, min(20, len(candles) - 1)): high = float(candles[i][2]) low = float(candles[i][3]) prev_close = float(candles[i+1][4]) tr = max(high - low, abs(high - prev_close), abs(low - prev_close)) tr_prev_list.append(tr) atr_prev = sum(tr_prev_list) / len(tr_prev_list) if tr_prev_list else atr_current # ATR 扩张比例 atr_ratio = atr_current / atr_prev if atr_prev > 0 else 1.0 # 方向判断（用近6根15m收盘价） closes = [float(c[4]) for c in reversed(candles[:6])] direction = 1 if closes[-1] > closes[0] else (-1 if closes[-1] < closes[0] else 0) chg = abs((closes[-1] - closes[0]) / closes[0]) if closes[0] > 0 else 0 score = 0.0 desc = "" price = float(candles[0][4]) if atr_ratio > 1.3 and chg > 0.002: score = min(1.0, (atr_ratio - 1) / 0.5 + chg / 0.01 * 0.3) desc = f"ATR扩张×{atr_ratio:.1f}" elif atr_ratio > 1.15 and chg > 0.001: score = min(1.0, (atr_ratio - 1) / 0.5 * 0.6) desc = f"ATR轻扩×{atr_ratio:.1f}" self.atr_history.append((time.time(), atr_current, atr_prev)) return {"direction": direction, "score": round(score, 3), "desc": desc, "atr_ratio": round(atr_ratio, 2), "atr_current": round(atr_current, 4)} # ═══════════════════════════════════════════════════════════════ # 3. 订单簿清算挂单分析 # ═══════════════════════════════════════════════════════════════ class OrderBookLiqAnalyzer: """ 用户规则： - 清算挂单占比 vs 真实挂单占比 → 推断波动幅度 - 只有巨量且持续的市价买入才能推动价格 - 做市商挂单主导成交时，会打掉大部分止损 """ def __init__(self): self.liq_ratio_history = deque(maxlen=10) def analyze(self, orderbook: dict, liq_data: dict, price: float) -> dict: """ orderbook: OKX订单簿 {"asks": [[px,sz,qty],...], "bids": [[px,sz,qty],...]} liq_data: Coinglass清算数据 返回: {"liq_vs_real": float, "is_fake_wall": bool, "score": float, "detail": str} """ asks = orderbook.get("asks", []) bids = orderbook.get("bids", []) # 计算挂单深度（±2%） ask_depth_2p = sum(float(a[1]) for a in asks if float(a[0]) <= price * 1.02) bid_depth_2p = sum(float(b[1]) for b in bids if float(b[0]) >= price * 0.98) total_depth = ask_depth_2p + bid_depth_2p # 清算数据中的待清算量（Coinglass 热力图数据） up_liq = liq_data.get("up_zones_total", 0) # 上方待清算总额 down_liq = liq_data.get("down_zones_total", 0) # 下方待清算总额 total_liq = up_liq + down_liq if total_liq <= 0 and total_depth <= 0: return {"liq_vs_real": 0, "is_fake_wall": False, "score": 0, "detail": "无数据"} # 清算挂单占比 # 将清算金额转换为近似挂单量（按当前价折算） liq_eq_depth = total_liq / price if price > 0 else 0 depth_total = total_depth if total_depth > 0 else 1 # 清算挂单占总挂单比 liq_ratio = liq_eq_depth / depth_total # 清算方向偏斜 liq_skew = 0 if total_liq > 0: liq_skew = (up_liq - down_liq) / total_liq # 评分逻辑： # 清算占比高 + 方向偏斜 → 猎杀行情概率高 # 清算占比低 → 自然成交，信号弱 score = 0.0 detail_parts = [] if liq_ratio > 0.5: score += 0.4 detail_parts.append("清算占比高") elif liq_ratio > 0.2: score += 0.2 detail_parts.append("清算占比中等") # 方向偏斜叠加 if abs(liq_skew) > 0.5: score += 0.3 dir_str = "上方偏斜" if liq_skew > 0 else "下方偏斜" detail_parts.append(dir_str) if abs(liq_skew) > 0.7: score += 0.3 detail_parts.append("极端偏斜") score = min(1.0, score) is_fake = liq_ratio > 0.7 or (liq_ratio > 0.3 and abs(liq_skew) > 0.7) self.liq_ratio_history.append(liq_ratio) return { "liq_vs_real": round(liq_ratio, 3), "is_fake_wall": is_fake, "liq_skew": round(liq_skew, 3), "score": round(score, 3), "detail": " | ".join(detail_parts) if detail_parts else "正常", "up_liq_m": round(up_liq / 1e6, 2), "down_liq_m": round(down_liq / 1e6, 2), } # ═══════════════════════════════════════════════════════════════ # 4. 清算衰竭信号检测 # ═══════════════════════════════════════════════════════════════ class LiquidationExhaustion: """ 用户规则：「在新的清算出现之前，不做单，全部视为假突破」 - 观察清算力度递减 → 行情难以推动 - 真实多头/空头开始止盈 → 进入震荡区间 → 不再开仓 """ def __init__(self, window_size: int = 6): self.window_size = window_size self.liq_intensity: deque = deque(maxlen=window_size) # (ts, total_liq_usd) self.consecutive_declines = 0 self.last_spike_ts = 0 # 上次清算高峰时间 def feed(self, liq_1h_long: float, liq_1h_short: float): """输入每小时的清算数据（美元）""" now = time.time() total = liq_1h_long + liq_1h_short self.liq_intensity.append((now, total)) # 检测清算力度趋势 if len(self.liq_intensity) >= 3: recent = list(self.liq_intensity)[-3:] if recent[2][1] < recent[1][1] < recent[0][1]: self.consecutive_declines += 1 else: self.consecutive_declines = max(0, self.consecutive_declines - 1) # 记录清算高峰 if recent[2][1] > max(recent[0][1], recent[1][1]) * 1.5: self.last_spike_ts = now @property def is_exhausted(self) -> bool: """清算是否已衰竭""" # 连续3次下降 → 衰竭 if self.consecutive_declines >= 3: return True # 最近清算量极低 if self.liq_intensity: recent_total = list(self.liq_intensity)[-1][1] if len(self.liq_intensity) >= 3: avg_total = sum(i[1] for i in list(self.liq_intensity)[-3:]) / 3 if avg_total < 100000 and recent_total < 50000: # < $100k return True return False @property def score(self) -> float: """ 清算衰竭评分：-1 到 1 -1 → 清算衰竭，不开仓 0 → 正常 1 → 清算活跃，可开仓 """ if self.is_exhausted: if self.consecutive_declines >= 5: return -1.0 elif self.consecutive_declines >= 3: return -0.7 return -0.4 if self.consecutive_declines == 0 and len(self.liq_intensity) >= 3: recent = list(self.liq_intensity)[-3:] if recent[2][1] > recent[1][1] > recent[0][1]: return 0.5 # 清算加速，可开仓 return 0.0 def should_trade(self, direction: str = "") -> tuple: """返回 (can_trade: bool, reason: str)""" if self.is_exhausted: return False, f"清算衰竭(连降{self.consecutive_declines}次)，假突破风险高" return True, "" # ═══════════════════════════════════════════════════════════════ # 5. 低杠杆账户检测 # ═══════════════════════════════════════════════════════════════ class LowLeverageDetector: """ 用户规则：「低杠杆账户出现 → 反转高概率信号」 低杠杆是专业资金的早期信号： - 降低杠杆 = 资金在保护仓位，预期行情可能反转 - 高杠杆 = 散户情绪驱动 """ def __init__(self): self.avg_leverage_history = deque(maxlen=10) # (ts, avg_leverage) self.low_lev_count = 0 def feed(self, avg_leverage: float): """输入当前市场平均杠杆倍数""" now = time.time() self.avg_leverage_history.append((now, avg_leverage)) # 低杠杆计数 if avg_leverage < 5: # 平均杠杆 < 5x self.low_lev_count += 1 else: self.low_lev_count = max(0, self.low_lev_count - 1) @property def score(self) -> float: """ 低杠杆信号评分：-1 到 1 >0 → 低杠杆，反转信号增强 <0 → 高杠杆，跟随趋势 """ if not self.avg_leverage_history: return 0.0 recent = list(self.avg_leverage_history)[-3:] if len(recent) < 2: return 0.0 avg_lev = sum(i[1] for i in recent) / len(recent) # 杠杆持续下降 → 反转信号 if len(recent) >= 3 and recent[2][1] > recent[1][1] > recent[0][1]: # 杠杆连续下降 drop = (recent[0][1] - recent[2][1]) / max(recent[2][1], 1) return min(1.0, abs(drop) * 8) # 降12.5% → 满分 # 低杠杆水平 if avg_lev < 3: return 0.8 elif avg_lev < 5: return 0.5 elif avg_lev < 8: return 0.2 elif avg_lev > 20: return -0.3 # 高杠杆 → 不追 return 0.0 @property def detail(self) -> str: if not self.avg_leverage_history: return "无数据" avg = sum(i[1] for i in list(self.avg_leverage_history)[-3:]) / max(len(self.avg_leverage_history), 1) if self.low_lev_count >= 3: return f"低杠杆持续({avg:.1f}x) → 反转信号" if avg < 5: return f"杠杆偏低({avg:.1f}x)" return f"杠杆正常({avg:.1f}x)" # ═══════════════════════════════════════════════════════════════ # 6. Bookmap 挂单历史分布真假墙识别 # ═══════════════════════════════════════════════════════════════ class BookmapFakeWallDetector: """ 用户规则： - 以市价 ±0.002 为价格分度，聚合挂单深度 - 对比历史挂单量 vs 真实成交量 - 接近 → 真实吸收，有效支撑/阻力 - 悬殊 → 虚假挂单，诱多/诱空 """ def __init__(self, bucket_pct: float = 0.002): self.bucket_pct = bucket_pct self.wall_history: deque = deque(maxlen=30) # (ts, price_bucket, order_qty, trade_qty) self.order_buckets: dict = {} # bucket_price → cumulative_order_qty self.trade_buckets: dict = {} # bucket_price → cumulative_trade_qty def feed_orderbook(self, bids: list, asks: list, price: float): """输入订单簿，聚合挂单""" self.order_buckets.clear() bucket_size = price * self.bucket_pct for px, sz, *_ in bids: bucket = round(float(px) / bucket_size) * bucket_size self.order_buckets[bucket] = self.order_buckets.get(bucket, 0) + float(sz) for px, sz, *_ in asks: bucket = round(float(px) / bucket_size) * bucket_size self.order_buckets[bucket] = self.order_buckets.get(bucket, 0) + float(sz) def feed_trades(self, trades: list, price: float): """输入成交记录，聚合成交量""" bucket_size = price * self.bucket_pct for trade in trades: px = float(trade.get("px", 0)) sz = float(trade.get("sz", 0)) if px <= 0: continue bucket = round(px / bucket_size) * bucket_size self.trade_buckets[bucket] = self.trade_buckets.get(bucket, 0) + sz def detect_fake_walls(self, price: float) -> dict: """ 检测虚假挂单墙 返回: {"fake_above": [...], "fake_below": [...], "real_above": [...], "real_below": [...]} """ fake_above, fake_below = [], [] real_above, real_below = [], [] for bucket_px, order_qty in self.order_buckets.items(): trade_qty = self.trade_buckets.get(bucket_px, 0) if order_qty <= 0: continue ratio = trade_qty / order_qty # 成交/挂单比 dist_pct = (bucket_px - price) / price entry = { "price": round(bucket_px, 2), "dist_pct": round(dist_pct * 100, 2), "order_qty": round(order_qty, 1), "trade_qty": round(trade_qty, 1), "ratio": round(ratio, 3), } # 比值过低 → 虚假挂单（挂了很多但没人成交） if order_qty > 1000 and ratio < 0.05: if dist_pct > 0: fake_above.append(entry) else: fake_below.append(entry) # 比值接近 → 真实吸收 elif order_qty > 500 and ratio > 0.4: if dist_pct > 0: real_above.append(entry) else: real_below.append(entry) # 排序 fake_above.sort(key=lambda x: x["dist_pct"]) fake_below.sort(key=lambda x: -x["dist_pct"]) real_above.sort(key=lambda x: x["dist_pct"]) real_below.sort(key=lambda x: -x["dist_pct"]) return { "fake_above": fake_above[:3], "fake_below": fake_below[:3], "real_above": real_above[:3], "real_below": real_below[:3], } @property def score(self) -> float: """ 真假墙评分：-1（假墙诱多/上方有大量假挂单）到 1（假墙诱空/下方有大量假挂单） 正值 → 下方假挂单多 → 真实支撑弱 → 偏空 负值 → 上方假挂单多 → 真实阻力弱 → 偏多 """ fake_above_total = sum(e["order_qty"] for e in self._last_detection.get("fake_above", [])) fake_below_total = sum(e["order_qty"] for e in self._last_detection.get("fake_below", [])) total = fake_above_total + fake_below_total if total <= 0: return 0.0 # 下方假单多 → 偏空（支撑是假的）；上方假单多 → 偏多（阻力是假的） return round(max(-1, min(1, (fake_above_total - fake_below_total) / total)), 3) _last_detection: dict = {}