tradingSystem/app/repo/pms_repo.py

# -*- coding: utf-8 -*-
"""
PMS 自有表数据访问 (pms_*, 153 代理侧, 严格单表)
=================================================
每个函数只碰一张表 —— 跨表编排一律在 services 层做。
所有 SQL 经 db.session 的单表守卫; 列名白名单防注入。
表结构见 ddl_pms_v1.sql。
"""
from __future__ import annotations

import json
from datetime import datetime

from app.db.session import execute, execute_many, fetch_all, fetch_one

# 允许动态更新的列白名单 (防注入; 与 DDL 保持一致)
POSITION_COLS = {
    "status", "frozen_reason", "total_qty", "avail_qty", "base_qty", "fill_qty", "add_qty",
    "dca_qty", "t0_qty", "avg_cost", "realized_t_profit", "cushion_pct", "cushion_state",
    "cushion_peak", "pct_of_scale", "target_pct", "stop_ref", "support_ref", "pressure_ref",
    "ref_source", "t0_enabled", "t0_ratio", "opened_date", "fill_count", "last_add_date",
    "dca_count", "t0_count_today",
}
LOT_COLS = {"qty", "closed_qty", "close_avg_price", "realized_pnl", "status", "open_price",
            "note", "lot_type"}

_NOW = lambda: datetime.now()   # noqa: E731  (容器时区 Asia/Shanghai)


def _dumps(v):
    return json.dumps(v, ensure_ascii=False) if not isinstance(v, (str, type(None))) else v


def _loads(v, default=None):
    if v in (None, ""):
        return default
    if isinstance(v, (dict, list)):
        return v
    try:
        return json.loads(v)
    except (ValueError, TypeError):
        return default


def _set_clause(fields: dict, allow: set) -> tuple:
    cols = [c for c in fields if c in allow]
    if not cols:
        return "", {}
    return ", ".join(f"{c} = :{c}" for c in cols), {c: fields[c] for c in cols}


# ================================================================ pms_runtime_param
def all_params() -> dict:
    rows = fetch_all("SELECT param_key, param_value, updated_by, updated_at FROM pms_runtime_param")
    return {r["param_key"]: r for r in rows}


def set_param(key: str, value, updated_by: str = "user") -> int:
    return execute(
        "INSERT INTO pms_runtime_param (param_key, param_value, updated_by, updated_at) "
        "VALUES (:k, :v, :by, :ts) "
        "ON DUPLICATE KEY UPDATE param_value = :v, updated_by = :by, updated_at = :ts",
        {"k": key, "v": str(value), "by": updated_by, "ts": _NOW()})


def get_param(key: str):
    r = fetch_one("SELECT param_value FROM pms_runtime_param WHERE param_key = :k", {"k": key})
    return r["param_value"] if r else None


# ================================================================ pms_command
def next_command_seq(ymd: int) -> int:
    r = fetch_one("SELECT COUNT(*) AS n FROM pms_command WHERE command_id LIKE :p",
                  {"p": f"CMD_{ymd}_%"})
    return int((r or {}).get("n") or 0) + 1


def insert_command(*, command_id, cmd_class, cmd_type, ts_code, params, status,
                   issued_by="user", note=None, progress=None) -> int:
    return execute(
        "INSERT INTO pms_command (command_id, cmd_class, cmd_type, ts_code, params_json, "
        "status, progress_json, issued_by, issued_at, note) VALUES "
        "(:cid, :cls, :ct, :code, :pj, :st, :pg, :by, :ts, :note)",
        {"cid": command_id, "cls": cmd_class, "ct": cmd_type, "code": ts_code,
         "pj": _dumps(params or {}), "st": status, "pg": _dumps(progress),
         "by": issued_by, "ts": _NOW(), "note": note})


def get_command(command_id: str):
    r = fetch_one("SELECT * FROM pms_command WHERE command_id = :cid", {"cid": command_id})
    return _cmd_row(r) if r else None


def list_commands(*, statuses=None, cmd_class=None, limit: int = 200) -> list:
    where, p = [], {"n": int(limit)}
    if statuses:
        keys = []
        for i, s in enumerate(statuses):
            keys.append(f":s{i}")
            p[f"s{i}"] = s
        where.append(f"status IN ({', '.join(keys)})")
    if cmd_class:
        where.append("cmd_class = :cls")
        p["cls"] = cmd_class
    sql = "SELECT * FROM pms_command"
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY id DESC LIMIT :n"
    return [_cmd_row(r) for r in fetch_all(sql, p)]


