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 logging
import os
from datetime import datetime, timedelta

from fastapi import Body, FastAPI, Query
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, param_store, portfolio

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)


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}"}


# ================================================================ 基础
@app.get("/health")
def 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.get("/api/overview")
def api_overview():
    return ok(portfolio.overview)


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


@app.post("/api/params")
def api_set_params(payload: dict = Body(...)):
    """单个 {key, value} 或批量 {items:[{key,value}...]}。逐项返回结果, 部分失败不整体回滚。"""
    items = payload.get("items") or [{"key": payload.get("key"), "value": payload.get("value")}]
    results = []
    for it in items:
        k = it.get("key")
        if not k:
            results.append({"ok": False, "error": "缺少 key"})
            continue
        results.append(param_store.set_param(k, it.get("value"),
                                             updated_by=payload.get("by") or "user"))
    return {"ok": all(r.get("ok") for r in results), "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)):
    statuses = [s for s in (status or "").split(",") if s] or None
    return ok(pms_repo.list_commands, statuses=statuses, limit=limit)


@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(payload: dict = Body(...)):
    return ok(command_service.issue, payload.get("cmd_type"), payload.get("params") or {},
              note=payload.get("note"), issued_by=payload.get("by") or "user",
              force_conflict=bool(payload.get("force")))


@app.post("/api/commands/{command_id}/cancel")
def api_cancel(command_id: str):
    return ok(command_service.cancel, 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(_replan)


@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():
    return ok(lambda: {"ok": True, **portfolio.positions_view(),
                       "stock_params": command_service.effective_stock_params()})


@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):
    statuses = [s for s in (status or "").split(",") if s] or None
    return ok(pms_repo.list_instructions, statuses=statuses, limit=limit)


@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/proposals")
def api_proposals(status: str = Query("WAIT_USER"), limit: int = 100):
    statuses = tuple(s for s in (status or "").split(",") if s) or ("WAIT_USER",)
    return ok(pms_repo.list_proposals, statuses=statuses, limit=limit)


@app.post("/api/proposals/{proposal_id}/decide")
def api_decide(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"}

    def _decide():
        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=payload.get("reason") or "页面人工裁决")
        instruction_id = None
        if decision == "ACCEPTED":
            instruction_id = cs.make_instruction_id(td.ymd(), p["ts_code"], p["action"], 1)
            side = "sell" if p["action"] in ("TRIM", "EXIT") else "buy"
            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=int(p["qty"] or 0),
                limit_price=hn.get("price"),
                window_tdays=param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3),
                status="PROPOSED", progress={"from_proposal": proposal_id})
        return {"ok": True, "decision": decision, "instruction_id": instruction_id}
    return ok(_decide)


@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.post("/api/ops/reconcile")
def api_reconcile(apply_fix: bool = Query(True)):
    return ok(ledger_service.reconcile, apply_fix=apply_fix)


@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(executor.cancel_instruction, instruction_id,
              payload.get("reason") or "页面人工撤销")


@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/ops/downstream-schema")
def api_downstream_schema():
    """导出下游三表的实际列定义 —— 用于回填 QMT_INTERFACE_REQUIREMENTS D1。"""
    def _schema():
        out = {}
        for t in ("trading_position", "trading_order", "trading_buy_plan"):
            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)}
        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)