tradingSystem/app/web/main.py

# -*- coding: utf-8 -*-
"""
PMS 管理页面 · Web 入口 (FastAPI + 单页)
=========================================
设计 §3.3 四块: 参数设置 / 命令台 / 持仓与账本 / 提议确认。

工程原则:
  * 任何后端异常都不得让页面开不了 —— 全部 API 走 `ok(...)` 包装, 失败返回
    {"ok": false, "error": "..."} 且 HTTP 200, 由前端在顶部横幅提示。
  * 页面只读参数一律经 ParamStore (表值优先), 改参即持久化到 pms_runtime_param。
  * 手动运维按钮 (回放/对账/日终/日报) 与调度器调用同一份服务函数, 便于未接调度时先验证。
"""
from __future__ import annotations

import json
import logging
import os
from datetime import datetime, timedelta

from fastapi import Body, FastAPI, Query, Request
from fastapi.responses import FileResponse, JSONResponse

from config.settings import settings
from app.core import command_spec as cs
from app.core import tradedays as td
from app.db import session as dbs
from app.repo import downstream_repo, pms_repo
from app.services import (command_service, industry, ledger_service, logic_state_service,
                          param_store, portfolio)
from app.web import auth as authmod

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)s [%(name)s] %(message)s")
logger = logging.getLogger("pms.web")

VERSION = "0.2.0-dev"
STATIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static")

app = FastAPI(title="PMS 持仓管理系统", version=VERSION)

# 把 static 目录挂到 /static —— 内网隔离时的离线兜底:
# 页面默认从 unpkg 取 Vue3 / ElementPlus / axios, 浏览器上不了外网就会白屏 (更准确地说是
# 只剩一个光秃秃的头部, 因为 el-* 组件全渲染不出来)。把三个库放进 static/vendor/ 再把
# index.html 头部那五个 URL 换成 /static/vendor/xxx 即可离线运行, 无需构建步骤。
# 具体做法见 README「管理页面」一节。
try:
    from fastapi.staticfiles import StaticFiles
    if os.path.isdir(STATIC_DIR):
        app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
except Exception as _e:            # 挂不上不影响任何 API, 页面照常从 CDN 取
    logger.warning("静态目录挂载失败 (不影响接口): %s", _e)


# ================================================================ 登录闸 (中间件)
@app.middleware("http")
async def _auth_guard(request: Request, call_next):
    """除公开白名单外都要有效会话票; 交易员碰管理员接口回 403。
    关键: 中间件里只 return JSONResponse, 不 raise —— 在这里抛异常会变成 500 而非 401/403。"""
    if not settings.PMS_AUTH_ENABLED:
        return await call_next(request)
    path = request.url.path
    if authmod.is_public_path(path):
        return await call_next(request)
    if not settings.PMS_SESSION_SECRET:
        # 开着登录却没配密钥: 失败即关门, 绝不用空密钥签/验票
        logger.error("PMS_AUTH_ENABLED 开着但 PMS_SESSION_SECRET 为空 —— 受保护请求一律拒绝")
        return JSONResponse({"ok": False, "error": "服务未配置会话密钥 PMS_SESSION_SECRET"},
                            status_code=503)
    sess = authmod.verify_session(request.cookies.get(settings.PMS_SESSION_COOKIE),
                                  settings.PMS_SESSION_SECRET)
    if not sess:
        return JSONResponse({"ok": False, "error": "未登录或登录已过期", "auth": "login_required"},
                            status_code=401)
    if (authmod.authz_decision(request.method, path) == "admin"
            and not authmod.is_admin(sess.get("roles"))):
        return JSONResponse({"ok": False, "error": "无权限: 该操作需要系统管理员", "auth": "forbidden"},
                            status_code=403)
    request.state.user = sess
    return await call_next(request)


def ok(fn, *args, **kw):
    """统一出参包装: 成功 {"ok":true, ...}; 失败 {"ok":false,"error":...} 且 HTTP 200。"""
    try:
        data = fn(*args, **kw)
        if isinstance(data, dict) and "ok" in data:
            return data
        return {"ok": True, "data": data}
    except Exception as e:
        logger.exception("API 失败: %s", getattr(fn, "__name__", fn))
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def _actor(request: Request, payload: dict = None) -> str:
    """操作人一律取**会话里的登录身份** (2026-08-28 审查修): 操作日志是问责凭证,
    原来所有写接口的操作人取自请求体 payload.by —— 任何登录用户都能在 JSON 里写别人的
    名字, 把下单撤单记到他人头上。认证身份中间件已放在 request.state.user, 就用它;
    关闭登录 (PMS_AUTH_ENABLED=False) 时才退回请求体与 "user" 兜底。"""
    u = getattr(request.state, "user", None) or {}
    name = str(u.get("username") or u.get("phone") or "").strip()
    if name:
        return name
    return str((payload or {}).get("by") or "user")


def _oplog(op, *, ok_flag, reason=None, ts_code=None, params=None, ref=None, by="user"):
    """交易员操作日志: 每个写操作落一行 (OK/BLOCKED 都记, 含原因)。日志失败不影响操作本体 (但记 logger)。"""
    try:
        pms_repo.insert_op_log(op=op, by=by, ts_code=ts_code,
                               result=("OK" if ok_flag else "BLOCKED"),
                               reason=reason, params=params, ref=ref)
    except Exception as e:
        logger.error("[op_log] 写操作日志失败 op=%s: %s (操作本体不受影响)", op, e)


