diff --git a/api.py b/api.py index b37f82a..44058fd 100644 --- a/api.py +++ b/api.py @@ -151,6 +151,53 @@ def _plan_cached(date, top, obs_top, theme_cap, nocache: bool): return data, False +def _attach_fresh_reviews(data: dict, ds: str) -> int: + """把个股深度评析按请求实时补一次，返回补了几只。 + + 为什么非补不可。评析是全天在跑的（数据基座每小时一批），而计划一天只装配 + 一次、这个接口又带十二小时缓存。于是装配那一刻还没出报告的票，当天就再也 + 补不上了 —— 2026-09-10 实测：早上七点十分那份计划里 154 行只有 7 只带评析； + 中午十二点二十三到三十五分之间新跑出五份报告，而这五只正是持仓管理系统当天 + 出提议的票，它当天一份都看不到，卡片上写的全是「数据基座还没出这家的报告」。 + + 三层原因叠在一起：评析的批次时刻晚于出计划、计划一天只装配一次、接口缓存 + 十二小时。这一段一次解掉后两层 —— 评析什么时候跑完都行，跑完下一次请求就带上。 + + 做法与上面 regime、market 两个键同一个路子：按请求补、不进缓存。代价是每次 + 请求多一条数据库查询（一次批量取，不是逐票取）。 + + 注意浅拷贝：get_plan 里的 dict(data) 只拷了顶层，main / observe 两个列表里的 + 行对象与缓存里那一份是同一批。要改行，必须先把那一行拷出来，否则会把当次 + 请求的结果写进缓存、污染全天。 + """ + rows = list(data.get("main") or []) + list(data.get("observe") or []) + codes = [r.get("code") for r in rows if isinstance(r, dict) and r.get("code")] + if not codes: + return 0 + try: + fresh = sources.company_reviews(codes, ds) + except Exception as e: # noqa: BLE001 —— 评析读不到绝不拖垮计划接口 + _access.warning("plan 评析实时补充失败, 本次沿用装配时那一份: %r", e) + return 0 + if not fresh: + return 0 + n = 0 + for key in ("main", "observe"): + out, changed = [], False + for r in (data.get(key) or []): + cr = fresh.get(r.get("code")) if isinstance(r, dict) else None + if cr: + r = dict(r) # 先拷贝再改, 别碰缓存里那一份 + r["company_review"] = cr + r["company_review_text"] = card.company_review_view(cr) + changed = True + n += 1 + out.append(r) + if changed: + data[key] = out + return n + + @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, nocache: int = 0): @@ -170,13 +217,14 @@ def get_plan(request: Request, date: str | None = None, format: str = "json", data["regime"] = reg or {"status": regime.UNKNOWN, "weak_day": None, "source": "当日快照无环境段（08:45 追加未跑或快照缺失）"} data["market"] = regime.read_section(ds, "market") or {} + n_cr = _attach_fresh_reviews(data, ds) _access.info("plan client=%s date=%s regime=%s generated_at=%s version=%s " - "top=%s obs_top=%s theme_cap=%s cache=%s source=%s 耗时=%.1fs", + "top=%s obs_top=%s theme_cap=%s cache=%s source=%s 评析补%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", data.get("plan_source") or "-", - time.time() - t0) + n_cr, time.time() - t0) if format == "md": return PlainTextResponse(plan.render_md(data), media_type="text/markdown; charset=utf-8") diff --git a/test_plan_reviews_live.py b/test_plan_reviews_live.py new file mode 100644 index 0000000..e92c3a4 --- /dev/null +++ b/test_plan_reviews_live.py @@ -0,0 +1,172 @@ +"""计划接口按请求实时补个股深度评析（2026-09-10）。 + +## 为什么要有这一段 + +个股深度评析是全天在跑的（数据基座每小时一批），而计划一天只装配一次、 +接口又带十二小时缓存。于是装配那一刻还没出报告的票，当天就再也补不上。 + +2026-09-10 实测：早上七点十分那份计划里 154 行只有 7 只带评析；中午十二点二十三 +到三十五分之间新跑出五份报告，而这五只（688556.SH、605598.SH、300118.SZ、 +688698.SH、300638.SZ）正是持仓管理系统当天出提议的票 —— 它当天一份都看不到， +卡片上写的全是「数据基座还没出这家的报告」。 + +三层原因叠在一起：评析的批次时刻晚于出计划、计划一天只装配一次、接口缓存十二小时。 +实时补充一次解掉后两层：评析什么时候跑完都行，跑完下一次请求就带上。 + +## 这里钉住什么 + +一、有新报告的行会被补上（结构化字段与整句都补）。 +二、**补充绝不能污染缓存**。get_plan 里的 dict(data) 只拷了顶层，main / observe + 里的行对象与缓存里那一份是同一批；不先拷贝就改，等于把当次请求的结果写进缓存， + 全天所有人都会拿到它。这一条是这个文件存在的主要理由。 +三、上游读不到时静默沿用装配时那一份，不抛异常、不拖垮计划接口。 +四、没有任何一只票有新报告时，行对象原样返回，不做无谓的拷贝。 + +离线，不连库不起服务（上游取数用替身）。但要在容器里跑 —— api.py 的依赖链里有 +`str | None` 这种写法，需要 Python 3.10 以上，而开发机是 3.9。 + +跑法（在 155 上）： + docker exec akg_factor_bridge python3 test_plan_reviews_live.py +预期最后一行是 ALL OK。 +""" +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 + +# 开发机上多半没装 fastapi（它只在容器里跑）。造一个够用的替身， +# 这样这份测试在开发机与容器里都能跑，测的仍然是 api.py 里的真代码。 +if "fastapi" not in sys.modules: + try: + __import__("fastapi") + except Exception: # noqa: BLE001 + _fa = types.ModuleType("fastapi") + + class _StubApp: + def get(self, *a, **k): + return lambda f: f + + def post(self, *a, **k): + return lambda f: f + + _fa.FastAPI = lambda *a, **k: _StubApp() + _fa.HTTPException = type("HTTPException", (Exception,), {}) + _fa.Request = type("Request", (), {}) + sys.modules["fastapi"] = _fa + _fr = types.ModuleType("fastapi.responses") + _fr.PlainTextResponse = type("PlainTextResponse", (), {}) + sys.modules["fastapi.responses"] = _fr + +import api # noqa: E402 +import sources # noqa: E402 + + +def t(name, cond, extra=""): + print((" ok " if cond else " FAIL ") + name + ((" " + str(extra)) if not cond else "")) + assert cond, name + + +def _plan(with_review_on=()): + """造一份计划：main 两行、observe 一行，指定哪些代码在装配时就已经有评析。""" + def row(code): + r = {"code": code, "name": code, "rank": 1} + if code in with_review_on: + r["company_review"] = {"overall": "旧", "period": "2025Q4"} + r["company_review_text"] = "公司深度（旧那一份）" + return r + return {"date": "2026-09-09", + "main": [row("SH688556"), row("SZ300118")], + "observe": [row("SZ300638")]} + + +def _fake_reviews(hit: dict): + """替身：sources.company_reviews 的返回形状是 {前缀码: 摘要字典}，没报告的票不在里面。""" + def f(codes, ds, **kw): + return {c: hit[c] for c in codes if c in hit} + return f + + +def main(): + print("=" * 62) + print("计划接口实时补评析") + print("=" * 62) + orig = sources.company_reviews + try: + # 一、有新报告的行被补上 + print("\n[A] 补上新报告") + data = _plan() + sources.company_reviews = _fake_reviews({ + "SH688556": {"overall": "差", "period": "2026Q1", "thesis": "论点一句"}, + "SZ300638": {"overall": "中", "period": "2026Q1"}, + }) + n = api._attach_fresh_reviews(data, "2026-09-09") + t("补了两只（主榜一只、观察档一只）", n == 2, n) + t("主榜那只拿到结构化字段", (data["main"][0].get("company_review") or {}).get("overall") == "差") + t("主榜那只拿到整句", "公司深度" in (data["main"][0].get("company_review_text") or "")) + t("观察档那只也补上了", (data["observe"][0].get("company_review") or {}).get("overall") == "中") + t("没报告的那只原样不动", "company_review" not in data["main"][1]) + + # 二、不污染缓存 —— 这一条是重点 + print("\n[B] 不许污染缓存（行对象与缓存共用，必须先拷贝再改）") + cached = _plan() + keep_main0 = cached["main"][0] # 记住缓存里那个行对象本身 + keep_obs0 = cached["observe"][0] + served = dict(cached) # get_plan 里就是这么浅拷一层 + sources.company_reviews = _fake_reviews({ + "SH688556": {"overall": "差", "period": "2026Q1"}, + "SZ300638": {"overall": "中", "period": "2026Q1"}, + }) + api._attach_fresh_reviews(served, "2026-09-09") + t("这次请求拿到了补充", (served["main"][0].get("company_review") or {}).get("overall") == "差") + t("缓存里那个行对象没被写进 company_review", "company_review" not in keep_main0) + t("缓存里观察档那行也没被写", "company_review" not in keep_obs0) + t("补过之后返回的是新的行对象", served["main"][0] is not keep_main0) + t("缓存里那份列表本身没被改", cached["main"][0] is keep_main0) + + # 三、覆盖旧版本 + print("\n[C] 装配时是旧报告，现在有新的 → 覆盖") + data = _plan(with_review_on=("SH688556",)) + sources.company_reviews = _fake_reviews({"SH688556": {"overall": "好", "period": "2026Q2"}}) + api._attach_fresh_reviews(data, "2026-09-09") + t("覆盖成新那一份", (data["main"][0].get("company_review") or {}).get("overall") == "好") + t("整句跟着换", "旧那一份" not in (data["main"][0].get("company_review_text") or "")) + + # 四、上游读不到 → 静默沿用，不抛 + print("\n[D] 上游读不到时不许拖垮计划接口") + data = _plan(with_review_on=("SH688556",)) + def _boom(codes, ds, **kw): + raise RuntimeError("PG 连不上") + sources.company_reviews = _boom + n = api._attach_fresh_reviews(data, "2026-09-09") + t("补了零只、没有抛异常", n == 0, n) + t("装配时那一份原样保留", (data["main"][0].get("company_review") or {}).get("overall") == "旧") + + # 五、一只都没有新报告 → 不做无谓拷贝 + print("\n[E] 一只都没有新报告时不动行对象") + data = _plan() + keep = data["main"][0] + sources.company_reviews = _fake_reviews({}) + n = api._attach_fresh_reviews(data, "2026-09-09") + t("补了零只", n == 0, n) + t("行对象原样，没有多余拷贝", data["main"][0] is keep) + + # 六、空计划不炸 + print("\n[F] 空计划") + n = api._attach_fresh_reviews({"date": "2026-09-09", "main": [], "observe": []}, "2026-09-09") + t("补了零只、没有抛异常", n == 0, n) + finally: + sources.company_reviews = orig + + print("\nALL OK") + + +if __name__ == "__main__": + main()