# -*- coding: utf-8 -*- """ MarketAnalyzer — 市场分析引擎 多因子综合评分：趋势 / 挂单 / 成交 / OI / 费率 / 期权 / 量变 / 大盘 / Gamma / IV / 衰竭 / 清算 / 均值回归 / 聪明钱 / MTF / OB清算 / 低杠杆 / 清算衰竭 """ import re import json import time import logging import os import requests from collections import deque from config import ( INSTRUMENTS, CG_API_KEY, CG_BASE_URL, ETHERSCAN_API_KEY, SCORE_THRESHOLD, MIN_FACTORS_AGREE, REQUIRE_TREND_ALIGN, PROXIES, W_TREND, W_ORDERBOOK, W_TAKER, W_OI_CHANGE, W_FUNDING, W_MAXPAIN, W_VOL_DELTA, W_BTC_CORR, W_GAMMA, W_IV, W_TAKER_EXHAUST, W_LIQ_COOLDOWN, W_MEAN_REVERT, W_SMART_MONEY, W_MTF_RESONANCE, W_OB_LIQ_RATIO, W_LOW_LEVERAGE, W_LIQ_EXHAUSTION, W_PRE_FILTER, PRE_FILTER_THRESHOLD, MTF_REQUIRE_ALL, MTF_MIN_SCORE, W_RETAIL_POSITION, W_MM_POSITION, W_LIQ_TRIGGER, W_TOXIC_FLOW, LIQ_EXHAUST_DECLINES, LOW_LEV_THRESHOLD, LIQ_PROXIMITY_BOOST, LIQ_PROXIMITY_MIN_USD, COPY_TRADE_ENABLED, COPY_TRADE_WEIGHT, COPY_TRADE_ADDRESSES, DB_ENABLED, ) from exchange.http import http_get from strategy.copy_trader import CopyTrader from strategy.hunting_signals import ( PreFilterGate, MultiTimeframeResonance, OrderBookLiqAnalyzer, LiquidationExhaustion, LowLeverageDetector, BookmapFakeWallDetector, ) from exchange.toxic_flow import ToxicFlowDetector logger = logging.getLogger("MyTrader") class MarketAnalyzer: def __init__(self, okx, deribit): self.okx = okx self.deribit = deribit self.ob_history = {sym: deque(maxlen=6) for sym in INSTRUMENTS} self.taker_history = {sym: deque(maxlen=6) for sym in INSTRUMENTS} self.oi_history = {sym: deque(maxlen=6) for sym in INSTRUMENTS} self.score_history = {sym: deque(maxlen=30) for sym in INSTRUMENTS} self.taker_abs_history = {sym: deque(maxlen=6) for sym in INSTRUMENTS} self.liq_history = {sym: deque(maxlen=10) for sym in INSTRUMENTS} self.factor_history = {sym: deque(maxlen=10) for sym in INSTRUMENTS} # 猎杀行情反转信号引擎 self.pre_filter = PreFilterGate() self.mtf_resonance = MultiTimeframeResonance() self.ob_liq_analyzer = OrderBookLiqAnalyzer() self.liq_exhaustion = {sym: LiquidationExhaustion(LIQ_EXHAUST_DECLINES) for sym in INSTRUMENTS} self.low_lev_detector = LowLeverageDetector() self.fake_wall_detector = BookmapFakeWallDetector(bucket_pct=0.002) # 有毒订单流检测器 self.toxic_detector = ToxicFlowDetector(max_history=100) # 链上跟单引擎 self.copy_trader = CopyTrader(COPY_TRADE_ADDRESSES, ETHERSCAN_API_KEY) if COPY_TRADE_ENABLED else None self._copy_last_poll = 0 self._copy_signal_cache: list = [] # 清算触发点追踪 self._liq_events = {sym: deque(maxlen=50) for sym in INSTRUMENTS} # [{price, side, ts}] # ────────────────────────────────────────── # 趋势分析（1H + 4H EMA交叉） # ────────────────────────────────────────── def _trend_score(self, inst_id): trend_1h, trend_4h = 0, 0 for bar in ["1H", "4H"]: candles = self.okx.candles(inst_id, bar=bar, limit=10) if len(candles) < 10: continue closes = [float(c[4]) for c in reversed(candles[:10])] ema5 = sum(closes[-5:]) / 5 ema10 = sum(closes) / 10 s = 1 if ema5 > ema10 * 1.001 else (-1 if ema5 < ema10 * 0.999 else 0) if bar == "1H": trend_1h = s else: trend_4h = s combined = trend_1h if trend_1h == trend_4h and trend_1h != 0 else \ (trend_1h if trend_1h != 0 else trend_4h) logger.info(f"[趋势] 1H:{trend_1h:+d} 4H:{trend_4h:+d} => {combined:+d}") return combined, trend_1h, trend_4h # ────────────────────────────────────────── # 清算密集区概率测算（Coinglass） # ────────────────────────────────────────── def _liquidation_bias(self, inst_id): try: tick = self.okx.ticker(inst_id) price = float(tick.get("last", 0)) if price == 0: return 0, {} cg_headers = {'X-Api-Key': CG_API_KEY} cg_base = CG_BASE_URL sym = inst_id.split("-")[0] # 1. 多时间框架清算统计 liq_1h_long = liq_1h_short = liq_4h_long = liq_4h_short = 0 liq_24h_long = liq_24h_short = 0 try: r = http_get(f'{cg_base}/api/futures/liquidation/coin-list?timeType=1', headers=cg_headers, timeout=8) d = r.json() if d.get('code') == '0' and d.get('data'): for coin in d['data']: if coin.get('symbol') == sym: liq_1h_long = coin.get('long_liquidation_usd_1h', 0) liq_1h_short = coin.get('short_liquidation_usd_1h', 0) liq_4h_long = coin.get('long_liquidation_usd_4h', 0) liq_4h_short = coin.get('short_liquidation_usd_4h', 0) liq_24h_long = coin.get('long_liquidation_usd_24h', 0) liq_24h_short = coin.get('short_liquidation_usd_24h', 0) break except: pass # 2. 清算密集区热力图（按价格桶化） up_zones = {} down_zones = {} bucket_size = price * 0.005 try: r2 = http_get(f'{cg_base}/api/futures/liquidation/order?symbol={sym}', headers=cg_headers, timeout=8) d2 = r2.json() if d2.get('code') == '0' and d2.get('data'): for order in d2['data']: op = order.get('price', 0) usd = order.get('usd_value', 0) if op <= 0 or usd <= 0: continue bucket = round(op / bucket_size) * bucket_size dist_pct = (op - price) / price if dist_pct > 0: up_zones[bucket] = up_zones.get(bucket, 0) + usd else: down_zones[bucket] = down_zones.get(bucket, 0) + usd except: pass # 3. 清算概率测算 up_total = sum(up_zones.values()) down_total = sum(down_zones.values()) up_nearest_dist, up_nearest_usd = 999, 0 for bp, usd in up_zones.items(): dist = abs(bp - price) / price if 0.002 < dist < 0.05 and usd > up_nearest_usd * 0.5: if dist < up_nearest_dist: up_nearest_dist = dist up_nearest_usd = usd down_nearest_dist, down_nearest_usd = 999, 0 for bp, usd in down_zones.items(): dist = abs(bp - price) / price if 0.002 < dist < 0.05 and usd > down_nearest_usd * 0.5: if dist < down_nearest_dist: down_nearest_dist = dist down_nearest_usd = usd up_magnet = up_nearest_usd / (up_nearest_dist * 1e6 + 1) if up_nearest_dist < 999 else 0 down_magnet = down_nearest_usd / (down_nearest_dist * 1e6 + 1) if down_nearest_dist < 999 else 0 # 3.5 清算墙邻近度 — 越近越大，强化磁吸效应 up_proximity = 0.0 down_proximity = 0.0 wall_threshold = 0.03 # 3%距离内有效 if up_nearest_dist < wall_threshold and up_nearest_usd >= LIQ_PROXIMITY_MIN_USD: up_proximity = (1.0 - up_nearest_dist / wall_threshold) * min(1.0, up_nearest_usd / (LIQ_PROXIMITY_MIN_USD * 5)) if down_nearest_dist < wall_threshold and down_nearest_usd >= LIQ_PROXIMITY_MIN_USD: down_proximity = (1.0 - down_nearest_dist / wall_threshold) * min(1.0, down_nearest_usd / (LIQ_PROXIMITY_MIN_USD * 5)) # 确定清算墙方向：上方墙近且大 → 看多(价格被吸上去清算空单)；下方近且大 → 看空 wall_direction = 0 # -1=看空, 0=中性, 1=看多 proximity = 0.0 if up_proximity > down_proximity * 1.15: wall_direction = 1 proximity = up_proximity elif down_proximity > up_proximity * 1.15: wall_direction = -1 proximity = down_proximity elif max(up_proximity, down_proximity) > 0: # 上下墙强度相近 → 震荡区间，方向中性但仍有邻近度 wall_direction = 0 proximity = (up_proximity + down_proximity) / 2 # 4. 多空比拥挤度 ls_ratio = 1.0 try: ls_data = self.okx.long_short_ratio(sym) if ls_data: ls_ratio = float(ls_data[0].get("ratio", 1.0)) except: pass crowd_sig = max(-1, min(1, -(ls_ratio - 1) / 0.4)) # 5. 综合评分 total_1h = liq_1h_long + liq_1h_short accel_sig = 0 if total_1h > 0: accel_sig = max(-1, min(1, -(liq_1h_long - liq_1h_short) / total_1h)) magnet_total = up_magnet + down_magnet magnet_sig = 0 if magnet_total > 0: magnet_sig = max(-1, min(1, (up_magnet - down_magnet) / magnet_total)) total_24h = liq_24h_long + liq_24h_short trend_sig = 0 if total_24h > 0: trend_sig = max(-1, min(1, -(liq_24h_long - liq_24h_short) / total_24h)) final = accel_sig * 0.4 + magnet_sig * 0.3 + crowd_sig * 0.2 + trend_sig * 0.1 score = max(-1, min(1, final)) down_prob = 50 + (-accel_sig) * 30 + (-magnet_sig) * 20 down_prob = max(10, min(90, down_prob)) up_prob = 100 - down_prob detail = { "hist": f"1h多清:${liq_1h_long/1e6:.2f}M 空清:${liq_1h_short/1e6:.2f}M", "zones": f"上方${up_total/1e6:.1f}M({up_nearest_dist*100:.1f}%) " f"下方${down_total/1e6:.1f}M({down_nearest_dist*100:.1f}%)", "prob": f"涨{up_prob:.0f}%/跌{down_prob:.0f}%", "ls_ratio": round(ls_ratio, 2), "final": round(final, 2), "wall_direction": wall_direction, "proximity": round(proximity, 3), } if proximity > 0.1: wall_label = {1: "↑吸向上", -1: "↓吸向下", 0: "↔震荡"}.get(wall_direction, "?") logger.info(f"[清算] {detail['hist']} | 密集区:{detail['zones']} | " f"概率:{detail['prob']} | 拥挤:{ls_ratio:.2f} | =>{score:+.2f}" f" | 墙邻近度:{proximity:.0%} {wall_label}") else: logger.info(f"[清算] {detail['hist']} | 密集区:{detail['zones']} | " f"概率:{detail['prob']} | 拥挤:{ls_ratio:.2f} | =>{score:+.2f}") return score, detail except Exception as e: logger.warning(f"清算地图错误: {e}") return 0, {} # ────────────────────────────────────────── # 主分析入口 # ────────────────────────────────────────── def analyze(self, symbol): cfg = INSTRUMENTS[symbol] inst_id = cfg["inst"] ccy = cfg.get("ccy", symbol) tick = self.okx.ticker(inst_id) price = float(tick.get("last", 0)) if price == 0: return 0, {"error": "no price", "direction": "WAIT", "price": 0, "total": 0, "scores": {}, "raw": {}, "funding_rate": 0} fr_data = self.okx.funding_rate(inst_id) fr = float(fr_data.get("fundingRate", 0)) ob = self.okx.orderbook(inst_id, 400) mp = self.deribit.max_pain(cfg.get("deribit")) if cfg.get("deribit") else None taker_vol_data = self.okx.taker_volume(ccy, "5m") oi_data = self.okx.oi_history(ccy, "5m") # 趋势 trend_score, t1h, t4h = self._trend_score(inst_id) # 订单簿深度失衡 asks = ob.get("asks", []) bids = ob.get("bids", []) ob_score_raw = 0 ask_near = sum(float(a[1]) for a in asks if float(a[0]) <= price * 1.005) bid_near = sum(float(b[1]) for b in bids if float(b[0]) >= price * 0.995) total_near = ask_near + bid_near if asks and bids: for pct, w in [(0.005, 0.5), (0.01, 0.3), (0.02, 0.2)]: a = sum(float(a[1]) for a in asks if float(a[0]) <= price * (1 + pct)) b = sum(float(b[1]) for b in bids if float(b[0]) >= price * (1 - pct)) t = a + b if t > 0: ob_score_raw += w * (b - a) / t self.ob_history[symbol].append(ob_score_raw) ob_avg = sum(self.ob_history[symbol]) / len(self.ob_history[symbol]) # OB反向：买盘多→短期消化压力→看空；卖盘多→支撑强→看多 ob_score = max(-1, min(1, -ob_avg * 2.5)) # 资金费率 # FR>0 → 多头付空头 → 多头拥挤 → 看空；FR<0 → 空头付多头 → 看多 fr_score = 0 if fr != 0: # 死区缩小到±0.00005（0.005%），超过即产生信号 if fr > 0.00005: fr_score = -min(1, fr / 0.0005) # 0.05%=满分看空 elif fr < -0.00005: fr_score = min(1, -fr / 0.0003) # -0.03%=满分看多 # 极端费率（>0.1%）加强 if abs(fr) > 0.001: fr_score = max(-1, min(1, fr_score * 1.5)) # Taker买卖比 taker_score = 0 if len(taker_vol_data) >= 6: r_buy = sum(float(d[1]) for d in taker_vol_data[:3]) r_sell = sum(float(d[2]) for d in taker_vol_data[:3]) p_buy = sum(float(d[1]) for d in taker_vol_data[3:6]) p_sell = sum(float(d[2]) for d in taker_vol_data[3:6]) r_total = r_buy + r_sell if r_total > 0: r_ratio = (r_buy - r_sell) / r_total p_total = p_buy + p_sell p_ratio = (p_buy - p_sell) / p_total if p_total > 0 else 0 accel = r_ratio - p_ratio taker_score = max(-1, min(1, r_ratio * 0.6 + max(-1, min(1, accel * 3)) * 0.4)) self.taker_history[symbol].append(taker_score) taker_avg = sum(self.taker_history[symbol]) / len(self.taker_history[symbol]) taker_score = max(-1, min(1, taker_avg)) # OI变化 oi_score = 0 if len(oi_data) >= 6: recent_oi = float(oi_data[0][1]) older_oi = float(oi_data[5][1]) if older_oi > 0: oi_change = (recent_oi - older_oi) / older_oi price_dir = 1 if taker_score > 0 else (-1 if taker_score < 0 else 0) if oi_change > 0: oi_score = price_dir * min(1, oi_change / 0.02) else: oi_score = -price_dir * min(1, abs(oi_change) / 0.02) self.oi_history[symbol].append(oi_score) oi_avg = sum(self.oi_history[symbol]) / len(self.oi_history[symbol]) oi_score = max(-1, min(1, oi_avg)) # 期权痛点 mp_score = 0 if mp is not None and mp > 0: mp_dist = (mp - price) / price # MP反向：痛点在上方→价格被压制回落→看空；痛点在下方→价格被托住→看多 mp_score = max(-1, min(1, -mp_dist / 0.03)) # 成交量Delta vd_score = 0 if len(taker_vol_data) >= 6: r_buy = sum(float(d[1]) for d in taker_vol_data[:3]) r_sell = sum(float(d[2]) for d in taker_vol_data[:3]) p_buy = sum(float(d[1]) for d in taker_vol_data[3:6]) p_sell = sum(float(d[2]) for d in taker_vol_data[3:6]) r_delta = r_buy - r_sell p_delta = p_buy - p_sell r_total = r_buy + r_sell dir_score = r_delta / r_total if r_total > 0 else 0 if p_delta != 0: accel = max(-2, min(2, (r_delta - p_delta) / abs(p_delta))) else: accel = 1 if r_delta > 0 else (-1 if r_delta < 0 else 0) vd_score = max(-1, min(1, dir_score * 0.5 + max(-1, min(1, accel * 0.3)) * 0.5)) logger.info(f"[{symbol}] VolDelta: recent={r_delta:.0f} prev={p_delta:.0f} " f"dir={dir_score:+.2f} accel={accel:+.2f} => {vd_score:+.2f}") # 宏观联动 btc_score = 0 try: # 用ETH作为加密货币市场宏观指标（BTC自身趋势由因子1TR捕获） macro_inst = "ETH-USDT-SWAP" if symbol == "BTC" else "BTC-USDT-SWAP" macro_candles = self.okx.candles(macro_inst, bar="1H", limit=10) if len(macro_candles) >= 10: macro_closes = [float(c[4]) for c in reversed(macro_candles[:10])] macro_diff = (sum(macro_closes[-5:]) / 5 - sum(macro_closes) / 10) / (sum(macro_closes) / 10) macro_trend = max(-1, min(1, macro_diff / 0.005)) else: macro_trend = 0 gold_signal = 0 try: paxg = http_get("https://api.binance.com/api/v3/klines", params={"symbol": "PAXGUSDT", "interval": "1h", "limit": 5}, timeout=5) paxg_data = paxg.json() if len(paxg_data) >= 5: gold_chg = (float(paxg_data[-1][4]) - float(paxg_data[0][1])) / float(paxg_data[0][1]) gold_signal = -max(-1, min(1, gold_chg / 0.005)) except: pass btc_score = max(-1, min(1, macro_trend * 0.7 + gold_signal * 0.3)) logger.info(f"[{symbol}] Macro:{macro_trend:+.2f}(via {macro_inst}) XAU:{gold_signal:+.2f} => {btc_score:+.2f}") except Exception as e: logger.warning(f"Macro error: {e}") # Gamma做市商行为 gamma_score = 0 gamma_detail = "" deribit_ccy = cfg.get("deribit") if deribit_ccy: try: net_gamma = self.deribit.net_gamma_exposure(deribit_ccy, price) mp = self.deribit.max_pain(deribit_ccy) mp_dist = (price - mp) / price if mp and mp > 0 else 0 if net_gamma > 100: gamma_score = max(-1, min(1, -mp_dist / 0.02)) gamma_detail = "正Gamma(稳定)" elif net_gamma < -100: gamma_strength = min(1, abs(net_gamma) / 5000) gamma_score = trend_score * gamma_strength if trend_score != 0 else 0 gamma_detail = "负Gamma(加速)" if mp and abs(mp_dist) < 0.02: magnet = max(-1, min(1, -mp_dist / 0.015)) gamma_score = gamma_score * 0.6 + magnet * 0.4 gamma_detail += "+磁吸" + ("↓" if mp_dist > 0 else "↑") if trend_score != 0 and gamma_score * trend_score < 0: gamma_score *= 0.5 gamma_detail += "(逆势减半)" gamma_score = max(-1, min(1, gamma_score)) logger.info(f"[{symbol}] Gamma: net={net_gamma:.0f} MP={mp} " f"dist={mp_dist:+.3f} {gamma_detail} => {gamma_score:+.2f}") except Exception as e: logger.warning(f"Gamma error: {e}") # 隐含波动率 iv_score = 0 try: eth_dvol, eth_dvol_prev = self.deribit.dvol("ETH") btc_dvol, _ = self.deribit.dvol("BTC") if eth_dvol and eth_dvol_prev and btc_dvol: dvol_chg = (eth_dvol - eth_dvol_prev) / eth_dvol_prev if eth_dvol_prev else 0 iv_dir = 0 if dvol_chg > 0: iv_dir = (-max(-1, min(1, dvol_chg / 0.05)) if trend_score >= 0 else max(-1, min(1, dvol_chg / 0.05))) else: iv_dir = trend_score * min(1, abs(dvol_chg) / 0.05) iv_ratio_sig = 0 if btc_dvol > 0: iv_ratio = eth_dvol / btc_dvol if iv_ratio > 1.5: iv_ratio_sig = -min(1, (iv_ratio - 1.5) / 0.3) elif iv_ratio < 1.1: iv_ratio_sig = min(1, (1.1 - iv_ratio) / 0.2) iv_score = max(-1, min(1, iv_dir * 0.6 + iv_ratio_sig * 0.4)) logger.info(f"[{symbol}] IV: ETH={eth_dvol:.1f}({dvol_chg:+.1%}) " f"BTC={btc_dvol:.1f} ratio={eth_dvol/btc_dvol:.2f} => {iv_score:+.2f}") except Exception as e: logger.warning(f"IV error: {e}") # 买/卖盘衰竭 exhaust_score = 0 if len(taker_vol_data) >= 6: r_total = sum(float(d[1]) + float(d[2]) for d in taker_vol_data[:3]) p_total = sum(float(d[1]) + float(d[2]) for d in taker_vol_data[3:6]) self.taker_abs_history[symbol].append(r_total) if p_total > 0: vol_decay = r_total / p_total if vol_decay < 1.0: decay_strength = min(1, (1.0 - vol_decay) / 0.6) exhaust_score = -trend_score * decay_strength if trend_score != 0 else 0 hist = list(self.taker_abs_history[symbol]) if len(hist) >= 3 and hist[-1] < hist[-2] < hist[-3]: exhaust_score = max(-1, min(1, exhaust_score * 1.3)) if abs(exhaust_score) > 0.2: logger.info(f"[{symbol}] TakerExhaust: decay={vol_decay:.2f} => {exhaust_score:+.2f}") # 清算方向 liq_cool_score, liq_detail = self._liquidation_bias(inst_id) liq_bias = liq_detail.get("final", 0) if liq_detail else 0 self.liq_history[symbol].append({"bias": liq_cool_score, "ts": time.time()}) # 短期过热均值回归 mr_score = 0 candles_5m = self.okx.candles(inst_id, bar="5m", limit=12) if len(candles_5m) >= 12: closes_5m = [float(c[4]) for c in reversed(candles_5m[:12])] chg_1h = (closes_5m[-1] - closes_5m[0]) / closes_5m[0] gains = [max(closes_5m[i] - closes_5m[i-1], 0) for i in range(1, len(closes_5m))] losses = [max(closes_5m[i-1] - closes_5m[i], 0) for i in range(1, len(closes_5m))] avg_gain = sum(gains) / len(gains) if gains else 0 avg_loss = sum(losses) / len(losses) if losses else 1 rsi = 100 - 100 / (1 + avg_gain / avg_loss) if avg_loss > 0 else 50 rsi_sig = (-min(1, (rsi - 60) / 25) if rsi > 60 else (min(1, (40 - rsi) / 25) if rsi < 40 else 0)) chg_sig = -max(-1, min(1, chg_1h / 0.02)) # MR反向：RSI超买/1h涨幅大 → 原来看空，现在作为反向指标取反 → 看多 mr_score = max(-1, min(1, -(rsi_sig * 0.6 + chg_sig * 0.4))) if abs(mr_score) > 0.3: logger.info(f"[{symbol}] MeanRevert: RSI={rsi:.0f} chg={chg_1h:+.2%} => {mr_score:+.2f}") # 聪明钱（四维） sm_score = 0 sm_detail = "" copy_score = 0.0 copy_detail = "" try: # 维度1: 合约vs现货taker分歧（30%） c_data = self.okx.taker_volume_ratio(symbol, "CONTRACTS", "5m", 6) s_data = self.okx.taker_volume_ratio(symbol, "SPOT", "5m", 6) taker_div = 0 if c_data and s_data: c_buy = sum(float(d[2]) for d in c_data) c_sell = sum(float(d[1]) for d in c_data) s_buy = sum(float(d[2]) for d in s_data) s_sell = sum(float(d[1]) for d in s_data) c_sig = max(-1, min(1, (c_buy / c_sell - 1) / 0.3 if c_sell > 0 else 0)) s_sig = max(-1, min(1, (s_buy / s_sell - 1) / 0.3 if s_sell > 0 else 0)) divergence = c_sig - s_sig taker_div = c_sig * 0.7 + max(-1, min(1, divergence)) * 0.3 # 维度2: 链上稳定币流入（20%，5分钟缓存） stablecoin_sig = getattr(self, '_sc_cache', 0) now_t = time.time() if now_t - getattr(self, '_sc_cache_ts', 0) > 300: try: sc_r = http_get( 'https://stablecoins.llama.fi/stablecoincharts/Ethereum?stablecoin=1', timeout=5) sc_data = sc_r.json() if len(sc_data) >= 2: curr_sc = sc_data[-1].get('totalCirculating', {}).get('peggedUSD', 0) prev_sc = sc_data[-2].get('totalCirculating', {}).get('peggedUSD', 0) if prev_sc > 0: stablecoin_sig = max(-1, min(1, (curr_sc - prev_sc) / prev_sc / 0.005)) self._sc_cache = stablecoin_sig self._sc_cache_ts = now_t except: pass # 维度3: 鲸鱼ETH余额变化（30%，5分钟缓存） whale_sig = getattr(self, '_whale_cache', 0) if now_t - getattr(self, '_whale_cache_ts', 0) > 300: try: whale_addrs = [ '0xf584F8728B874a6a5c7A8d4d387C9aae9172D621', # Jump Trading '0x00000000AE347930bD1E7B0F35588b92280f9e75', # Wintermute ] total_bal = 0 for addr in whale_addrs: wr = http_get('https://api.etherscan.io/v2/api', params={ 'chainid': 1, 'module': 'account', 'action': 'balance', 'address': addr, 'apikey': ETHERSCAN_API_KEY, }, timeout=5) wd = wr.json() if wd.get('status') == '1': total_bal += int(wd.get('result', '0')) / 1e18 prev_whale = getattr(self, '_whale_prev_bal', total_bal) if prev_whale > 0: whale_sig = max(-1, min(1, (total_bal - prev_whale) / prev_whale / 0.02)) self._whale_prev_bal = total_bal self._whale_cache = whale_sig self._whale_cache_ts = now_t except: pass # 维度4: ETH TVL变化（20%，5分钟缓存） tvl_sig = getattr(self, '_tvl_cache', 0) if now_t - getattr(self, '_tvl_cache_ts', 0) > 300: try: tvl_r = http_get('https://api.llama.fi/v2/historicalChainTvl/Ethereum', timeout=5) tvl_data = tvl_r.json() if len(tvl_data) >= 2: curr_tvl = tvl_data[-1].get('tvl', 0) prev_tvl = tvl_data[-2].get('tvl', 0) if prev_tvl > 0: tvl_sig = max(-1, min(1, (curr_tvl - prev_tvl) / prev_tvl / 0.03)) self._tvl_cache = tvl_sig self._tvl_cache_ts = now_t except: pass sm_score = max(-1, min(1, taker_div * 0.3 + stablecoin_sig * 0.2 + whale_sig * 0.3 + tvl_sig * 0.2)) sm_detail = "分歧%.2f 稳定币%.2f 鲸鱼%.2f TVL%.2f" % ( taker_div, stablecoin_sig, whale_sig, tvl_sig) # ── 链上跟单信号 ── if self.copy_trader: now_t = time.time() if now_t - self._copy_last_poll >= 60: self._copy_signal_cache = self.copy_trader.poll() self._copy_last_poll = now_t if self._copy_signal_cache: ct_result = self.copy_trader.get_score(self._copy_signal_cache) copy_score = ct_result["score"] copy_detail = ct_result["detail"] if abs(copy_score) > 0.2: logger.info("[%s] CopyTrade: %+.3f %s" % (symbol, copy_score, copy_detail)) # 跟单信号融入 SmartMoney (默认 10% 权重) if abs(copy_score) > 0.1: sm_score = sm_score * (1 - COPY_TRADE_WEIGHT) + copy_score * COPY_TRADE_WEIGHT sm_detail += " 跟单%.2f" % copy_score if abs(sm_score) > 0.1: logger.info("[%s] SmartMoney: %s => %+.2f" % (symbol, sm_detail, sm_score)) except Exception as e: logger.warning("SmartMoney error: %s" % e) # ── 散户点位 ── # 基于OI加权持仓倾向：OI↑+价格↑→散户追多；OI↑+价格↓→散户恐慌空 # 反向指标：散户过度拥挤时逆势 retail_score = 0.0 if len(oi_data) >= 6 and len(taker_vol_data) >= 6: oi_recent_3 = sum(float(d[1]) for d in oi_data[:3]) oi_prev_3 = sum(float(d[1]) for d in oi_data[3:6]) oi_chg = (oi_recent_3 - oi_prev_3) / oi_prev_3 if oi_prev_3 > 0 else 0 # 价格短期变化 candles_1h = self.okx.candles(inst_id, bar="15m", limit=4) if len(candles_1h) >= 4: px_chg = (price - float(candles_1h[-1][4])) / float(candles_1h[-1][4]) # OI↑+价格↑ → 散户追多，看空；OI↑+价格↓ → 散户恐慌，看多 retail_signal = oi_chg * px_chg # 同向=拥挤，反向=恐慌 # 费率极端加强信号 if abs(fr) > 0.001: retail_signal += -fr * 0.5 # 正费率=多头拥挤→看空 retail_score = max(-1, min(1, -retail_signal * 3.0)) if abs(retail_score) > 0.3: logger.info(f"[{symbol}] Retail: OIchg={oi_chg:+.3%} PXchg={px_chg:+.3%} FR={fr:+.4%} => {retail_score:+.2f}") # ── 做市商点位 ── # 使用ToxicFlowDetector的MM撤退检测 mm_score = 0.0 mm_detail = "" try: self.toxic_detector.feed_orderbook(asks, bids, price) mm_result = self.toxic_detector.detect_mm_retreat() if mm_result["retreating"]: # MM撤退=流动性下降，看空/看多取决于撤退方向 mm_score = -mm_result["score"] if mm_result.get("direction") == "sell" else mm_result["score"] mm_detail = mm_result.get("detail", "") if abs(mm_score) > 0.2: logger.info(f"[{symbol}] MM retreat: {mm_detail} => {mm_score:+.2f}") # 无撤退时用OB厚度作为做市商活跃度指标 else: ob_depth = (ask_near + bid_near) / (price * 2) if price > 0 else 0 # 厚度越大做市商越活跃，中性偏多 mm_score = max(0, min(0.3, ob_depth * 5 - 0.2)) except Exception as e: logger.warning(f"MM position error: {e}") # ── 清算触发点 ── # 追踪清算事件，识别清算密集价格区 liq_trigger_score = 0.0 try: liq_orders = self.okx.liquidations(inst_id) now_ts = time.time() for lo in liq_orders[:10]: liq_px = float(lo.get("bkPx", 0)) liq_side = lo.get("side", "") if liq_px > 0: self._liq_events[symbol].append({ "price": liq_px, "side": liq_side, "ts": now_ts}) # 统计清算密集区 if len(self._liq_events[symbol]) >= 5: events = list(self._liq_events[symbol]) recent = [e for e in events if now_ts - e["ts"] < 3600] # 1小时内 if recent: buy_liq = [e for e in recent if e["side"] == "buy"] sell_liq = [e for e in recent if e["side"] == "sell"] # 上方清算（空单被清）vs 下方清算（多单被清） above = [e for e in recent if e["price"] > price] below = [e for e in recent if e["price"] < price] # 价格靠近上方清算区→磁吸向上；靠近下方清算区→磁吸向下 if above: avg_above = sum(e["price"] for e in above) / len(above) dist_above = (avg_above - price) / price if dist_above < 0.02: # 2%以内 liq_trigger_score += 0.3 # 向上磁吸 if below: avg_below = sum(e["price"] for e in below) / len(below) dist_below = (price - avg_below) / price if dist_below < 0.02: liq_trigger_score -= 0.3 # 向下磁吸 liq_trigger_score = max(-1, min(1, liq_trigger_score)) if abs(liq_trigger_score) > 0.2: logger.info(f"[{symbol}] LiqTrigger: above={len(above)} below={len(below)} => {liq_trigger_score:+.2f}") except Exception as e: logger.warning(f"Liq trigger error: {e}") # ── 有毒订单流 ── toxic_score = 0.0 toxic_level = "normal" try: toxic_result = self.toxic_detector.analyze() toxic_score_raw = toxic_result.score # 0~1 toxic_level = toxic_result.level # 有毒流高→不确定性大→抑制开仓信号 if toxic_level == "high": toxic_score = -0.5 elif toxic_level == "medium": toxic_score = -0.2 if toxic_score_raw > 0.3: logger.info(f"[{symbol}] ToxicFlow: level={toxic_level} score={toxic_score_raw:.2f}") except Exception as e: logger.warning(f"ToxicFlow error: {e}") # ── 多时间框架共振 (1m/5m/15m) ── mtf_score = 0.0 mtf_detail = "" try: candles_1m = self.okx.candles(inst_id, bar="1m", limit=24) candles_5m = self.okx.candles(inst_id, bar="5m", limit=24) candles_15m = self.okx.candles(inst_id, bar="15m", limit=24) mtf_result = self.mtf_resonance.check(candles_1m, candles_5m, candles_15m, price) mtf_dir = mtf_result["resonance"] mtf_raw = mtf_result["score"] if mtf_dir == "LONG": mtf_score = mtf_raw elif mtf_dir == "SHORT": mtf_score = -mtf_raw mtf_detail = f"{mtf_dir}({mtf_raw:.2f}) " + \ f"1m:{mtf_result['details'].get('1m',{}).get('desc','?')} " + \ f"5m:{mtf_result['details'].get('5m',{}).get('desc','?')} " + \ f"15m:{mtf_result['details'].get('15m',{}).get('desc','?')}" if abs(mtf_score) > 0.2: logger.info(f"[{symbol}] MTF共振: {mtf_detail} => {mtf_score:+.2f}") except Exception as e: logger.warning(f"MTF resonance error: {e}") # ── 订单簿清算挂单占比 ── ob_liq_score = 0.0 ob_liq_detail = "" try: liq_for_ob = { "up_zones_total": sum(float(o.get("usd_value", 0)) for o in (liq_detail.get("up_zones", []) if isinstance(liq_detail, dict) else [])), "down_zones_total": sum(float(o.get("usd_value", 0)) for o in (liq_detail.get("down_zones", []) if isinstance(liq_detail, dict) else [])), } # Fallback: 用 liq_detail 中的 zones 信息提取 if liq_for_ob["up_zones_total"] == 0 and isinstance(liq_detail, dict): zones_str = liq_detail.get("zones", "") if zones_str: import re as _re up_m = _re.search(r'上方\$([\d.]+)M', zones_str) down_m = _re.search(r'下方\$([\d.]+)M', zones_str) if up_m: liq_for_ob["up_zones_total"] = float(up_m.group(1)) * 1e6 if down_m: liq_for_ob["down_zones_total"] = float(down_m.group(1)) * 1e6 ob_liq_result = self.ob_liq_analyzer.analyze( {"asks": asks, "bids": bids}, liq_for_ob, price) # 清算偏斜：上方清算多→价格被拉向清算区→看多；下方清算多→看空（反向信号） liq_skew = ob_liq_result.get("liq_skew", 0) ob_liq_score = -liq_skew # 上方清算多→负偏斜→取负=看多，下方多→正偏斜=取负=看空 ob_liq_detail = ob_liq_result.get("detail", "") if abs(ob_liq_score) > 0.2: logger.info(f"[{symbol}] OB清算比: 上方{ob_liq_result.get('up_liq_m',0)}M " f"下方{ob_liq_result.get('down_liq_m',0)}M " f"偏斜{liq_skew:+.2f} 真假墙{ob_liq_result.get('is_fake_wall','?')} " f"=> {ob_liq_score:+.2f}") except Exception as e: logger.warning(f"OB Liq analysis error: {e}") # ── 低杠杆检测 ── ll_score = 0.0 ll_detail = "" try: # OI变化与价格变化之比 → 近似杠杆方向 # OI增+价不动 → 杠杆积累（偏空）；OI减+价涨 → 去杠杆（偏多） if len(oi_data) >= 6: recent_oi = float(oi_data[0][1]) older_oi = float(oi_data[5][1]) if older_oi > 0: oi_chg_pct = (recent_oi - older_oi) / older_oi # 用5m蜡烛计算价格变化 px_5m_ago = float(candles_5m[5][4]) if len(candles_5m) > 5 else price px_chg_pct = (price - px_5m_ago) / px_5m_ago if px_5m_ago > 0 else 0 # OI减+价涨 → 去杠杆 → 健康（低杠杆信号）；OI增+价横 → 杠杆积累 if abs(oi_chg_pct) > 0 and abs(px_chg_pct) > 0: implied_lev = abs(oi_chg_pct) / abs(px_chg_pct) else: implied_lev = 10 self.low_lev_detector.feed(min(50, max(1, implied_lev * 100))) ll_score = self.low_lev_detector.score ll_detail = self.low_lev_detector.detail if abs(ll_score) > 0.2: logger.info(f"[{symbol}] LowLev: OIΔ{oi_chg_pct:+.2%} PxΔ{px_chg_pct:+.2%} {ll_detail} => {ll_score:+.2f}") except Exception as e: logger.warning(f"Low leverage error: {e}") # ── 前置过滤器门控 ── pre_filter_passed = True pre_filter_result = {"passed": True, "signals": [], "score": 0} try: fng_val = 50 # 估算OI值（最近6个数据点均值） oi_recent = float(oi_data[0][1]) if len(oi_data) > 0 else 0 pre_filter_result = self.pre_filter.check(price, fr, fng_val, oi_recent) if not pre_filter_result["passed"]: self._pre_filter_blocked = True if pre_filter_result.get("signals"): logger.info(f"[{symbol}] 前置过滤未通过: {' | '.join(pre_filter_result['signals'])}") pre_filter_passed = False except Exception as e: logger.warning(f"Pre-filter error: {e}") # ── 清算衰竭信号更新 ── liq_exhaust_score = 0.0 try: if isinstance(liq_detail, dict): hist_str = liq_detail.get("hist", "") import re as _re long_m = _re.search(r'多清:\$([\d.]+)M', hist_str) short_m = _re.search(r'空清:\$([\d.]+)M', hist_str) liq_1h_long = float(long_m.group(1)) * 1e6 if long_m else 0 liq_1h_short = float(short_m.group(1)) * 1e6 if short_m else 0 self.liq_exhaustion[symbol].feed(liq_1h_long, liq_1h_short) liq_exhaust_score = self.liq_exhaustion[symbol].score if liq_exhaust_score < -0.3: logger.info(f"[{symbol}] 清算衰竭: score={liq_exhaust_score:+.2f}") except Exception as e: logger.warning(f"Liq exhaustion error: {e}") # ── 清算墙邻近度 → 动态权重增强 ── liq_wall_direction = liq_detail.get("wall_direction", 0) if isinstance(liq_detail, dict) else 0 proximity = liq_detail.get("proximity", 0) if isinstance(liq_detail, dict) else 0 liq_weight = W_LIQ_COOLDOWN * (1.0 + proximity * LIQ_PROXIMITY_BOOST) if proximity > 0.15 and liq_wall_direction != 0: if liq_cool_score * liq_wall_direction > 0: # 清算评分方向与墙方向一致 → 强化 liq_cool_score = max(-1, min(1, liq_cool_score * 1.3)) liq_weight *= 1.2 else: # 评分与墙背离 → 磁吸部分反转（价格被吸向墙方向） blend = liq_cool_score * 0.6 + liq_wall_direction * 0.4 liq_cool_score = max(-1, min(1, blend)) if proximity > 0.1: logger.info(f"[{symbol}] 清算邻近度:{proximity:.0%} 墙方向:{liq_wall_direction:+d} " f"清算权重:{liq_weight:.3f}(基础{W_LIQ_COOLDOWN:.3f})") # ── 强趋势下均值回归因子压制 ── is_strong_trend = (t1h == t4h and t1h != 0) if is_strong_trend: for fname, fvar in [('exhaust', exhaust_score), ('gamma', gamma_score), ('mr', mr_score), ('mp', mp_score)]: if fvar * t1h < 0: if fname == 'exhaust': exhaust_score *= 0.5 elif fname == 'gamma': gamma_score *= 0.5 elif fname == 'mr': mr_score *= 0.5 elif fname == 'mp': mp_score *= 0.5 # ── 综合评分（归一化净差法）── mr_weight = W_MEAN_REVERT * (0.01 / 0.06 if t1h == t4h and t1h != 0 else 1.0) factor_pairs = [ (W_TREND, trend_score), (W_ORDERBOOK, ob_score), (W_FUNDING, fr_score), (W_TAKER, taker_score), (W_OI_CHANGE, oi_score), (W_MAXPAIN, mp_score), (W_VOL_DELTA, vd_score), (W_BTC_CORR, btc_score), (W_GAMMA, gamma_score), (W_IV, iv_score), (W_TAKER_EXHAUST, exhaust_score), (liq_weight, liq_cool_score), (mr_weight, mr_score), (W_SMART_MONEY, sm_score), (W_MTF_RESONANCE, mtf_score), # 新增: 多时间框架共振 (W_OB_LIQ_RATIO, ob_liq_score), # 新增: 订单簿清算占比 (W_LOW_LEVERAGE, ll_score), # 新增: 低杠杆检测 (W_LIQ_EXHAUSTION, liq_exhaust_score), # 新增: 清算衰竭 (W_RETAIL_POSITION, retail_score), (W_MM_POSITION, mm_score), (W_LIQ_TRIGGER, liq_trigger_score), (W_TOXIC_FLOW, toxic_score), ] bull_score = sum(w * max(0, s) for w, s in factor_pairs) bear_score = sum(w * abs(min(0, s)) for w, s in factor_pairs) total_weight = sum(w * abs(s) for w, s in factor_pairs) # ── 持仓管理用评分（忽略EX衰竭和清算衰竭，只影响开仓决策）── hold_pairs = [(w, s) for (w, s) in factor_pairs if w not in (W_TAKER_EXHAUST, W_LIQ_EXHAUSTION, W_TOXIC_FLOW)] h_bull = sum(w * max(0, s) for w, s in hold_pairs) h_bear = sum(w * abs(min(0, s)) for w, s in hold_pairs) h_weight = sum(w * abs(s) for w, s in hold_pairs) total_hold = round((h_bull - h_bear) / h_weight, 3) if h_weight > 0.01 else 0.0 if total_weight > 0.01: total = (bull_score - bear_score) / total_weight else: total = 0.0 total = round(max(-1, min(1, total)), 3) # ── 因子历史 & 动量 ── all_factors = { "trend": trend_score, "ob": ob_score, "fr": fr_score, "taker": taker_score, "oi": oi_score, "mp": mp_score, "vd": vd_score, "btc": btc_score, "gamma": gamma_score, "iv": iv_score, "exhaust": exhaust_score, "liq_cool": liq_cool_score, "mr": mr_score, "smart_money": sm_score, "copy_trade": copy_score, "mtf": mtf_score, "ob_liq": ob_liq_score, "low_lev": ll_score, "liq_ex": liq_exhaust_score, "retail": retail_score, "mm": mm_score, "liq_trigger": liq_trigger_score, "toxic": toxic_score, } self.factor_history[symbol].append(all_factors) self.score_history[symbol].append(total) momentum_adj = 0 if len(self.score_history[symbol]) >= 3: scores = list(self.score_history[symbol]) recent_3 = scores[-3:] if all(s > 0 for s in recent_3) and recent_3[-1] > recent_3[0]: momentum_adj = 0.03 elif all(s < 0 for s in recent_3) and recent_3[-1] < recent_3[0]: momentum_adj = -0.03 elif len(recent_3) >= 2 and recent_3[-1] * recent_3[-2] < 0: momentum_adj = -total * 0.1 flip_penalty = 0 if len(self.factor_history[symbol]) >= 2: prev = self.factor_history[symbol][-2] flips = sum(1 for k in all_factors if abs(all_factors[k]) > 0.2 and abs(prev.get(k, 0)) > 0.2 and all_factors[k] * prev.get(k, 0) < 0) if flips >= 4: flip_penalty = -abs(total) * 0.1 logger.info(f"[{symbol}] FactorFlip: {flips}个因子翻转，信号削弱") total = round(total + momentum_adj + flip_penalty, 3) # ── 前置过滤器门控：未通过→强制WAIT ── if not pre_filter_passed and abs(total) < SCORE_THRESHOLD + 0.05: pre_filter_blocked = True else: pre_filter_blocked = False direction = ("LONG" if total >= SCORE_THRESHOLD else ("SHORT" if total <= -SCORE_THRESHOLD else "WAIT")) # 前置过滤未通过且信号不够强时强制WAIT if pre_filter_blocked and direction != "WAIT": logger.info(f"[{symbol}] 前置过滤未通过(评分{pre_filter_result['score']:.2f}), 降级为WAIT") direction = "WAIT" # 趋势逆向惩罚 if direction == "LONG" and trend_score < 0: total = round(total * 0.7, 3) logger.info(f"[{symbol}] 趋势DOWN逆势LONG，削弱30% => {total:+.3f}") if total < SCORE_THRESHOLD: direction = "WAIT" elif direction == "SHORT" and trend_score > 0: total = round(total * 0.7, 3) logger.info(f"[{symbol}] 趋势UP逆势SHORT，削弱30% => {total:+.3f}") if abs(total) < SCORE_THRESHOLD: direction = "WAIT" # 清算地图逆向惩罚 if direction == "LONG" and liq_cool_score == -1: total = round(total * 0.8, 3) if total < SCORE_THRESHOLD: direction = "WAIT" elif direction == "SHORT" and liq_cool_score == 1: total = round(total * 0.8, 3) if abs(total) < SCORE_THRESHOLD: direction = "WAIT" # 1H+4H趋势必须同向（REQUIRE_TREND_ALIGN=True时） if direction != "WAIT" and REQUIRE_TREND_ALIGN: if t1h != t4h or t1h == 0: logger.info(f"[{symbol}] 1H({t1h:+d})/4H({t4h:+d})分歧，禁止开仓 → WAIT") direction = "WAIT" # 多因子确认 if direction != "WAIT": all_scores_list = [trend_score, ob_score, fr_score, taker_score, oi_score, mp_score, vd_score, btc_score, gamma_score, iv_score, exhaust_score, liq_cool_score, mr_score, sm_score, copy_score, mtf_score, ob_liq_score, ll_score, liq_exhaust_score, retail_score, mm_score, liq_trigger_score, toxic_score] dir_sign = 1 if direction == "LONG" else -1 agree = sum(1 for s in all_scores_list if s * dir_sign > 0.15) if agree < MIN_FACTORS_AGREE: logger.info(f"[{symbol}] 因子确认不足: {agree}/{MIN_FACTORS_AGREE}同向，降级为WAIT") direction = "WAIT" # 波动率过滤 if direction != "WAIT" and len(candles_5m) >= 6: atr = sum(float(c[2]) - float(c[3]) for c in candles_5m[:6]) / 6 if atr / price < 0.001: logger.info(f"[{symbol}] 波动率过低，降级为WAIT") direction = "WAIT" ob_skew_val = round((ask_near - bid_near) / total_near, 3) if total_near > 0 else 0 details = { "symbol": symbol, "inst_id": inst_id, "price": price, "funding_rate": fr, "max_pain": mp, "ob_skew": ob_skew_val, "scores": { "trend": trend_score, "orderbook": ob_score, "funding": fr_score, "taker": taker_score, "oi": oi_score, "maxpain": mp_score, "vol_delta": vd_score, "btc_corr": btc_score, "gamma": gamma_score, "iv": iv_score, "exhaust": exhaust_score, "liq_cool": liq_cool_score, "mean_revert": mr_score, "smart_money": sm_score, "copy_trade": copy_score, "mtf": mtf_score, "ob_liq": ob_liq_score, "low_lev": ll_score, "liq_ex": liq_exhaust_score, "retail": retail_score, "mm": mm_score, "liq_trigger": liq_trigger_score, "toxic": toxic_score, }, "raw": { "t1h": t1h, "t4h": t4h, "liq_bias": liq_bias, "liq_detail": liq_detail, "ob": round(ob_score_raw, 3), "ob_avg": round(ob_avg, 2), "ob_skew": ob_skew_val, "ask_near": round(ask_near, 1), "bid_near": round(bid_near, 1), "taker": round(taker_score, 3), "taker_avg": round(taker_avg, 2), "oi": round(oi_score, 3), "oi_avg": round(oi_avg, 2), "fr_raw": round(fr * 100, 4), "momentum_adj": round(momentum_adj, 3), "flip_penalty": round(flip_penalty, 3), "sm_detail": sm_detail, "copy_detail": copy_detail, "mtf_detail": mtf_detail, "ob_liq_detail": ob_liq_detail, "ll_detail": ll_detail, "pre_filter": pre_filter_result, }, "total": total, "total_hold": total_hold, "direction": direction, } liq_prox_info = f" 邻近度:{proximity:.0%}" if proximity > 0.1 else "" logger.info(f"[{symbol}] ${price:,.2f} | FR:{fr*100:.4f}% | MP:{mp} | Liq:{liq_cool_score:+.2f} " f"(w={liq_weight:.3f}{liq_prox_info}) 前置过滤:{'✓' if pre_filter_passed else '✗'}") logger.info(f"[{symbol}] TR:{trend_score:+d} OB:{ob_score:+.2f} FR:{fr_score:+.2f} " f"TK:{taker_score:+.2f} OI:{oi_score:+.2f} MP:{mp_score:+.2f} " f"VD:{vd_score:+.2f} BTC:{btc_score:+.2f} GM:{gamma_score:+.2f} IV:{iv_score:+.2f} " f"EX:{exhaust_score:+.2f} LC:{liq_cool_score:+.2f} MR:{mr_score:+.2f} SM:{sm_score:+.2f} CT:{copy_score:+.2f} " f"MTF:{mtf_score:+.2f} OBL:{ob_liq_score:+.2f} LL:{ll_score:+.2f} LEX:{liq_exhaust_score:+.2f} " f"RT:{retail_score:+.2f} MM:{mm_score:+.2f} LT:{liq_trigger_score:+.2f} TX:{toxic_score:+.2f} " f"mom:{momentum_adj:+.3f} flip:{flip_penalty:+.3f} => {total:+.3f} -> {direction}") # 写入 MySQL 因子快照 if DB_ENABLED: try: from storage.mysql_client import insert_factor_snapshot insert_factor_snapshot(symbol, price, direction, total, details.get('scores', {}), details.get('raw', {})) except Exception: pass return total, details