def ok_logged(op, fn, *args, ts_code=None, params=None, by="user", **kw):
    """跑 fn (经 ok 包装) 并按结果落一条操作日志。ok=False -> BLOCKED + error/errors 作原因。"""
    data = ok(fn, *args, **kw)
    okf = not (isinstance(data, dict) and data.get("ok") is False)
    reason = None
    if not okf:
        reason = "; ".join(data.get("errors") or []) or data.get("error") or "被约束挡下"
    ref = None
    if isinstance(data, dict):
        ref = (data.get("command_id") or data.get("instruction_id")
               or data.get("strategy_id") or data.get("proposal_id"))
    _oplog(op, ok_flag=okf, reason=reason, ts_code=ts_code, params=params, ref=ref, by=by)
    return data


# ================================================================ 基础
@app.get("/health")
def health():
    """容器健康检查 (**免登录, 只回布尔级状态**)。

    2026-08-28 审查修: 原来免登录就吐总操作规模、各仓位上限、自主档位和数据库错误串
    (可能带内网地址) —— 对一套下真钱的系统, 这些是未认证不该看到的。健康检查只需要
    活不活; 完整自证挪到管理员专属的 /api/ops/health-detail。"""
    db = dbs.ping("proxy")
    return {
        "status": "ok" if db["ok"] else "degraded",
        "version": VERSION, "service": "pms-web",
        "now": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "trade_day": td.is_trade_day(), "calendar_degraded": td.calendar_degraded(),
        "db_ok": bool(db.get("ok")),
    }


@app.get("/api/ops/health-detail")
def api_health_detail():
    """完整自证 (系统管理员): 配置装载、库连通、行业源、下发通道 —— 原 /health 的全量内容。"""
    db = dbs.ping("proxy")
    return {
        "status": "ok" if db["ok"] else "degraded",
        "version": VERSION, "service": "pms-web",
        "now": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "trade_day": td.is_trade_day(), "calendar_degraded": td.calendar_degraded(),
        "db_proxy": db,
        "config_loaded": {
            "total_scale": param_store.get("PMS_TOTAL_SCALE"),
            "portfolio_cap": param_store.get("PMS_PORTFOLIO_CAP"),
            "stock_cap": param_store.get("PMS_STOCK_CAP"),
            "max_names": param_store.get("PMS_MAX_NAMES"),
            "autonomy": param_store.get("PMS_AUTONOMY"),
            "web_port": settings.PMS_WEB_PORT,
        },
        "sector": industry.status(),
        "dispatch": _dispatch_health(),
    }


def _dispatch_health() -> dict:
    """下发通道健康快照。通道读不到不能让 /health 挂 —— 容器健康检查靠它。"""
    try:
        from app.services import dispatcher
        d = dispatcher.describe()
        ch = d.get("channel") or {}
        return {"mode": d["mode"], "online": ch.get("online"),
                "conn_state": ch.get("conn_state"), "queued": (ch.get("queue") or {}).get("QUEUED", 0),
                "hint": d.get("hint")}
    except Exception as e:
        return {"error": f"{type(e).__name__}: {e}"}


@app.get("/")
def index():
    path = os.path.join(STATIC_DIR, "index.html")
    if os.path.exists(path):
        return FileResponse(path, media_type="text/html; charset=utf-8")
    return JSONResponse({"hint": "页面文件缺失, 健康检查: /health"})


# ================================================================ 登录与权限接口
@app.post("/api/auth/login")
def api_auth_login(payload: dict = Body(default=None)):
    """手机号+密码 -> 调 bshop 校验 -> 判角色 -> 签会话票写 cookie。前端发 JSON。"""
    payload = payload or {}
    phone = str(payload.get("phone") or "").strip()
    password = str(payload.get("password") or "")
    if not phone or not password:
        return JSONResponse({"ok": False, "error": "手机号和密码都要填"}, status_code=400)
    res = authmod.call_bshop_login(settings.PMS_BSHOP_LOGIN_URL, phone, password,
                                   settings.PMS_BSHOP_TIMEOUT)
    if not res.get("ok"):
        return JSONResponse({"ok": False, "error": res.get("error") or "登录失败"}, status_code=401)
    roles = authmod.roles_from_permissions(res["permissions"], settings.PMS_ADMIN_PERMS,
                                           settings.PMS_TRADER_PERMS)
    if not roles:
        return JSONResponse({"ok": False, "error": "这个账号没有持仓管理系统的使用权限"}, status_code=403)
    if not settings.PMS_SESSION_SECRET:
        return JSONResponse({"ok": False, "error": "服务未配置会话密钥 PMS_SESSION_SECRET"}, status_code=503)
    token = authmod.make_session(res["phone"], res["username"], roles,
                                 settings.PMS_SESSION_SECRET, settings.PMS_SESSION_TTL_HOURS)
    resp = JSONResponse({"ok": True, "user": {"phone": res["phone"], "username": res["username"]},
                         "roles": roles, "orgs": res.get("orgs") or []})
    resp.set_cookie(settings.PMS_SESSION_COOKIE, token,
                    max_age=settings.PMS_SESSION_TTL_HOURS * 3600,
                    httponly=True, samesite="lax", path="/")
    return resp


@app.post("/api/auth/logout")
def api_auth_logout():
    """退出: 清掉会话 cookie。"""
    resp = JSONResponse({"ok": True})
    resp.delete_cookie(settings.PMS_SESSION_COOKIE, path="/")
    return resp