def update_command(command_id: str, *, status=None, progress=None, done_at=None,
                   note=None) -> int:
    sets, p = [], {"cid": command_id}
    if status is not None:
        sets.append("status = :st")
        p["st"] = status
    if progress is not None:
        sets.append("progress_json = :pg")
        p["pg"] = _dumps(progress)
    if done_at is not None:
        sets.append("done_at = :da")
        p["da"] = done_at
    if note is not None:
        sets.append("note = :note")
        p["note"] = note
    if not sets:
        return 0
    return execute(f"UPDATE pms_command SET {', '.join(sets)} WHERE command_id = :cid", p)


def supersede_param_commands(cmd_type: str, ts_code=None, keep_command_id=None) -> int:
    """同类型 (同标的) 的旧生效参数命令置 SUPERSEDED —— 参数命令当前值取最新 EFFECTIVE。"""
    sql = ("UPDATE pms_command SET status = 'SUPERSEDED' WHERE cmd_class = 'param' "
           "AND status = 'EFFECTIVE' AND cmd_type = :ct")
    p = {"ct": cmd_type}
    if ts_code:
        sql += " AND ts_code = :code"
        p["code"] = ts_code
    else:
        sql += " AND ts_code IS NULL"
    if keep_command_id:
        sql += " AND command_id <> :keep"
        p["keep"] = keep_command_id
    return execute(sql, p)


def latest_effective(cmd_type: str, ts_code=None):
    sql = ("SELECT * FROM pms_command WHERE cmd_class = 'param' AND status = 'EFFECTIVE' "
           "AND cmd_type = :ct")
    p = {"ct": cmd_type}
    if ts_code:
        sql += " AND ts_code = :code"
        p["code"] = ts_code
    sql += " ORDER BY id DESC LIMIT 1"
    r = fetch_one(sql, p)
    return _cmd_row(r) if r else None


def list_effective_stock_params(cmd_types=None) -> list:
    """个股参数命令的当前值 (黑白名单/止损价/目标价/做T授权/冻结 的事实源)。"""
    sql = ("SELECT * FROM pms_command WHERE cmd_class = 'param' AND status = 'EFFECTIVE' "
           "AND ts_code IS NOT NULL")
    p = {}
    if cmd_types:
        keys = []
        for i, t in enumerate(cmd_types):
            keys.append(f":t{i}")
            p[f"t{i}"] = t
        sql += f" AND cmd_type IN ({', '.join(keys)})"
    sql += " ORDER BY id DESC LIMIT 500"
    return [_cmd_row(r) for r in fetch_all(sql, p)]


def _cmd_row(r: dict) -> dict:
    d = dict(r)
    d["params"] = _loads(d.pop("params_json", None), {})
    d["progress"] = _loads(d.pop("progress_json", None), {})
    return d


# ================================================================ pms_plan
def insert_plans(rows: list) -> int:
    if not rows:
        return 0
    now = _NOW()
    payload = [{"pid": r["plan_id"], "cid": r["command_id"], "code": r["ts_code"],
                "act": r["action"], "qty": r.get("qty"), "amt": r.get("amount"),
                "pri": r.get("priority", 100), "dl": r.get("deadline"),
                "st": r.get("status", "PENDING"), "rsn": (r.get("reason") or "")[:300],
                "ts": now}
               for r in rows]
    return execute_many(
        "INSERT INTO pms_plan (plan_id, command_id, ts_code, action, qty, amount, priority, "
        "deadline, status, filled_qty, reason, created_at, updated_at) VALUES "
        "(:pid, :cid, :code, :act, :qty, :amt, :pri, :dl, :st, 0, :rsn, :ts, :ts)", payload)


def list_plans(*, command_id=None, statuses=None, ts_code=None, limit: int = 500) -> list:
    where, p = [], {"n": int(limit)}
    if command_id:
        where.append("command_id = :cid")
        p["cid"] = command_id
    if ts_code:
        where.append("ts_code = :code")
        p["code"] = ts_code
    if statuses:
        keys = []
        for i, s in enumerate(statuses):
            keys.append(f":s{i}")
            p[f"s{i}"] = s
        where.append(f"status IN ({', '.join(keys)})")
    sql = "SELECT * FROM pms_plan"
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY priority ASC, id ASC LIMIT :n"
    return fetch_all(sql, p)


def update_plan(plan_id: str, *, status=None, filled_qty=None) -> int:
    sets, p = ["updated_at = :ts"], {"pid": plan_id, "ts": _NOW()}
    if status is not None:
        sets.append("status = :st")
        p["st"] = status
    if filled_qty is not None:
        sets.append("filled_qty = :fq")
        p["fq"] = int(filled_qty)
    return execute(f"UPDATE pms_plan SET {', '.join(sets)} WHERE plan_id = :pid", p)


def cancel_plans_of_command(command_id: str) -> int:
    return execute("UPDATE pms_plan SET status = 'CANCELLED', updated_at = :ts "
                   "WHERE command_id = :cid AND status IN ('PENDING', 'EXECUTING')",
                   {"cid": command_id, "ts": _NOW()})


