From f25fa1d2f6700407b08de7c97e5f9eebf51a094c Mon Sep 17 00:00:00 2001 From: zlt Date: Tue, 11 Aug 2026 11:58:16 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=B0=E5=AF=8C=E4=BA=A4=E6=98=93=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/repo/pms_repo.py | 17 + app/scheduler.py | 13 +- app/services/strategy_runner.py | 609 ++++++++++++++++++++++++++++--- app/services/strategy_service.py | 136 +++++++ app/web/main.py | 107 +++++- app/web/static/index.html | 207 ++++++++++- ddl_pms_v1.sql | 18 + scripts/test_batch6_units.py | 4 +- 8 files changed, 1049 insertions(+), 62 deletions(-) create mode 100644 app/services/strategy_service.py diff --git a/app/repo/pms_repo.py b/app/repo/pms_repo.py index c5f5186..4e1282d 100644 --- a/app/repo/pms_repo.py +++ b/app/repo/pms_repo.py @@ -803,3 +803,20 @@ def update_strategy(strategy_id: str, **fields) -> int: p["ts"] = _NOW() return execute( f"UPDATE pms_strategy SET {clause}, updated_at = :ts WHERE strategy_id = :sid", p) + + +# ================================================================ pms_op_log (交易员操作日志) +def insert_op_log(*, op, result, by="user", ts_code=None, reason=None, params=None, ref=None) -> int: + return execute( + "INSERT INTO pms_op_log (at, op, by_user, ts_code, result, reason, params_json, ref) " + "VALUES (:at, :op, :by, :code, :res, :reason, :params, :ref)", + {"at": _NOW(), "op": str(op)[:40], "by": by, "code": ts_code, "res": result, + "reason": (str(reason)[:500] if reason else None), + "params": (_dumps(params) if params else None), "ref": ref}) + + +def list_op_log(*, limit: int = 200) -> list: + rows = fetch_all("SELECT * FROM pms_op_log ORDER BY id DESC LIMIT :n", {"n": int(limit)}) + for r in rows: + r["params"] = _loads(r.get("params_json"), {}) + return rows diff --git a/app/scheduler.py b/app/scheduler.py index 76264dc..da6f997 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -172,13 +172,20 @@ def signal_digest(): @celery_app.task(name="pms.t0_close") @guard(trade_day=True) def t0_close(): - """T 仓强制平回 (14:50)。做T为二期上线, 此处先留调度位并自证 T 仓应为 0。""" + """T 仓强制平回 (14:50): 对挂了做T策略且当日仍有未平腿的票, 立刻发对向平仓腿打平, + 绝不过夜 (设计 §六 rails)。平回后再自证账面 T 仓, 残留的记 WARN 交人工核查。 + + 正T 的账面 t0_qty 在平回后仍可能 > 0: 今日买入的那份 T0 批次 T+1 才可卖, 平回卖的是 + 底仓存量, 净持仓已打平, 该 T0 批次次日由对账并入底仓 —— 这属正常, 不是没平回。""" from app.repo import pms_repo + from app.services import strategy_runner + r = strategy_runner.force_t0_close() left = [p["ts_code"] for p in pms_repo.list_positions(only_open=True) if int(p.get("t0_qty") or 0) > 0] if left: - logger.warning("[t0_close] 仍有 T 仓未平回: %s (做T为二期功能, 请人工核查)", left) - return {"t0_open": left, "phase": "二期"} + logger.warning("[t0_close] 平回后账面仍有 T0 批次: %s —— 多为今日买入次日才可卖(T+1)" + "或平仓腿尚未成交, 请人工核查", left) + return {"forced": r, "t0_open_after": left} @celery_app.task(name="pms.daily_settle") diff --git a/app/services/strategy_runner.py b/app/services/strategy_runner.py index afb9467..e69c48e 100644 --- a/app/services/strategy_runner.py +++ b/app/services/strategy_runner.py @@ -1,105 +1,622 @@ # -*- coding: utf-8 -*- """ -个股交易方案 (策略) 运行器 —— PER_STOCK_STRATEGY_PLAN.md §四 -============================================================ +个股交易方案 (策略) 运行器 —— PER_STOCK_STRATEGY_PLAN.md §四/§六/§七/§七B +============================================================================ 每分钟一跳 (挂在 scheduler.intraday_exec 里, 与 run_tick 并列)。载入 ACTIVE 策略, 按类型 -评估, 触发就**发一张短窗口指令** (window_tdays=1, is_command=True, origin_type='strategy'), -交 executor.run_tick 用现有管线执行 (择时 / 规则闸 / T+1 / 下发 / 账本 一道不重写); -autonomy=confirm 的落一条提议进「等我拍板」。 +评估 (做T / 网格 / 跟踪止盈), 触发就**发一张短窗口指令** (window_tdays=1, is_command=True, +origin_type='strategy'), 交 executor.run_tick 用现有管线执行 —— 择时 / 规则闸 / T+1 可卖封顶 / +下发 / 账本一道不重写; autonomy=confirm 的落一条提议进「等我拍板」, 人点采纳后再由本层发指令。 -安全 (设计 §九): - * 全局开关 PMS_STRATEGY_ENABLED (默认 False) —— 关着时本模块整体空转。 - * 挂了 ACTIVE 策略的票由 action_engine.scan 排除 (见 active_codes)。 - * 策略动作走命令口径 (过规则闸、不过研判闸)。 +**只做加法**: 本模块不改 executor / rule_gate / action_engine 任何一行, 唯一的引擎触点是 +action_engine.scan 早已加好的「有 ACTIVE 策略的票跳过」那一条 skip (设计 §四)。 + +一条铁律定了整个框架 —— T+1 (设计 §三) +-------------------------------------------------- +当日买入不可当日卖出, 所以做T / 网格 / 跟踪止盈本质都是「在一只**底仓**上、用 T+1 可卖的存量股 +做买卖」= 同一自动机的三种配置。卖出腿一律经 run_tick 按 avail_qty 封顶, T+1 天然被挡在那里。 + +批次口径 (与 app/core/recon.ACTION_TO_LOT 对齐, 不另立) +-------------------------------------------------- + 做T —— 买卖两腿都用 action='T0_ROUND' → 记 T0 批次; 卖出核销次序 T0→ADD→DCA→FILL→BASE + 先把 T0 批次对冲掉, 底仓与摊薄成本不动, 做T利润自然摊入 realized_t_profit。 + 网格 —— 买腿 action='ADD' (记 ADD 批次), 卖腿 action='TRIM' (从 avail 卖)。 + 跟踪止盈 —— 卖腿 action='TRIM' (部分) / 'EXIT' (全清)。只卖不买。 + +安全 (设计 §九) +-------------------------------------------------- + * 全局开关 PMS_STRATEGY_ENABLED (默认 False) —— 关着时本模块整体空转、一条指令都不发。 + * 挂了 ACTIVE 策略的票由 action_engine.scan 排除 (两个大脑不抢同一只)。 + * 策略动作走命令口径 (is_command=True, 过规则闸、不过研判闸)。 * scheduler 的 @guard(session=True) 兜住: 非交易时段 / 休假模式不跑。 - -**本文件是第 1 步骨架**: 三个评估器 (做T / 网格 / 跟踪止盈) 为占位, 一律返回 None, 全链空跑通; -规则在第 2~4 步按设计 §六 / §七 / §七B 接入, 接入点在 tick() 里已注明。 + * 影子/实盘由 PMS_DISPATCH_MODE 决定 (shadow=只落影子出口不碰真 QMT), 与本层无关、自动继承。 + * **双保险**: 本层 rails 先拦 (熔断 / 当日次数 / 上限 / 平回 / 下界), 规则闸再拦一道 (合规)。 + * **关键路径禁止丢弃返回值**: 发不出指令 / 落不了库一律进 out["errors"], 绝不静默当成功。 """ from __future__ import annotations import logging +from datetime import datetime +from app.core import tradedays as td +from app.core.sizer import LOT from app.repo import pms_repo -from app.services import param_store +from app.services import market, param_store, portfolio logger = logging.getLogger("pms.strategy") +# 指令在途 (未终态) 的状态集 —— 与 executor.LIVE 一致; 本层据此判断「上一腿还没走完, 先别再发」。 +LIVE_INS = ("PROPOSED", "RULE_PASSED", "DISPATCHED") -# 评估器签名 (第 2~4 步接入): fn(st, pos, day, now) -> -# None 或 {"side": "buy|sell", "qty": int, "limit": float, "reason": str, -# "leg": str, "state_patch": dict} -def _eval_t0(st, pos, day, now): - return None # 第 2 步: 做T 正T/反T + 3 次/日 + 14:50 平回 + T亏熔断 (设计 §六) +# 批次动作 (与 recon.ACTION_TO_LOT 对齐) +A_T0 = "T0_ROUND" # 做T 买卖两腿 → T0 批次 +A_GRID_BUY = "ADD" # 网格买腿 → ADD 批次 +A_SELL = "TRIM" # 网格 / 跟踪止盈 部分卖 +A_EXIT = "EXIT" # 跟踪止盈 全清 (允许零股一次性清出) + +# 触发用的小额贴近带 (设计只说「近支撑 / 近压力 / 滞涨」, 未给具体数; 这里取保守小带并写明)。 +NEAR_BAND = 0.005 # 现价距支撑/压力 0.5% 以内算「贴近」 +OFF_HIGH_BAND = 0.003 # 距当日高点回落 0.3% 以上算「滞涨」(反T 用) -def _eval_grid(st, pos, day, now): - return None # 第 3 步: 网格 高抛低吸 + 下界=继续持有不再买 (设计 §七) +# ================================================================ 取数小工具 +def _f(v, d=0.0): + try: + return float(v) + except (TypeError, ValueError): + return d -def _eval_trail(st, pos, day, now): - return None # 第 4 步: 跟踪止盈 高水位回撤触发卖出 (设计 §七B) +def _pos_of(view: dict, code: str) -> dict: + for x in view.get("positions") or []: + if x.get("ts_code") == code: + return x + return {} + + +def _round_lot(qty) -> int: + """向下取整到一手 (100 股)。不足一手返回 0。""" + n = int(_f(qty)) + return (n // LOT) * LOT + + +def _today() -> int: + return td.ymd() + + +# ================================================================ 指令 / 提议下发 +def _emit_instruction(st: dict, dec: dict, *, forced: bool = False) -> str: + """按决策发一张短窗口命令指令 (window_tdays=1, is_command=True), 交 run_tick 执行。 + + 返回 instruction_id。limit_price 不在这里定 —— 交由 run_tick 的择时按实时行情现算 + (与 materialize_plans 同口径: 那里也是 limit_price=None)。 + """ + code = st["ts_code"] + now = datetime.now() + iid = f"STR{_today()}{now.strftime('%H%M%S')}_{code.replace('.', '')}_{dec['leg'][:1]}"[:40] + prog = {"is_command": True, "deadline": str(_today()), "children": [], + "origin": "strategy", "strategy_id": st["strategy_id"], "stype": st.get("type"), + "leg": dec["leg"], "reason": dec.get("reason"), "forced": bool(forced)} + pms_repo.insert_instruction( + instruction_id=iid, origin_type="strategy", origin_id=st["strategy_id"], + ts_code=code, action=dec["action"], side=dec["side"], qty=int(dec["qty"]), + limit_price=None, window_tdays=1, status="PROPOSED", progress=prog) + return iid + + +def _emit_proposal(st: dict, dec: dict) -> str: + """autonomy=confirm: 落一条提议进「等我拍板」。hard_numbers 里带全腿谱, 人采纳后由 + main._decide 识别 kind=strategy 再回调本层发指令 (不走通用物化, 因为 T0 两腿同 action、 + side 无法由 action 反推)。""" + code = st["ts_code"] + now = datetime.now() + pid = f"STP{_today()}{now.strftime('%H%M%S')}_{code.replace('.', '')}"[:40] + hard = {"kind": "strategy", "strategy_id": st["strategy_id"], "stype": st.get("type"), + "side": dec["side"], "leg": dec["leg"], "action": dec["action"], + "qty": int(dec["qty"]), "reason": dec.get("reason"), "entry": dec.get("entry")} + expire = now.replace(hour=15, minute=0, second=0, microsecond=0) + pms_repo.insert_proposal(proposal_id=pid, ts_code=code, action=dec["action"], + qty=int(dec["qty"]), hard_numbers=hard, expire_at=expire, + judge_verdict="STRATEGY", + judge_reason=f"{st.get('type')} · {dec.get('reason')}") + return pid + + +def emit_from_spec(spec: dict) -> dict: + """confirm 提议被采纳后的回调 (main._decide 调): 按 hard_numbers 里存的腿谱发指令。 + 返回 {ok, instruction_id} 或 {ok:false, error}。""" + try: + st = pms_repo.get_strategy(spec.get("strategy_id")) + if not st: + return {"ok": False, "error": "策略不存在或已撤下"} + if st.get("status") != "ACTIVE": + return {"ok": False, "error": f"策略处于 {st.get('status')}, 不再发指令"} + dec = {"side": spec["side"], "action": spec["action"], "qty": int(spec["qty"]), + "leg": spec.get("leg") or "open", "reason": spec.get("reason"), + "entry": spec.get("entry")} + iid = _emit_instruction(st, dec) + state = dict(st.get("state") or {}) + state["pending"] = {"iid": iid, "leg": dec["leg"], "dir": spec.get("dir"), + "qty": dec["qty"], "entry": dec.get("entry")} + pms_repo.update_strategy(st["strategy_id"], state=state) + return {"ok": True, "instruction_id": iid} + except Exception as e: # noqa: BLE001 + logger.exception("[strategy] 采纳提议发指令失败") + return {"ok": False, "error": f"{type(e).__name__}: {e}"} + + +# ================================================================ 在途腿的收敛 +def _pending_terminal(pending: dict): + """上一腿是否已终态。返回 (已终态?, 指令行 or None)。无 pending 视为已终态。""" + if not pending: + return True, None + iid = pending.get("iid") + if not iid: # confirm 提议还没被采纳 —— 看提议是否还在等 + pid = pending.get("pid") + if not pid: + return True, None + pr = pms_repo.get_proposal(pid) + if pr and pr.get("status") == "WAIT_USER": + return False, None # 还在等人拍板, 先别再发 + return True, None # 已采纳(转指令,另有pending.iid)/驳回/过期 + ins = pms_repo.get_instruction(iid) + if not ins: + return True, None + return (ins.get("status") not in LIVE_INS), ins + + +def _reconcile(st: dict, state: dict, pos: dict): + """把已终态的上一腿并进 state: 开仓腿成交 → 记 open_leg; 平仓腿成交 → 记一次完成、清 open_leg。 + + 做T 用; 网格 / 跟踪止盈 的 filled_levels / high_water 在各自评估器里按现价推进, 不依赖这里。 + """ + pending = state.get("pending") + done, ins = _pending_terminal(pending) + if not done: + return False # 上一腿未走完 + state["pending"] = None + if not ins or not pending or not pending.get("iid"): + return True + filled = int(ins.get("exec_qty") or 0) > 0 + leg = pending.get("leg") + if st.get("type") == "T0": + if leg == "open" and filled: + state["open_leg"] = {"dir": pending.get("dir"), "qty": int(ins.get("exec_qty") or 0), + "entry": _f(pending.get("entry")) or _f(ins.get("limit_price")), + "opened_at": ins.get("updated_at") and str(ins["updated_at"])} + elif leg == "close": + # 平仓腿终态: 只有真成交才算一轮完成; 一股没成 (到期/被拒) 要**保留 open_leg**, + # 让下一跳与 14:50 平回继续补平 —— 绝不能把一条还开着的 T 仓静默丢掉。 + filled_qty = int(ins.get("exec_qty") or 0) + ol = dict(state.get("open_leg") or {}) + rem = _round_lot(_f(ol.get("qty")) - filled_qty) + if rem >= LOT: + ol["qty"] = rem + state["open_leg"] = ol + if filled_qty > 0: + logger.info("[strategy] %s 平仓腿部分成交 %s, 余 %s 股待续平", + st.get("strategy_id"), filled_qty, rem) + else: + state["t_count_today"] = int(state.get("t_count_today") or 0) + 1 + state["open_leg"] = None + return True + + +# ================================================================ 评估器: 做T (设计 §六) +def _eval_t0(st, pos, day, now, ctx): + """做T: 正T (回落近支撑 → 买, 目标价差高卖) / 反T (近压力或滞涨 → 卖, 低买回)。 + + rails: 当日 ≤ 3 次; 单票 / 全局当日T亏熔断后当日禁开新T (仍允许平回已开的腿); + 14:50 强制平回由 force_t0_close 走 (本函数不管平回时点, 只管盘中触发)。 + """ + state = ctx["state"] + price = _f(day.get("price")) + if price <= 0: + return None + total = int(pos.get("total_qty") or 0) + avail = int(pos.get("avail_qty") or 0) + params = st.get("params") or {} + t_ratio = min(_f(params.get("t_ratio")), param_store.get_float("PMS_T0_RATIO_MAX", 0.333)) + t_qty = _round_lot(total * t_ratio) + + open_leg = state.get("open_leg") + round_target = param_store.get_float("PMS_T0_ROUND_TARGET", 0.015) + + # ---- 有未平的腿 → 只找平回机会 (熔断不挡平回) ---- + if open_leg: + entry = _f(open_leg.get("entry")) or price + q = _round_lot(open_leg.get("qty")) + if open_leg.get("dir") == "long": # 正T: 已买, 等高卖 + pressure = _f(pos.get("pressure_ref")) + hit = price >= entry * (1 + round_target) or (pressure > 0 and price >= pressure * (1 - NEAR_BAND)) + if hit and q > 0: + return {"side": "sell", "action": A_T0, "qty": min(q, avail), "leg": "close", + "reason": f"正T平回: 现价 {price} 达目标 {entry * (1 + round_target):.2f}(买价 {entry})"} + else: # 反T: 已卖, 等低买回 + support = _f(pos.get("support_ref")) + hit = price <= entry * (1 - round_target) or (support > 0 and price <= support * (1 + NEAR_BAND)) + if hit and q > 0: + return {"side": "buy", "action": A_T0, "qty": q, "leg": "close", + "reason": f"反T平回: 现价 {price} 回到目标 {entry * (1 - round_target):.2f}(卖价 {entry})"} + return None + + # ---- 无未平的腿 → 看要不要开新的一轮 (受熔断 / 3 次 / 存量约束) ---- + if ctx.get("halted") or int(state.get("t_count_today") or 0) >= 3: + return None + if t_qty < LOT: + return None + high = _f(day.get("high")) + support = _f(pos.get("support_ref")) + pressure = _f(pos.get("pressure_ref")) + pull = param_store.get_float("PMS_T0_PULLBACK_PCT", 0.03) + rally = param_store.get_float("PMS_T0_RALLY_PCT", 0.05) + dayup = _f(day.get("day_chg_from_open")) + + # 正T: 距当日高点回落 ≥ 回落阈 且 (近支撑 或 无支撑参照时仅凭回落, 写明) + if high > 0 and price <= high * (1 - pull): + near_sup = support > 0 and price <= support * (1 + NEAR_BAND) + if near_sup or support <= 0: + note = "近支撑" if near_sup else "无支撑参照, 仅凭回落(偏保守)" + return {"side": "buy", "action": A_T0, "qty": t_qty, "leg": "open", "dir": "long", + "entry": price, + "reason": f"正T开仓: 距高点 {high} 回落 {1 - price / high:.1%} 且{note}, 买 {t_qty} 股"} + + # 反T: 近压力 或 (日内涨 ≥ 反T阈 且 已从高点滞涨) —— 卖 avail 的 t_qty, 待低买回 + near_pre = pressure > 0 and price >= pressure * (1 - NEAR_BAND) + stalled = dayup >= rally and high > 0 and price <= high * (1 - OFF_HIGH_BAND) + if (near_pre or stalled) and avail >= LOT: + q = min(t_qty, _round_lot(avail)) + if q >= LOT: + why = "近压力" if near_pre else f"日内涨 {dayup:.1%} 滞涨" + return {"side": "sell", "action": A_T0, "qty": q, "leg": "open", "dir": "short", + "entry": price, "reason": f"反T开仓: {why}, 卖 {q} 股待低买回"} + return None + + +# ================================================================ 评估器: 网格 (设计 §七) +def _grid_levels(params: dict) -> list: + """按 中枢/档距/上下界 生成一组网格价位 (由低到高)。档距支持百分比(step_pct)或绝对值(step)。""" + lo, hi = _f(params.get("lower")), _f(params.get("upper")) + center = _f(params.get("center")) + step = _f(params.get("step")) + step_pct = _f(params.get("step_pct")) + if lo <= 0 or hi <= lo: + return [] + base = center if center > 0 else (lo + hi) / 2 + if step <= 0 and step_pct > 0: + step = base * step_pct + if step <= 0: + return [] + levels, p, guard = [], lo, 0 + while p <= hi + 1e-9 and guard < 200: + levels.append(round(p, 3)) + p += step + guard += 1 + return levels + + +def _eval_grid(st, pos, day, now, ctx): + """网格: 现价跌破未买档 → 买 1 份; 现价涨破已买档 → 卖 1 份(从 avail); 越上界停做; + 跌破下界 = 继续持有、不再买 (设计四点拍板①)。""" + state = ctx["state"] + params = st.get("params") or {} + price = _f(day.get("price")) + if price <= 0: + return None + levels = _grid_levels(params) + if not levels: + return None + lo, hi = levels[0], levels[-1] + step = round(levels[1] - levels[0], 3) if len(levels) > 1 else 0 + per_lot = _round_lot(params.get("per_lot")) or LOT + max_capital = _f(params.get("max_capital")) + filled = {int(k): v for k, v in (state.get("filled_levels") or {}).items()} + invested = _f(state.get("invested")) + avail = int(pos.get("avail_qty") or 0) + + # 跌破下界: 停买入腿、保留已买、告警 (已买档回升仍按规则卖) + if price < lo: + if not state.get("below_floor"): + state["below_floor"] = True + ctx["notes"].append(f"{st['ts_code']} 跌破网格下界 {lo}, 已停止网格买入(继续持有已买档)") + # 下界之下不买, 但仍可卖 (若有已买档且价格回升) —— 落到下面卖出分支 + else: + state["below_floor"] = False + + # 卖出腿: 现价涨破某已买档 (买价 + 一档) → 卖那一份 (从 avail, T+1 由 run_tick 封顶) + sell_idx, sell_buyprice = None, -1.0 + for idx, info in filled.items(): + bp = _f(info.get("price")) + if bp > 0 and price >= bp + step and bp > sell_buyprice: + sell_idx, sell_buyprice = idx, bp + if sell_idx is not None and avail >= LOT: + q = min(per_lot, _round_lot(avail)) + if q >= LOT: + return {"side": "sell", "action": A_SELL, "qty": q, "leg": f"grid_sell:{sell_idx}", + "grid_sell_idx": sell_idx, + "reason": f"网格卖: 现价 {price} 涨破买档 {sell_buyprice}(+一档 {step}), 卖 {q} 股"} + + # 买入腿: 现价跌破某未买档 → 买一份 (受 max_capital 与 单股上限[规则闸] 双约束) + if state.get("below_floor") or price < lo or price > hi: + return None + buy_idx, buy_level = None, -1.0 + for i, lv in enumerate(levels): + if i in filled: + continue + if price <= lv and lv > buy_level: # 现价已跌到/跌破该档 + buy_idx, buy_level = i, lv + if buy_idx is not None: + need = per_lot * price + if max_capital > 0 and invested + need > max_capital + 1e-6: + if not state.get("cap_hit"): + state["cap_hit"] = True + ctx["notes"].append(f"{st['ts_code']} 网格已达最大投入 {max_capital:.0f} 元, 暂停买入") + return None + state["cap_hit"] = False + return {"side": "buy", "action": A_GRID_BUY, "qty": per_lot, "leg": f"grid_buy:{buy_idx}", + "grid_buy_idx": buy_idx, "grid_buy_price": price, + "reason": f"网格买: 现价 {price} 跌破档位 {buy_level}, 买 {per_lot} 股"} + return None + + +# ================================================================ 评估器: 跟踪止盈 (设计 §七B) +def _eval_trail(st, pos, day, now, ctx): + """跟踪止盈: 创新高抬止盈线, 从高点回落 ≥ giveback 就卖 (从 avail); 命中硬止盈目标直接全清。 + 只卖不买 —— 纯离场保护。""" + state = ctx["state"] + params = st.get("params") or {} + price = _f(day.get("price")) + if price <= 0: + return None + avg = _f(pos.get("avg_cost")) + avail = int(pos.get("avail_qty") or 0) + profit = _f(pos.get("cushion_pct")) if pos.get("cushion_pct") is not None else ( + (price / avg - 1) if avg > 0 else 0.0) + + start_line = _f(params.get("start_line")) or param_store.get_float("PMS_CUSHION_SOLID", 0.03) + giveback = _f(params.get("giveback")) + sell_ratio = _f(params.get("sell_ratio")) or 1.0 + hard_target = _f(params.get("hard_target")) + + # 高水位每跳只抬不降 + hw = max(_f(state.get("high_water")), price) + state["high_water"] = round(hw, 3) + if not state.get("armed") and profit >= start_line: + state["armed"] = True + ctx["notes"].append(f"{st['ts_code']} 跟踪止盈已武装(浮盈 {profit:.1%} ≥ 启动线 {start_line:.1%})") + + if avail < LOT: + return None # 无 T+1 可卖, 只更新高水位 + + # 硬止盈目标: 直接全清 + if hard_target > 0 and profit >= hard_target: + return {"side": "sell", "action": A_EXIT, "qty": _round_lot(avail) or avail, "leg": "trail_hard", + "reason": f"跟踪止盈-硬目标: 浮盈 {profit:.1%} ≥ {hard_target:.1%}, 全清 avail {avail} 股"} + + # 已武装且从高点回落到设定比例 → 卖 + if state.get("armed") and hw > 0 and price <= hw * (1 - giveback) and giveback > 0: + q = _round_lot(avail * sell_ratio) if sell_ratio < 1 else (_round_lot(avail) or avail) + if q >= LOT or (sell_ratio >= 1 and q > 0): + act = A_EXIT if sell_ratio >= 1 else A_SELL + return {"side": "sell", "action": act, "qty": q, "leg": "trail_sell", + "reason": f"跟踪止盈: 现价 {price} 自高点 {hw} 回落 {1 - price / hw:.1%} ≥ {giveback:.1%}, 卖 {q} 股"} + return None EVALUATORS = {"T0": _eval_t0, "GRID": _eval_grid, "TRAIL": _eval_trail} +# ================================================================ 供 action_engine 排除 def active_codes() -> set: - """有 ACTIVE 策略的 ts_code —— 供 action_engine 排除。读库失败按空集 (不误排除)。""" + """有 ACTIVE 策略的 ts_code —— 供 action_engine 排除。读库失败按空集 (不误排除全体持仓)。""" try: return pms_repo.active_strategy_codes() - except Exception as e: # noqa: BLE001 —— 读不到不能把全体持仓都从动作引擎排除 + except Exception as e: # noqa: BLE001 logger.warning("[strategy] 读取 ACTIVE 策略集失败 (按空集): %s", e) return set() -def tick(*, now=None, dry_run: bool = False) -> dict: - """盘中每分钟一跳。 +# ================================================================ 每分钟主跳 +def _roll_day(state: dict, pos: dict, today: int) -> dict: + """做T 的按日重置: 当日次数归零, 记下当日 realized_t_profit 基线 (算当日T盈亏用)。 + 网格 filled_levels / 跟踪止盈 high_water 是跨日的, 不在这里动。""" + if int(state.get("day") or 0) != today: + state["day"] = today + state["t_count_today"] = 0 + state["rt_base"] = _f(pos.get("realized_t_profit")) + # 隔夜后原则上不该留未平的腿 (14:50 已平回); 万一留了, 清掉 open_leg 交对账兜底 + state["open_leg"] = None + return state - 第 1 步只做载入与分发: 评估器占位返回 None, 不取行情、不发指令、不落库 —— 全链空跑通。 - 第 2~4 步在下面标注的接入点补: 取持仓行与当日行情 → 评估 → 过 rails → 发短窗口指令或落提议 → 更新 state。 + +def _day_t_pnl(state: dict, pos: dict) -> float: + return _f(pos.get("realized_t_profit")) - _f(state.get("rt_base")) + + +def tick(*, now=None, dry_run: bool = False) -> dict: + """盘中每分钟一跳: 载入 ACTIVE 策略 → 逐只评估 → 触发就发短窗口指令 / 落提议 → 更新 state。 + + dry_run=True 只算不发不落库 (页面「试算」用)。 """ - out = {"enabled": False, "checked": 0, "fired": [], "queued": [], "skipped": [], "errors": []} + now = now or datetime.now() + out = {"enabled": False, "checked": 0, "fired": [], "queued": [], "skipped": [], + "notes": [], "errors": [], "dry_run": dry_run} if not param_store.get_bool("PMS_STRATEGY_ENABLED", False): out["skipped"].append("PMS_STRATEGY_ENABLED=False, 策略层整体停用") return out out["enabled"] = True + try: strategies = pms_repo.active_strategies() except Exception as e: # noqa: BLE001 logger.exception("[strategy] 载入 ACTIVE 策略失败") - return {**out, "errors": [f"载入失败: {type(e).__name__}: {e}"]} + return {**out, "ok": False, "errors": [f"载入失败: {type(e).__name__}: {e}"]} + if not strategies: + out["ok"] = True + return out + + try: + view = portfolio.positions_view() + except Exception as e: # noqa: BLE001 + logger.exception("[strategy] 取持仓快照失败") + return {**out, "ok": False, "errors": [f"取持仓失败: {type(e).__name__}: {e}"]} + scale = _f(view.get("totals", {}).get("scale")) + today = _today() + + # 全局当日T亏熔断: 汇总所有做T策略的当日T盈亏 (realized_t_profit 相对日初基线的增量) + global_t_pnl = 0.0 + for st in strategies: + if st.get("type") == "T0": + global_t_pnl += _day_t_pnl(st.get("state") or {}, _pos_of(view, st["ts_code"])) + global_loss_cap = param_store.get_float("PMS_T0_GLOBAL_DAY_LOSS", 0.01) * scale + global_halt = scale > 0 and global_t_pnl <= -global_loss_cap + if global_halt: + out["notes"].append(f"全局当日T亏 {global_t_pnl:.0f} 元 达熔断线 {global_loss_cap:.0f} 元, 今日不再开新T") for st in strategies: out["checked"] += 1 + code = st.get("ts_code") fn = EVALUATORS.get(st.get("type")) if not fn: out["skipped"].append({"strategy_id": st.get("strategy_id"), "why": f"未知策略类型 {st.get('type')}"}) continue - # —— 第 2~4 步接入点 —— - # pos = portfolio.positions_view() 里该股的行 (或 pms_repo.get_position) - # day = market.day_snapshot(st["ts_code"]) - # decision = fn(st, pos, day, now); 过策略层 rails (熔断/上限/平回/次数/下界) - # auto: _emit_instruction(st, decision) 发短窗口指令走 run_tick - # confirm: _emit_proposal(st, decision) 落一条提议进「等我拍板」 - # 最后 pms_repo.update_strategy(st["strategy_id"], state=<新 state>) try: - decision = fn(st, None, None, now) + pos = _pos_of(view, code) + if not pos or int(pos.get("total_qty") or 0) <= 0: + out["skipped"].append({"strategy_id": st.get("strategy_id"), + "why": f"{code} 已无持仓, 策略空转 (可撤下)"}) + continue + state = dict(st.get("state") or {}) + if st.get("type") == "T0": + state = _roll_day(state, pos, today) + + # 上一腿还没走完就别再发; 走完了先把结果并进 state + if not _reconcile(st, state, pos): + if not dry_run: + pms_repo.update_strategy(st["strategy_id"], state=state) + out["skipped"].append({"strategy_id": st["strategy_id"], "why": "上一腿在途, 等它走完"}) + continue + + day = market.day_snapshot(code) + if not day or not day.get("price"): + out["skipped"].append({"strategy_id": st["strategy_id"], + "why": f"{code} 无实时行情 (停牌/盘前), 顺延"}) + if not dry_run: + pms_repo.update_strategy(st["strategy_id"], state=state) + continue + + # 单票当日T亏熔断 (做T) + halted = global_halt + if st.get("type") == "T0" and scale > 0: + stock_cap = param_store.get_float("PMS_T0_STOCK_DAY_LOSS", 0.003) * scale + if _day_t_pnl(state, pos) <= -stock_cap: + halted = True + out["notes"].append(f"{code} 当日T亏达单票熔断线 {stock_cap:.0f} 元, 今日不再开新T") + + ctx = {"state": state, "scale": scale, "halted": halted, "notes": out["notes"]} + dec = fn(st, pos, day, now, ctx) + + if dec: + # 网格成交态在这里落 (评估器只给意图, 落 state 由这里做, 保证与发指令原子) + if not dry_run: + _apply_grid_state(state, dec) + if (st.get("autonomy") or "auto") == "confirm": + pid = _emit_proposal(st, dec) + state["pending"] = {"pid": pid, "leg": dec["leg"], "dir": dec.get("dir"), + "qty": dec["qty"], "entry": dec.get("entry")} + out["queued"].append({"strategy_id": st["strategy_id"], "proposal_id": pid, + "reason": dec["reason"]}) + else: + iid = _emit_instruction(st, dec, forced=False) + state["pending"] = {"iid": iid, "leg": dec["leg"], "dir": dec.get("dir"), + "qty": dec["qty"], "entry": dec.get("entry")} + out["fired"].append({"strategy_id": st["strategy_id"], "instruction_id": iid, + "side": dec["side"], "qty": dec["qty"], "reason": dec["reason"]}) + else: + out["fired"].append({"strategy_id": st["strategy_id"], "dry_run": True, + "side": dec["side"], "qty": dec["qty"], "reason": dec["reason"]}) + + if not dry_run: + pms_repo.update_strategy(st["strategy_id"], state=state) except Exception as e: # noqa: BLE001 —— 单策略异常不拖垮整轮 logger.exception("[strategy] 评估失败 %s", st.get("strategy_id")) out["errors"].append(f"{st.get('strategy_id')}: {type(e).__name__}: {e}") - continue - if decision is None: - continue - # 骨架阶段评估器不会返回非 None; 真返回了说明有人提前接了规则却没接下发 —— 明着记一条, 不静默下单 - out["skipped"].append({"strategy_id": st.get("strategy_id"), - "why": "评估器已产出决策, 但下发/落提议在第 2~4 步接入 (骨架阶段不下单)"}) out["ok"] = not out["errors"] return out +def _apply_grid_state(state: dict, dec: dict): + """网格发单同时更新 filled_levels / invested (买入占用一档, 卖出释放一档)。""" + if "grid_buy_idx" in dec: + filled = dict(state.get("filled_levels") or {}) + filled[str(dec["grid_buy_idx"])] = {"price": _f(dec.get("grid_buy_price")), + "qty": int(dec["qty"])} + state["filled_levels"] = filled + state["invested"] = _f(state.get("invested")) + _f(dec.get("grid_buy_price")) * int(dec["qty"]) + elif "grid_sell_idx" in dec: + filled = dict(state.get("filled_levels") or {}) + info = filled.pop(str(dec["grid_sell_idx"]), None) + state["filled_levels"] = filled + if info: + state["invested"] = max(0.0, _f(state.get("invested")) - _f(info.get("price")) * int(info.get("qty") or 0)) + + +# ================================================================ 14:50 强制平回 (设计 §六 rails) def force_t0_close(*, now=None) -> dict: - """14:50 做T 强制平回 (设计 §六 rails) —— 第 2 步接入。骨架阶段: 自证无 T 仓残留占位。""" - return {"phase": "骨架", "note": "做T平回在第 2 步接入 (scheduler.t0_close 已留调度位)"} + """做T 强制平回 (scheduler.t0_close 在 PMS_T0_CLOSE_TIME 调): 对每只有未平腿的做T策略, + 立刻发对向平仓腿把当日T仓打平, 绝不过夜。已有在途腿的先撤后不重复 —— 这里只补「还没平」的。""" + out = {"closed": [], "skipped": [], "errors": []} + if not param_store.get_bool("PMS_STRATEGY_ENABLED", False): + out["skipped"].append("PMS_STRATEGY_ENABLED=False") + return out + try: + strategies = [s for s in pms_repo.active_strategies() if s.get("type") == "T0"] + except Exception as e: # noqa: BLE001 + return {**out, "errors": [f"载入做T策略失败: {type(e).__name__}: {e}"]} + if not strategies: + return out + try: + view = portfolio.positions_view() + except Exception as e: # noqa: BLE001 + return {**out, "errors": [f"取持仓失败: {type(e).__name__}: {e}"]} + + for st in strategies: + code = st["ts_code"] + try: + state = dict(st.get("state") or {}) + # 先把在途腿收敛掉 + pos = _pos_of(view, code) + _reconcile(st, state, pos) + open_leg = state.get("open_leg") + if not open_leg: + out["skipped"].append({"strategy_id": st["strategy_id"], "why": "无未平腿"}) + pms_repo.update_strategy(st["strategy_id"], state=state) + continue + if state.get("pending"): + out["skipped"].append({"strategy_id": st["strategy_id"], "why": "平仓腿已在途"}) + continue + q = _round_lot(open_leg.get("qty")) + if q < LOT: + out["skipped"].append({"strategy_id": st["strategy_id"], "why": "残量不足一手"}) + continue + if open_leg.get("dir") == "long": # 正T 已买 → 卖平 + dec = {"side": "sell", "action": A_T0, "qty": min(q, int(pos.get("avail_qty") or 0)), + "leg": "close", "reason": "14:50 强制平回(正T)"} + else: # 反T 已卖 → 买平 + dec = {"side": "buy", "action": A_T0, "qty": q, "leg": "close", + "reason": "14:50 强制平回(反T)"} + if int(dec["qty"]) < LOT: + out["skipped"].append({"strategy_id": st["strategy_id"], "why": "可平量不足 (avail 不够)"}) + continue + iid = _emit_instruction(st, dec, forced=True) + state["pending"] = {"iid": iid, "leg": "close", "dir": open_leg.get("dir"), + "qty": dec["qty"], "entry": open_leg.get("entry")} + pms_repo.update_strategy(st["strategy_id"], state=state) + out["closed"].append({"strategy_id": st["strategy_id"], "instruction_id": iid, + "side": dec["side"], "qty": dec["qty"]}) + except Exception as e: # noqa: BLE001 + logger.exception("[strategy] 强制平回失败 %s", st.get("strategy_id")) + out["errors"].append(f"{st.get('strategy_id')}: {type(e).__name__}: {e}") + out["ok"] = not out["errors"] + return out diff --git a/app/services/strategy_service.py b/app/services/strategy_service.py new file mode 100644 index 0000000..ccfff28 --- /dev/null +++ b/app/services/strategy_service.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +""" +个股交易方案 (策略) 管理侧: 校验 / 挂载 / 暂停撤下 —— PER_STOCK_STRATEGY_PLAN.md §五/§十 +运行侧 (每分钟评估、下单) 见 strategy_runner.py。 + +挂载前**先过约束校验** (validate): 违反仓位 / 存量 / 上限就挡下、给明确中文原因、**不写库** +—— 落实「理论上不允许违反, 无法操作要在页面提示」。真正下单时的合规由 rule_gate 再兜一道 (双保险)。 +校验不写库、不抛异常; 读持仓失败按「暂不能校验」挡下 (不放行未校验的挂载)。 +""" +from __future__ import annotations + +import logging +from datetime import datetime + +from app.core import tradedays as td +from app.repo import pms_repo +from app.services import param_store, portfolio + +logger = logging.getLogger("pms.strategy_svc") + +TYPES = {"T0", "GRID", "TRAIL"} +AUTONOMY = {"auto", "confirm"} +STATUSES = {"ACTIVE", "PAUSED", "CANCELLED", "DONE"} + + +def _f(v, d=0.0): + try: + return float(v) + except (TypeError, ValueError): + return d + + +def _pos(view, code): + for x in view["positions"]: + if x["ts_code"] == code: + return x + return None + + +def validate(cfg: dict, view=None) -> dict: + """校验一条策略配置。返回 {ok, reasons:[中文原因]}。不写库、不抛异常。""" + cfg = cfg or {} + reasons = [] + code = cfg.get("ts_code") + typ = cfg.get("type") + autonomy = cfg.get("autonomy") or "auto" + params = cfg.get("params") or {} + if not code: + reasons.append("未指定股票") + if typ not in TYPES: + reasons.append(f"未知策略类型 {typ}(仅支持 做T=T0 / 网格=GRID / 跟踪止盈=TRAIL)") + if autonomy not in AUTONOMY: + reasons.append(f"自主档 {autonomy} 非法(仅 auto / confirm)") + if reasons: + return {"ok": False, "reasons": reasons} + + try: + view = view or portfolio.positions_view() + except Exception as e: # noqa: BLE001 —— 读不到持仓不能放行未校验的挂载 + return {"ok": False, "reasons": [f"读持仓失败, 暂不能校验(不放行): {type(e).__name__}: {e}"]} + pos = _pos(view, code) + if not pos or int(pos.get("total_qty") or 0) <= 0: + return {"ok": False, "reasons": [ + "该股当前没有持仓 —— 交易方案要在已有底仓上做差价(A股 T+1, 当日买入不可当日卖)"]} + + # 同股已有 ACTIVE 策略 → 不重复挂 + try: + dup = pms_repo.list_strategies(ts_code=code, statuses=["ACTIVE"], limit=5) + except Exception: # noqa: BLE001 + dup = [] + if dup: + reasons.append(f"该股已挂着一条 {dup[0].get('type')} 策略(生效中), 先撤下再挂新的") + + prm = view["params"] + scale = _f(prm.get("scale")) + stock_cap = _f(prm.get("stock_cap")) + mv = _f(pos.get("market_value")) + avail = int(pos.get("avail_qty") or 0) + cap_room = max(0.0, stock_cap * scale - mv) # 这只票离单股上限还差多少钱 + + if typ == "T0": + tr = _f(params.get("t_ratio")) + if not (0 < tr <= 0.3334): + reasons.append("做T 的 T 仓比例 t_ratio 必须在 0~1/3 之间(硬上限 1/3)") + if avail <= 0: + reasons.append("做T 需要 T+1 可卖的存量股(当前可卖为 0), 无法在存量上做差价") + elif typ == "GRID": + lo, hi = _f(params.get("lower")), _f(params.get("upper")) + mid = _f(params.get("center")) or _f(pos.get("price")) + step = _f(params.get("step")) + max_cap = _f(params.get("max_capital")) + if not (0 < lo < mid < hi): + reasons.append("网格上下界不成立: 需 0 < 下界 < 中枢 < 上界") + if step <= 0 and _f(params.get("step_pct")) <= 0: + reasons.append("网格档距必须 > 0 (绝对档距 step 或百分比档距 step_pct 至少给一个)") + if max_cap <= 0: + reasons.append("网格最大投入额必须 > 0") + elif max_cap > cap_room + 1e-6: + reasons.append( + f"网格最大投入 {max_cap:,.0f} 元超过该股离单股上限的余量 {cap_room:,.0f} 元" + f"(单股上限 {stock_cap:.0%}×规模 {scale:,.0f}, 已占市值 {mv:,.0f})") + elif typ == "TRAIL": + gb = _f(params.get("giveback")) + if not (0 < gb < 1): + reasons.append("跟踪止盈的回撤比例 giveback 必须在 0~1 之间(如 0.05=从高点回落 5% 就卖)") + if avail <= 0: + reasons.append("跟踪止盈触发时要卖出, 但当前 T+1 可卖为 0") + + return {"ok": not reasons, "reasons": reasons} + + +def attach(cfg: dict, by: str = "user") -> dict: + """挂载一条策略。先校验, 违反返回 {ok:false, errors}; 通过则写 pms_strategy(ACTIVE)。""" + v = validate(cfg) + if not v["ok"]: + return {"ok": False, "errors": v["reasons"]} + code = cfg["ts_code"] + ymd = td.ymd() + sid = f"STR_{ymd}_{code.replace('.', '')}_{int(datetime.now().timestamp()) % 1000000}" + pms_repo.insert_strategy(strategy_id=sid, ts_code=code, stype=cfg["type"], + autonomy=cfg.get("autonomy") or "auto", + params=cfg.get("params") or {}, state={}, + status="ACTIVE", note=cfg.get("note")) + return {"ok": True, "strategy_id": sid, "ts_code": code} + + +def set_status(strategy_id: str, status: str, by: str = "user") -> dict: + """暂停(PAUSED) / 恢复(ACTIVE) / 撤下(CANCELLED)。""" + status = (status or "").upper() + if status not in STATUSES: + return {"ok": False, "error": f"非法状态 {status}"} + st = pms_repo.get_strategy(strategy_id) + if not st: + return {"ok": False, "error": "策略不存在"} + pms_repo.update_strategy(strategy_id, status=status) + return {"ok": True, "strategy_id": strategy_id, "ts_code": st.get("ts_code"), "status": status} diff --git a/app/web/main.py b/app/web/main.py index 9fce73b..cb459e5 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -60,6 +60,31 @@ def ok(fn, *args, **kw): return {"ok": False, "error": f"{type(e).__name__}: {e}"} +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(): @@ -135,7 +160,12 @@ def api_set_params(payload: dict = Body(...)): 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} + _okf = all(r.get("ok") for r in results) + _oplog("set_params", ok_flag=_okf, + params={it.get("key"): it.get("value") for it in items}, + reason=(None if _okf else "; ".join((r.get("error") or "") for r in results if not r.get("ok"))), + by=payload.get("by") or "user") + return {"ok": _okf, "results": results} # ================================================================ ② 命令台 @@ -169,14 +199,18 @@ def api_command_detail(command_id: str): @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"))) + 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=payload.get("by") or "user", + force_conflict=bool(payload.get("force")), + ts_code=(payload.get("params") or {}).get("ts_code"), + params=payload, by=payload.get("by") or "user") @app.post("/api/commands/{command_id}/cancel") def api_cancel(command_id: str): - return ok(command_service.cancel, command_id) + return ok_logged("cancel_command", command_service.cancel, command_id, + params={"command_id": command_id}) @app.post("/api/commands/{command_id}/replan") @@ -190,7 +224,7 @@ def api_replan(command_id: str): return {"ok": False, "error": f"命令处于 {c['status']}, 不可重规划"} pms_repo.cancel_plans_of_command(command_id) return command_service.plan_command(c) - return ok(_replan) + return ok_logged("replan_command", _replan, params={"command_id": command_id}) @app.get("/api/plans") @@ -257,6 +291,14 @@ def api_decide(proposal_id: str, payload: dict = Body(default={})): ref_id=proposal_id, reason=payload.get("reason") or "页面人工裁决") 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" pms_repo.insert_instruction( @@ -275,7 +317,9 @@ def api_decide(proposal_id: str, payload: dict = Body(default={})): "warning": f"一次性守卫计数器未写入 ({g.get('error')}) —— " f"{p['ts_code']} 的 {p['action']}「只做一次」本轮失效"} return {"ok": True, "decision": decision, "instruction_id": instruction_id} - return ok(_decide) + return ok_logged("decide_proposal", _decide, + params={"proposal_id": proposal_id, "decision": decision}, + by=payload.get("by") or "user") @app.post("/api/proposals") @@ -369,8 +413,9 @@ def api_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 "页面人工撤销") + 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") @@ -539,3 +584,47 @@ def api_industry_import(payload: dict = Body(...)): 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)): + statuses = [s for s in (status or "").split(",") if s] or None + return ok(lambda: {"ok": True, + "strategies": pms_repo.list_strategies(statuses=statuses, limit=300), + "enabled": param_store.get_bool("PMS_STRATEGY_ENABLED", False)}) + + +@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(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=payload.get("by") or "user") + + +@app.post("/api/strategies/{strategy_id}/status") +def api_strategy_status(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=payload.get("by") or "user") + + +@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)}) diff --git a/app/web/static/index.html b/app/web/static/index.html index 83ab0d1..18c5b3f 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -201,6 +201,7 @@ 设目标价 {{ (s.row.frozen_reason && s.row.frozen_reason!=='NONE') ? '解冻' : '冻结(禁增持)' }} 加入黑名单 + 挂交易方案（做T/网格/跟踪止盈） @@ -228,6 +229,130 @@ + +
+

