dekun
32079bb4c2
日历数据改为安全 JSON 内嵌，仅统计页构建；构建失败时降级为空，避免拖垮其他页面。 Co-authored-by: Cursor <cursoragent@cursor.com>
116 lines
3.5 KiB
Python
116 lines
3.5 KiB
Python
"""按交易日聚合实例 trade_records 盈亏，供统计分析页日历 API 使用。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from typing import Any, Callable
|
|
|
|
|
|
def build_trade_stats_calendar(
|
|
pnls: list[tuple],
|
|
year: int,
|
|
month: int,
|
|
segment_key: str,
|
|
row_matches_fn: Callable[[Any, str], bool],
|
|
*,
|
|
reset_hour: int = 8,
|
|
) -> dict[str, Any]:
|
|
"""pnls: _load_completed_trade_pnls 返回值 (pnl, close_dt, trading_day, row)。"""
|
|
y = int(year)
|
|
m = int(month)
|
|
if m < 1 or m > 12:
|
|
raise ValueError("month 无效")
|
|
first = f"{y:04d}-{m:02d}-01"
|
|
if m == 12:
|
|
next_first = datetime(y + 1, 1, 1)
|
|
else:
|
|
next_first = datetime(y, m + 1, 1)
|
|
last = (next_first - timedelta(days=1)).strftime("%Y-%m-%d")
|
|
seg = (segment_key or "all").strip() or "all"
|
|
days: dict[str, dict[str, Any]] = {}
|
|
for pnl, _close_dt, td, row in pnls:
|
|
if not td or td < first or td > last:
|
|
continue
|
|
if not row_matches_fn(row, seg):
|
|
continue
|
|
bucket = days.setdefault(
|
|
td,
|
|
{
|
|
"trading_day": td,
|
|
"open_count": 0,
|
|
"pnl_total": 0.0,
|
|
"turnover_total": 0.0,
|
|
"commission_total": 0.0,
|
|
"has_sick": False,
|
|
"sick_count": 0,
|
|
},
|
|
)
|
|
bucket["open_count"] += 1
|
|
bucket["pnl_total"] += float(pnl or 0)
|
|
try:
|
|
bucket["turnover_total"] += float(row["exchange_turnover_usdt"] or 0)
|
|
except (TypeError, ValueError, KeyError):
|
|
pass
|
|
try:
|
|
bucket["commission_total"] += float(row["exchange_commission_usdt"] or 0)
|
|
except (TypeError, ValueError, KeyError):
|
|
pass
|
|
for d in days.values():
|
|
d["pnl_total"] = round(float(d["pnl_total"]), 4)
|
|
d["turnover_total"] = round(float(d["turnover_total"]), 4)
|
|
d["commission_total"] = round(float(d["commission_total"]), 4)
|
|
month_pnl = sum(float(d["pnl_total"]) for d in days.values())
|
|
month_count = sum(int(d["open_count"]) for d in days.values())
|
|
return {
|
|
"year": y,
|
|
"month": m,
|
|
"date_from": first,
|
|
"date_to": last,
|
|
"segment": seg,
|
|
"reset_hour": int(reset_hour),
|
|
"days": days,
|
|
"month_pnl_total": round(month_pnl, 4),
|
|
"month_open_count": month_count,
|
|
}
|
|
|
|
|
|
def build_initial_stats_calendar(
|
|
pnls: list[tuple],
|
|
now_dt: datetime,
|
|
row_matches_fn: Callable[[Any, str], bool],
|
|
*,
|
|
reset_hour: int = 8,
|
|
segment_key: str = "all",
|
|
) -> dict[str, Any]:
|
|
"""统计页首屏内嵌日历（当前自然月、默认品类）。"""
|
|
return build_trade_stats_calendar(
|
|
pnls,
|
|
now_dt.year,
|
|
now_dt.month,
|
|
segment_key,
|
|
row_matches_fn,
|
|
reset_hour=reset_hour,
|
|
)
|
|
|
|
|
|
def build_stats_calendar_bootstrap(
|
|
pnls: list[tuple],
|
|
now_dt: datetime,
|
|
row_matches_fn: Callable[[Any, str], bool],
|
|
*,
|
|
reset_hour: int = 8,
|
|
segment_key: str = "all",
|
|
) -> tuple[dict[str, Any] | None, str | None]:
|
|
"""返回 (payload, json_str)；失败时 (None, None)，供模板安全内嵌。"""
|
|
try:
|
|
payload = build_initial_stats_calendar(
|
|
pnls,
|
|
now_dt,
|
|
row_matches_fn,
|
|
reset_hour=reset_hour,
|
|
segment_key=segment_key,
|
|
)
|
|
return payload, json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
|
except Exception:
|
|
return None, None
|