@app.get("/api/me")
def api_me(request: Request):
    """当前登录人与角色。关掉登录时回一个合成的管理员, 前端就不弹登录框也不藏东西。"""
    if not settings.PMS_AUTH_ENABLED:
        return {"ok": True, "auth_enabled": False,
                "user": {"phone": "-", "username": "本地(未开登录)"}, "roles": [authmod.ROLE_ADMIN, authmod.ROLE_TRADER]}
    secret = settings.PMS_SESSION_SECRET
    sess = (authmod.verify_session(request.cookies.get(settings.PMS_SESSION_COOKIE), secret)
            if secret else None)
    if not sess:
        return JSONResponse({"ok": False, "auth": "login_required"}, status_code=401)
    return {"ok": True, "auth_enabled": True,
            "user": {"phone": sess.get("phone"), "username": sess.get("username")},
            "roles": sess.get("roles") or []}


@app.get("/api/overview")
def api_overview():
    return ok(portfolio.overview)


@app.get("/api/names")
def api_names(codes: str = Query("")):
    """代码 → 中文名 (gp_code_all)。codes 逗号分隔点式代码。绝不抛错 —— 交易员视图靠它显示中文名。"""
    cs_list = [c.strip() for c in (codes or "").split(",") if c.strip()]
    return ok(lambda: {"ok": True, "names": downstream_repo.fetch_names(cs_list)})


# ================================================================ ① 参数设置
@app.get("/api/params")
def api_params():
    return ok(param_store.snapshot)


# 只有系统管理员能改的开关 (2026-08-28 审查加): 把下发通道从影子切到实盘直连是
# 运维级动作, 不属于交易员的"参数微调"。其余键的边界维持用户定的原则不动。
_ADMIN_ONLY_PARAM_KEYS = {"PMS_DISPATCH_MODE"}


@app.post("/api/params")
def api_set_params(request: Request, payload: dict = Body(...)):
    """单个 {key, value} 或批量 {items:[{key,value}...]}。逐项返回结果, 部分失败不整体回滚。"""
    items = payload.get("items") or [{"key": payload.get("key"), "value": payload.get("value")}]
    actor = _actor(request, payload)
    sess = getattr(request.state, "user", None) or {}
    is_admin = (not settings.PMS_AUTH_ENABLED) or authmod.is_admin(sess.get("roles"))
    results = []
    for it in items:
        if not isinstance(it, dict):
            # 元素不是对象直接给出明确错误, 不再冒 500 (2026-08-28 审查修)
            results.append({"ok": False, "error": f"条目格式非法 (应为 {{key, value}}): {it!r}"})
            continue
        k = it.get("key")
        if not k:
            results.append({"ok": False, "error": "缺少 key"})
            continue
        if k in _ADMIN_ONLY_PARAM_KEYS and not is_admin:
            results.append({"ok": False, "error": f"{k} 需要系统管理员才能修改"})
            continue
        results.append(param_store.set_param(k, it.get("value"), updated_by=actor))
    _okf = all(r.get("ok") for r in results)
    _oplog("set_params", ok_flag=_okf,
           params={(it.get("key") if isinstance(it, dict) else str(it)):
                   (it.get("value") if isinstance(it, dict) else None) for it in items},
           reason=(None if _okf else "; ".join((r.get("error") or "") for r in results if not r.get("ok"))),
           by=actor)
    return {"ok": _okf, "results": results}


# ================================================================ ② 命令台
@app.get("/api/commands/catalog")
def api_catalog():
    return ok(lambda: {"commands": cs.list_commands(sector_source_ready=industry.ready()),
                       "sector": industry.status()})


@app.get("/api/commands/active")
def api_commands_active():
    return ok(command_service.active_commands)


@app.get("/api/commands")
def api_commands(status: str = Query(None), limit: int = Query(100),
                 include_archived: bool = Query(False)):
    statuses = [s for s in (status or "").split(",") if s] or None
    return ok(pms_repo.list_commands, statuses=statuses, limit=limit,
              include_archived=include_archived)


@app.get("/api/commands/{command_id}")
def api_command_detail(command_id: str):
    def _detail():
        c = pms_repo.get_command(command_id)
        if not c:
            return {"ok": False, "error": f"命令 {command_id} 不存在"}
        return {"ok": True, "command": c,
                "plans": pms_repo.list_plans(command_id=command_id, limit=500)}
    return ok(_detail)


@app.post("/api/commands")
def api_issue(request: Request, payload: dict = Body(...)):
    actor = _actor(request, payload)
    return ok_logged("issue_command:" + str(payload.get("cmd_type")),
                     command_service.issue, payload.get("cmd_type"), payload.get("params") or {},
                     note=payload.get("note"), issued_by=actor,
                     force_conflict=bool(payload.get("force")),
                     ts_code=(payload.get("params") or {}).get("ts_code"),
                     params=payload, by=actor)


@app.post("/api/commands/{command_id}/cancel")
def api_cancel(command_id: str):
    return ok_logged("cancel_command", command_service.cancel, command_id,
                     params={"command_id": command_id})


@app.post("/api/commands/{command_id}/replan")
def api_replan(command_id: str):
    """重新生成方案 (窗口内行情变化后可重算; 旧方案作废)。"""
    def _replan():
        c = pms_repo.get_command(command_id)
        if not c:
            return {"ok": False, "error": "命令不存在"}
        if c["status"] not in cs.ACTIVE_TASK_STATES:
            return {"ok": False, "error": f"命令处于 {c['status']}, 不可重规划"}
        pms_repo.cancel_plans_of_command(command_id)
        return command_service.plan_command(c)
    return ok_logged("replan_command", _replan, params={"command_id": command_id})


