crypto_monitor/order_monitor_display_lib.py

"""实时持仓展示：开仓快照盈亏比、交易所止损是否已保本。"""
from __future__ import annotations

from typing import Any, Callable, Optional


def _positive_float(value: Any) -> Optional[float]:
    try:
        v = float(value)
        return v if v > 0 else None
    except (TypeError, ValueError):
        return None


def snapshot_stop_loss(initial_stop_loss: Any, stop_loss: Any) -> Optional[float]:
    """展示盈亏比时优先用开仓时止损快照。"""
    sl = _positive_float(initial_stop_loss)
    if sl is not None:
        return sl
    return _positive_float(stop_loss)


def snapshot_rr(
    calc_rr_ratio_fn: Callable[..., Optional[float]],
    direction: str,
    trigger_price: Any,
    initial_stop_loss: Any,
    stop_loss: Any,
    take_profit: Any,
) -> Optional[float]:
    entry = _positive_float(trigger_price)
    sl = snapshot_stop_loss(initial_stop_loss, stop_loss)
    tp = _positive_float(take_profit)
    if entry is None or sl is None or tp is None:
        return None
    return calc_rr_ratio_fn(direction or "long", entry, sl, tp)


def tpsl_slot_trigger_price(slot: Any) -> Optional[float]:
    if not isinstance(slot, dict):
        return None
    for key in ("trigger_price", "trigger_display"):
        v = _positive_float(slot.get(key))
        if v is not None:
            return v
    return None


def stop_is_profit_protecting(direction: str, entry_price: Any, stop_loss: Any) -> bool:
    """
    止损是否已在盈利侧（保本/锁盈），不再适用「开仓盈亏比」风控。
    做空：止损 < 成交价；做多：止损 > 成交价。
    """
    entry = _positive_float(entry_price)
    sl = _positive_float(stop_loss)
    if entry is None or sl is None:
        return False
    d = (direction or "long").strip().lower()
    if d == "short":
        return sl < entry
    return sl > entry


def tpsl_update_passes_rr_gate(
    direction: str,
    entry_price: Any,
    stop_loss: Any,
    take_profit: Any,
    min_rr: float,
    calc_rr_ratio_fn: Callable[..., Optional[float]],
) -> tuple[bool, Optional[str]]:
    """持仓委托改价：盈利侧止损跳过最低盈亏比；否则按开仓价几何校验。"""
    if stop_is_profit_protecting(direction, entry_price, stop_loss):
        return True, None
    rr = calc_rr_ratio_fn(direction or "long", entry_price, stop_loss, take_profit)
    if rr is not None and rr >= float(min_rr):
        return True, None
    rr_txt = f"{rr:.4f}" if rr is not None else "无法计算"
    return False, f"计划盈亏比 {rr_txt}:1 低于最低要求 {min_rr}:1（盈利侧保本止损不受此限）"


def is_sl_breakeven_secured(direction: str, entry_price: Any, exchange_sl_price: Any) -> bool:
    """
    交易所当前止损相对开仓成交价是否已保本。
    做多：止损 >= 成交价；做空：止损 <= 成交价。
    """
    entry = _positive_float(entry_price)
    sl = _positive_float(exchange_sl_price)
    if entry is None or sl is None:
        return False
    d = (direction or "long").strip().lower()
    if d == "short":
        return sl <= entry
    return sl >= entry


def sl_breakeven_from_exchange_tpsl(
    direction: str,
    entry_price: Any,
    exchange_tpsl: Any,
) -> bool:
    if not isinstance(exchange_tpsl, dict):
        return False
    sl_px = tpsl_slot_trigger_price(exchange_tpsl.get("sl"))
    if sl_px is None:
        return False
    return is_sl_breakeven_secured(direction, entry_price, sl_px)


def enrich_order_display_fields(item: dict[str, Any], calc_rr_ratio_fn: Callable[..., Optional[float]]) -> dict[str, Any]:
    item["rr_ratio"] = snapshot_rr(
        calc_rr_ratio_fn,
        item.get("direction") or "long",
        item.get("trigger_price"),
        item.get("initial_stop_loss"),
        item.get("stop_loss"),
        item.get("take_profit"),
    )
    return item


