# -*- coding: utf-8 -*- """ 报告格式化 - format_analysis : 详细因子报告（有交易时发送） - format_cycle : 简洁周期报告 """ from datetime import datetime def format_analysis(result): """精简报告 — 非零因子详细，零分因子一行带过""" lines = [f"📊 ({datetime.now().strftime('%H:%M:%S')}) 💰${result['equity']} 可用${result['available']}"] dd = result.get('drawdown_pct', 0) if dd > 1: lines.append(f"⚠️ 回撤 {dd:.1f}%") if result.get("positions"): for p in result["positions"]: em = "🟢多" if p["side"] == "long" else "🔴空" pnl = p["upl"] pnl_s = "+$%.4f" % pnl if pnl >= 0 else "-$%.4f" % abs(pnl) lines.append( f"{em} {p['qty']}张 ${p['entry']:.2f}→${p.get('last',0):.2f} " f"{pnl_s} {p.get('lever','?')}x" ) for sig in result.get("signals", []): sig_dir = sig.get("direction", "") if not sig_dir or sig_dir == "PAUSED": # PAUSED / 无direction信号，简单显示action/reason action = sig.get("action", "") reason = sig.get("reason", "") lines.append(f"⏸ {action}: {reason}" if action else f"⏸ {reason}") if "trade" in sig: t = sig["trade"] if t.get("success"): lines.append(f" ✅ 开仓 {t.get('size','')}张 {t.get('leverage','')}x TP${t.get('tp','')} SL${t.get('sl','')}") elif t.get("action") in ("BLOCKED", "COOLDOWN"): lines.append(f" ⏸ {t.get('reason', '')}") continue em = {"LONG": "🟢", "SHORT": "🔴", "WAIT": "⏳"}.get(sig_dir, "") f = sig["factors"] raw = sig.get("raw", {}) price = sig['price'] lines.append(f"\n{em} ETH ${price:,.2f} → {sig['direction']} ({sig['score']:+.3f})") factor_info = [ ("趋势", f['trend'], 0.04, f"1H:{'↑' if raw.get('t1h',0)>0 else '↓' if raw.get('t1h',0)<0 else '→'} " f"4H:{'↑' if raw.get('t4h',0)>0 else '↓' if raw.get('t4h',0)<0 else '→'}"), ("挂单", f['orderbook'], 0.10, f"卖{raw.get('ask_near',0):.0f}/买{raw.get('bid_near',0):.0f} 偏度{raw.get('ob_skew',0):+.2f}"), ("成交", f['taker'], 0.02, f"本次{raw.get('taker',0):+.2f} 均值{raw.get('taker_avg',0):.2f}"), ("持仓量", f['oi'], 0.01, f"本次{raw.get('oi',0):+.2f} 均值{raw.get('oi_avg',0):.2f}"), ("费率", f['funding'], 0.10, f"{raw.get('fr_raw',0):.4f}%"), ("痛点", f['maxpain'], 0.08, f"MP:{sig.get('max_pain') or '无'}"), ("量变", f.get('vol_delta', 0), 0.03, ""), ("大盘", f.get('btc_corr', 0), 0.02, ""), ("Gamma", f.get('gamma', 0), 0.02, ""), ("波动", f.get('iv', 0), 0.02, ""), ("衰竭", f.get('exhaust', 0), 0.18, ""), ("清算", f.get('liq_cool', 0), 0.04, f"{raw.get('liq_detail',{}).get('prob','')} {raw.get('liq_detail',{}).get('hist','')}" if raw.get('liq_detail') else ""), ("回归", f.get('mean_revert', 0), 0.01, ""), ("聪明钱", f.get('smart_money', 0), 0.03, raw.get('sm_detail', '')), ("MTF共振", f.get('mtf', 0), 0.10, raw.get('mtf_detail', '')), ("OB清算比", f.get('ob_liq', 0), 0.08, raw.get('ob_liq_detail', '')), ("低杠杆", f.get('low_lev', 0), 0.04, raw.get('ll_detail', '')), ("清算衰竭", f.get('liq_ex', 0), 0.06, ""), ("散户点位", f.get('retail', 0), 0.05, ""), ("做市商", f.get('mm', 0), 0.05, ""), ("清算磁吸", f.get('liq_trigger', 0), 0.06, ""), ("有毒流", f.get('toxic', 0), 0.06, ""), ] active = [] inactive_names = [] for name, score, weight, detail in factor_info: if abs(score) > 0.05: contrib = score * weight word = "📈" if score > 0 else "📉" line = f" {word} {name}: {score:+.2f}×{weight}={contrib:+.3f}" if detail: line += f" ({detail})" active.append(line) else: inactive_names.append(name) if active: lines.append("\n🔥 有信号:") lines.extend(active) if inactive_names: lines.append(f"⚪ 中性: {', '.join(inactive_names)}") mom = raw.get('momentum_adj', 0) flip = raw.get('flip_penalty', 0) if mom != 0 or flip != 0: lines.append(f"⚙️ 动量{mom:+.3f} 翻转{flip:+.3f}") # 幌骗检测状态 sp = sig.get("spoof") if sp and sp.get("risk_level") != "normal": risk_em = "‼️" if sp["risk_level"] == "danger" else "⚠️" wall_s = f"{sp['wall_side']}墙" if sp.get("wall_side") else "" score_s = f"{sp['score']:+.2f}" if sp.get("score") else "0.00" lines.append( f"\n{risk_em} 订单簿幌骗: {sp['risk_level'].upper()} " f"{wall_s} 评分{score_s}" ) if sp.get("wall_active"): lines.append(f" 大{wall_s}监控中，撤单预警") if sp.get("last_alert"): la = sp["last_alert"] lines.append( f" 最近告警: {la.get('withdraw_pct',0):.0f}%撤单 " f"预期{la.get('direction','')} 耗时{la.get('elapsed',0):.0f}s" ) elif sig.get("action","").startswith("BLOCKED_SPOOF"): lines.append(f"\n⛔ {sig['action']}") if "trade" in sig: t = sig["trade"] if t.get("success"): lines.append(f"\n✅ 开仓 {t['size']}张 {t['leverage']}x TP${t['tp']} SL${t['sl']}") elif t.get("action") in ("BLOCKED", "COOLDOWN"): lines.append(f"⏸ {t.get('reason', '')}") return "\n".join(lines) def format_gold_analysis(result): """XAU 黄金分析报告 — 精简版""" lines = [f"🥇 XAU黄金 ({datetime.now().strftime('%H:%M:%S')}) ${result['price']:,.1f}"] direction = result.get("direction", "WAIT") score = result.get("total", 0) em = {"LONG": "🟢", "SHORT": "🔴", "WAIT": "⏳"}.get(direction, "") lines.append(f"{em} → {direction} (评分 {score:+.3f})") scores = result.get("scores", {}) raw = result.get("raw", {}) # 分类展示因子 sections = [ ("趋势与动量", [("XTR", "趋势"), ("XMF", "多周期共振"), ("XMA", "均线排列"), ("XMO", "动量")]), ("量仓分析", [("XVL", "成交量"), ("XOI", "持仓变化"), ("XCF", "COT持仓"), ("XVX", "量价背离")]), ("宏观驱动", [("XRD", "实际利率"), ("XDX", "美元反向"), ("XEF", "ETF资金流"), ("XCR", "信用风险")]), ("技术指标", [("XRSI", "RSI回归"), ("XBB", "布林带"), ("XEX", "量能衰竭")]), ] for section_name, factors in sections: active = [(tag, name, scores.get(tag, 0)) for tag, name in factors if abs(scores.get(tag, 0)) > 0.1] if active: lines.append(f"\n{section_name}:") for tag, name, s in active: arrow = "📈" if s > 0 else "📉" lines.append(f" {arrow} {name}: {s:+.2f}") # 宏观数据 if raw: extras = [] if raw.get("t1h", 0) != 0 or raw.get("t4h", 0) != 0: t1h = "↑" if raw["t1h"] > 0 else ("↓" if raw["t1h"] < 0 else "→") t4h = "↑" if raw["t4h"] > 0 else ("↓" if raw["t4h"] < 0 else "→") t1d = "↑" if raw.get("t1d", 0) > 0 else ("↓" if raw.get("t1d", 0) < 0 else "→") extras.append(f"趋势 1H:{t1h} 4H:{t4h} 1D:{t1d}") if raw.get("cot_net_pct"): extras.append(f"COT净多:{raw['cot_net_pct']:.1f}%") if extras: lines.append(" " + " | ".join(extras)) return "\n".join(lines) def format_cycle(result): """简洁报告""" lines = [f"📊 ({datetime.now().strftime('%H:%M')}) 💰${result['equity']}"] if result.get("positions"): for p in result["positions"]: em = "🟢" if p["side"] == "long" else "🔴" lines.append(f"{em}{p['symbol']} {p['side']} {p['qty']}张 P&L:${p['upl']:.4f}") for sig in result.get("signals", []): f = sig["factors"] lines.append(f"{sig['symbol']} ${sig['price']:,.2f} → {sig['direction']} ({sig['score']:+.3f})") lines.append( f" TR:{f['trend']:+d} OB:{f.get('orderbook',0):+.1f} FR:{f.get('funding',0):+.1f} " f"TK:{f.get('taker',0):+.1f} OI:{f.get('oi',0):+.1f} MP:{f.get('maxpain',0):+.1f} " f"VD:{f.get('vol_delta',0):+.1f} BTC:{f.get('btc_corr',0):+.1f} GM:{f.get('gamma',0):+.1f} " f"IV:{f.get('iv',0):+.1f} EX:{f.get('exhaust',0):+.1f} LC:{f.get('liq_cool',0):+.1f} " f"MR:{f.get('mean_revert',0):+.1f} SM:{f.get('smart_money',0):+.1f} " f"MTF:{f.get('mtf',0):+.1f} OBL:{f.get('ob_liq',0):+.1f} LL:{f.get('low_lev',0):+.1f} LEX:{f.get('liq_ex',0):+.1f}" ) return "\n".join(lines)