akg-factor-bridge/api.py

"""桥侧计划 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()
# 2026-09-10: 默认从 600 秒（10 分钟）改成 12 小时。
# 起因: 09-10 开盘前 PMS 在 08:40 预热了一次（花 39 秒），缓存 08:50 就过期；用户 09:02
# 打开页面正好撞上冷启动，页面几路并发一起等，叠加破了 60 秒超时。
# 为什么可以放这么长: 缓存键里带的是**真实数据日**，上游换日后旧那份立刻失效，不靠时间
# 保新鲜。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）。

    键里放**真实数据日**而不是请求里的 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

    # 同一个键同时只算一次 (2026-09-10)。原来 plan.collect 在锁外面, 缓存冷的时候页面
    # 几路并发进来会**各跑一遍**这趟三十多秒的装配, 还互相抢同一批库连接, 实际耗时远超
    # 单跑一次 —— 这正是 09-10 早上页面破 60 秒超时的那一半原因。
    # 现在让第一个请求去算, 后到的等它算完直接吃缓存。等待用的是每个键自己的锁,
    # 不同参数组合互不阻塞。
    with _PLAN_CACHE_LOCK:
        lock = _PLAN_BUILD_LOCKS.get(key)
        if lock is None:
            lock = _PLAN_BUILD_LOCKS[key] = threading.Lock()
    with lock:
        # 双重检查: 排在后面的请求进到这里时, 前一个多半已经把结果写进缓存了
        if not nocache and PLAN_CACHE_SEC > 0:
            with _PLAN_CACHE_LOCK:
                hit = _PLAN_CACHE.get(key)
                if hit and time.time() - hit[0] < PLAN_CACHE_SEC:
                    return hit[1], True
        data = _plan_build(date, top, obs_top, theme_cap)
        if PLAN_CACHE_SEC > 0:
            with _PLAN_CACHE_LOCK:
                _PLAN_CACHE[key] = (time.time(), 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)
                        _PLAN_BUILD_LOCKS.pop(k, None)
    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):
    """向下兼容承诺（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 {}
    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 评析补%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 "-",
                 n_cr, 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}