"""桥侧计划 API（07-30 用户需求）：对外提供每日选股计划。 容器常驻命令改为 uvicorn 后随容器启动（docker-compose 已配端口，默认 8300）； cron 的 docker exec 构建/出计划照旧，互不影响。局域网内部服务，v1 无鉴权。 GET /health 存活 + 最新计划日 GET /plan 最新一天的计划（JSON） GET /plan?date=2026-07-30 指定日期 GET /plan?format=md Markdown 原文（浏览器直接可读） GET /plan/dates 可用日期列表 POST /plan/refresh?date=... 重新生成该日计划文件（data/plan/*.md） 统一任务调度平台（XXL-JOB）触发入口挂在 /api/v1/xxl/*（见 xxl.py，2026-08-03）： 盘前链（build → plan → push-pool）可由平台拉起并回调结案，.env 配 XXL_TRIGGER_KEY 才启用。 """ import logging import os import time import threading import pandas as pd from fastapi import FastAPI, HTTPException, Request from fastapi.responses import PlainTextResponse import card import sources import config import db import logic_state_daily import plan import plan_reconcile import regime from xxl import router as xxl_router _access = logging.getLogger("plan.access") app = FastAPI(title="akg-factor-bridge · 每日选股计划", version="0.1") app.include_router(xxl_router) @app.get("/health") def health(): """存活探针。2026-09-03 起带 pool_top 与 pool_max：PMS 的池深探针拿它与自身的计划深度 比较（池深不变式，台账 008），库连不上时也照样返回这两项。""" depth = {"pool_top": config.POOL_TOP, "pool_max": config.POOL_MAX} try: d = plan._latest_date("t_factor_akg_score") # noqa: SLF001 —— 桥内自用 except Exception as e: # noqa: BLE001 —— 库连不上也要能回答"我还活着" return {"ok": False, "error": repr(e), **depth} return {"ok": True, "latest_plan_date": d, **depth} @app.get("/plan/dates") def plan_dates(limit: int = 30): df = db.read_mysql( "factor", "SELECT DISTINCT trade_date FROM t_factor_akg_score " "ORDER BY trade_date DESC LIMIT %s", (int(limit),)) if df.empty: return {"dates": []} return {"dates": [pd.Timestamp(x).date().isoformat() 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, 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, cached = _plan_cached(date, top, obs_top, theme_cap, bool(nocache)) except RuntimeError as e: raise HTTPException(status_code=404, detail=str(e)) 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" data["regime"] = reg or {"status": regime.UNKNOWN, "weak_day": None, "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", 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) if format == "md": return PlainTextResponse(plan.render_md(data), media_type="text/markdown; charset=utf-8") return data @app.post("/plan/refresh") def refresh(date: str | None = None): try: 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} @app.get("/plan/verdict") def plan_verdict(codes: str | None = None, code: str | None = None, date: str | None = None): """逐票『计划判决』（只读）——今日页 / 机会线索页对齐用。 GET /plan/verdict?codes=300750,600438,SH688041 多只（逗号分隔） GET /plan/verdict?code=300750&date=2026-08-04 单只 + 指定档位日 每只票返回 decision（main 主榜 / observe 观察档 / reject 不采纳 / absent 无此票）、 与命令行 plan_reconcile 完全同口径的 verdict_text（页面直接展示的一行解释）， 以及 score/rank/tier/upside/热度/传导/赛道/图谱证据等明细。 date 缺省=当日档位日。三种代码形态都收（600000.SH / SH600000 / 600000）。纯 SELECT，不写任何库。 """ raw = (codes or code or "").strip() want = [c.strip() for c in raw.split(",") if c.strip()] if not want: raise HTTPException(status_code=400, detail="缺少 code / codes 参数") d = date or plan_reconcile.latest_date() if not d: raise HTTPException(status_code=404, detail="档位表为空——先跑当日构建") try: L = plan_reconcile._load(d) # noqa: SLF001 —— 桥内自用只读加载 except Exception as e: # noqa: BLE001 —— 数据层异常统一收成 500 raise HTTPException(status_code=500, detail=f"加载档位数据失败: {e!r}") verdicts = [] for c in want: try: k = plan_reconcile._norm_code(c) # noqa: SLF001 except SystemExit as e: # _norm_code 认不出的形态会 raise SystemExit verdicts.append({"input": c, "error": str(e)}) continue try: v = plan_reconcile.verdict(k, L) except Exception as e: # noqa: BLE001 —— 单票数据异常（如 score 缺行）不崩整批， verdicts.append({"input": c, "code": k, # 与上面认不出形态的处理对称 "error": f"{type(e).__name__}: {e}"}) continue v["input"] = c verdicts.append(v) return {"date": d, "stale": L.get("stale", ""), "count": len(verdicts), "verdicts": verdicts} @app.get("/logic_state") def logic_state_lookup(codes: str | None = None, code: str | None = None, date: str | None = None): """逐票逻辑状态四态（只读）——给 PMS 早上拉完计划后查在持票用（2026-09-07 第三件桥侧前置）。 GET /logic_state?codes=300750,SH688041 多只（逗号分隔） GET /logic_state?code=300750&date=2026-09-04 单只 + 指定数据日 持仓票在候选筛选第一步就被整行剔掉，/plan 里读不到它，所以要有这个入口。 每只票优先回逐票日频表里当日那一行（早上生成计划时落定的，带代码版本，可回溯）， 标 source=daily；当日表里没有这只票（不在档位表、或当日还没生成）就按此刻的数据现算、 按表里的历史做抗抖动，标 source=computed——现算的是"此刻"不是"早上"，复盘别把它当当日态。 date 缺省=档位表最新日（与 /plan 的数据日同口径）。三种代码形态都收。纯 SELECT，不写任何库。 """ raw = (codes or code or "").strip() want = [c.strip() for c in raw.split(",") if c.strip()] if not want: raise HTTPException(status_code=400, detail="缺少 code / codes 参数") d = date or plan._latest_date("t_factor_akg_score") # noqa: SLF001 —— 桥内自用 if not d: raise HTTPException(status_code=404, detail="档位表为空——先跑当日构建") keyed, bad = {}, [] for c in want: try: keyed.setdefault(plan_reconcile._norm_code(c), c) # noqa: SLF001 except SystemExit as e: # _norm_code 认不出的形态会 raise SystemExit bad.append({"input": c, "error": str(e)}) stored = logic_state_daily.lookup(list(keyed), d) todo = [k for k in keyed if k not in stored] computed = {} if todo: try: computed = plan.logic_states_for(todo, d) except Exception as e: # noqa: BLE001 —— 数据层异常统一收成 500 raise HTTPException(status_code=500, detail=f"现算逻辑状态失败: {e!r}") out = [] for k, c in keyed.items(): if k in stored: row = stored[k] out.append({"input": c, "code": k, "date": d, "source": "daily", "verdict": row.get("verdict"), "card_rank": row.get("card_rank"), "plan_version": row.get("plan_version"), **logic_state_daily.row_to_out(row)}) else: out.append({"input": c, "code": k, "date": d, "source": "computed", "verdict": None, "card_rank": None, "plan_version": None, **(plan._state_out(computed.get(k)) or {})}) # noqa: SLF001 # 安全边际三情景随状态一起回（第四件；2026-09-07 拍板持仓页显示中性情景作参考目标价，只显示不触发）。 # 逐票日频表里不存它，每次按此刻现算；算失败整批不带这两键，状态照回。 try: vals = plan.valuations_for(list(keyed), d) for item in out: v = vals.get(item["code"]) item["valuation"] = v item["valuation_text"] = card.valuation_view(v) except Exception as e: # noqa: BLE001 —— 估值算不出不该拖垮状态查询 _access.warning("logic_state valuation failed date=%s err=%r", d, e) # 公司深度两键随状态一起回（2026-09-09 接入方案）：持仓页"证据还在不在"旁显示质地与失效条件。读不到整批不带。 try: crs = sources.company_reviews(list(keyed), d) for item in out: cr = crs.get(item["code"]) item["company_review"] = cr item["company_review_text"] = card.company_review_view(cr) except Exception as e: # noqa: BLE001 _access.warning("logic_state company_review failed date=%s err=%r", d, e) return {"date": d, "count": len(out) + len(bad), "states": out + bad}