"""中控 AI:今日总结生成。""" from __future__ import annotations from typing import Any from hub_ai.client import generate_text, model_label from hub_ai.context import build_daily_context from hub_ai.prompts import SUMMARY_SYSTEM, build_summary_user_prompt from hub_ai.store import append_summary, get_latest_summary, list_summaries def generate_daily_summary( exchanges: list[dict], *, trading_day: str | None = None, force: bool = False, ) -> dict[str, Any]: ctx = build_daily_context(exchanges, trading_day=trading_day) day = ctx["trading_day"] if not force: latest = get_latest_summary(day) if latest and latest.get("context_hash") == ctx.get("context_hash"): return { "ok": True, "cached": True, "trading_day": day, "summary": latest, "model": latest.get("model") or model_label(), } system = SUMMARY_SYSTEM.replace("{trading_day}", day) user = build_summary_user_prompt(ctx["text"], day) content = generate_text(system=system, user=user, temperature=0.15) if content.startswith("AI 调用失败"): return {"ok": False, "msg": content, "trading_day": day} stats_snapshot = { "totals": ctx.get("totals"), "by_account": { str(ac.get("key") or ac.get("id")): { "name": ac.get("name"), "status": ac.get("status"), "pnl_u": (ac.get("trade_stats") or {}).get("total_pnl_u"), "closed_count": (ac.get("trade_stats") or {}).get("closed_count"), "float_pnl_u": ac.get("float_pnl_u"), "issues": ac.get("issues") or [], } for ac in ctx.get("accounts") or [] }, } row = append_summary( trading_day=day, content_md=content, model=model_label(), context_hash=ctx.get("context_hash") or "", stats_snapshot=stats_snapshot, ) return { "ok": True, "cached": False, "trading_day": day, "summary": row, "model": model_label(), "context": ctx, } def summary_list(trading_day: str | None = None) -> list[dict]: return list_summaries(trading_day=trading_day)