@app.get("/api/plans")
def api_plans(command_id: str = Query(None), status: str = Query(None), limit: int = 300):
    statuses = [s for s in (status or "").split(",") if s] or None
    return ok(pms_repo.list_plans, command_id=command_id, statuses=statuses, limit=limit)


# ================================================================ ③ 持仓与账本
@app.get("/api/positions")
def api_positions():
    def _view():
        v = portfolio.positions_view()
        sp = command_service.effective_stock_params()
        # 用户设的止损价与目标价, 事实源在命令表, 与持仓行上系统每天早上重算的参考位
        # (support_ref / pressure_ref / stop_ref) 是两回事。在这里合到一起, 是为了让页面
        # 把两个数分列显示 —— 只是显示, 判断那一侧照旧各读各的事实源, 见
        # command_service.attach_user_prices 的说明。
        command_service.attach_user_prices(v["positions"], sp)
        # 入场论点随持仓走 (2026-09-07 第三件): 每行加「当初为什么买」(entry) 与「现在证据还在不在」
        # (logic_now)。只是显示; 任何一行取不到都写原因, 页面不能因此塌掉。
        try:
            logic_state_service.decorate_positions(v["positions"])
        except Exception as e:  # noqa: BLE001
            logger.warning("[持仓] 入场论点与逻辑状态两栏取不到: %s", e)
        return {"ok": True, **v, "stock_params": sp}
    return ok(_view)


@app.get("/api/positions/{ts_code}/lots")
def api_lots(ts_code: str, status: str = Query("OPEN")):
    return ok(pms_repo.list_lots, ts_code, status=(status or None))


@app.get("/api/instructions")
def api_instructions(status: str = Query(None), limit: int = 200,
                     include_archived: bool = Query(False)):
    statuses = [s for s in (status or "").split(",") if s] or None
    return ok(pms_repo.list_instructions, statuses=statuses, limit=limit,
              include_archived=include_archived)


@app.get("/api/ledger")
def api_ledger(ts_code: str = Query(None), limit: int = 100):
    return ok(pms_repo.list_ledger, ts_code=ts_code, limit=limit)


@app.get("/api/report")
def api_report(ymd: int = Query(None)):
    return ok(lambda: (pms_repo.get_report(ymd) if ymd else pms_repo.latest_report())
              or {"ymd": None, "report": {}})


@app.get("/api/export/publish-xlsx")
def api_export_publish_xlsx():
    """公示表导出 (2026-08-28): 持仓/平仓/净值一张表, 版式对齐公司《量化数据》模板,
    数据只含系统接管后的账本 (口径见 publish_export 模块头)。二进制下载不走 ok()
    包装; 失败按站内惯例回 JSON (HTTP 200), 新开的下载页会直接显示错误原因。"""
    from urllib.parse import quote

    from fastapi.responses import Response
    try:
        from app.services import publish_export
        fname, blob = publish_export.export_xlsx()
    except Exception as e:
        logger.exception("公示表导出失败")
        return JSONResponse({"ok": False, "error": f"{type(e).__name__}: {e}"})
    return Response(
        content=blob,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(fname)}"})


# ================================================================ ④ 提议确认
@app.get("/api/proposals")
def api_proposals(status: str = Query("WAIT_USER"), limit: int = 100,
                  include_archived: bool = Query(False)):
    statuses = tuple(s for s in (status or "").split(",") if s) or ("WAIT_USER",)
    return ok(pms_repo.list_proposals, statuses=statuses, limit=limit,
              include_archived=include_archived)


