diff --git a/app/services/signal_service.py b/app/services/signal_service.py index b9d7a67..911a23d 100644 --- a/app/services/signal_service.py +++ b/app/services/signal_service.py @@ -109,6 +109,10 @@ def consume(*, batch: int = 50, dry_run: bool = False) -> dict: "trim_ratio": param_store.get_float("PMS_SIGNAL_TRIM_RATIO", 1 / 3)} seen = _load_seen() ymd = td.ymd() + try: + strat_codes = pms_repo.active_strategy_codes() + except Exception: + strat_codes = set() for db, key, parser in streams(): try: @@ -120,7 +124,7 @@ def consume(*, batch: int = 50, dry_run: bool = False) -> dict: for msg_id, fields in msgs: try: sig = parser(fields, msg_id=msg_id) - _handle(sig, view, prm, seen, ymd, dry_run, out) + _handle(sig, view, prm, seen, ymd, dry_run, out, strat_codes) if not dry_run: _ack(db, key, msg_id) except Exception as e: @@ -133,7 +137,7 @@ def consume(*, batch: int = 50, dry_run: bool = False) -> dict: return out -def _handle(sig, view, prm, seen, ymd, dry_run, out): +def _handle(sig, view, prm, seen, ymd, dry_run, out, strat_codes=frozenset()): code = sig.get("ts_code") pos = _pos_of(view, code) if code else None d = sr.digest(sig, pos, prm) @@ -199,14 +203,31 @@ def _handle(sig, view, prm, seen, ymd, dry_run, out): # 但这一天的去重键已经烧掉了 —— 同一条风控卖出信号后面再来多少次都被当成重复丢弃, # 指令一条都不会落。失败长得像成功: 页面只多一行 error, 而该卖的票就那么留着了。 # 2026-07-31 修。 - if act == sr.ACT_EXIT: + on_strategy = bool(code) and code in strat_codes + if act == sr.ACT_EXIT and not on_strategy: iid = _make_exit(code, d, pos) seen.add(key) out["exits"].append({**brief, "instruction_id": iid}) else: - pid = _make_proposal(code, d, pos, sig) + # 挂了策略的票: 决策系统的风控卖出只提示、不自动清仓 (强制离场会推翻你特意设的策略); + # 一律落提议进「等我拍板」由你定"维持 / 采纳即撤策略并清仓", 采纳的是全清 (as_exit)。 + pid = _make_proposal(code, d, pos, sig, on_strategy=on_strategy, + as_exit=(on_strategy and act == sr.ACT_EXIT)) seen.add(key) - out["proposals"].append({**brief, "proposal_id": pid}) + out["proposals"].append({**brief, "proposal_id": pid, "on_strategy": on_strategy}) + if on_strategy and not dry_run: + # 同时暂停该票策略的买入这一侧 (不平仓、不动卖出、页面可恢复) —— 挡住"资金在流出、 + # 网格还在逢跌买入", 又不替你做清仓这种不可逆的事。 + try: + from app.services import strategy_service + affected = strategy_service.pause_buy(code, reason=d["reason"], + source=(sig.get("source") or "signal")) + if affected: + out.setdefault("strategy_buy_paused", []).extend(affected) + logger.warning("[信号消化] %s 挂着策略, 决策系统卖出只提示不自动清仓; " + "已暂停该票策略买入(页面可恢复): %s", code, affected) + except Exception as e: + logger.error("[信号消化] 暂停策略买入失败 %s: %s", code, e) def _make_exit(code, d, pos) -> str: @@ -229,15 +250,21 @@ def _make_exit(code, d, pos) -> str: return iid -def _make_proposal(code, d, pos, sig) -> str: +def _make_proposal(code, d, pos, sig, *, on_strategy=False, as_exit=False) -> str: ttl = param_store.get_int("PMS_PROPOSAL_TTL_HOURS", 24) pid = f"PRP_{td.ymd()}_{code.replace('.', '')}_SIGSELL" + action = "EXIT" if as_exit else "TRIM" + reason = d["reason"] + if on_strategy: + reason = ("这只票挂着交易方案(策略): 决策系统卖出信号只提示、未自动清仓; 该票策略买入已暂停。" + "你定: 维持观察 / 采纳即撤策略并清仓。 —— " + reason) hn = {**d["hard_numbers"], "price": float((pos or {}).get("price") or 0), - "reason": d["reason"], "signal_source": sig.get("source")} - pms_repo.insert_proposal(proposal_id=pid, ts_code=code, action="TRIM", qty=d["qty"], + "reason": reason, "signal_source": sig.get("source"), "on_strategy": on_strategy} + pms_repo.insert_proposal(proposal_id=pid, ts_code=code, action=action, qty=d["qty"], hard_numbers=hn, expire_at=datetime.now() + timedelta(hours=ttl), - judge_verdict=None, judge_reason=d["reason"][:500]) + judge_verdict=("STRATEGY_RISK" if on_strategy else None), + judge_reason=reason[:500]) return pid diff --git a/app/services/strategy_runner.py b/app/services/strategy_runner.py index e69c48e..8237b91 100644 --- a/app/services/strategy_runner.py +++ b/app/services/strategy_runner.py @@ -13,14 +13,14 @@ action_engine.scan 早已加好的「有 ACTIVE 策略的票跳过」那一条 s 一条铁律定了整个框架 —— T+1 (设计 §三) -------------------------------------------------- 当日买入不可当日卖出, 所以做T / 网格 / 跟踪止盈本质都是「在一只**底仓**上、用 T+1 可卖的存量股 -做买卖」= 同一自动机的三种配置。卖出腿一律经 run_tick 按 avail_qty 封顶, T+1 天然被挡在那里。 +做买卖」= 同一自动机的三种配置。卖出一律经 run_tick 按 avail_qty 封顶, T+1 天然被挡在那里。 批次口径 (与 app/core/recon.ACTION_TO_LOT 对齐, 不另立) -------------------------------------------------- - 做T —— 买卖两腿都用 action='T0_ROUND' → 记 T0 批次; 卖出核销次序 T0→ADD→DCA→FILL→BASE + 做T —— 买入与卖出都用 action='T0_ROUND' → 记 T0 批次; 卖出核销次序 T0→ADD→DCA→FILL→BASE 先把 T0 批次对冲掉, 底仓与摊薄成本不动, 做T利润自然摊入 realized_t_profit。 - 网格 —— 买腿 action='ADD' (记 ADD 批次), 卖腿 action='TRIM' (从 avail 卖)。 - 跟踪止盈 —— 卖腿 action='TRIM' (部分) / 'EXIT' (全清)。只卖不买。 + 网格 —— 买入 action='ADD' (记 ADD 批次), 卖出 action='TRIM' (从 avail 卖)。 + 跟踪止盈 —— 卖出 action='TRIM' (部分) / 'EXIT' (全清)。只卖不买。 安全 (设计 §九) -------------------------------------------------- @@ -28,9 +28,19 @@ action_engine.scan 早已加好的「有 ACTIVE 策略的票跳过」那一条 s * 挂了 ACTIVE 策略的票由 action_engine.scan 排除 (两个大脑不抢同一只)。 * 策略动作走命令口径 (is_command=True, 过规则闸、不过研判闸)。 * scheduler 的 @guard(session=True) 兜住: 非交易时段 / 休假模式不跑。 - * 影子/实盘由 PMS_DISPATCH_MODE 决定 (shadow=只落影子出口不碰真 QMT), 与本层无关、自动继承。 + * 影子/实盘由 PMS_DISPATCH_MODE 决定, 与本层无关、自动继承。 * **双保险**: 本层 rails 先拦 (熔断 / 当日次数 / 上限 / 平回 / 下界), 规则闸再拦一道 (合规)。 * **关键路径禁止丢弃返回值**: 发不出指令 / 落不了库一律进 out["errors"], 绝不静默当成功。 + +决策系统对策略票的边界 (2026-08-11 用户定) +-------------------------------------------------- +挂了策略的票, 决策系统的风控卖出信号只提示、不自动清仓 (强制离场会推翻你特意设的策略); +`signal_service` 会把它落成提议、并把该票策略的**买入这一侧暂停** (不平仓、不动卖出、可页面恢复)。 +暂停标记存在运行参数 `PMS_STRATEGY_BUYPAUSE` (按 ts_code 映射), 由 strategy_service 维护、 +本模块只读 —— 特意不放进策略 state_json: 那份 state 每跳都被本模块重写, 放进去会被并发的 +signal_digest / intraday_exec 互相覆盖 (与 portfolio.neg_streak_map 用独立参数同一个道理)。 +暂停只挡「开新仓 / 加仓」这一侧, 卖出、平回、跟踪止盈照常 —— 挡的是"资金在流出、网格还在 +逢跌买入", 不替你做清仓这种不可逆的事。 """ from __future__ import annotations @@ -44,12 +54,12 @@ from app.services import market, param_store, portfolio logger = logging.getLogger("pms.strategy") -# 指令在途 (未终态) 的状态集 —— 与 executor.LIVE 一致; 本层据此判断「上一腿还没走完, 先别再发」。 +# 指令在途 (未终态) 的状态集 —— 与 executor.LIVE 一致; 本层据此判断「上一笔还没走完, 先别再发」。 LIVE_INS = ("PROPOSED", "RULE_PASSED", "DISPATCHED") # 批次动作 (与 recon.ACTION_TO_LOT 对齐) -A_T0 = "T0_ROUND" # 做T 买卖两腿 → T0 批次 -A_GRID_BUY = "ADD" # 网格买腿 → ADD 批次 +A_T0 = "T0_ROUND" # 做T 买入与卖出 → T0 批次 +A_GRID_BUY = "ADD" # 网格买入 → ADD 批次 A_SELL = "TRIM" # 网格 / 跟踪止盈 部分卖 A_EXIT = "EXIT" # 跟踪止盈 全清 (允许零股一次性清出) @@ -83,12 +93,22 @@ def _today() -> int: return td.ymd() +def _buypause_codes() -> set: + """当前被暂停买入的 ts_code 集合 (决策系统对策略票的风控预警触发)。读失败按空集。""" + try: + from app.services import strategy_service + return set(strategy_service.buypause_map().keys()) + except Exception as e: # noqa: BLE001 + logger.warning("[strategy] 读取买入暂停集失败 (按空集): %s", e) + return set() + + # ================================================================ 指令 / 提议下发 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)。 + limit_price 不在这里定 —— 交由 run_tick 的择时按实时行情现算 (与 materialize_plans + 同口径: 那里也是 limit_price=None)。 """ code = st["ts_code"] now = datetime.now() @@ -104,8 +124,8 @@ def _emit_instruction(st: dict, dec: dict, *, forced: bool = False) -> str: def _emit_proposal(st: dict, dec: dict) -> str: - """autonomy=confirm: 落一条提议进「等我拍板」。hard_numbers 里带全腿谱, 人采纳后由 - main._decide 识别 kind=strategy 再回调本层发指令 (不走通用物化, 因为 T0 两腿同 action、 + """autonomy=confirm: 落一条提议进「等我拍板」。hard_numbers 里带完整动作参数, 人采纳后由 + main._decide 识别 kind=strategy 再回调本层发指令 (不走通用物化, 因为 T0 买卖同 action、 side 无法由 action 反推)。""" code = st["ts_code"] now = datetime.now() @@ -122,7 +142,7 @@ def _emit_proposal(st: dict, dec: dict) -> str: def emit_from_spec(spec: dict) -> dict: - """confirm 提议被采纳后的回调 (main._decide 调): 按 hard_numbers 里存的腿谱发指令。 + """confirm 提议被采纳后的回调 (main._decide 调): 按 hard_numbers 里存的动作参数发指令。 返回 {ok, instruction_id} 或 {ok:false, error}。""" try: st = pms_repo.get_strategy(spec.get("strategy_id")) @@ -144,9 +164,9 @@ def emit_from_spec(spec: dict) -> dict: return {"ok": False, "error": f"{type(e).__name__}: {e}"} -# ================================================================ 在途腿的收敛 +# ================================================================ 在途委托的收敛 def _pending_terminal(pending: dict): - """上一腿是否已终态。返回 (已终态?, 指令行 or None)。无 pending 视为已终态。""" + """上一笔委托是否已终态。返回 (已终态?, 指令行 or None)。无 pending 视为已终态。""" if not pending: return True, None iid = pending.get("iid") @@ -165,14 +185,14 @@ def _pending_terminal(pending: dict): def _reconcile(st: dict, state: dict, pos: dict): - """把已终态的上一腿并进 state: 开仓腿成交 → 记 open_leg; 平仓腿成交 → 记一次完成、清 open_leg。 + """把已终态的上一笔并进 state: 开仓成交 → 记 open_leg; 平回成交 → 记一次完成、清 open_leg。 做T 用; 网格 / 跟踪止盈 的 filled_levels / high_water 在各自评估器里按现价推进, 不依赖这里。 """ pending = state.get("pending") done, ins = _pending_terminal(pending) if not done: - return False # 上一腿未走完 + return False # 上一笔未走完 state["pending"] = None if not ins or not pending or not pending.get("iid"): return True @@ -184,7 +204,7 @@ def _reconcile(st: dict, state: dict, pos: dict): "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**, + # 平回终态: 只有真成交才算一轮完成; 一股没成 (到期/被拒) 要**保留 open_leg**, # 让下一跳与 14:50 平回继续补平 —— 绝不能把一条还开着的 T 仓静默丢掉。 filled_qty = int(ins.get("exec_qty") or 0) ol = dict(state.get("open_leg") or {}) @@ -193,7 +213,7 @@ def _reconcile(st: dict, state: dict, pos: dict): ol["qty"] = rem state["open_leg"] = ol if filled_qty > 0: - logger.info("[strategy] %s 平仓腿部分成交 %s, 余 %s 股待续平", + 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 @@ -205,8 +225,8 @@ def _reconcile(st: dict, state: dict, pos: dict): def _eval_t0(st, pos, day, now, ctx): """做T: 正T (回落近支撑 → 买, 目标价差高卖) / 反T (近压力或滞涨 → 卖, 低买回)。 - rails: 当日 ≤ 3 次; 单票 / 全局当日T亏熔断后当日禁开新T (仍允许平回已开的腿); - 14:50 强制平回由 force_t0_close 走 (本函数不管平回时点, 只管盘中触发)。 + rails: 当日 ≤ 3 次; 单票 / 全局当日T亏熔断后当日禁开新T (仍允许平回已开的仓); + 买入暂停 (决策系统风控预警) 时同样禁开新T, 但平回照常; 14:50 强制平回由 force_t0_close 走。 """ state = ctx["state"] price = _f(day.get("price")) @@ -221,7 +241,7 @@ def _eval_t0(st, pos, day, now, ctx): 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")) @@ -239,8 +259,8 @@ def _eval_t0(st, pos, day, now, ctx): "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: + # ---- 无未平的仓 → 看要不要开新的一轮 (受熔断 / 买入暂停 / 3 次 / 存量约束) ---- + if ctx.get("halted") or ctx.get("buy_paused") or int(state.get("t_count_today") or 0) >= 3: return None if t_qty < LOT: return None @@ -296,7 +316,7 @@ def _grid_levels(params: dict) -> list: def _eval_grid(st, pos, day, now, ctx): """网格: 现价跌破未买档 → 买 1 份; 现价涨破已买档 → 卖 1 份(从 avail); 越上界停做; - 跌破下界 = 继续持有、不再买 (设计四点拍板①)。""" + 跌破下界 = 继续持有、不再买 (设计四点拍板①)。买入暂停时只停买入、卖出照常。""" state = ctx["state"] params = st.get("params") or {} price = _f(day.get("price")) @@ -313,16 +333,15 @@ def _eval_grid(st, pos, day, now, ctx): 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 封顶) + # 卖出: 现价涨破某已买档 (买价 + 一档) → 卖那一份 (从 avail, T+1 由 run_tick 封顶) sell_idx, sell_buyprice = None, -1.0 for idx, info in filled.items(): bp = _f(info.get("price")) @@ -335,8 +354,9 @@ def _eval_grid(st, pos, day, now, ctx): "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: + # 买入: 现价跌破某未买档 → 买一份 (受 max_capital 与 单股上限[规则闸] 双约束) + # 下界之下 / 越上界 / 决策系统风控预警暂停买入 时, 都不再买 (卖出不受影响, 已在上面处理)。 + if state.get("below_floor") or price < lo or price > hi or ctx.get("buy_paused"): return None buy_idx, buy_level = None, -1.0 for i, lv in enumerate(levels): @@ -361,7 +381,7 @@ def _eval_grid(st, pos, day, now, ctx): # ================================================================ 评估器: 跟踪止盈 (设计 §七B) def _eval_trail(st, pos, day, now, ctx): """跟踪止盈: 创新高抬止盈线, 从高点回落 ≥ giveback 就卖 (从 avail); 命中硬止盈目标直接全清。 - 只卖不买 —— 纯离场保护。""" + 只卖不买 —— 纯离场保护, 不受买入暂停影响。""" state = ctx["state"] params = st.get("params") or {} price = _f(day.get("price")) @@ -423,7 +443,7 @@ def _roll_day(state: dict, pos: dict, today: int) -> dict: state["day"] = today state["t_count_today"] = 0 state["rt_base"] = _f(pos.get("realized_t_profit")) - # 隔夜后原则上不该留未平的腿 (14:50 已平回); 万一留了, 清掉 open_leg 交对账兜底 + # 隔夜后原则上不该留未平的仓 (14:50 已平回); 万一留了, 清掉 open_leg 交对账兜底 state["open_leg"] = None return state @@ -461,6 +481,7 @@ def tick(*, now=None, dry_run: bool = False) -> dict: return {**out, "ok": False, "errors": [f"取持仓失败: {type(e).__name__}: {e}"]} scale = _f(view.get("totals", {}).get("scale")) today = _today() + paused_codes = _buypause_codes() # 决策系统风控预警暂停买入的票 (本模块只读) # 全局当日T亏熔断: 汇总所有做T策略的当日T盈亏 (realized_t_profit 相对日初基线的增量) global_t_pnl = 0.0 @@ -490,11 +511,11 @@ def tick(*, now=None, dry_run: bool = False) -> dict: if st.get("type") == "T0": state = _roll_day(state, pos, today) - # 上一腿还没走完就别再发; 走完了先把结果并进 state + # 上一笔还没走完就别再发; 走完了先把结果并进 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": "上一腿在途, 等它走完"}) + out["skipped"].append({"strategy_id": st["strategy_id"], "why": "上一笔还在途, 等它走完"}) continue day = market.day_snapshot(code) @@ -513,11 +534,11 @@ def tick(*, now=None, dry_run: bool = False) -> dict: halted = True out["notes"].append(f"{code} 当日T亏达单票熔断线 {stock_cap:.0f} 元, 今日不再开新T") - ctx = {"state": state, "scale": scale, "halted": halted, "notes": out["notes"]} + ctx = {"state": state, "scale": scale, "halted": halted, "notes": out["notes"], + "buy_paused": code in paused_codes} 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": @@ -564,8 +585,10 @@ def _apply_grid_state(state: dict, dec: dict): # ================================================================ 14:50 强制平回 (设计 §六 rails) def force_t0_close(*, now=None) -> dict: - """做T 强制平回 (scheduler.t0_close 在 PMS_T0_CLOSE_TIME 调): 对每只有未平腿的做T策略, - 立刻发对向平仓腿把当日T仓打平, 绝不过夜。已有在途腿的先撤后不重复 —— 这里只补「还没平」的。""" + """做T 强制平回 (scheduler.t0_close 在 PMS_T0_CLOSE_TIME 调): 对每只有未平仓的做T策略, + 立刻发反方向的平回委托把当日T仓打平, 绝不过夜。已在途的不重复 —— 这里只补「还没平」的。 + + 平回不受买入暂停影响: 暂停挡的是开新仓, 平回是把已开的打平, 必须放行 (反T 平回是买回)。""" out = {"closed": [], "skipped": [], "errors": []} if not param_store.get_bool("PMS_STRATEGY_ENABLED", False): out["skipped"].append("PMS_STRATEGY_ENABLED=False") @@ -585,16 +608,15 @@ def force_t0_close(*, now=None) -> dict: 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": "无未平腿"}) + 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": "平仓腿已在途"}) + out["skipped"].append({"strategy_id": st["strategy_id"], "why": "平回委托已在途"}) continue q = _round_lot(open_leg.get("qty")) if q < LOT: diff --git a/app/services/strategy_service.py b/app/services/strategy_service.py index ccfff28..22dcda0 100644 --- a/app/services/strategy_service.py +++ b/app/services/strategy_service.py @@ -9,6 +9,7 @@ """ from __future__ import annotations +import json import logging from datetime import datetime @@ -134,3 +135,62 @@ def set_status(strategy_id: str, status: str, by: str = "user") -> dict: 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} + + +# ================================================================ 买入暂停表 (决策系统风控预警) +# 挂了策略的票, 决策系统的风控卖出只提示、不自动清仓 (强制离场会推翻你特意设的策略); +# 这里把该票策略的**买入这一侧**暂停 —— 不平仓、不动卖出、可页面恢复。 +# 用独立运行参数 PMS_STRATEGY_BUYPAUSE (按 ts_code 映射), **不放进策略 state_json**: +# 那份 state 每跳被 strategy_runner 重写, 放进去会被并发的 signal_digest / intraday_exec +# 互相覆盖 (与 portfolio.neg_streak_map 用独立参数、避开读改写竞态同一个道理)。 +BUYPAUSE_KEY = "PMS_STRATEGY_BUYPAUSE" + + +def buypause_map() -> dict: + """当前被暂停买入的票 {ts_code: {reason, source, at}}。读失败按空 (不误暂停)。""" + try: + raw = pms_repo.get_param(BUYPAUSE_KEY) + return json.loads(raw) if raw else {} + except Exception: # noqa: BLE001 + return {} + + +def _save_buypause(m: dict) -> dict: + try: + pms_repo.set_param(BUYPAUSE_KEY, json.dumps(m, ensure_ascii=False), "system") + return {"ok": True} + except Exception as e: # noqa: BLE001 —— 写不进要吭声, 别让"已暂停"变成静默没暂停 + logger.error("[strategy] 买入暂停表写入失败: %s —— 该票策略买入可能仍在跑", e) + return {"ok": False, "error": f"{type(e).__name__}: {e}"} + + +def pause_buy(ts_code: str, *, reason: str = "", source: str = "signal") -> list: + """暂停某票所有 ACTIVE 策略的买入这一侧 (决策系统风控预警触发)。返回受影响的 strategy_id。 + 只标记、不平仓、不动卖出; 由页面「恢复买入」(resume_buy) 解除。""" + if not ts_code: + return [] + m = buypause_map() + if ts_code not in m: + m[ts_code] = {"reason": (str(reason)[:300] if reason else ""), "source": source, + "at": str(datetime.now())[:19]} + _save_buypause(m) + try: + return [s.get("strategy_id") for s in + pms_repo.list_strategies(ts_code=ts_code, statuses=["ACTIVE"], limit=10)] + except Exception: # noqa: BLE001 + return [] + + +def resume_buy(strategy_id: str, by: str = "user") -> dict: + """恢复买入 (按策略解除该票的买入暂停)。只恢复买入, 不影响卖出/平回。""" + st = pms_repo.get_strategy(strategy_id) + if not st: + return {"ok": False, "error": "策略不存在"} + code = st.get("ts_code") + m = buypause_map() + if code in m: + m.pop(code, None) + r = _save_buypause(m) + if not r.get("ok"): + return {"ok": False, "error": r.get("error")} + return {"ok": True, "strategy_id": strategy_id, "ts_code": code} diff --git a/app/web/main.py b/app/web/main.py index cb459e5..32ebc69 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -591,9 +591,19 @@ def api_industry_import(payload: dict = Body(...)): @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)}) + from app.services import strategy_service + + def _load(): + rows = pms_repo.list_strategies(statuses=statuses, limit=300) + try: + bp = strategy_service.buypause_map() + except Exception: + bp = {} + for r in rows: + r["buy_paused"] = bp.get(r.get("ts_code")) or None + return {"ok": True, "strategies": rows, + "enabled": param_store.get_bool("PMS_STRATEGY_ENABLED", False)} + return ok(_load) @app.post("/api/strategies/validate") @@ -628,3 +638,11 @@ def api_strategy_status(strategy_id: str, payload: dict = Body(...)): def api_op_log(limit: int = Query(200)): """交易员操作日志 (每个页面写操作一行, 含 OK/BLOCKED 与原因)。""" return ok(lambda: {"ok": True, "rows": pms_repo.list_op_log(limit=limit)}) + + +@app.post("/api/strategies/{strategy_id}/resume-buy") +def api_strategy_resume_buy(strategy_id: str, payload: dict = Body(default={})): + """恢复该策略的买入 (决策系统风控预警触发的暂停由你手动解除; 只恢复买入, 不影响卖出/平回)。""" + from app.services import strategy_service + return ok_logged("resume_strategy_buy", strategy_service.resume_buy, strategy_id, + params={"strategy_id": strategy_id}, by=payload.get("by") or "user") diff --git a/app/web/static/index.html b/app/web/static/index.html index 18c5b3f..b1358c5 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -242,11 +242,15 @@ - + -