def set_plans_deadline(command_id: str, deadline) -> int:
    return execute("UPDATE pms_plan SET deadline = :dl, updated_at = :ts "
                   "WHERE command_id = :cid AND status IN ('PENDING', 'EXECUTING')",
                   {"cid": command_id, "dl": deadline, "ts": _NOW()})


# ================================================================ pms_position
def list_positions(*, only_open: bool = False) -> list:
    sql = "SELECT * FROM pms_position"
    if only_open:
        sql += " WHERE total_qty > 0"
    sql += " ORDER BY ts_code"
    return fetch_all(sql)


def get_position(ts_code: str):
    return fetch_one("SELECT * FROM pms_position WHERE ts_code = :code", {"code": ts_code})


def ensure_position(ts_code: str) -> int:
    return execute(
        "INSERT INTO pms_position (ts_code, status, updated_at) VALUES (:code, 'PLANNED', :ts) "
        "ON DUPLICATE KEY UPDATE updated_at = :ts", {"code": ts_code, "ts": _NOW()})


def update_position(ts_code: str, **fields) -> int:
    clause, p = _set_clause(fields, POSITION_COLS)
    if not clause:
        return 0
    p.update({"code": ts_code, "ts": _NOW()})
    return execute(f"UPDATE pms_position SET {clause}, updated_at = :ts WHERE ts_code = :code", p)


def bump_position_qty(ts_code: str, *, total_delta: int = 0, avail_delta: int = 0) -> int:
    return execute(
        "UPDATE pms_position SET total_qty = GREATEST(0, total_qty + :td), "
        "avail_qty = GREATEST(0, avail_qty + :ad), updated_at = :ts WHERE ts_code = :code",
        {"td": int(total_delta), "ad": int(avail_delta), "code": ts_code, "ts": _NOW()})


def reset_avail_all() -> int:
    """日初 T+1 重置: 全部持仓可卖 (设计 §4)。"""
    return execute("UPDATE pms_position SET avail_qty = total_qty, t0_count_today = 0, "
                   "updated_at = :ts WHERE total_qty >= 0", {"ts": _NOW()})


# ================================================================ pms_lot
def list_lots(ts_code=None, *, status: str = "OPEN", limit: int = 1000) -> list:
    where, p = [], {"n": int(limit)}
    if ts_code:
        where.append("ts_code = :code")
        p["code"] = ts_code
    if status:
        where.append("status = :st")
        p["st"] = status
    sql = "SELECT * FROM pms_lot"
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY open_date ASC, id ASC LIMIT :n"
    return fetch_all(sql, p)


def insert_lot(*, ts_code, lot_type, qty, open_price, open_date, instruction_id=None,
               note=None) -> int:
    now = _NOW()
    return execute(
        "INSERT INTO pms_lot (ts_code, lot_type, qty, open_price, open_date, closed_qty, "
        "realized_pnl, status, instruction_id, note, created_at, updated_at) VALUES "
        "(:code, :lt, :qty, :px, :od, 0, 0, 'OPEN', :iid, :note, :ts, :ts)",
        {"code": ts_code, "lt": lot_type, "qty": int(qty), "px": float(open_price),
         "od": open_date, "iid": instruction_id, "note": note, "ts": now})


def close_lot_qty(lot_id: int, *, qty: int, close_price: float, realized_pnl: float) -> int:
    """核销批次数量。close_avg_price 按加权平均累计 (多次部分卖出不覆盖)。

    注: MySQL 的 UPDATE ... SET 按书写顺序求值且后项可见前项新值 —— 故
    close_avg_price 必须写在 closed_qty 之前 (用旧 closed_qty 加权),
    status 必须写在 qty 之后 (用新 qty 判断是否核销完)。顺序不可随意调整。
    """
    return execute(
        "UPDATE pms_lot SET "
        "close_avg_price = CASE WHEN (closed_qty + :q) > 0 "
        "  THEN (COALESCE(close_avg_price, 0) * closed_qty + :px * :q) / (closed_qty + :q) "
        "  ELSE :px END, "
        "qty = GREATEST(0, qty - :q), "
        "closed_qty = closed_qty + :q, "
        "realized_pnl = realized_pnl + :pnl, "
        "status = CASE WHEN qty <= 0 THEN 'CLOSED' ELSE 'OPEN' END, "
        "updated_at = :ts WHERE id = :id",
        {"q": int(qty), "px": float(close_price), "pnl": float(realized_pnl),
         "id": int(lot_id), "ts": _NOW()})


def update_lot(lot_id: int, **fields) -> int:
    clause, p = _set_clause(fields, LOT_COLS)
    if not clause:
        return 0
    p.update({"id": int(lot_id), "ts": _NOW()})
    return execute(f"UPDATE pms_lot SET {clause}, updated_at = :ts WHERE id = :id", p)


