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 pandas as pd
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import PlainTextResponse

import card
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"]]}


@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):
    """向下兼容承诺（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 按字段名取值、忽略未知键。"""
    try:
        data = plan.collect(date, top, obs_top, theme_cap)
    except RuntimeError as e:
        raise HTTPException(status_code=404, detail=str(e))
    data.pop("_full", None)
    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",
                 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)
    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))
    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)
    return {"date": d, "count": len(out) + len(bad), "states": out + bad}