# -*- coding: utf-8 -*- """ 个股交易方案 (策略) 运行器 —— PER_STOCK_STRATEGY_PLAN.md §四/§六/§七/§七B ============================================================================ 每分钟一跳 (挂在 scheduler.intraday_exec 里, 与 run_tick 并列)。载入 ACTIVE 策略, 按类型 评估 (做T / 网格 / 跟踪止盈), 触发就**发一张短窗口指令** (window_tdays=1, is_command=True, origin_type='strategy'), 交 executor.run_tick 用现有管线执行 —— 择时 / 规则闸 / T+1 可卖封顶 / 下发 / 账本一道不重写; autonomy=confirm 的落一条提议进「等我拍板」, 人点采纳后再由本层发指令。 **只做加法**: 本模块不改 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) 兜住: 非交易时段 / 休假模式不跑。 * 影子/实盘由 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 market, param_store, portfolio logger = logging.getLogger("pms.strategy") # 指令在途 (未终态) 的状态集 —— 与 executor.LIVE 一致; 本层据此判断「上一腿还没走完, 先别再发」。 LIVE_INS = ("PROPOSED", "RULE_PASSED", "DISPATCHED") # 批次动作 (与 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 _f(v, d=0.0): try: return float(v) except (TypeError, ValueError): return d 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 排除。读库失败按空集 (不误排除全体持仓)。""" try: return pms_repo.active_strategy_codes() except Exception as e: # noqa: BLE001 logger.warning("[strategy] 读取 ACTIVE 策略集失败 (按空集): %s", e) return set() # ================================================================ 每分钟主跳 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 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 只算不发不落库 (页面「试算」用)。 """ 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, "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 try: 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}") 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: """做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