# ================================================================ pms_instruction
def insert_instruction(*, instruction_id, origin_type, origin_id, ts_code, action, side, qty,
                       limit_price=None, window_tdays=3, status="PROPOSED",
                       progress=None) -> int:
    now = _NOW()
    return execute(
        "INSERT INTO pms_instruction (instruction_id, origin_type, origin_id, ts_code, action, "
        "side, qty, limit_price, window_tdays, status, exec_qty, progress_json, created_at, "
        "updated_at) VALUES (:iid, :ot, :oid, :code, :act, :side, :qty, :lp, :w, :st, 0, :pg, "
        ":ts, :ts)",
        {"iid": instruction_id, "ot": origin_type, "oid": origin_id, "code": ts_code,
         "act": action, "side": side, "qty": int(qty), "lp": limit_price,
         "w": int(window_tdays), "st": status, "pg": _dumps(progress), "ts": now})


def list_instructions(*, statuses=None, side=None, ts_code=None, limit: int = 300) -> list:
    where, p = [], {"n": int(limit)}
    if statuses:
        keys = []
        for i, s in enumerate(statuses):
            keys.append(f":s{i}")
            p[f"s{i}"] = s
        where.append(f"status IN ({', '.join(keys)})")
    if side:
        where.append("side = :side")
        p["side"] = side
    if ts_code:
        where.append("ts_code = :code")
        p["code"] = ts_code
    sql = "SELECT * FROM pms_instruction"
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY id DESC LIMIT :n"
    rows = fetch_all(sql, p)
    for r in rows:
        r["progress"] = _loads(r.get("progress_json"), {})
    return rows


def get_instruction(instruction_id: str):
    r = fetch_one("SELECT * FROM pms_instruction WHERE instruction_id = :iid",
                  {"iid": instruction_id})
    if r:
        r["progress"] = _loads(r.get("progress_json"), {})
    return r


def update_instruction(instruction_id: str, *, status=None, exec_qty=None, exec_avg_price=None,
                       dispatch_ref=None, progress=None) -> int:
    sets, p = ["updated_at = :ts"], {"iid": instruction_id, "ts": _NOW()}
    if status is not None:
        sets.append("status = :st")
        p["st"] = status
    if exec_qty is not None:
        sets.append("exec_qty = :eq")
        p["eq"] = int(exec_qty)
    if exec_avg_price is not None:
        sets.append("exec_avg_price = :ep")
        p["ep"] = float(exec_avg_price)
    if dispatch_ref is not None:
        sets.append("dispatch_ref = :dr")
        p["dr"] = dispatch_ref
    if progress is not None:
        sets.append("progress_json = :pg")
        p["pg"] = _dumps(progress)
    return execute(f"UPDATE pms_instruction SET {', '.join(sets)} WHERE instruction_id = :iid", p)


def add_instruction_exec(instruction_id: str, qty: int) -> int:
    return execute("UPDATE pms_instruction SET exec_qty = exec_qty + :q, updated_at = :ts "
                   "WHERE instruction_id = :iid",
                   {"q": int(qty), "iid": instruction_id, "ts": _NOW()})


# ================================================================ pms_proposal
def insert_proposal(*, proposal_id, ts_code, action, qty, hard_numbers, expire_at,
                    judge_verdict=None, judge_reason=None, status="WAIT_USER") -> int:
    return execute(
        "INSERT INTO pms_proposal (proposal_id, ts_code, action, qty, hard_numbers_json, "
        "judge_verdict, judge_reason, status, expire_at, created_at) VALUES "
        "(:pid, :code, :act, :qty, :hn, :jv, :jr, :st, :exp, :ts)",
        {"pid": proposal_id, "code": ts_code, "act": action, "qty": int(qty),
         "hn": _dumps(hard_numbers or {}), "jv": judge_verdict, "jr": judge_reason,
         "st": status, "exp": expire_at, "ts": _NOW()})


def list_proposals(*, statuses=("WAIT_USER",), limit: int = 200) -> list:
    p, keys = {"n": int(limit)}, []
    for i, s in enumerate(statuses or ()):
        keys.append(f":s{i}")
        p[f"s{i}"] = s
    sql = "SELECT * FROM pms_proposal"
    if keys:
        sql += f" WHERE status IN ({', '.join(keys)})"
    sql += " ORDER BY id DESC LIMIT :n"
    rows = fetch_all(sql, p)
    for r in rows:
        r["hard_numbers"] = _loads(r.get("hard_numbers_json"), {})
    return rows


def get_proposal(proposal_id: str):
    r = fetch_one("SELECT * FROM pms_proposal WHERE proposal_id = :pid", {"pid": proposal_id})
    if r:
        r["hard_numbers"] = _loads(r.get("hard_numbers_json"), {})
    return r