@app.post("/api/proposals/{proposal_id}/decide")
def api_decide(request: Request, proposal_id: str, payload: dict = Body(default={})):
    """采纳/驳回一条自主提议。采纳 = 先落指令表 (先记账后动作), 下发由择时执行器负责。"""
    decision = str(payload.get("decision") or "").upper()
    if decision not in ("ACCEPTED", "DECLINED"):
        return {"ok": False, "error": "decision 必须是 ACCEPTED 或 DECLINED"}
    # 理由在服务端强制 (2026-09-07): 页面 09-03 起就弹必填框, 但只在浏览器里成立 ——
    # 任何脚本、旧缓存页面都能发一条没有理由的裁决, 账本上落一句「页面人工裁决」的
    # 默认文案, 与真裁决看不出区别。复盘要的是「人为什么这么判」, 没理由的不收。
    reason = str(payload.get("reason") or "").strip()
    if not reason:
        return {"ok": False, "error": "采纳或驳回都要写理由, 空理由不收"}

    def _decide():
        from app.core import action_engine as ae
        p = pms_repo.get_proposal(proposal_id)
        if not p:
            return {"ok": False, "error": "提议不存在"}
        if p["status"] != "WAIT_USER":
            return {"ok": False, "error": f"提议已处于 {p['status']}"}
        if not pms_repo.decide_proposal(proposal_id, decision):
            return {"ok": False, "error": "提议状态已变更, 请刷新"}
        hn = p.get("hard_numbers") or {}
        pms_repo.insert_ledger(ts_code=p["ts_code"], action=p["action"], arbiter="user",
                               verdict="PASS" if decision == "ACCEPTED" else "REJECT",
                               price_at=float(hn.get("price") or 0), hard_numbers=hn,
                               ref_id=proposal_id, reason=reason)
        instruction_id = None
        if decision == "ACCEPTED":
            # 策略(confirm 档)提议: 两腿同 action、side 无法由 action 反推, 交策略层按腿谱发指令
            if str((hn or {}).get("kind")) == "strategy":
                from app.services import strategy_runner
                rr = strategy_runner.emit_from_spec(hn)
                if not rr.get("ok"):
                    return {"ok": False, "error": rr.get("error") or "策略提议发指令失败"}
                return {"ok": True, "decision": decision,
                        "instruction_id": rr.get("instruction_id"), "strategy": True}
            instruction_id = cs.make_instruction_id(td.ymd(), p["ts_code"], p["action"], 1)
            side = "sell" if p["action"] in ("TRIM", "EXIT") else "buy"
            _win = param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3)
            qty = int(p["qty"] or 0)
            if side == "sell":
                # 按拍板这一刻的可卖量重算 (2026-09-07 审查修): 提议记的是扫描那一刻的股数,
                # 等人点采纳时持仓可能已被别的路卖少, 照原数落单会被卖出前的检查整条驳回,
                # 人点了「清掉」结果一股没卖, 且那条指令会永久在途、挡住后续所有清仓。
                cur = pms_repo.get_position(p["ts_code"]) or {}
                qty = ae.clamp_sell_qty(qty, cur)
                if qty <= 0:
                    return {"ok": True, "decision": decision, "instruction_id": None,
                            "note": "这只票现在没有可卖的股数 (已被清掉或今天刚买入), 未落单"}
            pms_repo.insert_instruction(
                instruction_id=instruction_id, origin_type="proposal", origin_id=proposal_id,
                ts_code=p["ts_code"], action=p["action"], side=side, qty=qty,
                limit_price=hn.get("price"),
                window_tdays=_win,
                status="PROPOSED",
                progress={"from_proposal": proposal_id, "is_command": False, "children": [],
                          "deadline": str(td.window_deadline(None, _win))})
            # 与自主执行同一口径: 采纳即算「做过一次」, 计数器要跟着走
            # (否则页面采纳的那条动作绕开了 §6 的一次性约束)
            from app.services import proposal_service
            g = proposal_service.bump_once_guards(p["ts_code"], p["action"], hn) or {}
            if not g.get("ok"):
                # 指令已落表, 不回滚; 但要让页面看见"这条一次性纪律本轮没锁上"
                return {"ok": True, "decision": decision, "instruction_id": instruction_id,
                        "warning": f"一次性守卫计数器未写入 ({g.get('error')}) —— "
                                   f"{p['ts_code']} 的 {p['action']}「只做一次」本轮失效"}
        return {"ok": True, "decision": decision, "instruction_id": instruction_id}
    return ok_logged("decide_proposal", _decide,
                     params={"proposal_id": proposal_id, "decision": decision},
                     by=_actor(request, payload))


@app.post("/api/proposals")
def api_create_proposal(payload: dict = Body(...)):
    """人工补录一条待确认提议 (影子运行期造数与联调用)。"""
    def _create():
        pid = payload.get("proposal_id") or f"PRP_{td.ymd()}_{int(datetime.now().timestamp())}"
        ttl = param_store.get_int("PMS_PROPOSAL_TTL_HOURS", 24)
        pms_repo.insert_proposal(
            proposal_id=pid, ts_code=cs.normalize_code(payload.get("ts_code") or ""),
            action=payload.get("action") or "ADD", qty=int(payload.get("qty") or 0),
            hard_numbers=payload.get("hard_numbers") or {},
            expire_at=datetime.now() + timedelta(hours=ttl))
        return {"ok": True, "proposal_id": pid}
    return ok(_create)


# ================================================================ 运维操作 (与调度器同一实现)
@app.post("/api/ops/replay")
def api_replay(limit: int = Query(500)):
    return ok(ledger_service.replay_fills, limit=limit)


@app.get("/api/ops/rebuild-preflight")
def api_rebuild_preflight():
    """账本重建的只读预检 (README 待办 #4)。不写任何东西, 随时可点。

    成本价能不能用是这一步的成败所在 —— 详见 app/core/rebuild_check.py 的模块说明。
    """
    def _pf():
        from app.services import ledger_service
        return ledger_service.rebuild_preflight()
    return ok(_pf)


@app.get("/api/ops/rebuild-accept")
def api_rebuild_accept():
    """重建之后的只读判收: 安全垫分布 / 批次账 / 行业集中度。"""
    def _ac():
        from app.services import ledger_service
        return ledger_service.rebuild_accept()
    return ok(_ac)


@app.post("/api/ops/reconcile")
def api_reconcile(apply_fix: bool = Query(True), force: bool = Query(False)):
    """force=true 绕过「差异面过大不自动改账」的限制, 仅在人工确认下游读数无误后使用。"""
    return ok(ledger_service.reconcile, apply_fix=apply_fix, force=force)


@app.post("/api/ops/premarket")
def api_premarket():
    return ok(ledger_service.premarket)


@app.post("/api/ops/daily-settle")
def api_daily_settle():
    return ok(ledger_service.daily_settle)


@app.post("/api/ops/report")
def api_build_report():
    return ok(ledger_service.build_daily_report)


@app.post("/api/ops/plan-pending")
def api_plan_pending():
    return ok(command_service.plan_pending)


@app.post("/api/ops/materialize")
def api_materialize():
    """方案 → 指令 (先记账后动作)。"""
    from app.services import executor
    return ok(executor.materialize_plans)


@app.post("/api/ops/exec-tick")
def api_exec_tick(dry_run: bool = Query(False)):
    """择时出手一跳。dry_run=true 只试算不下发, 用来在盘中先看「现在会怎么动」。"""
    from app.services import executor
    return ok(executor.run_tick, dry_run=dry_run)


@app.post("/api/ops/sweep-windows")
def api_sweep_windows():
    from app.services import executor
    return ok(executor.sweep_windows)


