Files
crypto_monitor/instance_embed_context_lib.py
dekun 4742a0bb9d refactor: 移除四所统计分析页交易日历
删除日历 UI、bootstrap 与 /api/stats/calendar 注册;保留日/周/月统计表。内照明心档案日历不受影响。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-30 09:51:10 +08:00

85 lines
2.8 KiB
Python

"""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"})
@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",
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 trade_records_summary(conn, start_bj: str, end_bj: str, tr_ts: str) -> dict[str, Any]:
"""顶栏统计用 COUNT,避免 embed 壳拉 1000 行交易记录。"""
from 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 result = '错过' THEN 1 ELSE 0 END) AS miss_count,
SUM(CASE WHEN {pnl_sql} > 0 THEN 1 ELSE 0 END) AS wins,
SUM(CASE WHEN result = '错过' AND COALESCE(miss_reason,'') LIKE '%持仓占用%' THEN 1 ELSE 0 END) AS occupied_miss
FROM trade_records
WHERE {tr_ts} >= ? AND {tr_ts} <= ?
""",
(start_bj, end_bj),
).fetchone()
total = int(row["total"] or 0) if row else 0
miss_count = int(row["miss_count"] or 0) if row else 0
wins = int(row["wins"] or 0) if row else 0
occupied_miss_total = int(row["occupied_miss"] or 0) if row else 0
rate = round(wins / total * 100, 2) if total else 0
return {
"records": [],
"total": total,
"miss_count": miss_count,
"rate": rate,
"occupied_miss_total": occupied_miss_total,
}
def minimal_stats_bundle(reset_hour: int) -> dict[str, Any]:
return {"stats_reset_hour": reset_hour, "segments": []}