我的策略 · {{ strategies.length }} 个 + + {{ stratEnabled ? '策略层已启用' : '策略层总开关未开（PMS_STRATEGY_ENABLED）' }} +

+ +
还没有挂任何交易方案。到「我的持仓」某行「更多 ▾ → 挂交易方案」给它挂一个。
+ + + + + + + + + +

操作日志 · 最近 {{ opLog.length }} 条

+
每一次页面写操作都留一行；被仓位 / 盈亏 / T+1 可卖 / 单股上限挡下的，连原因一起记，不静默。
+
还没有操作记录。
+
+ {{ (o.at||'').slice(11,16) }} + {{ o.result }} + {{ o.op }} · {{ nm(o.ts_code) }} + — {{ o.reason }} +
+
+ + + +
+ 现价 {{ stratDlg.price }} · 持仓 {{ stratDlg.total_qty }} 股（T+1 可卖 {{ stratDlg.avail_qty }}）· 浮盈 {{ pct(stratDlg.cushion_pct) }}。参数按这些现值给了默认，不用盲填。 +
+ + + + 做T + 网格 + 跟踪止盈 + + + + + + + + + + + + 自动执行 + 待我拍板 + + + + + +
· {{ r }}
+ + + + +

组合操作

@@ -1008,6 +1133,11 @@ createApp({ const catalog = ref([]), commands = ref([]), plans = ref([]), plansOf = ref(''); const positions = ref([]), lots = ref([]), lotsOf = ref(''); const instructions = ref([]), ledger = ref([]), proposals = ref([]); + const strategies = ref([]), stratEnabled = ref(false), opLog = ref([]); + const stratDlg = reactive({ visible:false, busy:false, reasons:[], ts_code:'', price:0, + total_qty:0, avail_qty:0, cushion_pct:null, type:'T0', autonomy:'auto', + p:{ t_ratio:0.2, lower:0, upper:0, center:0, step_pct:0.02, per_lot:100, max_capital:0, + start_line:0.03, giveback:0.05, sell_ratio:1.0, hard_target:0 } }); const report = ref({}), reportDrawer = ref(false), dm = ref({}); const opsDrawer = ref(false), opsResult = ref(''), opsLoading = ref(false); // 上游选股计划: 单独取数, 不塞进 /api/overview —— overview 是轮询的, 上游不通时 @@ -1129,6 +1259,8 @@ createApp({ CONFIRMED:'已成交', REJECTED:'被拒', EXPIRED:'到期作废', CANCELLED:'已撤销' }, arbiter: { rule:'规则闸', judge:'研判闸', user:'你' }, verdict: { PASS:'放行', REJECT:'驳回', NOTE:'记录', UNAVAILABLE:'研判暂不可用' }, + stype: { T0:'做T', GRID:'网格', TRAIL:'跟踪止盈' }, + stratStatus: { ACTIVE:'运行中', PAUSED:'已暂停', CANCELLED:'已撤下', DONE:'已完成' }, }; const tx = (g, c) => { const m = T[g]; @@ -1247,6 +1379,7 @@ createApp({ else if (c === 'target') actTargetPrice(row); else if (c === 'freeze') actFreeze(row); else if (c === 'black') actBlacklist(row); + else if (c === 'strategy') openStrategy(row); } // 组合 function pHaltBuy() { quickCmd('HALT_BUY', {}, '暂停一切新增买入并撤销在途买入指令？卖出与止损不受影响。'); } @@ -1411,7 +1544,7 @@ createApp({ loading.value = true; err.value = ''; await Promise.all([loadOverview(), loadParams(), loadCatalog(), loadCommands(), loadPositions(), loadInstructions(), loadLedger(), loadProposals(), - loadDispatchMode(), loadWs()]); + loadDispatchMode(), loadWs(), loadStrategies(), loadOpLog()]); await loadNames(); loading.value = false; wsPollSync(); @@ -1494,6 +1627,74 @@ createApp({ else ElementPlus.ElMessage.error(d.error); await Promise.all([loadProposals(), loadInstructions(), loadLedger()]); } + async function loadStrategies() { + const d = await call('get', '/api/strategies'); + strategies.value = d.strategies || (d.data && d.data.strategies) || []; + stratEnabled.value = !!(d.enabled != null ? d.enabled : (d.data && d.data.enabled)); + } + async function loadOpLog() { + const d = await call('get', '/api/op-log?limit=80'); + opLog.value = d.rows || (d.data && d.data.rows) || []; + } + function openStrategy(row) { + const px = Number(row.price) || 0; + Object.assign(stratDlg, { + visible: true, busy: false, reasons: [], ts_code: row.ts_code, + price: px, total_qty: row.total_qty, avail_qty: row.avail_qty, + cushion_pct: row.cushion_pct, type: 'T0', autonomy: 'auto', + p: { t_ratio: 0.2, lower: +(px * 0.92).toFixed(2), upper: +(px * 1.08).toFixed(2), + center: px, step_pct: 0.02, per_lot: 100, max_capital: 0, + start_line: 0.03, giveback: 0.05, sell_ratio: 1.0, hard_target: 0 } + }); + } + function stratPayload() { + const t = stratDlg.type, p = stratDlg.p; + const out = { ts_code: stratDlg.ts_code, type: t, autonomy: stratDlg.autonomy, params: {} }; + if (t === 'T0') out.params = { t_ratio: +p.t_ratio }; + else if (t === 'GRID') out.params = { lower:+p.lower, upper:+p.upper, center:+p.center, + step_pct:+p.step_pct, per_lot:+p.per_lot, max_capital:+p.max_capital }; + else out.params = { start_line:+p.start_line, giveback:+p.giveback, + sell_ratio:+p.sell_ratio, hard_target:+p.hard_target }; + return out; + } + async function validateStrategy() { + stratDlg.busy = true; + const d = await call('post', '/api/strategies/validate', stratPayload()); + stratDlg.reasons = d.ok ? [] : (d.reasons || [d.error].filter(Boolean)); + stratDlg.busy = false; + if (d.ok) ElementPlus.ElMessage.success('校验通过，可挂载'); + } + async function attachStrategy() { + stratDlg.busy = true; + const d = await call('post', '/api/strategies', stratPayload()); + stratDlg.busy = false; + if (d.ok) { ElementPlus.ElMessage.success('已挂载 ' + (d.strategy_id || '')); + stratDlg.visible = false; await Promise.all([loadStrategies(), loadOpLog()]); } + else { stratDlg.reasons = d.errors || d.reasons || [d.error].filter(Boolean); + ElementPlus.ElMessage.error('挂载被挡下，见下方原因'); } + } + async function setStrategyStatus(sid, status) { + const label = { PAUSED:'暂停', ACTIVE:'恢复', CANCELLED:'撤下' }[status] || status; + try { await ElementPlus.ElMessageBox.confirm(label + '该策略？', '确认', { type:'warning' }); } + catch (e) { return; } + const d = await call('post', '/api/strategies/' + sid + '/status', { status }); + ElementPlus.ElMessage[(d.ok ? 'success' : 'error')](d.ok ? (label + '成功') : (d.error || '失败')); + await Promise.all([loadStrategies(), loadOpLog()]); + } + const stratQty = computed(() => { + const p = stratDlg.p, total = stratDlg.total_qty || 0; + if (stratDlg.type === 'T0') return Math.floor(total * (+p.t_ratio || 0) / 100) * 100; + return 0; + }); + function stratStateText(s) { + const st = s.state || {}; + if (s.type === 'T0') return '今日 ' + (st.t_count_today || 0) + '/3 次' + + (st.open_leg ? (' · 有未平腿（' + (st.open_leg.dir === 'long' ? '正T待卖' : '反T待买') + '）') : ' · 无未平腿'); + if (s.type === 'GRID') return '已买 ' + Object.keys(st.filled_levels || {}).length + ' 档 · 投入 ' + + money(st.invested || 0) + (st.below_floor ? ' · 跌破下界已停买' : ''); + if (s.type === 'TRAIL') return '高点 ' + (st.high_water || '—') + ' · ' + (st.armed ? '已武装' : '未武装'); + return ''; + } async function ops(name) { opsLoading.value = true; const d = await call('post', '/api/ops/' + name); @@ -1548,7 +1749,9 @@ createApp({ propWhy, nextStep, scrollTo, actExitStock, actReduceStock, rowMore, pHaltBuy, pResumeBuy, pHaltAll, pResumeAll, pReduce, pIncrease, pLiquidate, pSectorExit, pctOf, tgtPos, posMoveValid, posMovePreview, doPosMove, heldSectors, secSel, doSectorExit, - pmap, pval, dcaOn, sumPosition, sumDca, sumAutonomy }; + pmap, pval, dcaOn, sumPosition, sumDca, sumAutonomy, + strategies, stratEnabled, opLog, stratDlg, stratQty, loadStrategies, loadOpLog, + openStrategy, validateStrategy, attachStrategy, setStrategyStatus, stratStateText }; } }).use(ElementPlus).mount('#app'); diff --git a/ddl_pms_v1.sql b/ddl_pms_v1.sql index 1fee208..52bfa85 100644 --- a/ddl_pms_v1.sql +++ b/ddl_pms_v1.sql @@ -355,3 +355,21 @@ CREATE TABLE IF NOT EXISTS pms_strategy ( KEY idx_code_status (ts_code, status), KEY idx_status (status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='个股交易方案 (策略)'; + + +-- 17. 交易员操作日志 —— 每个页面写操作落一行 (OK/BLOCKED/ERROR + 原因)。见 PER_STOCK_STRATEGY_PLAN.md §九 +-- 「理论上不允许违反, 无法操作要在页面提示」的留痕面: 被约束挡下的操作连同原因一起记, 不静默。 +CREATE TABLE IF NOT EXISTS pms_op_log ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + at DATETIME NOT NULL, + op VARCHAR(40) NOT NULL COMMENT '操作: issue_command / decide_proposal / attach_strategy / set_params ...', + by_user VARCHAR(32) NOT NULL DEFAULT 'user', + ts_code VARCHAR(16) NULL, + result VARCHAR(8) NOT NULL COMMENT 'OK / BLOCKED / ERROR', + reason VARCHAR(500) NULL COMMENT '被约束挡下或出错的原因', + params_json TEXT NULL, + ref VARCHAR(64) NULL COMMENT '关联 command/proposal/strategy/instruction id', + KEY idx_at (at), + KEY idx_op (op), + KEY idx_code (ts_code) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='交易员操作日志'; diff --git a/scripts/test_batch6_units.py b/scripts/test_batch6_units.py index ec033fa..f42af20 100644 --- a/scripts/test_batch6_units.py +++ b/scripts/test_batch6_units.py @@ -557,7 +557,7 @@ def run(): "DEFAULT 'NONE' COMMENT 'NONE/REQUESTED/SENT'"): eq(find_adjacent_literals(good), [], f"误报: {good[:40]}") - @case("DDL 文件本身体检通过 (16 张表 + 1 条初始行)") + @case("DDL 文件本身体检通过 (17 张表 + 1 条初始行)") def _(): sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from init_db import DDL_FILE, find_adjacent_literals, parse_statements @@ -569,7 +569,7 @@ def run(): eq(find_adjacent_literals(s), [], f"{tbl} 有相邻字面量") # 加表就要来这里 +1 —— 这一行是 DDL 与代码之间唯一的哨兵, 它不动就说明新表没进 # ddl_pms_v1.sql (init_db 只认这个文件, 建不出来的表在实机上才会报"表不存在") - eq(len([1 for k, _, _ in stmts if k == "table"]), 16) + eq(len([1 for k, _, _ in stmts if k == "table"]), 17) eq(len([1 for k, _, _ in stmts if k == "seed"]), 1) @case("通道三表的 SQL 全部单表合规")