def decide_proposal(proposal_id: str, status: str) -> int:
    return execute("UPDATE pms_proposal SET status = :st, decided_at = :ts "
                   "WHERE proposal_id = :pid AND status = 'WAIT_USER'",
                   {"st": status, "pid": proposal_id, "ts": _NOW()})


def expire_proposals(now=None) -> int:
    return execute("UPDATE pms_proposal SET status = 'EXPIRED' "
                   "WHERE status = 'WAIT_USER' AND expire_at < :ts", {"ts": now or _NOW()})


# ================================================================ pms_action_ledger
def insert_ledger(*, ts_code, action, arbiter, verdict, price_at, hard_numbers=None,
                  failed_checks=None, reason=None, ref_id=None) -> int:
    return execute(
        "INSERT INTO pms_action_ledger (ts_code, decided_at, action, arbiter, verdict, "
        "price_at, hard_numbers_json, failed_checks_json, reason, ref_id, outcome_scored) "
        "VALUES (:code, :ts, :act, :arb, :vd, :px, :hn, :fc, :rsn, :ref, 0)",
        {"code": ts_code, "ts": _NOW(), "act": action, "arb": arbiter, "vd": verdict,
         "px": float(price_at or 0), "hn": _dumps(hard_numbers), "fc": _dumps(failed_checks),
         "rsn": (reason or "")[:500], "ref": ref_id})


# ================================================================ pms_cash_flow
def insert_cash_flow(*, ymd, kind, amount, ts_code=None, estimated=0, trade_no=None,
                     instruction_id=None, note=None) -> int:
    """记一笔现金流水 (费用不入成本, 只入现金账 —— 协议 §5.5)。

    (kind, trade_no) 上有唯一键做兜底去重。注意**不要拿返回值判断"是不是新插的"**:
    SQLAlchemy 的 MySQL 方言默认开 CLIENT_FOUND_ROWS, 重复插入照样回 1 (同 inbox_put 的坑)。
    调用方的去重靠 inbox 的 processed 标记, 这里的唯一键只是最后一道保险。
    """
    return execute(
        "INSERT INTO pms_cash_flow (ymd, kind, ts_code, amount, estimated, trade_no, "
        "instruction_id, note, created_at) VALUES (:y, :k, :code, :amt, :est, :tn, :iid, "
        ":note, :ts) ON DUPLICATE KEY UPDATE id = id",
        {"y": int(ymd), "k": kind, "code": ts_code, "amt": float(amount),
         "est": 1 if estimated else 0, "tn": trade_no, "iid": instruction_id,
         "note": (str(note)[:300] if note else None), "ts": _NOW()})


def sum_cash_flow(ymd, kind=None) -> float:
    sql = "SELECT COALESCE(SUM(amount), 0) AS s FROM pms_cash_flow WHERE ymd = :y"
    p = {"y": int(ymd)}
    if kind:
        sql += " AND kind = :k"
        p["k"] = kind
    return float((fetch_one(sql, p) or {}).get("s") or 0)


def list_cash_flow(*, ymd=None, kind=None, limit: int = 200) -> list:
    where, p = [], {"n": int(limit)}
    if ymd:
        where.append("ymd = :y")
        p["y"] = int(ymd)
    if kind:
        where.append("kind = :k")
        p["k"] = kind
    sql = "SELECT * FROM pms_cash_flow"
    if where:
        sql += " WHERE " + " AND ".join(where)
    return fetch_all(sql + " ORDER BY id DESC LIMIT :n", p)


def rule_rejected_today(since) -> set:
    """今天已被**规则闸**拒过的 (代码, 动作) —— 自主扫描据此当日不再重复评估。

    评审账本是给事后判分用的 (「拒了的后来涨了多少」), 同一件事一天记一条足矣。而扫描每分钟
    一跳, 不去重的话一条持续不通过的候选 —— 比如已超总仓上限时的补仓 —— 一天能写进 240 行
    一模一样的记录, 把真正有信息量的行淹掉。判分锚被噪声埋了就不再是锚。
    """
    rows = fetch_all("SELECT DISTINCT ts_code, action FROM pms_action_ledger "
                     "WHERE verdict = 'REJECT' AND arbiter = 'rule' AND decided_at >= :d",
                     {"d": since})
    return {(r["ts_code"], r["action"]) for r in rows}


