计划接口改读当日全量快照，不再每次现场装配（台账 057）
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
zlt
parent
d631a31ebb
commit
abd7fec110
31
api.py
31
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")
|
||||
|
|
|
|||
62
plan.py
62
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:
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
Loading…
Reference in New Issue