crypto_monitor/lib/instance/instance_embed_context_lib.py

"""embed 壳/片段:按 tab 裁剪 render_main_page 的数据加载,降内存与 API 压力."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

EMBED_STRATEGY_PAGES = frozenset({"strategy", "strategy_trend", "strategy_roll", "strategy_records"})

_WIN_EPS = 1e-9


@dataclass(frozen=True)
class EmbedRenderPlan:
    exchange_capitals: bool
    records_rows: bool
    records_summary: bool
    key_history: bool
    key_list: bool
    orders: bool
    stats_bundle: bool
    strategy: bool
    orphan_live: bool


def embed_render_plan(page: str, embed_mode: str | None) -> EmbedRenderPlan:
    if embed_mode not in ("fragment", "shell"):
        return EmbedRenderPlan(
            exchange_capitals=True,
            records_rows=True,
            records_summary=False,
            key_history=True,
            key_list=True,
            orders=True,
            stats_bundle=True,
            strategy=True,
            orphan_live=True,
        )
    is_shell = embed_mode == "shell"
    is_strategy = page in EMBED_STRATEGY_PAGES
    return EmbedRenderPlan(
        exchange_capitals=is_shell,
        records_rows=page == "records",
        # 顶栏常驻:设置/风控/env 也要统计,否则首屏 SSR 为 0 后软切 tab 不会重绘顶栏
        records_summary=is_shell and page != "records",
        key_history=page == "key_monitor",
        key_list=page in ("key_monitor", "trade") or is_strategy,
        orders=page == "trade" or is_strategy,
        stats_bundle=page == "stats",
        strategy=is_strategy,
        orphan_live=page == "trade" and is_shell,
    )


def profit_loss_ratio_from_averages(avg_win: float | None, avg_loss: float | None) -> float | None:
    """盈亏比 = 平均盈利 / |平均亏损|."""
    if avg_win is None or avg_loss is None:
        return None
    try:
        aw = float(avg_win)
        al = float(avg_loss)
    except (TypeError, ValueError):
        return None
    if al == 0:
        return None
    return round(aw / abs(al), 2)


def profit_loss_ratio_from_trades(trades: list[dict[str, Any]] | None) -> float | None:
    wins: list[float] = []
    losses: list[float] = []
    for row in trades or []:
        if not isinstance(row, dict):
            continue
        try:
            pnl = float(row.get("effective_pnl_amount") or row.get("pnl_amount") or 0)
        except (TypeError, ValueError):
            continue
        if pnl > _WIN_EPS:
            wins.append(pnl)
        elif pnl < -_WIN_EPS:
            losses.append(pnl)
    avg_win = sum(wins) / len(wins) if wins else None
    avg_loss = sum(losses) / len(losses) if losses else None
    return profit_loss_ratio_from_averages(avg_win, avg_loss)


def options_funding_label(
    funding_usdc: float | None,
    funding_usdt: float | None = None,
) -> str:
    parts: list[str] = []
    if funding_usdc is not None and float(funding_usdc) > 0:
        parts.append(f"{float(funding_usdc):.2f} USDC")
    if funding_usdt is not None and float(funding_usdt) > 0:
        parts.append(f"{float(funding_usdt):.2f} USDT")
    return " · ".join(parts) if parts else "—"


def total_funds_usdt(
    funding_usdt: float | None,
    trading_usdt: float | None,
    options_trading_usdc: float | None = None,
    options_funding_usdc: float | None = None,
    options_funding_usdt: float | None = None,
    options_trading_usdt: float | None = None,
) -> float | None:
    parts = [
        funding_usdt,
        trading_usdt,
        options_funding_usdc,
        options_funding_usdt,
        options_trading_usdc,
        options_trading_usdt,
    ]
    if all(v is None for v in parts):
        return None
    try:
        total = 0.0
        for v in parts:
            if v is not None:
                total += float(v)
        return round(total, 2)
    except (TypeError, ValueError):
        return None


def trade_records_summary(conn, start_bj: str, end_bj: str, tr_ts: str) -> dict[str, Any]:
    """顶栏统计用 COUNT,避免 embed 壳拉 1000 行交易记录."""
    from lib.trade.trade_result_lib import sql_effective_pnl_expr

    pnl_sql = sql_effective_pnl_expr()
    row = conn.execute(
        f"""
        SELECT
            COUNT(*) AS total,
            SUM(CASE WHEN {pnl_sql} > 0 THEN 1 ELSE 0 END) AS wins,
            AVG(CASE WHEN {pnl_sql} > 0 THEN {pnl_sql} END) AS avg_win,
            AVG(CASE WHEN {pnl_sql} < 0 THEN {pnl_sql} END) AS avg_loss
        FROM trade_records
        WHERE {tr_ts} >= ? AND {tr_ts} <= ?
          AND COALESCE(result, '') != '错过'
          AND COALESCE(reviewed_result, '') != '错过'
        """,
        (start_bj, end_bj),
    ).fetchone()
    total = int(row["total"] or 0) if row else 0
    wins = int(row["wins"] or 0) if row else 0
    rate = round(wins / total * 100, 2) if total else 0
    avg_win = float(row["avg_win"]) if row and row["avg_win"] is not None else None
    avg_loss = float(row["avg_loss"]) if row and row["avg_loss"] is not None else None
    return {
        "records": [],
        "total": total,
        "rate": rate,
        "profit_loss_ratio": profit_loss_ratio_from_averages(avg_win, avg_loss),
    }


def header_trade_stats_for_window(conn, list_window: dict[str, Any], app_tz) -> dict[str, Any]:
    """account_snapshot / 顶栏刷新:按当前列表窗返回总交易/胜率/盈亏比."""
    from lib.common.history_window_lib import sql_list_time_field, utc_window_to_bj_sql_strings

    start_bj, end_bj = utc_window_to_bj_sql_strings(
        list_window["start_utc"], list_window["end_utc"], app_tz
    )
    tr_ts = sql_list_time_field("closed_at", "created_at", "opened_at")
    summary = trade_records_summary(conn, start_bj, end_bj, tr_ts)
    return {
        "total": summary["total"],
        "rate": summary["rate"],
        "profit_loss_ratio": summary.get("profit_loss_ratio"),
    }


def minimal_stats_bundle(reset_hour: int) -> dict[str, Any]:
    return {"stats_reset_hour": reset_hour, "segments": []}