def judge_rejected_today(since) -> set:
    """今天已被**研判闸**驳回的 (代码, 动作)。

    与上面那条是刻意分开的两个函数, 因为**只有新建仓会用它**。
    加仓类 (FILL/ADD/DCA) 对研判驳回**有意**不做当日去重 —— 契约里那句「研判结论会变」,
    节流责任放在决策系统那侧的半小时缓存上。这条纪律不动。

    新建仓不一样, 有两点不同:
      1. 「这只票今天不该从零建仓」这个结论当天基本不会翻转, 每分钟重问没有新信息;
      2. 候选池是几十只的量级, 而加仓类的分母只有持仓那几只。不去重的话, 决策系统那侧
         虽然靠缓存不烧大模型, PMS 这边却会每分钟往评审账本写一行一模一样的驳回记录 ——
         半小时三十行, 几只票就是几百行。**这正是 2026-07-29 那次把判分锚淹掉的教训**
         (十六只深亏票的补仓候选每分钟被拒一次, 一天几千行), 只是换了个入口重来一遍。
    """
    rows = fetch_all("SELECT DISTINCT ts_code, action FROM pms_action_ledger "
                     "WHERE verdict = 'REJECT' AND arbiter = 'judge' AND decided_at >= :d",
                     {"d": since})
    return {(r["ts_code"], r["action"]) for r in rows}


def buy_signals_today(since) -> dict:
    """今天决策系统判过「盘中转多」的票 → {代码: {reason, price, at}}。

    读的是 signal_service 落下的那些留痕行 (action='SIGNAL_BUY', verdict='NOTE')。
    动作引擎那条路拿它给候选**排序**用: 同样在候选池里的票, 今天被决策系统盘中判过转多的
    排在前面。**只影响先后, 不影响资格** —— 不在候选池里的票不会因为有信号就被建仓,
    候选层那一整套过滤 (ST、黑名单、分数下限、主题限额、预期空间) 一道都不绕。

    为什么走账本而不是让 signal_service 直接调建仓: 两边各管各的一件事 ——
    信号消化管「收到了、记下来」, 动作引擎管「买不买、买多少」, 中间靠账本这个既有的事实源
    传递, 不新增跨模块调用、不复制闸门逻辑。代价是最多差一分钟 (两个调度位各自每分钟一跳),
    而后面还要等择时区间, 这点延迟无所谓。
    """
    rows = fetch_all(
        "SELECT ts_code, MAX(decided_at) AS at, "
        "       SUBSTRING_INDEX(GROUP_CONCAT(reason ORDER BY decided_at DESC SEPARATOR '\\n'), "
        "                       '\\n', 1) AS reason, "
        "       MAX(price_at) AS price "
        "FROM pms_action_ledger "
        "WHERE action = 'SIGNAL_BUY' AND decided_at >= :d GROUP BY ts_code",
        {"d": since})
    return {r["ts_code"]: {"reason": r.get("reason"), "at": str(r.get("at") or ""),
                           "price": float(r.get("price") or 0)} for r in rows}


def opened_names_today(since) -> list:
    """今天已经落了新建仓指令、或还在提议队列里等确认的票 (去重后的代码列表)。

    **只用来显示与留痕, 不做任何上限。** 每天开几只由名额、资金、上限与择时区间自己收敛,
    不由一个人为的数字决定 —— 但「今天开了哪几只」得让人一眼看得见, 否则无人值守就成了黑箱。
    """
    rows = fetch_all(
        "SELECT DISTINCT ts_code FROM pms_instruction "
        "WHERE action = 'OPEN' AND created_at >= :d "
        "UNION "
        "SELECT DISTINCT ts_code FROM pms_proposal "
        "WHERE action = 'OPEN' AND created_at >= :d",
        {"d": since})
    return [r["ts_code"] for r in rows]


def list_ledger(*, ts_code=None, limit: int = 200) -> list:
    sql = "SELECT * FROM pms_action_ledger"
    p = {"n": int(limit)}
    if ts_code:
        sql += " WHERE ts_code = :code"
        p["code"] = ts_code
    sql += " ORDER BY id DESC LIMIT :n"
    rows = fetch_all(sql, p)
    for r in rows:
        r["hard_numbers"] = _loads(r.get("hard_numbers_json"), {})
        r["failed_checks"] = _loads(r.get("failed_checks_json"), [])
    return rows


# ================================================================ pms_daily_report
def upsert_report(ymd: int, report: dict) -> int:
    return execute(
        "INSERT INTO pms_daily_report (ymd, report_json, created_at) VALUES (:y, :r, :ts) "
        "ON DUPLICATE KEY UPDATE report_json = :r, created_at = :ts",
        {"y": int(ymd), "r": _dumps(report), "ts": _NOW()})


def get_report(ymd: int):
    r = fetch_one("SELECT * FROM pms_daily_report WHERE ymd = :y", {"y": int(ymd)})
    return {"ymd": r["ymd"], "report": _loads(r["report_json"], {})} if r else None


