dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Slim embed tab rendering to cut memory use and restore calculator.

Browse Source 
Load only per-tab data for embed fragments, skip exchange capital fetches on tab switches, and harden calculator market imports/timeouts.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-24 01:33:42 +08:00
parent bced61b9d7
commit 384d404bb3
8 changed files with 380 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_monitor_binance/app.py
+64 -28
View File
@@ -6836,37 +6836,73 @@ def render_main_page(page="trade", embed_mode=None):
conn = get_db()
session_row = ensure_session(conn, trading_day)
local_current_capital = float(session_row["current_capital"])
funding_capital, trading_capital = get_exchange_capitals()
from instance_embed_context_lib import (
embed_render_plan,
minimal_stats_bundle,
trade_records_summary,
)
plan = embed_render_plan(page, embed_mode)
if plan.exchange_capitals:
funding_capital, trading_capital = get_exchange_capitals()
else:
funding_capital, trading_capital = None, None
# 资金账户：仅展示交易所读取结果（含 0）。不可用 TOTAL_CAPITAL 兜底，否则会与实盘不符。
funding_usdt = round(funding_capital, FUNDS_DECIMALS) if funding_capital is not None else None
current_capital = round(trading_capital, FUNDS_DECIMALS) if trading_capital is not None else round(local_current_capital, FUNDS_DECIMALS)
recommended_capital = get_recommended_capital(current_capital)
key_list = conn.execute("SELECT * FROM key_monitors").fetchall()
key_history = conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
stats_bundle = compute_stats_bundle(conn, trading_day, now)
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
order_list = []
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
tr_ts = sql_list_time_field("closed_at", "created_at", "opened_at")
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {tr_ts} >= ? AND {tr_ts} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(1 for r in records if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈"))
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
key_list = (
conn.execute("SELECT * FROM key_monitors").fetchall() if plan.key_list else []
)
rate = round(win/total*100,2) if total else 0
key_history = (
conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
if plan.key_history
else []
)
stats_bundle = (
compute_stats_bundle(conn, trading_day, now)
if plan.stats_bundle
else minimal_stats_bundle(TRADING_DAY_RESET_HOUR)
)
order_list = []
if plan.orders:
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
tr_ts = sql_list_time_field("closed_at", "created_at", "opened_at")
if plan.records_rows:
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {tr_ts} >= ? AND {tr_ts} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(
1
for r in records
if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈")
)
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
)
rate = round(win / total * 100, 2) if total else 0
elif plan.records_summary:
summary = trade_records_summary(conn, start_bj, end_bj, tr_ts)
records = summary["records"]
total = summary["total"]
miss_count = summary["miss_count"]
rate = summary["rate"]
occupied_miss_total = summary["occupied_miss_total"]
else:
records = []
total = miss_count = rate = occupied_miss_total = 0
active_count = len(order_list)
opens_today = count_opens_for_trading_day(conn, trading_day)
risk_status = hub_account_risk_status(conn)
@@ -6895,7 +6931,7 @@ def render_main_page(page="trade", embed_mode=None):
trigger_entry_validity_hours=TRIGGER_ENTRY_VALIDITY_HOURS,
)
strategy_extra = {}
if page in ("strategy", "strategy_trend", "strategy_roll", "strategy_records"):
if plan.strategy:
from strategy_ui import strategy_render_extras
strategy_extra = strategy_render_extras(
@@ -6906,7 +6942,7 @@ def render_main_page(page="trade", embed_mode=None):
trend_cfg=app.extensions.get("strategy_trend_cfg"),
)
orphan_live_positions = []
if not order_list and exchange_private_api_configured():
if plan.orphan_live and not order_list and exchange_private_api_configured():
orphan_live_positions = list_orphan_live_positions(conn)
conn.close()
from instance_embed_lib import embed_context_extras
crypto_monitor_gate/app.py
+62 -26
View File
@@ -6710,21 +6710,42 @@ def render_main_page(page="trade", embed_mode=None):
conn = get_db()
session_row = ensure_session(conn, trading_day)
local_current_capital = float(session_row["current_capital"])
funding_capital, trading_capital = get_exchange_capitals()
from instance_embed_context_lib import (
embed_render_plan,
minimal_stats_bundle,
trade_records_summary,
)
plan = embed_render_plan(page, embed_mode)
if plan.exchange_capitals:
funding_capital, trading_capital = get_exchange_capitals()
else:
funding_capital, trading_capital = None, None
# 资金账户：仅展示交易所读取结果（含 0）。不可用 TOTAL_CAPITAL 兜底，否则会与实盘不符。
funding_usdt = round(funding_capital, 2) if funding_capital is not None else None
current_capital = round(trading_capital, 2) if trading_capital is not None else round(local_current_capital, 2)
recommended_capital = round(float(get_recommended_capital(current_capital)), 2)
key_list = conn.execute("SELECT * FROM key_monitors").fetchall()
key_history = conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
stats_bundle = compute_stats_bundle(conn, trading_day, now)
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
key_list = (
conn.execute("SELECT * FROM key_monitors").fetchall() if plan.key_list else []
)
key_history = (
conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
if plan.key_history
else []
)
stats_bundle = (
compute_stats_bundle(conn, trading_day, now)
if plan.stats_bundle
else minimal_stats_bundle(TRADING_DAY_RESET_HOUR)
)
order_list = []
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
if plan.orders:
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
exchange_pnl_sync = {}
if exchange_private_api_configured() and not request_is_hub_soft_nav() and embed_mode not in (
"fragment",
@@ -6735,21 +6756,36 @@ def render_main_page(page="trade", embed_mode=None):
except Exception as e:
exchange_pnl_sync = {"ok": False, "reason": str(e)}
tr_ts = sql_list_time_field("closed_at", "created_at", "opened_at")
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {tr_ts} >= ? AND {tr_ts} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(1 for r in records if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈"))
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
)
rate = round(win/total*100,2) if total else 0
if plan.records_rows:
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {tr_ts} >= ? AND {tr_ts} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(
1
for r in records
if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈")
)
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
)
rate = round(win / total * 100, 2) if total else 0
elif plan.records_summary:
summary = trade_records_summary(conn, start_bj, end_bj, tr_ts)
records = summary["records"]
total = summary["total"]
miss_count = summary["miss_count"]
rate = summary["rate"]
occupied_miss_total = summary["occupied_miss_total"]
else:
records = []
total = miss_count = rate = occupied_miss_total = 0
active_count = len(order_list)
opens_today = count_opens_for_trading_day(conn, trading_day)
risk_status = hub_account_risk_status(conn)
@@ -6778,7 +6814,7 @@ def render_main_page(page="trade", embed_mode=None):
trigger_entry_validity_hours=TRIGGER_ENTRY_VALIDITY_HOURS,
)
strategy_extra = {}
if page in ("strategy", "strategy_trend", "strategy_roll", "strategy_records"):
if plan.strategy:
from strategy_ui import strategy_render_extras
strategy_extra = strategy_render_extras(
crypto_monitor_gate_bot/app.py
+62 -26
View File
@@ -6710,21 +6710,42 @@ def render_main_page(page="trade", embed_mode=None):
conn = get_db()
session_row = ensure_session(conn, trading_day)
local_current_capital = float(session_row["current_capital"])
funding_capital, trading_capital = get_exchange_capitals()
from instance_embed_context_lib import (
embed_render_plan,
minimal_stats_bundle,
trade_records_summary,
)
plan = embed_render_plan(page, embed_mode)
if plan.exchange_capitals:
funding_capital, trading_capital = get_exchange_capitals()
else:
funding_capital, trading_capital = None, None
# 资金账户：仅展示交易所读取结果（含 0）。不可用 TOTAL_CAPITAL 兜底，否则会与实盘不符。
funding_usdt = round(funding_capital, 2) if funding_capital is not None else None
current_capital = round(trading_capital, 2) if trading_capital is not None else round(local_current_capital, 2)
recommended_capital = round(float(get_recommended_capital(current_capital)), 2)
key_list = conn.execute("SELECT * FROM key_monitors").fetchall()
key_history = conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
stats_bundle = compute_stats_bundle(conn, trading_day, now)
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
key_list = (
conn.execute("SELECT * FROM key_monitors").fetchall() if plan.key_list else []
)
key_history = (
conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
if plan.key_history
else []
)
stats_bundle = (
compute_stats_bundle(conn, trading_day, now)
if plan.stats_bundle
else minimal_stats_bundle(TRADING_DAY_RESET_HOUR)
)
order_list = []
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
if plan.orders:
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
exchange_pnl_sync = {}
if exchange_private_api_configured() and not request_is_hub_soft_nav() and embed_mode not in (
"fragment",
@@ -6735,21 +6756,36 @@ def render_main_page(page="trade", embed_mode=None):
except Exception as e:
exchange_pnl_sync = {"ok": False, "reason": str(e)}
tr_ts = sql_list_time_field("closed_at", "created_at", "opened_at")
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {tr_ts} >= ? AND {tr_ts} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(1 for r in records if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈"))
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
)
rate = round(win/total*100,2) if total else 0
if plan.records_rows:
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {tr_ts} >= ? AND {tr_ts} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(
1
for r in records
if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈")
)
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
)
rate = round(win / total * 100, 2) if total else 0
elif plan.records_summary:
summary = trade_records_summary(conn, start_bj, end_bj, tr_ts)
records = summary["records"]
total = summary["total"]
miss_count = summary["miss_count"]
rate = summary["rate"]
occupied_miss_total = summary["occupied_miss_total"]
else:
records = []
total = miss_count = rate = occupied_miss_total = 0
active_count = len(order_list)
opens_today = count_opens_for_trading_day(conn, trading_day)
risk_status = hub_account_risk_status(conn)
@@ -6778,7 +6814,7 @@ def render_main_page(page="trade", embed_mode=None):
trigger_entry_validity_hours=TRIGGER_ENTRY_VALIDITY_HOURS,
)
strategy_extra = {}
if page in ("strategy", "strategy_trend", "strategy_roll", "strategy_records"):
if plan.strategy:
from strategy_ui import strategy_render_extras
strategy_extra = strategy_render_extras(
crypto_monitor_okx/app.py
+63 -27
View File
@@ -6215,20 +6215,41 @@ def render_main_page(page="trade", embed_mode=None):
conn = get_db()
session_row = ensure_session(conn, trading_day)
local_current_capital = float(session_row["current_capital"])
funding_capital, trading_capital = get_exchange_capitals()
from instance_embed_context_lib import (
embed_render_plan,
minimal_stats_bundle,
trade_records_summary,
)
plan = embed_render_plan(page, embed_mode)
if plan.exchange_capitals:
funding_capital, trading_capital = get_exchange_capitals()
else:
funding_capital, trading_capital = None, None
funding_usdt = round(funding_capital, FUNDS_DECIMALS) if funding_capital is not None else None
current_capital = round(trading_capital, FUNDS_DECIMALS) if trading_capital is not None else round(local_current_capital, FUNDS_DECIMALS)
recommended_capital = get_recommended_capital(current_capital)
key_list = conn.execute("SELECT * FROM key_monitors").fetchall()
key_history = conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
stats_bundle = compute_stats_bundle(conn, trading_day, now)
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
key_list = (
conn.execute("SELECT * FROM key_monitors").fetchall() if plan.key_list else []
)
key_history = (
conn.execute(
"SELECT * FROM key_monitor_history WHERE closed_at >= ? AND closed_at <= ? ORDER BY id DESC LIMIT 500",
(start_bj, end_bj),
).fetchall()
if plan.key_history
else []
)
stats_bundle = (
compute_stats_bundle(conn, trading_day, now)
if plan.stats_bundle
else minimal_stats_bundle(TRADING_DAY_RESET_HOUR)
)
order_list = []
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
if plan.orders:
raw_order_list = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
for o in raw_order_list:
order_list.append(enrich_order_item(row_to_dict(o), current_capital))
exchange_pnl_sync = {}
if exchange_private_api_configured() and not request_is_hub_soft_nav() and embed_mode not in (
"fragment",
@@ -6238,22 +6259,37 @@ def render_main_page(page="trade", embed_mode=None):
exchange_pnl_sync = sync_trade_records_from_exchange(conn) or {}
except Exception as e:
exchange_pnl_sync = {"ok": False, "reason": str(e)}
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {sql_list_time_field('closed_at', 'created_at', 'opened_at')} >= ? "
f"AND {sql_list_time_field('closed_at', 'created_at', 'opened_at')} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(1 for r in records if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈"))
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
)
rate = round(win/total*100,2) if total else 0
tr_ts = sql_list_time_field("closed_at", "created_at", "opened_at")
if plan.records_rows:
raw_records = conn.execute(
f"SELECT * FROM trade_records WHERE {tr_ts} >= ? AND {tr_ts} <= ? ORDER BY id DESC LIMIT 1000",
(start_bj, end_bj),
).fetchall()
records = [to_effective_trade_dict(r) for r in raw_records]
total = len(records)
miss_count = sum(1 for r in records if (r.get("effective_result") or "") == "错过")
win = sum(
1
for r in records
if (r.get("effective_result") or "") in ("止盈", "保本止盈", "移动止盈")
)
occupied_miss_total = sum(
1
for r in records
if (r.get("effective_result") or "") == "错过"
and ("持仓占用" in str(r.get("effective_miss_reason") or ""))
)
rate = round(win / total * 100, 2) if total else 0
elif plan.records_summary:
summary = trade_records_summary(conn, start_bj, end_bj, tr_ts)
records = summary["records"]
total = summary["total"]
miss_count = summary["miss_count"]
rate = summary["rate"]
occupied_miss_total = summary["occupied_miss_total"]
else:
records = []
total = miss_count = rate = occupied_miss_total = 0
active_count = len(order_list)
open_guard_enabled = get_trading_day_reset_open_guard_enabled(conn)
open_guard_blocks_now = open_guard_enabled and now.hour < TRADING_DAY_RESET_HOUR
@@ -6284,7 +6320,7 @@ def render_main_page(page="trade", embed_mode=None):
trigger_entry_validity_hours=TRIGGER_ENTRY_VALIDITY_HOURS,
)
strategy_extra = {}
if page in ("strategy", "strategy_trend", "strategy_roll", "strategy_records"):
if plan.strategy:
from strategy_ui import strategy_render_extras
strategy_extra = strategy_render_extras(
hub_calculator_market_lib.py
+5 -2
View File
@@ -10,12 +10,15 @@ import urllib.request
from typing import Any, Callable, Optional, Tuple
from urllib.parse import urlencode
from manual_trading_hub.settings_store import enabled_exchanges, load_settings
try:
from settings_store import enabled_exchanges, load_settings
except ImportError:
from manual_trading_hub.settings_store import enabled_exchanges, load_settings
MARKET_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
MARKET_LOCK = threading.Lock()
MARKET_TTL_SEC = 300.0
HUB_FLASK_TIMEOUT = float(__import__("os").getenv("HUB_FLASK_TIMEOUT", "12"))
HUB_FLASK_TIMEOUT = float(__import__("os").getenv("HUB_FLASK_TIMEOUT", "20"))
def normalize_base_symbol(text: str) -> str:
instance_embed_context_lib.py
+81
View File
@@ -0,0 +1,81 @@
"""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 行交易记录。"""
row = conn.execute(
f"""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN result = '错过' THEN 1 ELSE 0 END) AS miss_count,
SUM(CASE WHEN result IN ('止盈','保本止盈','移动止盈') 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": []}
instance_embed_lib.py
+15
View File
@@ -55,8 +55,23 @@ def path_to_embed_tab(path: str) -> str | None:
return PATH_TO_EMBED_TAB.get(base)
def embed_shell_enabled() -> bool:
return (os.getenv("HUB_EMBED_SHELL") or "1").strip().lower() in ("1", "true", "yes", "on")
def rewrite_embed_dest(path: str, hub_theme: str | None = None) -> str:
"""embed=1 打开时：/trade → /embed?tab=trade&embed=1"""
if not embed_shell_enabled():
split = urlsplit(path or "/")
q = dict(parse_qsl(split.query, keep_blank_values=True))
q["embed"] = "1"
ht = (hub_theme or q.get("hub_theme") or "").strip().lower()
if ht in ("light", "dark"):
q["hub_theme"] = ht
dest = split.path or "/"
if q:
return f"{dest}?{urlencode(q)}"
return dest + "?embed=1"
split = urlsplit(path or "/")
tab = path_to_embed_tab(split.path)
q = dict(parse_qsl(split.query, keep_blank_values=True))
tests/test_instance_embed_context_lib.py
+28
View File
@@ -0,0 +1,28 @@
from instance_embed_context_lib import embed_render_plan, trade_records_summary
def test_embed_fragment_trade_is_light():
plan = embed_render_plan("trade", "fragment")
assert plan.exchange_capitals is False
assert plan.records_rows is False
assert plan.records_summary is False
assert plan.orders is True
assert plan.key_history is False
def test_embed_shell_trade_summary_only():
plan = embed_render_plan("trade", "shell")
assert plan.exchange_capitals is True
assert plan.records_summary is True
assert plan.records_rows is False
def test_embed_records_page_loads_rows():
plan = embed_render_plan("records", "fragment")
assert plan.records_rows is True
def test_full_page_unchanged():
plan = embed_render_plan("trade", None)
assert plan.records_rows is True
assert plan.exchange_capitals is True