def apply_order_live_price_display(
    payload: dict[str, Any],
    symbol: Any,
    ticker_price: Any,
    exchange_mark_price: Any,
    format_price_fn: Callable[[Any, Any], str],
) -> dict[str, Any]:
    """标记价/现价展示：与交易所 price_to_precision 对齐，避免前端 toFixed(8)。"""
    px_for_fmt = ticker_price
    mark_raw = exchange_mark_price
    if mark_raw is not None:
        try:
            px_for_fmt = float(mark_raw)
        except (TypeError, ValueError):
            pass
    px_disp = format_price_fn(symbol, px_for_fmt)
    payload["price_display"] = px_disp
    if mark_raw is not None:
        try:
            payload["exchange_mark_price_display"] = format_price_fn(symbol, float(mark_raw))
        except (TypeError, ValueError):
            payload["exchange_mark_price_display"] = px_disp
    else:
        payload["exchange_mark_price_display"] = None
    return payload


def resolve_live_tpsl_prices(
    plan_sl: Any,
    plan_tp: Any,
    exchange_tpsl: Any,
) -> tuple[Optional[float], Optional[float], Optional[float], Optional[float]]:
    """返回 (展示用止损, 展示用止盈, 交易所止损, 交易所止盈)。"""
    ex_sl = ex_tp = None
    if isinstance(exchange_tpsl, dict):
        ex_sl = tpsl_slot_trigger_price(exchange_tpsl.get("sl"))
        ex_tp = tpsl_slot_trigger_price(exchange_tpsl.get("tp"))
    disp_sl = ex_sl if ex_sl is not None else _positive_float(plan_sl)
    disp_tp = ex_tp if ex_tp is not None else _positive_float(plan_tp)
    return disp_sl, disp_tp, ex_sl, ex_tp


def order_monitor_tpsl_needs_sync(
    plan_sl: Any,
    plan_tp: Any,
    exchange_tpsl: Any,
    *,
    eps: float = 1e-12,
) -> tuple[Optional[float], Optional[float], bool]:
    """若交易所 TP/SL 与库中不一致，返回应写回的 (sl, tp) 及是否需更新。"""
    _, _, ex_sl, ex_tp = resolve_live_tpsl_prices(plan_sl, plan_tp, exchange_tpsl)
    try:
        cur_sl = float(plan_sl or 0)
        cur_tp = float(plan_tp or 0)
    except (TypeError, ValueError):
        cur_sl, cur_tp = 0.0, 0.0
    new_sl = ex_sl if ex_sl is not None else cur_sl
    new_tp = ex_tp if ex_tp is not None else cur_tp
    changed = (
        (ex_sl is not None and abs(new_sl - cur_sl) > eps)
        or (ex_tp is not None and abs(new_tp - cur_tp) > eps)
    )
    return new_sl, new_tp, changed


def apply_order_price_display_fields(
    payload: dict[str, Any],
    *,
    direction: str,
    entry_price: Any,
    initial_stop_loss: Any,
    stop_loss: Any,
    take_profit: Any,
    calc_rr_ratio_fn: Callable[..., Optional[float]],
    exchange_tpsl: Any = None,
    format_price_fn: Optional[Callable[[Any, Any], str]] = None,
    symbol: Any = None,
) -> dict[str, Any]:
    disp_sl, disp_tp, _, _ = resolve_live_tpsl_prices(stop_loss, take_profit, exchange_tpsl)
    payload["rr_ratio"] = snapshot_rr(
        calc_rr_ratio_fn,
        direction,
        entry_price,
        initial_stop_loss,
        stop_loss,
        take_profit,
    )
    payload["sl_breakeven_secured"] = sl_breakeven_from_exchange_tpsl(
        direction, entry_price, exchange_tpsl
    )
    payload["stop_loss"] = disp_sl
    payload["take_profit"] = disp_tp
    if disp_sl is not None and disp_tp is not None:
        payload["display_rr_ratio"] = calc_rr_ratio_fn(
            direction or "long", entry_price, disp_sl, disp_tp
        )
    else:
        payload["display_rr_ratio"] = None
    if format_price_fn is not None and symbol is not None:
        payload["stop_loss_display"] = (
            format_price_fn(symbol, disp_sl) if disp_sl is not None else "—"
        )
        payload["take_profit_display"] = (
            format_price_fn(symbol, disp_tp) if disp_tp is not None else "—"
        )
    return payload