diff --git a/api.py b/api.py index 3b206cc..b37f82a 100644 --- a/api.py +++ b/api.py @@ -78,10 +78,33 @@ _PLAN_CACHE_LOCK = threading.Lock() # 保新鲜。12 小时是为了覆盖整个交易日（08:40 预热能管到 20:40），不是为了省算力。 # 当天要强刷仍然走 nocache 参数。 PLAN_CACHE_SEC = int(os.environ.get("PLAN_CACHE_SEC", "43200")) +# 2026-09-10（台账 057）：接口优先读当日全量快照、按请求参数截取，不再每次现场装配。 +# 装配那三十三秒本来就不该在请求里花 —— 计划是日频的，出计划时已经落了一份全量底本。 +# 关掉这个开关就回到每次现算的老路（缓存仍在），是这条路出问题时的一键回退。 +PLAN_FROM_SNAPSHOT = os.environ.get("PLAN_FROM_SNAPSHOT", "1").lower() not in ("0", "false", "no") _PLAN_CACHE_MAX = 8 # 不同参数组合最多留几份，防内存慢涨 _PLAN_BUILD_LOCKS: dict = {} # 每个缓存键一把「正在算」的锁，见 _plan_cached 里的说明 +def _plan_build(date, top, obs_top, theme_cap) -> dict: + """出一份计划。优先读当日全量快照（毫秒级），读不到就现场装配（三十几秒）。 + + 回落是自动的、无声的对下游而言，但**一定会记一行警告**说明为什么没走快照 —— + 如果快照那条路悄悄失效了，只会表现成"页面偶尔变慢"，不记原因就查不出来。 + 返回里的 plan_source 说明这一份是哪来的，snapshot 或 live。 + """ + if PLAN_FROM_SNAPSHOT: + try: + return plan.from_snapshot(date, top, obs_top, theme_cap) + except Exception as e: # noqa: BLE001 —— 任何原因读不到快照都回落，不能让接口挂掉 + _access.warning("plan 快照读取失败, 本次回落到实时装配 (date=%s): %s: %s", + date or "最新", type(e).__name__, e) + data = plan.collect(date, top, obs_top, theme_cap) + data.pop("_full", None) + data["plan_source"] = "live" + return data + + def _plan_cached(date, top, obs_top, theme_cap, nocache: bool): """返回 (data, 命中与否)。data 是 plan.collect 的结果（已去掉 _full）。 @@ -117,8 +140,7 @@ def _plan_cached(date, top, obs_top, theme_cap, nocache: bool): hit = _PLAN_CACHE.get(key) if hit and time.time() - hit[0] < PLAN_CACHE_SEC: return hit[1], True - data = plan.collect(date, top, obs_top, theme_cap) - data.pop("_full", None) + data = _plan_build(date, top, obs_top, theme_cap) if PLAN_CACHE_SEC > 0: with _PLAN_CACHE_LOCK: _PLAN_CACHE[key] = (time.time(), data) @@ -149,11 +171,12 @@ 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 cache=%s 耗时=%.1fs", + "top=%s obs_top=%s theme_cap=%s cache=%s source=%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, - "hit" if cached else "miss", time.time() - t0) + "hit" if cached else "miss", data.get("plan_source") or "-", + time.time() - t0) if format == "md": return PlainTextResponse(plan.render_md(data), media_type="text/markdown; charset=utf-8") diff --git a/plan.py b/plan.py index c70d657..c36b555 100644 --- a/plan.py +++ b/plan.py @@ -855,6 +855,68 @@ def render_md(d: dict) -> str: return "\n".join(L) +def pick_rows(rows, n: int, theme_cap: int) -> list: + """从**已按分数排好序**的行里取 n 条，每个传导主题最多 theme_cap 条（0=不设限）。 + + 与 collect 里那个同名的内部函数是同一套规则，差别只在取主题的来源：那边回查证据映射， + 这边直接读行里的 evidence.theme —— 装配时就把主题写进每一行了，快照里也带着。 + 两处规则必须一致，test_plan_snapshot.py 拿真实快照逐行比对钉住这件事。 + """ + out, cnt = [], {} + for r in rows: + theme = ((r.get("evidence") or {}).get("theme")) or "（无传导）" + if theme_cap and cnt.get(theme, 0) >= theme_cap: + continue + cnt[theme] = cnt.get(theme, 0) + 1 + out.append(r) + if len(out) >= n: + break + return out + + +def from_snapshot(date: str | None = None, top: int = 20, obs_top: int = 10, + theme_cap: int = 5) -> dict: + """从当日 JSON 快照读计划，按请求参数截取。**不做任何装配、不查任何库。** + + 为什么要这条路（2026-09-10，台账 057）: collect 每次请求都现场装配整池，实测三十三秒， + 而选股计划是日频的、用的全是昨收与昨夜数据 —— 一天算一次就够，不该放在请求里现算。 + 出计划时本来就落了一份全量快照（不裁剪、不设主题限额），注释里写明它是"复盘与对账的 + 唯一底本"，只是接口一直没读它。这条路就是把接口接到那份底本上。 + + 与 collect 的输出逐字一致，只差三处、都是设计使然: + · main / observe 按本次请求的参数从全量截取，名次重编成连续的 + · theme_cap 按本次请求写，不用落盘时那个 + · 多两个键 plan_source 与 snapshot_generated_at，让人一眼看出这份是哪来的、多新 + 快照不存在、读不动、或数据日对不上，一律抛异常，由调用方回落到 collect。 + """ + ds = date or _latest_date("t_factor_akg_score") + if not ds: + raise RuntimeError("t_factor_akg_score 还没有数据——先 build akg_score。") + path = os.path.join(config.PLAN_SNAPSHOT_DIR, f"plan_{ds}.json") + if not os.path.exists(path): + raise FileNotFoundError(f"{ds} 的计划快照不存在: {path}") + with open(path, encoding="utf-8") as f: + snap = json.load(f) + if str(snap.get("date") or "") != str(ds): + raise RuntimeError(f"快照里的数据日 {snap.get('date')} 与请求的 {ds} 对不上: {path}") + full_main, full_obs = snap.get("main"), snap.get("observe") + if not isinstance(full_main, list) or not isinstance(full_obs, list): + raise RuntimeError(f"快照缺全量主榜或观察档: {path}") + + data = {k: v for k, v in snap.items() + # 前三个是落盘那次的裁剪结果与参数，与本次请求无关； + # 后两个由接口每次实时读，留在这里会被覆盖，删掉免得看的人以为快照说了算。 + if k not in ("main_shown", "observe_shown", "shown_params", "market", "regime")} + data["main"] = [{**r, "rank": i} + for i, r in enumerate(pick_rows(full_main, top, theme_cap), 1)] + data["observe"] = [{**r, "rank": i} + for i, r in enumerate(pick_rows(full_obs, obs_top, theme_cap), 1)] + data["theme_cap"] = theme_cap + data["plan_source"] = "snapshot" + data["snapshot_generated_at"] = snap.get("generated_at") + return data + + def generate(date: str | None = None, top: int = 20, obs_top: int = 10, theme_cap: int = 5) -> str: try: diff --git a/test_plan_snapshot.py b/test_plan_snapshot.py new file mode 100644 index 0000000..b7ed3e4 --- /dev/null +++ b/test_plan_snapshot.py @@ -0,0 +1,142 @@ +"""接口改读当日全量快照（2026-09-10，台账 057）：截取规则、名次重编、回落、字段一致。 + +要解决的问题：计划接口每次请求都现场装配整池，实测三十三秒。而计划是日频的、用的全是 +昨收与昨夜数据，出计划时本来就落了一份全量快照（不裁剪、不设主题限额），接口一直没读它。 + +离线，不连库不起服务。跑法：python3 test_plan_snapshot.py，预期最后一行是 ALL OK。 +真实快照存在时会额外做一轮逐行比对，不存在就跳过那一轮并说明。 +""" +import glob +import json +import os +import sys +import types + +for _n in ("pandas", "pymysql", "psycopg"): + if _n not in sys.modules: + try: + __import__(_n) + except Exception: # noqa: BLE001 + _m = types.ModuleType(_n) + if _n == "pandas": + _m.DataFrame = type("DataFrame", (), {}) + sys.modules[_n] = _m + +import config # noqa: E402 +import plan # noqa: E402 + +FAILED = [] + + +def t(name, cond, extra=""): + print((" ok " if cond else " FAIL ") + name + ((" " + str(extra)) if not cond else "")) + if not cond: + FAILED.append(name) + + +def _row(code, theme, rank): + return {"code": code, "rank": rank, "score": 100.0 - rank, + "evidence": ({"theme": theme} if theme else None)} + + +def test_pick_rules(): + print("[截取规则]") + rows = [_row("A", "光伏", 1), _row("B", "光伏", 2), _row("C", "光伏", 3), + _row("D", "军工", 4), _row("E", None, 5)] + t("不设限就按顺序取", [r["code"] for r in plan.pick_rows(rows, 3, 0)] == ["A", "B", "C"]) + got = plan.pick_rows(rows, 4, 2) + t("每个主题最多两条，第三条光伏被跳过", + [r["code"] for r in got] == ["A", "B", "D", "E"], [r["code"] for r in got]) + t("没有主题的行归到同一个桶，不当成各自独立", + [r["code"] for r in plan.pick_rows( + [_row("X", None, 1), _row("Y", None, 2), _row("Z", "光伏", 3)], 3, 1)] == ["X", "Z"]) + t("要的条数比总行数多就全给", len(plan.pick_rows(rows, 99, 0)) == 5) + t("要零条就给零条", plan.pick_rows(rows, 0, 0) == []) + + +def test_replay_against_real_snapshot(): + print("[拿真实快照逐行比对 —— 这条是这次改动成立的根据]") + fs = sorted(glob.glob(os.path.join(config.PLAN_SNAPSHOT_DIR, "plan_*.json"))) + if not fs: + print(" -- 本机没有快照文件，跳过这一轮（在部署机上跑才有）") + return + p = fs[-1] + snap = json.load(open(p, encoding="utf-8")) + sp = snap.get("shown_params") or {} + if not sp or not isinstance(snap.get("main"), list): + print(" -- 最新快照没有全量段或落盘参数，跳过：" + os.path.basename(p)) + return + print(" 用 %s（落盘参数 %s）" % (os.path.basename(p), sp)) + for seg, shown_key, n_key in (("main", "main_shown", "top"), + ("observe", "observe_shown", "obs_top")): + got = plan.pick_rows(snap[seg], sp[n_key], sp["theme_cap"]) + shown = snap[shown_key] + t("%s 截出来的代码序列与当时下发的完全一致" % seg, + [r["code"] for r in got] == [r["code"] for r in shown], + ([r["code"] for r in got][:6], [r["code"] for r in shown][:6])) + # 名次是唯一该不一样的字段：全量快照里存的是全量排名，下发时要重编成连续名次 + diff = {k for a, b in zip(got, shown) for k in set(a) | set(b) if a.get(k) != b.get(k)} + t("%s 除名次外逐字段一致" % seg, diff <= {"rank"}, diff) + renum = [{**r, "rank": i} for i, r in enumerate(got, 1)] + t("%s 名次重编后与当时下发的一模一样" % seg, renum == shown) + + +def test_from_snapshot(tmpdir=None): + print("[from_snapshot 的输出与回落]") + fs = sorted(glob.glob(os.path.join(config.PLAN_SNAPSHOT_DIR, "plan_*.json"))) + if not fs: + print(" -- 本机没有快照文件，跳过这一轮") + return + ds = os.path.basename(fs[-1])[len("plan_"):-len(".json")] + d = plan.from_snapshot(ds, top=5, obs_top=3, theme_cap=2) + t("名次是连续的 1 到 n", [r["rank"] for r in d["main"]] == list(range(1, len(d["main"]) + 1))) + t("条数不超过请求的", len(d["main"]) <= 5 and len(d["observe"]) <= 3) + t("主题限额按本次请求生效，不是落盘那次的", d["theme_cap"] == 2) + t("标明了这份是从快照来的", d.get("plan_source") == "snapshot") + t("带上快照的生成时刻，一眼看得出多新", bool(d.get("snapshot_generated_at"))) + for k in ("main_shown", "observe_shown", "shown_params"): + t("不把落盘那次的裁剪结果带出去（%s）" % k, k not in d) + for k in ("market", "regime"): + t("不把快照里的 %s 带出去（接口每次实时读）" % k, k not in d) + for k in ("date", "generated_at", "plan_version", "counts", "candidates", "watch", + "segments_pointed", "card_counts", "card_params", "changes", "encoding"): + t("顶层保留 %s" % k, k in d) + # 全量计数不随裁剪变化：counts 说的是整池有多少，不是这次给了几条 + t("全量计数原样保留", (d.get("counts") or {}).get("main", 0) >= len(d["main"])) + # 认不出的日期、对不上的日期一律抛异常，交给调用方回落 + try: + plan.from_snapshot("1999-01-01", 5, 3, 2) + t("快照不存在要抛异常", False) + except Exception as e: # noqa: BLE001 + t("快照不存在要抛异常", isinstance(e, (FileNotFoundError, RuntimeError)), type(e).__name__) + + +def test_api_falls_back(): + print("[接口回落：读不到快照不能让接口挂掉]") + src = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "api.py"), + encoding="utf-8").read() + t("有总开关，能一键回到每次现算的老路", "PLAN_FROM_SNAPSHOT" in src) + t("快照那条路包了兜底，任何异常都回落", "except Exception" in src and "回落到实时装配" in src) + t("回落一定记一行警告说明原因（不然只会表现成偶尔变慢）", + "_access.warning" in src and "快照读取失败" in src) + i_try = src.index("plan.from_snapshot") + i_live = src.index("plan.collect(date, top, obs_top, theme_cap)") + t("先试快照、后现算", i_try < i_live) + t("访问日志写明这次走的哪条路", "source=%s" in src) + + +def main(): + test_pick_rules() + test_replay_against_real_snapshot() + test_from_snapshot() + test_api_falls_back() + print("-" * 70) + if FAILED: + print("FAILED %d: %s" % (len(FAILED), "; ".join(FAILED))) + return 1 + print("ALL OK — 计划读快照：截取规则 / 与真实快照逐行一致 / 输出与回落 / 接口接线") + return 0 + + +if __name__ == "__main__": + sys.exit(main())