crypto_okx/lib/trade/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 monitor_open_stop_loss(row: Any) -> Optional[float]:
    """从 order_monitors 行取开仓止损快照."""
    try:
        keys = row.keys() if hasattr(row, "keys") else ()
    except Exception:
        keys = ()
    init = row["initial_stop_loss"] if "initial_stop_loss" in keys else None
    cur = row["stop_loss"] if "stop_loss" in keys else None
    if init is None and isinstance(row, dict):
        init = row.get("initial_stop_loss")
        cur = row.get("stop_loss")
    return snapshot_stop_loss(init, cur)


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 resolve_breakeven_entry_price(entry_price: Any, avg_entry_price: Any = None) -> Optional[float]:
    """保本判断基准价:有持仓加权均价时优先(滚仓后),否则用首仓成交价."""
    avg = _positive_float(avg_entry_price)
    if avg is not None:
        return avg
    return _positive_float(entry_price)


def stale_breakeven_armed(direction: str, entry_price: Any, stop_loss: Any, breakeven_armed: Any) -> bool:
    """止损已回到亏损侧时 breakeven_armed 视为过期(如滚仓下移止损)."""
    try:
        armed = int(breakeven_armed or 0) != 0
    except (TypeError, ValueError):
        return False
    if not armed:
        return False
    return not stop_is_profit_protecting(direction, entry_price, stop_loss)


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 calc_risk_fraction(direction: str, entry_price: Any, stop_loss: Any) -> Optional[float]:
    """|入场-止损|/入场;盈利侧止损返回 0."""
    entry = _positive_float(entry_price)
    sl = _positive_float(stop_loss)
    if entry is None or sl is None:
        return None
    d = (direction or "long").strip().lower()
    if d == "short":
        risk = sl - entry
    else:
        risk = entry - sl
    if risk <= 0:
        return 0.0
    return risk / entry


def calc_latest_risk_amount(
    direction: str,
    entry_price: Any,
    stop_loss: Any,
    *,
    margin_capital: Any = None,
    leverage: Any = None,
    exchange_notional: Any = None,
    contracts: Any = None,
    contract_size: Any = None,
    mark_price: Any = None,
    funds_decimals: int = 2,
) -> Optional[float]:
    """按当前止损与持仓名义价值估算最新风险(U)."""
    rf = calc_risk_fraction(direction, entry_price, stop_loss)
    if rf is None:
        return None
    if rf <= 0:
        return 0.0
    notional = _positive_float(exchange_notional)
    if notional is None:
        try:
            mc = float(margin_capital or 0)
            lev = float(leverage or 0)
            if mc > 0 and lev > 0:
                notional = mc * lev
        except (TypeError, ValueError):
            pass
    if notional is None:
        try:
            c = abs(float(contracts or 0))
            cs = float(contract_size or 1)
            if cs <= 0:
                cs = 1.0
            px = _positive_float(mark_price) or _positive_float(entry_price)
            if c > 0 and px is not None:
                notional = c * cs * px
        except (TypeError, ValueError):
            pass
    if notional is None or notional <= 0:
        return None
    return round(notional * rf, funds_decimals)


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,
    margin_capital: Any = None,
    leverage: Any = None,
    exchange_notional: Any = None,
    contracts: Any = None,
    contract_size: Any = None,
    mark_price: Any = None,
    avg_entry_price: Any = None,
    funds_decimals: int = 2,
) -> dict[str, Any]:
    disp_sl, disp_tp, _, _ = resolve_live_tpsl_prices(stop_loss, take_profit, exchange_tpsl)
    payload["stop_loss_raw"] = _positive_float(stop_loss)
    payload["take_profit_raw"] = _positive_float(take_profit)
    payload["rr_ratio"] = snapshot_rr(
        calc_rr_ratio_fn,
        direction,
        entry_price,
        initial_stop_loss,
        stop_loss,
        take_profit,
    )
    risk_entry = resolve_breakeven_entry_price(entry_price, avg_entry_price)
    payload["avg_entry_price"] = risk_entry
    payload["sl_breakeven_secured"] = sl_breakeven_from_exchange_tpsl(
        direction, risk_entry, 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 contracts is not None:
        try:
            from lib.market.position_metrics_lib import normalize_contracts_qty

            c = normalize_contracts_qty(contracts)
            if c > 0:
                payload["contracts"] = c
        except (TypeError, ValueError):
            pass
    payload["latest_risk_amount"] = calc_latest_risk_amount(
        direction,
        risk_entry,
        disp_sl if disp_sl is not None else stop_loss,
        margin_capital=margin_capital,
        leverage=leverage,
        exchange_notional=exchange_notional,
        contracts=payload.get("contracts") if payload.get("contracts") is not None else contracts,
        contract_size=contract_size,
        mark_price=mark_price,
        funds_decimals=funds_decimals,
    )
    tp_for_reward = disp_tp if disp_tp is not None else _positive_float(take_profit)
    qty_for_reward = payload.get("contracts")
    if qty_for_reward is None and contracts is not None:
        try:
            qty_for_reward = abs(float(contracts))
        except (TypeError, ValueError):
            qty_for_reward = None
    if risk_entry is not None and tp_for_reward is not None and qty_for_reward:
        try:
            def reward_at_tp_usdt(direction, entry, tp, contracts, contract_size, fee_rate=0.0):
                try:
                    d = (direction or "long").lower()
                    e, t, c, cs = float(entry), float(tp), float(contracts), float(contract_size or 1)
                    if e <= 0 or t <= 0 or c <= 0:
                        return None
                    raw = (t - e) * c * cs if d == "long" else (e - t) * c * cs
                    return round(raw, 2)
                except Exception:
                    return None

            reward = reward_at_tp_usdt(
                direction,
                risk_entry,
                tp_for_reward,
                float(qty_for_reward),
                contract_size=float(contract_size or 1.0),
            )
            payload["reward_at_tp_usdt"] = (
                round(reward, funds_decimals) if reward is not None else None
            )
        except Exception:
            payload["reward_at_tp_usdt"] = None
    else:
        payload["reward_at_tp_usdt"] = 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 "—"
        )
    mark_raw = mark_price if mark_price is not None else None
    if mark_raw is not None and format_price_fn is not None and symbol is not None:
        try:
            payload["exchange_mark_price_display"] = format_price_fn(symbol, float(mark_raw))
        except (TypeError, ValueError):
            payload["exchange_mark_price_display"] = None
    return payload


