qihuo/modules/trading/trade_notify.py

# Copyright (c) 2025-2026 马建军. All rights reserved.
# 专有软件 — 未经授权禁止复制、传播、转售。
# 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。
# 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md

"""交易事件推送：企业微信 + AI 分析。"""
from __future__ import annotations

from typing import Callable, Optional

from modules.core.contract_specs import calc_position_metrics, get_contract_spec
from modules.trading.sl_tp_guard import monitor_source_label
from modules.notify.wechat_notify import format_close_done, format_key_open_success, format_open_success


def _risk_amount(capital: float, risk_percent: float) -> Optional[float]:
    try:
        return round(float(capital) * float(risk_percent) / 100.0, 2)
    except (TypeError, ValueError):
        return None


def notify_manual_open_filled(
    *,
    send_wechat: Callable[[str], None],
    get_setting: Callable[[str, str], str],
    mode_label: str,
    sym: str,
    symbol_name: str,
    direction: str,
    entry: float,
    sl: Optional[float],
    tp: Optional[float],
    lots: int,
    capital: float,
    order_id: str = "",
    trailing_be: bool = False,
    be_tick_buffer: int = 2,
    schedule_ai_fn=None,
    db_path: str = "",
) -> None:
    if not sl:
        return
    spec = get_contract_spec(sym)
    tick = float(spec.get("tick_size") or 1.0)
    try:
        rp = float(get_setting("risk_percent", "1") or 1)
    except (TypeError, ValueError):
        rp = 1.0
    metrics = calc_position_metrics(direction, entry, sl, tp or entry, lots, entry, capital, sym)
    msg = format_open_success(
        symbol_name=symbol_name,
        symbol=sym,
        direction=direction,
        mode_label=mode_label,
        order_id=order_id,
        entry=entry,
        stop_loss=float(sl),
        take_profit=float(tp) if tp else None,
        lots=lots,
        capital=capital,
        margin=metrics.get("margin"),
        margin_pct=metrics.get("position_pct"),
        risk_percent=rp,
        risk_amount=_risk_amount(capital, rp),
        trailing_be=trailing_be,
        be_tick_buffer=be_tick_buffer,
        tick_size=tick,
        source="期货下单",
    )
    send_wechat(msg)
    if schedule_ai_fn and db_path:
        schedule_ai_fn(
            db_path=db_path,
            get_setting_fn=get_setting,
            kind="open",
            title=f"{symbol_name or sym} 开仓",
            payload={
                "symbol": sym,
                "direction": direction,
                "entry": entry,
                "stop_loss": sl,
                "take_profit": tp,
                "lots": lots,
                "capital": capital,
            },
            send_wechat_fn=None,
        )


def notify_key_breakout_open(
    *,
    send_wechat: Callable[[str], None],
    get_setting: Callable[[str, str], str],
    mode_label: str,
    row: dict,
    break_side: str,
    bar_time: str,
    direction: str,
    entry: float,
    sl: float,
    tp: float,
    lots: int,
    capital: float,
    order_id: str = "",
    schedule_ai_fn=None,
    db_path: str = "",
) -> None:
    sym = row.get("symbol") or ""
    name = row.get("symbol_name") or sym
    trailing_be = bool(int(row.get("trailing_be") or 0))
    try:
        rp = float(get_setting("risk_percent", "1") or 1)
        be_buf = int(float(get_setting("trailing_be_tick_buffer", "2") or 2))
    except (TypeError, ValueError):
        rp, be_buf = 1.0, 2
    spec = get_contract_spec(sym)
    tick = float(spec.get("tick_size") or 1.0)
    metrics = calc_position_metrics(direction, entry, sl, tp, lots, entry, capital, sym)
    msg = format_key_open_success(
        symbol_name=name,
        symbol=sym,
        monitor_type=row.get("monitor_type") or "",
        trade_mode=row.get("trade_mode") or "顺势",
        bar_time=bar_time,
        break_side=break_side,
        direction=direction,
        mode_label=mode_label,
        order_id=order_id,
        entry=entry,
        stop_loss=sl,
        take_profit=tp,
        lots=lots,
        capital=capital,
        margin=metrics.get("margin"),
        margin_pct=metrics.get("position_pct"),
        risk_percent=rp,
        risk_amount=_risk_amount(capital, rp),
        trailing_be=trailing_be,
        be_tick_buffer=be_buf,
        tick_size=tick,
    )
    send_wechat(msg)
    if schedule_ai_fn and db_path:
        schedule_ai_fn(
            db_path=db_path,
            get_setting_fn=get_setting,
            kind="key_open",
            title=f"{name} 关键位开仓",
            payload={
                "monitor_type": row.get("monitor_type"),
                "trade_mode": row.get("trade_mode"),
                "break_side": break_side,
                "entry": entry,
                "stop_loss": sl,
                "take_profit": tp,
                "lots": lots,
            },
        )


def notify_trade_log_close(
    *,
    send_wechat: Callable[[str], None],
    get_setting: Callable[[str, str], str],
    mode_label: str,
    capital: float,
    sym: str,
    symbol_name: str,
    direction: str,
    entry: float,
    close_price: float,
    sl: Optional[float],
    tp: Optional[float],
    lots: float,
    pnl_net: float,
    equity_after: Optional[float],
    holding_minutes: int,
    result: str,
    monitor_type: str = "",
    schedule_ai_fn=None,
    db_path: str = "",
) -> None:
    src = monitor_source_label(monitor_type) if monitor_type else "期货下单"
    note = ""
    if tp and sl:
        if direction == "long":
            if close_price > tp or close_price < sl:
                note = "成交价不在计划止盈/止损带内（可能为手动或其他类型平仓）"
        else:
            if close_price < tp or close_price > sl:
                note = "成交价不在计划止盈/止损带内（可能为手动或其他类型平仓）"
    msg = format_close_done(
        symbol_name=symbol_name,
        symbol=sym,
        mode_label=mode_label,
        direction=direction,
        result=result,
        pnl_net=pnl_net,
        equity_after=equity_after,
        capital=capital,
        entry=entry,
        close_price=close_price,
        stop_loss=sl,
        take_profit=tp,
        lots=lots,
        holding_minutes=holding_minutes,
        note=note,
    )
    send_wechat(msg)
    if schedule_ai_fn and db_path:
        schedule_ai_fn(
            db_path=db_path,
            get_setting_fn=get_setting,
            kind="close",
            title=f"{symbol_name or sym} 平仓",
            payload={
                "source": src,
                "result": result,
                "pnl_net": pnl_net,
                "entry": entry,
                "close_price": close_price,
                "lots": lots,
            },
        )