@app.post("/api/instructions/{instruction_id}/cancel")
def api_cancel_instruction(instruction_id: str, payload: dict = Body(default={})):
    from app.services import executor
    return ok_logged("cancel_instruction", executor.cancel_instruction, instruction_id,
                     payload.get("reason") or "页面人工撤销",
                     params={"instruction_id": instruction_id})


@app.post("/api/ops/scan-proposals")
def api_scan_proposals(dry_run: bool = Query(False)):
    """自主提议扫描一轮 (动作引擎 → 规则闸 → 研判闸 → 按档位分流)。"""
    from app.services import proposal_service
    return ok(proposal_service.scan_and_route, dry_run=dry_run)


@app.post("/api/ops/digest-signals")
def api_digest_signals(dry_run: bool = Query(False)):
    """消化一批决策系统盘中信号。dry_run=true 只解析判定, 不落表也不 ACK。"""
    from app.services import signal_service
    return ok(signal_service.consume, dry_run=dry_run)


@app.get("/api/signal-status")
def api_signal_status():
    from app.services import signal_service
    return ok(signal_service.status)


@app.get("/api/dispatch-mode")
def api_dispatch_mode():
    from app.services import dispatcher, judge
    return ok(lambda: {"ok": True, **dispatcher.describe(), "judge": judge.status()})


@app.get("/api/ws-channel")
def api_ws_channel(limit: int = Query(50)):
    """ws 通道运维视图: 连接状态 / seq 水位 / 出口队列 / 最近上行消息。

    这四样凑一起才看得出通道到底"通没通" —— 心跳说明进程在、conn_state 说明连接在、
    水位说明消息没断层、队列深度说明指令有没有卡住。少看一样都可能误判。
    """
    from app.repo import qmt_repo
    from app.services import dispatcher

    def _view():
        out = {"ok": True, "mode": dispatcher.mode(),
               "channel": dispatcher.channel_status()}
        try:
            out["orders"] = qmt_repo.list_orders(limit=limit)
            out["inbox"] = qmt_repo.inbox_list(limit=limit)
        except Exception as e:
            out["error"] = f"{type(e).__name__}: {e}"
        return out
    return ok(_view)


@app.post("/api/ws-channel/clear-resync")
def api_clear_resync():
    """人工确认全量对账已做完后清 resync 标记 (协议 §6.2)。

    这个标记只能人来清 —— 它的含义是「对端补发不全, 中间那段消息我们永远拿不到了」,
    自动清等于假装没发生过。
    """
    from app.repo import qmt_repo

    def _clear():
        st = qmt_repo.get_state()
        qmt_repo.set_conn(st.get("conn_state") or "OFFLINE", resync=False, beat=False)
        return {"ok": True, "message": "已清除 resync 标记 —— 请确认全量对账确实已完成"}
    return ok(_clear)


# ================================================================ 上游选股计划
@app.get("/api/open-scan")
def api_open_scan():
    """候选池处置快照 (只读): 每只候选今天为什么下单 / 没下单。

    复用每分钟那条自主扫描的 scan_open 口径, 但止步于规则闸之前、不送研判、不落任何表 ——
    补上「名额/资金/上限/不足一手/在途/已持有」这些算完即弃、评审账本里查不到的原因。
    「已下单/已生成提议」由前端拿它已在手的 instructions/proposals 按股票对齐, 这里不重复查。
    """
    from app.services import proposal_service
    return ok(proposal_service.disposition_snapshot)


@app.get("/api/upstream-signals")
def api_upstream_signals():
    """上游信号只读快照 (三系统盘中信号: 决策系统买卖 / 择时层 BUY / 双向告警 / 资金异动 / 实况分)。
    只展示、不进任何下单或决策逻辑; 内部一律 XREVRANGE/ZREVRANGE, 不建消费组、不 ACK、不写。"""
    from app.services import upstream_signals
    return ok(upstream_signals.snapshot)


@app.get("/api/upstream/signals/{ts_code}")
def api_upstream_signals_by_code(ts_code: str):
    """一只票**今天的全部**上游信号 (2026-09-09, 台账 055)。

    与 /api/upstream-signals 的分工: 那个是全场最近 N 条的快照, 告警每类只留最新 20 条,
    到下午一只活跃票早盘的信号早被挤掉; 这个按代码倒序回扫到当天开盘, 所以才叫"全部"。

    **只能点击触发, 绝不能进页面轮询** —— 进轮询就是每 30 秒对上游做一次全流扫描,
    把只读观察变成压力源。内部纪律与快照相同: 一律 XREVRANGE/ZREVRANGE, 不建消费组、不 ACK、不写。
    """
    from app.services import upstream_signals
    return ok(upstream_signals.by_code, ts_code)


# ================================================================ 宏观择时 (MACRO_TIMING_PLAN.md)
@app.get("/api/macro/status")
def api_macro_status():
    """宏观择时面板快照: 各信号当前值/区域/近 20 日序列/当日建议/最近命令/冷却 + 闸状态。"""
    from app.services import macro_service
    return ok(macro_service.status)


@app.post("/api/ops/macro-scan")
def api_macro_scan(dry_run: bool = Query(False)):
    """手动宏观扫描 (= 09:35 调度位)。dry_run=true 只算不落表不下达, 供判收与盘中复核。"""
    from app.services import macro_service
    return ok_logged("macro_scan", macro_service.scan, dry_run=dry_run,
                     params={"dry_run": dry_run})


@app.post("/api/macro/adopt")
def api_macro_adopt(payload: dict = Body(default={})):
    """采纳当日宏观建议 → 以用户名义下命令 (幂等, 建议隔日作废)。"""
    from app.services import macro_service
    key = (payload or {}).get("signal_key") or "hedge_fx"
    return ok_logged("macro_adopt", macro_service.adopt, key, params={"signal_key": key})


