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 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,
) -> dict[str, Any]:
    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
    )
    return payload