def enrich_active_monitor_tpsl_json(
    row: Any,
    stop_loss: Any,
    take_profit: Any,
    exchange_tpsl: Any,
    *,
    position_row: Any = None,
    exchange_notional: Any = None,
    contracts: Any = None,
    contract_size: float = 1.0,
    mark_price: Any = None,
    calc_rr_ratio_fn: Callable[..., Optional[float]],
    format_price_fn: Optional[Callable[[Any, Any], str]] = None,
    symbol: Any = None,
    funds_decimals: int = 2,
) -> dict[str, Any]:
    """place_tpsl 响应:展示用 TP/SL,最新风险,当前盈亏比."""
    def _row_val(key: str, default=None):
        try:
            if hasattr(row, "keys") and key in row.keys():
                return row[key]
        except Exception:
            pass
        if isinstance(row, dict):
            return row.get(key, default)
        return default

    direction = _row_val("direction") or "long"
    entry = _row_val("trigger_price")
    init_sl = _row_val("initial_stop_loss")
    margin = _row_val("margin_capital")
    leverage = _row_val("leverage")
    if position_row is not None:
        from lib.market.position_metrics_lib import parse_position_entry_price, position_contracts

        live_c = position_contracts(position_row)
        if abs(live_c) >= 1e-12:
            contracts = abs(live_c)
        avg_entry = parse_position_entry_price(position_row)
    else:
        avg_entry = None
    payload: dict[str, Any] = {
        "stop_loss": stop_loss,
        "take_profit": take_profit,
    }
    apply_order_price_display_fields(
        payload,
        direction=direction,
        entry_price=entry,
        initial_stop_loss=init_sl,
        stop_loss=stop_loss,
        take_profit=take_profit,
        calc_rr_ratio_fn=calc_rr_ratio_fn,
        exchange_tpsl=exchange_tpsl,
        format_price_fn=format_price_fn,
        symbol=symbol or _row_val("symbol"),
        margin_capital=margin,
        leverage=leverage,
        exchange_notional=exchange_notional,
        contracts=contracts,
        contract_size=contract_size,
        mark_price=mark_price,
        avg_entry_price=avg_entry,
        funds_decimals=funds_decimals,
    )
    return payload