crypto_monitor/focus_chart_lib.py

"""实盘/关键位放大 K 线：订单元数据与交易所浮盈、价格展示精度。"""
from __future__ import annotations

from typing import Any, Callable, Optional

from hub_ohlcv_lib import (
    normalize_price_tick,
    price_tick_from_market,
    round_ohlcv_bars_to_tick,
)
from order_monitor_display_lib import (
    apply_order_live_price_display,
    apply_order_price_display_fields,
)


def resolve_kline_price_tick(
    exchange: Any,
    exchange_symbol: str,
    *,
    ensure_markets_fn: Callable[[], None],
) -> Optional[float]:
    """交易所最小价格变动单位，供 lightweight-charts 右侧刻度与标记线对齐。"""
    if not exchange_symbol:
        return None
    try:
        ensure_markets_fn()
        return normalize_price_tick(price_tick_from_market(exchange, exchange_symbol))
    except Exception:
        return None


def align_candles_to_price_tick(
    candles: list[dict[str, Any]],
    price_tick: Optional[float],
) -> None:
    if price_tick is not None and candles:
        round_ohlcv_bars_to_tick(candles, price_tick)


def kline_api_price_fields(
    exchange: Any,
    exchange_symbol: str,
    candles: list[dict[str, Any]],
    *,
    ensure_markets_fn: Callable[[], None],
) -> dict[str, Any]:
    tick = resolve_kline_price_tick(
        exchange, exchange_symbol, ensure_markets_fn=ensure_markets_fn
    )
    align_candles_to_price_tick(candles, tick)
    return {"price_tick": tick}


def load_swap_positions_for_order_kline(
    exchange: Any,
    *,
    private_configured: bool,
    ensure_markets_fn: Callable[[], None],
    settle: str = "usdt",
) -> list:
    if not private_configured:
        return []
    try:
        ensure_markets_fn()
        try:
            return exchange.fetch_positions(None, {"settle": settle}) or []
        except Exception:
            return exchange.fetch_positions() or []
    except Exception:
        return []


def metrics_for_order_item(
    order_item: dict[str, Any],
    positions: list,
    *,
    resolve_ex_sym_fn: Callable[[Any], str],
    select_live_fn: Callable[[list, str, str], Any],
    parse_metrics_fn: Callable[..., Optional[dict]],
) -> Optional[dict]:
    if not positions:
        return None
    ex_sym = resolve_ex_sym_fn(order_item)
    direction = order_item.get("direction") or "long"
    prow = select_live_fn(positions, ex_sym, direction)
    if not prow:
        return None
    lev = order_item.get("leverage")
    return parse_metrics_fn(prow, order_leverage=lev)


def build_order_kline_order_payload(
    order_item: dict[str, Any],
    *,
    ticker_price: Any,
    format_price_fn: Callable[[Any, Any], str],
    calc_pnl_fn: Callable[..., float],
    calc_rr_ratio_fn: Callable[..., Optional[float]],
    ex_metrics: Optional[dict] = None,
) -> dict[str, Any]:
    sym = order_item.get("symbol") or ""
    direction = order_item.get("direction") or "long"
    margin = float(order_item.get("margin_capital") or 0)
    leverage = float(order_item.get("leverage") or 0)
    entry = float(order_item.get("trigger_price") or 0)

    float_pnl = 0.0
    float_pct = 0.0
    if ticker_price and entry > 0:
        float_pnl = float(
            calc_pnl_fn(direction, entry, ticker_price, margin, leverage)
        )
        float_pct = round((float_pnl / margin * 100), 4) if margin > 0 else 0.0

    px_for_fmt = ticker_price
    mark_raw = None
    if ex_metrics and ex_metrics.get("mark_price") is not None:
        mark_raw = ex_metrics["mark_price"]
        try:
            px_for_fmt = float(mark_raw)
        except (TypeError, ValueError):
            pass

    if ex_metrics and ex_metrics.get("unrealized_pnl") is not None:
        float_pnl = round(float(ex_metrics["unrealized_pnl"]), 2)
        denom = ex_metrics.get("initial_margin") or margin
        float_pct = (
            round((float_pnl / float(denom)) * 100, 4)
            if denom and float(denom) > 0
            else float_pct
        )

    payload: dict[str, Any] = {
        "id": order_item["id"],
        "symbol": sym,
        "direction": direction,
        "trigger_price": order_item.get("trigger_price"),
        "stop_loss": order_item.get("stop_loss"),
        "take_profit": order_item.get("take_profit"),
        "trigger_price_display": format_price_fn(sym, order_item.get("trigger_price")),
        "stop_loss_display": format_price_fn(sym, order_item.get("stop_loss")),
        "take_profit_display": format_price_fn(sym, order_item.get("take_profit")),
        "margin_capital": order_item.get("margin_capital"),
        "leverage": order_item.get("leverage"),
        "position_ratio": order_item.get("position_ratio"),
        "breakeven_enabled": bool(int(order_item.get("breakeven_enabled") or 0)),
        "current_price": round(float(px_for_fmt), 8) if px_for_fmt is not None else None,
        "float_pnl": round(float(float_pnl), 2),
        "float_pct": float_pct,
    }
    apply_order_price_display_fields(
        payload,
        direction=direction,
        entry_price=order_item.get("trigger_price"),
        initial_stop_loss=order_item.get("initial_stop_loss"),
        stop_loss=order_item.get("stop_loss"),
        take_profit=order_item.get("take_profit"),
        calc_rr_ratio_fn=calc_rr_ratio_fn,
    )
    apply_order_live_price_display(
        payload,
        sym,
        ticker_price,
        mark_raw,
        format_price_fn,
    )
    payload["current_price_display"] = payload.get("price_display") or (
        format_price_fn(sym, px_for_fmt) if px_for_fmt is not None else None
    )
    return payload


def enrich_key_kline_response(
    *,
    symbol: str,
    current_price: Any,
    key_info: Optional[dict[str, Any]],
    format_price_fn: Callable[[Any, Any], str],
) -> tuple[Any, Optional[dict[str, Any]]]:
    price_display = format_price_fn(symbol, current_price) if current_price is not None else None
    if key_info is None:
        return price_display, None
    enriched = dict(key_info)
    enriched["upper_display"] = format_price_fn(symbol, key_info.get("upper"))
    enriched["lower_display"] = format_price_fn(symbol, key_info.get("lower"))
    return price_display, enriched