dekun
251 lines
8.0 KiB
Python
251 lines
8.0 KiB
Python
"""embed 壳/片段:按 tab 裁剪 render_main_page 的数据加载,降内存与 API 压力."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
EMBED_STRATEGY_PAGES = frozenset({"strategy", "strategy_trend", "strategy_roll", "strategy_records"})
|
|
|
|
_WIN_EPS = 1e-9
|
|
|
|
|
|
def env_truthy(raw: str | None, default: bool = False) -> bool:
|
|
if raw is None or str(raw).strip() == "":
|
|
return default
|
|
return str(raw).strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
def show_perp_funds_enabled(*, exchange_key: str | None = None) -> bool:
|
|
"""OKX:是否在顶栏显示永续资金账户/交易账户.其他所恒为 True."""
|
|
ex = (exchange_key or "").strip().lower()
|
|
if ex and ex != "okx":
|
|
return True
|
|
# 默认关闭:产品以期权/期期对冲为主;币本位买币时由调用方另或 True
|
|
return env_truthy(os.getenv("OKX_SHOW_PERP_FUNDS"), default=False)
|
|
|
|
|
|
@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,
|
|
funding_eth: float | None = None,
|
|
margin_mode: str | None = None,
|
|
underly: str = "ETH",
|
|
) -> str:
|
|
"""期权侧顶栏文案(仅 USDC 模式使用;币本位不展示期权资金/交易两列)."""
|
|
if funding_usdc is None:
|
|
return "—"
|
|
try:
|
|
return f"{float(funding_usdc):.2f} USDC"
|
|
except (TypeError, ValueError):
|
|
return "—"
|
|
|
|
|
|
def _fmt_coin_amount(v: float | None, *, min_amt: float = 1e-6) -> str | None:
|
|
if v is None:
|
|
return None
|
|
try:
|
|
n = float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if n < min_amt:
|
|
return None
|
|
txt = f"{n:.6f}".rstrip("0").rstrip(".")
|
|
return txt or None
|
|
|
|
|
|
def trading_account_label(
|
|
usdt: float | None,
|
|
eth: float | None = None,
|
|
btc: float | None = None,
|
|
*,
|
|
margin_mode: str | None = None,
|
|
) -> str:
|
|
"""交易账户顶栏文案.
|
|
|
|
币本位:USDT / ETH / BTC(有余额才带上,不显示其它币种).
|
|
其它模式:xx.xxU.
|
|
"""
|
|
try:
|
|
from lib.options.options_margin_mode_lib import normalize_options_margin_mode
|
|
|
|
mode = normalize_options_margin_mode(margin_mode)
|
|
except Exception:
|
|
mode = str(margin_mode or "coin").strip().lower() or "coin"
|
|
if mode != "coin":
|
|
if usdt is None:
|
|
return "—"
|
|
try:
|
|
return f"{float(usdt):.2f}U"
|
|
except (TypeError, ValueError):
|
|
return "—"
|
|
parts: list[str] = []
|
|
if usdt is not None:
|
|
try:
|
|
parts.append(f"{float(usdt):.2f} USDT")
|
|
except (TypeError, ValueError):
|
|
pass
|
|
eth_txt = _fmt_coin_amount(eth, min_amt=1e-6)
|
|
if eth_txt is not None:
|
|
parts.append(f"{eth_txt} ETH")
|
|
btc_txt = _fmt_coin_amount(btc, min_amt=1e-7)
|
|
if btc_txt is not None:
|
|
parts.append(f"{btc_txt} BTC")
|
|
return "\n".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": []}
|