# -*- coding: utf-8 -*- """ 规则闸 · 一级关口 (纯逻辑, 无外部依赖, 可单测) ================================================ 设计 POSITION_MGMT_DESIGN.md §7: 「全指令必过, PMS 纯代码。下发前用最新数据终检 —— 命令参数上限、一手检查、 T+1 可卖、冻结状态、刹车状态、行业集中度、(买入)不追高。 任一不过或必要输入缺失 → 拒绝 (宁可不动), 未通过项落账本。」 两条容易搞反的口径, 在此钉死: 1. **冻结与刹车只挡增持, 不挡减持** —— 卖出、止损任何时候都放行 (设计 §3.1/§5)。 2. **命令驱动不受刹车限制** —— 刹车是给自主提议踩的, 用户命令优先, 只提示不拦 (设计 §5「命令类不受限, 执行前提示」)。上限/一手/可卖量这类硬约束则一视同仁。 输出结构固定为 {"passed", "failed", "warnings", "hard_numbers"}, 直接喂 pms_action_ledger 的 failed_checks_json / hard_numbers_json —— 「拒了的后来涨了多少」靠这份留痕做判分。 """ from __future__ import annotations from app.core.sizer import LOT from app.core.planner import check_all_caps BUY, SELL = "buy", "sell" def _num(v, default=0.0): try: return float(v) except (TypeError, ValueError): return default def check(*, side: str, action: str, qty: int, price: float, ctx: dict) -> dict: """指令下发前终检。 ctx 由调用方备齐 (services/executor.py): position {total_qty, avail_qty, frozen_reason} 该股账本快照 caps check_all_caps 所需组合上下文 (仅买入用) day {price, vwap, ma5, day_chg_from_open, halted, limit_up, limit_down} params {no_chase_ma5, buy_halt_dayup, sector_source_ready} flags {buy_halt, exec_halt, brake_active, blacklisted, is_command} """ pos = ctx.get("position") or {} day = ctx.get("day") or {} prm = ctx.get("params") or {} flg = ctx.get("flags") or {} is_cmd = bool(flg.get("is_command")) side = str(side or "").lower() failed, warns = [], [] qty = int(qty or 0) price = _num(price) total_qty = int(pos.get("total_qty") or 0) avail_qty = int(pos.get("avail_qty") or 0) hard = {"side": side, "action": action, "qty": qty, "price": price, "total_qty": total_qty, "avail_qty": avail_qty, "frozen_reason": pos.get("frozen_reason") or "NONE", "day_chg_from_open": day.get("day_chg_from_open"), "vwap": day.get("vwap"), "ma5": day.get("ma5"), "is_command": is_cmd} # ---- 通用: 必要输入缺失一律拒绝 (宁可不动) ---- if qty <= 0: failed.append(f"QTY_INVALID: 数量 {qty} 非法") if price <= 0: failed.append("PRICE_MISSING: 取不到现价, 无法定限价与校验 (宁可不动)") if day.get("halted"): failed.append("HALTED: 停牌, 当日跳过顺延") if flg.get("exec_halt"): failed.append("EXEC_HALT: 全局暂停执行 (休假模式) 生效中") if side == SELL: _check_sell(failed, warns, qty, total_qty, avail_qty, day) elif side == BUY: _check_buy(failed, warns, qty, price, ctx, pos, day, prm, flg, is_cmd, hard) else: failed.append(f"SIDE_INVALID: 方向 {side!r} 非法") return {"passed": not failed, "failed": failed, "warnings": warns, "hard_numbers": hard} def _check_sell(failed, warns, qty, total_qty, avail_qty, day): """减持方向: 冻结/刹车/上限一概不拦, 只看「卖得出去吗」。""" if qty > total_qty: failed.append(f"OVER_SELL: 卖出 {qty} > 持仓 {total_qty}") elif qty > avail_qty: failed.append(f"T1_UNAVAILABLE: 卖出 {qty} > T+1 可卖 {avail_qty} (当日买入次日才可卖)") # 清仓允许卖零股 (A股规则); 部分减持必须整百 if qty % LOT != 0 and qty != total_qty: failed.append(f"LOT_INVALID: 部分减持 {qty} 股非整百 (零股只能在清仓时一次性卖出)") if day.get("limit_down"): warns.append("LIMIT_DOWN: 跌停封板, 大概率成交不了, 已按限价挂出") def _check_buy(failed, warns, qty, price, ctx, pos, day, prm, flg, is_cmd, hard): """增持方向: 全部硬约束都要过。""" if qty % LOT != 0: failed.append(f"LOT_INVALID: 买入 {qty} 股非整百") if (pos.get("frozen_reason") or "NONE") != "NONE": failed.append(f"FROZEN: 该股处于 {pos['frozen_reason']}, 禁止增持") if flg.get("buy_halt"): failed.append("BUY_HALT: 全局暂停买入生效中") if flg.get("blacklisted"): failed.append("BLACKLIST: 该股在黑名单 (永不买入)") if flg.get("brake_active"): if is_cmd: warns.append("BRAKE_ACTIVE: 组合刹车中 —— 命令驱动不受限, 仅提示") else: failed.append("BRAKE_ACTIVE: 组合刹车中, 自主增持暂停") if day.get("limit_up"): failed.append("LIMIT_UP: 涨停封板, 不追买") # 不追高: 当日涨幅 与 距 MA5 两道 dayup = day.get("day_chg_from_open") cap_dayup = _num(prm.get("buy_halt_dayup"), 0.05) if dayup is None: # 缺输入**必须留痕**, 与下面 MA5_MISSING 同口径。原来只有 `is not None` 一个条件, # 缺了就整道闸无声跳过 —— 而自主提议那条路一直硬编码 None, 于是动作引擎产出的每一笔 # 买单都没过过这道闸, 返回值还是 `passed=True, failed=[], warnings=[]`, 与 # 「涨幅 2%、检查通过」一字不差, 账本里也查不出来。 warns.append("DAYUP_MISSING: 取不到当日涨幅, 不追高(涨幅)一项未校验") elif _num(dayup) > cap_dayup: failed.append(f"NO_CHASE_DAYUP: 当日涨幅 {_num(dayup):.2%} > 上限 {cap_dayup:.0%}") ma5 = _num(day.get("ma5")) cap_ma5 = _num(prm.get("no_chase_ma5"), 0.06) if ma5 > 0: gap = price / ma5 - 1 hard["ma5_gap"] = round(gap, 4) if gap > cap_ma5: failed.append(f"NO_CHASE_MA5: 距 MA5 {gap:.2%} > 上限 {cap_ma5:.0%}") else: warns.append("MA5_MISSING: 取不到 MA5, 距均线一项未校验") caps = ctx.get("caps") if caps: bad = check_all_caps(ts_code=ctx.get("ts_code") or "-", add_amount=qty * price, ctx=caps) failed.extend(bad) if not caps.get("sector_source_ready", True): warns.append("SECTOR_OFF: 行业数据源未配置, 行业集中度未校验 (页面已提示)") _check_cash(failed, warns, qty, price, caps, hard) else: failed.append("CAPS_MISSING: 缺组合上下文, 无法校验上限 (宁可不动)") def _check_cash(failed, warns, qty, price, caps, hard): """买入前的资金校验 (QMT_INTERFACE_REQUIREMENTS A3 的原意)。 与组合上限是**两个不同的约束**, 所以单列一条而不是塞进 check_all_caps: 上限管「该不该买这么多」—— 仓位纪律, 基准是 PMS_TOTAL_SCALE 这个用户命令参数 资金管「买不买得起」 —— 账户事实, 基准是 ws 资金快照的 available_cash + 当日回笼 2026-07-30 之前只有前者: scale=200 万而模拟账户实际只有 98 万, 于是排出的方案规则闸一路 放行, 要等 QMT 回 `INSUFFICIENT_CASH` 才被拒; 而 reject 不自动重发, 择时下一跳又算又发 又拒 —— 页面看着一切正常, 实际一单也下不去。 **拿不到真实资金时降级不拦, 但必须留痕。** 与研判闸「拿不到不算通过、也不当拒绝, 降级 并记录」同一口径 —— 一律拒会让资金快照一断就把所有买入停掉, 那是把通道故障升级成业务 停摆。降级期间 QMT 那层的 `INSUFFICIENT_CASH` 仍是最后一道防线, 只是会吵一点。 """ src = str(caps.get("cash_source") or "estimate") avail = caps.get("cash_avail") need = round(qty * price, 2) hard.update({"cash_need": need, "cash_source": src, "cash_avail": None if avail is None else round(_num(avail), 2)}) if src != "ws" or avail is None: warns.append("CASH_ESTIMATED: 拿不到 ws 资金快照, 本单未按真实可用资金校验 " "(scale−市值 的估算值不能当真钱用)") return if need > _num(avail): failed.append(f"INSUFFICIENT_CASH: 需 {need:.0f} 元 > 可用 {_num(avail):.0f} 元 " f"(含当日卖出回笼)") def summarize(results: list) -> dict: """一批指令的闸门统计 (日报关注区用)。""" passed = [r for r in results if r.get("passed")] reasons = {} for r in results: for f in r.get("failed") or []: k = f.split(":")[0] reasons[k] = reasons.get(k, 0) + 1 return {"total": len(results), "passed": len(passed), "rejected": len(results) - len(passed), "by_reason": reasons}