@app.post("/api/ops/strategy-attach-scan")
def api_strategy_attach_scan(dry_run: bool = Query(False)):
    """手动策略自动挂载 (= 09:40 调度位)。dry_run=true 只判不写 —— 不挂、不留痕、
    不动买入暂停表, 返回本轮会做什么, 供上线判收与日常复核。"""
    from app.services import strategy_advisor
    return ok_logged("strategy_attach_scan", strategy_advisor.scan, dry_run=dry_run,
                     params={"dry_run": dry_run})


@app.get("/api/upstream/plan")
def api_upstream_plan(limit: int = Query(30), date: str = Query(None),
                      bucket: str = Query("main")):
    """上游 /plan 预览。绝不抛错 —— 接口不通时 status.ok=false + hint, 页面照常渲染。

    candidates_raw 是**不扣持仓/黑名单**的原始筛选结果 (纯看参数选出来什么),
    真正下命令时的候选池还会再剔除已持有、黑名单与取不到价的票。
    """
    def _plan():
        from app.services import plan_feed
        st = plan_feed.status()
        out = {"ok": True, "status": st, "rows": [], "candidates_raw": None}
        if not st.get("ok"):
            return out
        try:
            plan = plan_feed.get_plan(date=(date or None))
            rows = plan.get("observe" if bucket == "observe" else "main") or []
            out["rows"] = rows[:max(1, int(limit or 30))]
            out["candidates_raw"] = plan_feed.candidates()
        except Exception as e:
            out["status"] = {**st, "ok": False, "hint": f"{type(e).__name__}: {e}"}
        return out
    return ok(_plan)


@app.post("/api/ops/plan-refresh")
def api_plan_refresh(date: str = Query(None)):
    """强刷上游计划, 并把 evidence.theme 灌进 pms_industry_map (行业源 custom_table 的数据来源)。"""
    def _refresh():
        from app.services import plan_feed
        plan_feed.invalidate()
        plan = plan_feed.get_plan(force=True, date=(date or None))
        return {"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"),
                "returned": plan["returned"], "theme_sync": plan.get("theme_sync"),
                "snapshot": plan.get("snapshot"), "industry": industry.status()}
    return ok(_refresh)


@app.get("/api/upstream/plan-changes")
def api_plan_changes(limit: int = Query(50)):
    """榜单变化 (PMS 自算, 上游 `changes` 字段恒为 null —— UPSTREAM_PLAN_API.md §9)。

    绝不抛错: 变化提示是锦上添花, 它坏了不该让「上游计划」抽屉打不开。
    `log.revised` 里某个 plan_date 出现多版, 就是 §7.3 那个"同一 date 会变"。

    子状态**嵌在 changes 里**而不是顶层 —— 顶层 `ok` 一为假, 页面会挂全局红条。
    「快照关着」「还没有快照」都不是接口故障, 不该长成那副样子 (同 status 的处理)。
    """
    def _changes():
        from app.services import plan_feed
        return {"ok": True, "changes": plan_feed.changes(),
                "log": plan_feed.snapshot_log(limit=max(1, int(limit or 50)))}
    return ok(_changes)


@app.get("/api/ops/downstream-schema")
def api_downstream_schema():
    """导出下游三表的实际列定义 —— 用于回填 QMT_INTERFACE_REQUIREMENTS D1。"""
    def _schema():
        out = {}
        for t in ("trading_position", "trading_order", "trading_buy_plan",
                  "gp_stock_category"):
            try:
                out[t] = downstream_repo.describe(t)
            except Exception as e:
                out[t] = {"error": f"{type(e).__name__}: {e}"}
        try:
            out["_position_probe"] = downstream_repo.fetch_positions()["columns"]
        except Exception as e:
            out["_position_probe"] = {"error": str(e)}
        try:
            out["_category_probe"] = downstream_repo.category_columns(force=True)
        except Exception as e:
            out["_category_probe"] = {"error": str(e)}
        return out
    return ok(_schema)


# ================================================================ 行业映射
@app.get("/api/industry")
def api_industry(limit: int = 2000):
    return ok(lambda: {"ok": True, "status": industry.status(),
                       "rows": pms_repo.list_industry(limit=limit)})


@app.post("/api/industry/import")
def api_industry_import(payload: dict = Body(...)):
    """导入行业映射。支持 {rows:[{ts_code,industry}]} 或 {text:"600000.SH,银行\\n..."}。"""
    def _imp():
        rows = payload.get("rows")
        if not rows and payload.get("text"):
            rows = []
            for line in str(payload["text"]).splitlines():
                parts = [x.strip() for x in line.replace("\t", ",").split(",") if x.strip()]
                if len(parts) >= 2:
                    rows.append({"ts_code": cs.normalize_code(parts[0]), "industry": parts[1]})
        rows = [r for r in (rows or []) if r.get("ts_code") and r.get("industry")]
        n = pms_repo.upsert_industry(rows)
        industry.invalidate()
        return {"ok": True, "imported": len(rows), "affected": n, "status": industry.status()}
    return ok(_imp)



