From 12a2ab159d0c862caad23b08180f1427141dd809 Mon Sep 17 00:00:00 2001 From: zlt Date: Wed, 9 Sep 2026 13:43:22 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E6=97=B6=E8=AE=A1=E5=88=92=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=8A=A0=E5=BD=93=E6=97=A5=E7=BC=93=E5=AD=98=EF=BC=9A?= =?UTF-8?q?=E8=A3=85=E9=85=8D=E8=A6=81=E4=B8=89=E5=9B=9B=E5=8D=81=E7=A7=92?= =?UTF-8?q?=EF=BC=8CPMS=20=E9=A1=B5=E9=9D=A2=E4=B8=80=E6=AC=A1=E5=8A=A0?= =?UTF-8?q?=E8=BD=BD=E6=8B=89=E5=87=A0=E8=B7=AF=E5=B0=B1=E8=B6=85=E6=97=B6?= =?UTF-8?q?=EF=BC=9B=E9=94=AE=E5=B8=A6=E7=9C=9F=E5=AE=9E=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=97=A5=EF=BC=8C=E9=BB=98=E8=AE=A4=E5=8D=81=E5=88=86=E9=92=9F?= =?UTF-8?q?=EF=BC=8Cnocache=3D1=20=E5=BC=BA=E5=88=B6=E9=87=8D=E7=AE=97?= =?UTF-8?q?=EF=BC=8C=E9=87=8D=E6=96=B0=E7=94=9F=E6=88=90=E8=AE=A1=E5=88=92?= =?UTF-8?q?=E6=97=B6=E6=B8=85=E7=A9=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5 --- api.py | 59 +++++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 54 insertions(+), 5 deletions(-) diff --git a/api.py b/api.py index d3f8f45..67ebfe5 100644 --- a/api.py +++ b/api.py @@ -15,6 +15,8 @@ cron 的 docker exec 构建/出计划照旧，互不影响。局域网内部服 """ import logging import os +import time +import threading import pandas as pd from fastapi import FastAPI, HTTPException, Request @@ -59,18 +61,62 @@ def plan_dates(limit: int = 30): for x in df["trade_date"]]} +# --------------------------------------------------------------------------- +# 实时计划的进程内缓存（2026-09-09）。装配一次要三四十秒：读基座因果论断 20 秒、四类券商 +# 事件 7.6 秒、事件日字段 6 秒（09-08 加的两路），PMS 页面一次加载要拉几路，累计超过它 +# 60 秒的超时，人在页面上看到的就是"上游的选股计划读不到"。 +# 同一数据日、同一参数的装配结果在一天里是确定的（用的全是昨收与昨夜的数据，盘中不变）， +# 所以缓存是安全的：键带数据日与三个参数，过期时间可配，默认十分钟；nocache=1 强制重算。 +# 计划文件每早重新生成，数据日一变缓存自然失效。 +# --------------------------------------------------------------------------- +_PLAN_CACHE: dict = {} +_PLAN_CACHE_LOCK = threading.Lock() +PLAN_CACHE_SEC = int(os.environ.get("PLAN_CACHE_SEC", "600")) +_PLAN_CACHE_MAX = 8 # 不同参数组合最多留几份，防内存慢涨 + + +def _plan_cached(date, top, obs_top, theme_cap, nocache: bool): + """返回 (data, 命中与否)。data 是 plan.collect 的结果（已去掉 _full）。 + + 键里放**真实数据日**而不是请求里的 date：不传 date 时数据日随构建更新，明早换日后 + 旧那一份必须立刻失效，不能等缓存到期。取最新数据日是一条毫秒级查询。""" + ds_key = date + if not ds_key: + try: + ds_key = plan._latest_date("t_factor_akg_score") # noqa: SLF001 —— 同仓自用 + except Exception: # noqa: BLE001 —— 取不到就退回按请求参数缓存 + ds_key = "" + key = (ds_key or "", int(top), int(obs_top), int(theme_cap)) + now = time.time() + if not nocache and PLAN_CACHE_SEC > 0: + with _PLAN_CACHE_LOCK: + hit = _PLAN_CACHE.get(key) + if hit and now - hit[0] < PLAN_CACHE_SEC: + return hit[1], True + data = plan.collect(date, top, obs_top, theme_cap) + data.pop("_full", None) + if PLAN_CACHE_SEC > 0: + with _PLAN_CACHE_LOCK: + _PLAN_CACHE[key] = (now, data) + if len(_PLAN_CACHE) > _PLAN_CACHE_MAX: + for k in sorted(_PLAN_CACHE, key=lambda x: _PLAN_CACHE[x][0])[:-_PLAN_CACHE_MAX]: + _PLAN_CACHE.pop(k, None) + return data, False + + @app.get("/plan") def get_plan(request: Request, date: str | None = None, format: str = "json", - top: int = 20, obs_top: int = 10, theme_cap: int = 5): + top: int = 20, obs_top: int = 10, theme_cap: int = 5, nocache: int = 0): """向下兼容承诺（2026-09-02 方案 2.7）：main / observe 的装配、排序、裁剪与既有字段 一字不动，每行只多联入判决类字段；顶层只新增 generated_at、plan_version、regime、 card_counts、candidates、watch、segments_pointed、snapshot，2026-09-03 再加 market （环境段市场四项，与 regime 一样只从当日快照读，快照缺失为空字典）。PMS 按字段名取值、忽略未知键。""" + t0 = time.time() try: - data = plan.collect(date, top, obs_top, theme_cap) + data, cached = _plan_cached(date, top, obs_top, theme_cap, bool(nocache)) except RuntimeError as e: raise HTTPException(status_code=404, detail=str(e)) - data.pop("_full", None) + data = dict(data) # 下面几个键按请求写，不污染缓存里那一份 ds = data["date"] reg = regime.read_from_snapshot(ds) data["snapshot"] = "present" if os.path.exists(regime.snapshot_path(ds)) else "missing" @@ -78,10 +124,11 @@ def get_plan(request: Request, date: str | None = None, format: str = "json", "source": "当日快照无环境段（08:45 追加未跑或快照缺失）"} data["market"] = regime.read_section(ds, "market") or {} _access.info("plan client=%s date=%s regime=%s generated_at=%s version=%s " - "top=%s obs_top=%s theme_cap=%s", + "top=%s obs_top=%s theme_cap=%s cache=%s 耗时=%.1fs", request.client.host if request.client else "-", ds, data["regime"].get("status"), data.get("generated_at"), - data.get("plan_version"), top, obs_top, theme_cap) + data.get("plan_version"), top, obs_top, theme_cap, + "hit" if cached else "miss", time.time() - t0) if format == "md": return PlainTextResponse(plan.render_md(data), media_type="text/markdown; charset=utf-8") @@ -94,6 +141,8 @@ def refresh(date: str | None = None): out = plan.generate(date) except SystemExit as e: raise HTTPException(status_code=404, detail=str(e)) + with _PLAN_CACHE_LOCK: # 重新生成了计划，实时接口的缓存整体作废 + _PLAN_CACHE.clear() return {"ok": True, "file": out}