def latest_report():
    r = fetch_one("SELECT * FROM pms_daily_report ORDER BY ymd DESC LIMIT 1")
    return {"ymd": r["ymd"], "report": _loads(r["report_json"], {})} if r else None


# ================================================================ pms_industry_map
def get_industry(ts_code: str):
    r = fetch_one("SELECT industry FROM pms_industry_map WHERE ts_code = :code",
                  {"code": ts_code})
    return r["industry"] if r else None


def list_industry(limit: int = 5000) -> list:
    return fetch_all("SELECT ts_code, industry, updated_at FROM pms_industry_map "
                     "ORDER BY ts_code LIMIT :n", {"n": int(limit)})


def upsert_industry(rows: list) -> int:
    if not rows:
        return 0
    now = _NOW()
    payload = [{"code": r["ts_code"], "ind": r["industry"], "ts": now} for r in rows]
    return execute_many(
        "INSERT INTO pms_industry_map (ts_code, industry, updated_at) VALUES (:code, :ind, :ts) "
        "ON DUPLICATE KEY UPDATE industry = :ind, updated_at = :ts", payload)


# ================================================================ pms_plan_snapshot
def latest_plan_digest():
    """最新一份快照的 (plan_date, digest); 没有则 None。判"这份榜变没变"只该拿它比。"""
    r = fetch_one("SELECT plan_date, digest FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1")
    return (r["plan_date"], r["digest"]) if r else None


def insert_plan_snapshot(*, plan_date, digest, roster, meta=None, n_main=0, n_observe=0,
                         capped_main=False, capped_obs=False, fetched_at=None) -> int:
    """落一份名册快照; 内容与**最新那份**相同则跳过, 返回 0。

    两个坑都在这一句"跟最新那份比"上, 值得写清楚:

    1. **不能拿 rowcount 判是不是新插的。** 本文件 `inbox_put` 那里已经栽过一次:
       SQLAlchemy 的 MySQL 方言默认开 `CLIENT_FOUND_ROWS`, `ON DUPLICATE KEY UPDATE id = id`
       命中重复照样回 1。所以先查后插, 由**查的结果**决定 `stored`, 不看 rowcount。
    2. **去重的对象是"最新那份", 不是"曾经出现过的任何一份"。** 早先给
       `(plan_date, digest)` 加了唯一键当去重手段, 结果是: 同一天上游 A→B→A 地改回去时,
       第三次的 A 因为"曾经存过"而落不下去, 库里最新仍是 B —— 于是页面把 B→A 这次**真实的
       回退**显示成 A→B, 方向整个反了, 还会顺带报一句"手上这份没落进快照 (落库失败?)"。
       现在唯一键去掉, 只跟最新那行比: 变了就存, 没变就跳过。表因此可能出现同 digest 的
       多行, 那正是想记的事实 —— 上游确实改回来过。

    并发下两个协程可能同时判定"变了"而落两行相同的快照。代价是多一行 (比对结果为"无变化",
    不产生任何假信号), 换掉的是一整类方向错误, 这笔交换是划算的。
    """
    now = fetched_at or _NOW()
    cur = latest_plan_digest()
    if cur and cur[0] == str(plan_date) and cur[1] == str(digest):
        return 0
    execute(
        "INSERT INTO pms_plan_snapshot "
        "  (plan_date, digest, fetched_ymd, fetched_at, n_main, n_observe, "
        "   capped_main, capped_obs, roster_json, meta_json) "
        "VALUES (:d, :g, :ymd, :ts, :nm, :no, :cm, :co, :roster, :meta)",
        {"d": str(plan_date), "g": str(digest), "ymd": int(now.strftime("%Y%m%d")), "ts": now,
         "nm": int(n_main or 0), "no": int(n_observe or 0),
         "cm": 1 if capped_main else 0, "co": 1 if capped_obs else 0,
         "roster": _dumps(roster or []), "meta": _dumps(meta or {})})
    return 1


def latest_plan_snapshots(limit: int = 2) -> list:
    """最近若干份快照, **新的在前**。带名册 —— 比对要用。"""
    rows = fetch_all(
        "SELECT id, plan_date, digest, fetched_ymd, fetched_at, n_main, n_observe, "
        "       capped_main, capped_obs, roster_json, meta_json "
        "FROM pms_plan_snapshot ORDER BY id DESC LIMIT :n", {"n": max(1, int(limit))})
    return [_snapshot_row(r) for r in rows]


