# Copyright (c) 2025-2026 马建军. All rights reserved. # 专有软件 — 未经授权禁止复制、传播、转售。 # 严禁用于:带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。 # 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md """企业微信推送:开仓 / 平仓 / 关键位 结构化消息。""" from __future__ import annotations from typing import Optional def _dir_label(direction: str) -> str: d = (direction or "long").strip().lower() return "多头(long)" if d == "long" else "空头(short)" def _dir_emoji(direction: str) -> str: d = (direction or "long").strip().lower() return "📈" if d == "long" else "📉" def fmt_holding(minutes: int) -> str: m = max(0, int(minutes or 0)) if m >= 1440: return f"{m // 1440}天{m % 1440 // 60}小时{m % 60}分钟" if m >= 60: return f"{m // 60}小时{m % 60}分钟" return f"{m}分钟" def calc_rr(entry: float, sl: float, tp: float, direction: str) -> Optional[float]: try: entry_f, sl_f, tp_f = float(entry), float(sl), float(tp) except (TypeError, ValueError): return None risk = abs(entry_f - sl_f) if risk <= 0: return None reward = (tp_f - entry_f) if direction == "long" else (entry_f - tp_f) if reward <= 0: return None return round(reward / risk, 2) def format_open_success( *, symbol_name: str, symbol: str, direction: str, mode_label: str, order_id: str = "", entry: float, stop_loss: float, take_profit: Optional[float], lots: int, capital: float, margin: Optional[float], margin_pct: Optional[float], risk_percent: float, risk_amount: Optional[float], trailing_be: bool = False, be_tick_buffer: int = 2, tick_size: float = 1.0, source: str = "期货下单", extra_lines: Optional[list[str]] = None, ) -> str: """正常 / 关键位开仓成功推送。""" name = symbol_name or symbol emoji = _dir_emoji(direction) rr = calc_rr(entry, stop_loss, take_profit, direction) if take_profit else None lines = [ f"{emoji} {name} 开仓成功", f"💼 账户:{mode_label}", "", "🧾 订单基础信息", f"📌 来源:{source}", ] if order_id: lines.append(f"🔖 委托号:{order_id}") lines.extend([ f"📈 方向:{_dir_label(direction)}", f"⚠ 单笔风控:{risk_percent:g}%" + (f"≈{risk_amount:.2f}元" if risk_amount is not None else ""), "", "📊 仓位配置", f"账户权益:{capital:.2f} 元", f"开仓手数:{lots} 手", ]) if margin is not None: lines.append(f"占用保证金:{margin:.2f} 元") if margin_pct is not None: lines.append(f"仓位占比:{margin_pct:.2f}%") lines.extend(["", "🎯 价位 & 盈亏比", f"开仓价:{entry:g}", f"止损价:{stop_loss:g}"]) if take_profit is not None: lines.append(f"止盈价:{take_profit:g}") if rr is not None: lines.append(f"计划盈亏比:RR {rr:g} : 1") if trailing_be: be_px = entry - be_tick_buffer * tick_size if direction == "long" else entry + be_tick_buffer * tick_size lines.append(f"移动保本:1.0R → {be_px:g}(缓冲 {be_tick_buffer} 跳)") lines.extend(["", "📌 状态", "✅ 已进入下单监控,本地 SL/TP 守护"]) if extra_lines: lines.extend(extra_lines) return "\n".join(lines) def format_key_open_success( *, symbol_name: str, symbol: str, monitor_type: str, trade_mode: str, bar_time: str, break_side: str, **kwargs, ) -> str: side_label = "向上突破" if break_side == "upper" else "向下突破" extra = [ "", "📎 关键位触发", f"类型:{monitor_type}", f"模式:{trade_mode} · {side_label}", f"5m 收盘:{bar_time}", ] source = f"{monitor_type}·{trade_mode}" return format_open_success( symbol_name=symbol_name, symbol=symbol, source=source, extra_lines=extra, **kwargs, ) def format_close_done( *, symbol_name: str, symbol: str, mode_label: str, direction: str, result: str, pnl_net: float, equity_after: Optional[float], capital: float, entry: float, close_price: float, stop_loss: Optional[float], take_profit: Optional[float], lots: float, holding_minutes: int = 0, order_id: str = "", note: str = "", ) -> str: """平仓完成推送。""" name = symbol_name or symbol emoji = "📈" if pnl_net >= 0 else "📉" pnl_sign = "+" if pnl_net >= 0 else "" lines = [ f"{emoji} {name} 平仓完成", f"💼 账户:{mode_label}", "", "🧾 平仓概要", ] if order_id: lines.append(f"🔖 平仓单号:{order_id}") lines.extend([ f"📌 方向:{_dir_label(direction)}", f"📌 平仓结果:{result}", f"💰 本单净盈亏:{pnl_sign}{pnl_net:.2f} 元", f"⏱ 持仓时长:{fmt_holding(holding_minutes)}", f"💵 账户权益:{equity_after if equity_after is not None else capital:.2f} 元", "", "🎯 价位(计划)", f"开仓价:{entry:g}", f"平仓价:{close_price:g}", ]) if take_profit is not None: lines.append(f"止盈价:{take_profit:g}") if stop_loss is not None: lines.append(f"止损价:{stop_loss:g}") if note: lines.extend(["", "📎 备注", note]) return "\n".join(lines)