# ================================================================ 个股交易方案 (策略) + 操作日志
@app.get("/api/strategies")
def api_strategies(status: str = Query(None), include_archived: bool = Query(False)):
    statuses = [s for s in (status or "").split(",") if s] or None
    from app.services import strategy_service

    def _load():
        rows = pms_repo.list_strategies(statuses=statuses, limit=300,
                                        include_archived=include_archived)
        try:
            bp = strategy_service.buypause_map()
        except Exception:
            bp = {}
        for r in rows:
            r["buy_paused"] = bp.get(r.get("ts_code")) or None
        # 自动挂载仪表 (2026-08-26): 总开关 + 今天的正式一跳跑没跑。摘要由 advisor 真跑后
        # 写进 PMS_AUTO_LAST_SCAN, 这里只读; 解析不了按没跑显示, 不猜。
        last, ran_today = None, False
        try:
            raw = param_store.get("PMS_AUTO_LAST_SCAN") or ""
            last = json.loads(raw) if raw else None
            ran_today = bool(last and int(last.get("ymd") or 0) == td.ymd())
        except (ValueError, TypeError):
            last = None
        return {"ok": True, "strategies": rows,
                "enabled": param_store.get_bool("PMS_STRATEGY_ENABLED", False),
                "auto": {"enabled": param_store.get_bool("PMS_AUTO_STRATEGY_ENABLED", False),
                         "ran_today": ran_today, "last_scan": last}}
    return ok(_load)


@app.post("/api/strategies/validate")
def api_strategy_validate(payload: dict = Body(...)):
    """挂载前约束校验 (不写库): 返回 {ok, reasons}。违反仓位/存量/上限就给中文原因, 页面红字提示、不落库。"""
    from app.services import strategy_service
    return ok(lambda: strategy_service.validate(payload))


@app.post("/api/strategies")
def api_strategy_attach(request: Request, payload: dict = Body(...)):
    """挂载一条策略 (先校验再落库)。校验不过返回 {ok:false, errors}, 并落一条 BLOCKED 操作日志。"""
    from app.services import strategy_service
    return ok_logged("attach_strategy:" + str(payload.get("type")),
                     strategy_service.attach, payload,
                     ts_code=payload.get("ts_code"), params=payload,
                     by=_actor(request, payload))


@app.post("/api/strategies/{strategy_id}/status")
def api_strategy_status(request: Request, strategy_id: str, payload: dict = Body(...)):
    """暂停(PAUSED)/恢复(ACTIVE)/撤下(CANCELLED): payload {status}。"""
    from app.services import strategy_service
    return ok_logged("set_strategy_status:" + str(payload.get("status")),
                     strategy_service.set_status, strategy_id,
                     str(payload.get("status") or "").upper(),
                     params={"strategy_id": strategy_id, "status": payload.get("status")},
                     by=_actor(request, payload))


# ================================ 软归档: 把终态记录从在办/在途/待确认/策略列表移除 (不删行, 可恢复)
def _arch_result(n: int, on: bool) -> dict:
    if on:
        note = "已移除, 从视图隐藏 (行仍在库, 对账与审计不受影响)" if n > 0 \
            else "未改动: 该记录不是终态、不存在、或早已移除"
    else:
        note = "已恢复显示" if n > 0 else "未改动: 不存在或本就未移除"
    return {"ok": n > 0, "affected": n, "archived": on, "note": note}


@app.post("/api/commands/{command_id}/archive")
def api_archive_command(command_id: str, payload: dict = Body(default={})):
    on = bool(payload.get("archived", True))
    return ok_logged("archive_command" if on else "unarchive_command",
                     lambda: _arch_result(
                         (pms_repo.archive_command if on else pms_repo.unarchive_command)(command_id), on),
                     params={"command_id": command_id, "archived": on})


@app.post("/api/instructions/{instruction_id}/archive")
def api_archive_instruction(instruction_id: str, payload: dict = Body(default={})):
    on = bool(payload.get("archived", True))
    return ok_logged("archive_instruction" if on else "unarchive_instruction",
                     lambda: _arch_result(
                         (pms_repo.archive_instruction if on else pms_repo.unarchive_instruction)(instruction_id), on),
                     params={"instruction_id": instruction_id, "archived": on})


@app.post("/api/proposals/{proposal_id}/archive")
def api_archive_proposal(proposal_id: str, payload: dict = Body(default={})):
    on = bool(payload.get("archived", True))
    return ok_logged("archive_proposal" if on else "unarchive_proposal",
                     lambda: _arch_result(
                         (pms_repo.archive_proposal if on else pms_repo.unarchive_proposal)(proposal_id), on),
                     params={"proposal_id": proposal_id, "archived": on})


@app.post("/api/strategies/{strategy_id}/archive")
def api_archive_strategy(strategy_id: str, payload: dict = Body(default={})):
    on = bool(payload.get("archived", True))
    return ok_logged("archive_strategy" if on else "unarchive_strategy",
                     lambda: _arch_result(
                         (pms_repo.archive_strategy if on else pms_repo.unarchive_strategy)(strategy_id), on),
                     params={"strategy_id": strategy_id, "archived": on})


@app.get("/api/op-log")
def api_op_log(limit: int = Query(200)):
    """交易员操作日志 (每个页面写操作一行, 含 OK/BLOCKED 与原因)。"""
    return ok(lambda: {"ok": True, "rows": pms_repo.list_op_log(limit=limit)})


@app.post("/api/strategies/{strategy_id}/resume-buy")
def api_strategy_resume_buy(request: Request, strategy_id: str, payload: dict = Body(default={})):
    """恢复该策略的买入 (决策系统风控预警触发的暂停由你手动解除; 只恢复买入, 不影响卖出/平回)。"""
    from app.services import strategy_service
    return ok_logged("resume_strategy_buy", strategy_service.resume_buy, strategy_id,
                     params={"strategy_id": strategy_id}, by=_actor(request, payload))