def list_plan_snapshots(*, plan_date=None, limit: int = 50) -> list:
    """快照台账 (不带名册 —— 页面列表不需要那 20KB)。

    同一 plan_date 有几行, 就是上游那天重算过几版 (UPSTREAM_PLAN_API.md §7.3)。
    """
    sql = ("SELECT id, plan_date, digest, fetched_ymd, fetched_at, n_main, n_observe, "
           "       capped_main, capped_obs, meta_json FROM pms_plan_snapshot ")
    params = {"n": max(1, int(limit))}
    if plan_date:
        sql += "WHERE plan_date = :d "
        params["d"] = str(plan_date)
    rows = fetch_all(sql + "ORDER BY id DESC LIMIT :n", params)
    for r in rows:
        r["meta"] = _loads(r.pop("meta_json", None), {})
        r["capped_main"] = bool(r.get("capped_main"))   # 与 latest_plan_snapshots 保持同型
        r["capped_obs"] = bool(r.get("capped_obs"))
    return rows


def prune_plan_snapshots(keep: int = 200) -> int:
    """只留最近 keep 份。

    单表守卫不许子查询跨表, 所以先查出保留水位的 id 再按 id 删 —— 两条单表语句,
    比一条 `DELETE ... WHERE id NOT IN (SELECT ...)` 干净, 也不会被守卫拦下。

    下限是 **2 份不是 1 份**: 比对天生要两份, 留一份等于把榜单变化永久锁死在"首次"上,
    而且页面上看起来一切正常。参数填了 0 或 1 也按 2 算。
    """
    keep = max(2, int(keep or 0))
    r = fetch_one("SELECT id FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1 OFFSET :k",
                  {"k": keep})
    if not r:
        return 0
    return execute("DELETE FROM pms_plan_snapshot WHERE id <= :id", {"id": int(r["id"])})


def _snapshot_row(r: dict) -> dict:
    out = dict(r)
    out["roster"] = _loads(out.pop("roster_json", None), [])
    out["meta"] = _loads(out.pop("meta_json", None), {})
    out["capped_main"] = bool(out.get("capped_main"))
    out["capped_obs"] = bool(out.get("capped_obs"))
    return out


# ================================================================ pms_strategy (个股交易方案)
# 只允许更新的列白名单 (与 DDL 一致; `type` 只在建时设, 不入更新集, 避开保留字与误改)
STRATEGY_COLS = {"status", "autonomy", "params_json", "state_json", "note"}


def list_strategies(*, ts_code=None, statuses=None, limit: int = 500) -> list:
    where, p = [], {"n": int(limit)}
    if ts_code:
        where.append("ts_code = :code")
        p["code"] = ts_code
    if statuses:
        keys = []
        for i, s in enumerate(statuses):
            keys.append(f":st{i}")
            p[f"st{i}"] = s
        where.append(f"status IN ({', '.join(keys)})")
    sql = "SELECT * FROM pms_strategy"
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY id DESC LIMIT :n"
    rows = fetch_all(sql, p)
    for r in rows:
        r["params"] = _loads(r.get("params_json"), {})
        r["state"] = _loads(r.get("state_json"), {})
    return rows


def get_strategy(strategy_id: str):
    r = fetch_one("SELECT * FROM pms_strategy WHERE strategy_id = :sid", {"sid": strategy_id})
    if r:
        r["params"] = _loads(r.get("params_json"), {})
        r["state"] = _loads(r.get("state_json"), {})
    return r


def active_strategies() -> list:
    return list_strategies(statuses=["ACTIVE"], limit=1000)


def active_strategy_codes() -> set:
    """有 ACTIVE 策略的 ts_code 集合 —— 动作引擎据此排除这些票。"""
    rows = fetch_all("SELECT DISTINCT ts_code FROM pms_strategy WHERE status = 'ACTIVE'")
    return {r["ts_code"] for r in rows if r.get("ts_code")}


def insert_strategy(*, strategy_id, ts_code, stype, autonomy="auto", params=None,
                    state=None, status="ACTIVE", note=None) -> int:
    now = _NOW()
    return execute(
        "INSERT INTO pms_strategy (strategy_id, ts_code, `type`, status, autonomy, "
        "params_json, state_json, note, created_at, updated_at) "
        "VALUES (:sid, :code, :type, :status, :auto, :params, :state, :note, :ts, :ts)",
        {"sid": strategy_id, "code": ts_code, "type": stype, "status": status,
         "auto": autonomy, "params": _dumps(params or {}), "state": _dumps(state or {}),
         "note": note, "ts": now})


def update_strategy(strategy_id: str, **fields) -> int:
    """更新策略。params/state 传 dict 会自动 json 化到 *_json 列。"""
    upd = dict(fields)
    if "params" in upd:
        upd["params_json"] = _dumps(upd.pop("params"))
    if "state" in upd:
        upd["state_json"] = _dumps(upd.pop("state"))
    clause, p = _set_clause(upd, STRATEGY_COLS)
    if not clause:
        return 0
    p["sid"] = strategy_id
    p["ts"] = _NOW()
    return execute(
        f"UPDATE pms_strategy SET {clause}, updated_at = :ts WHERE strategy_id = :sid", p)