tradingSystem/app/services/strategy_runner.py

810 lines
43 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- 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 决定, 与本层无关、自动继承。
* **双保险**: 本层 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
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 _min_lot(code) -> int:
"""最小申报数量: 科创板 (688/689 开头) 买卖都是 200 股起, 其余 100 股。
2026-08-25 补: 持仓里出现科创板票后发现全库取整都按 100 股, 而科创板 100 股的申报
会被券商直接拒掉。规则还有一条例外: 持仓不足 200 股时允许**一次性全部卖出**,
网格与止盈的各卖出落点分别处理了这一条。做T是命令授权的手工策略, 科创板做T的
开腿与平回数量怎么处理需要单独拍板, 本次不动 (见 DEVLOG)。"""
return 200 if str(code or "").startswith(("688", "689")) else 100
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 _macro_gate_active() -> bool:
"""宏观偏热闸 (MACRO_TIMING_PLAN.md §5.2): 生效时策略的买入开腿一并暂停。
与 buypause 走同一个判定点 (ctx.buy_paused), 天然继承「只挡开新仓/加仓腿,
卖出/平回/跟踪止盈照常」的既有语义 —— 反T 的买回是平回腿, 不会被拦。
读不到按不生效: 闸的安全方向是不额外拦。"""
try:
from app.services import macro_service
return bool((macro_service.gate_state() or {}).get("active"))
except Exception as e: # noqa: BLE001
logger.warning("[strategy] 读宏观闸状态失败 (按不生效): %s", e)
return False
# ================================================================ 指令 / 提议下发
def _emit_instruction(st: dict, dec: dict, *, forced: bool = False) -> str:
"""按决策发一张短窗口命令指令 (window_tdays=1, is_command=True), 交 run_tick 执行。
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]
is_cmd = (st.get("type") == "T0") # 做T必须当日轧平→命令口径(含14:45强制平回);
# 网格/跟踪止盈到价即成交、当日没成交则作废, 不强制
prog = {"is_command": is_cmd, "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") == "GRID" and str(leg or "").startswith("grid_sell"):
# 网格卖单当日未成交作废 → 把发单时弹掉的档位放回去 (2026-08-28 审查修)。
# 部分成交则按剩余量放回; 全成交不放。invested 同步加回放回部分。
exec_qty = int(ins.get("exec_qty") or 0)
info = dict(pending.get("grid_restore") or {})
idx = pending.get("grid_sell_idx")
left = int(_f(info.get("qty"))) - exec_qty
if idx is not None and info and left >= LOT:
fl = dict(state.get("filled_levels") or {})
fl[str(idx)] = {"price": _f(info.get("price")), "qty": left}
state["filled_levels"] = fl
state["invested"] = _f(state.get("invested")) + _f(info.get("price")) * left
logger.info("[strategy] %s 网格卖档 %s 未全成交 (成 %s), 档位按余量 %s 放回",
st.get("strategy_id"), idx, exec_qty, left)
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 (仍允许平回已开的仓);
买入暂停 (决策系统风控预警) 时同样禁开新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:
# 平回数量取整 (2026-08-28 审查修): avail 可能是零股 (其它卖出把可卖量
# 消耗成 137 之类), 非整百的部分卖出会被券商拒单并每跳重发。整百部分先平,
# 尾巴交 14:50 兜底与对账。
q_close = min(q, avail)
if q_close % LOT and q_close != int(pos.get("total_qty") or 0):
q_close = _round_lot(q_close)
if q_close <= 0:
return None
return {"side": "sell", "action": A_T0, "qty": q_close, "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 ctx.get("buy_paused") 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 _band(levels: list, price: float) -> int:
"""现价所处的档位下标: 满足 levels[i] <= price 的最大 i; 低于最低档返回 -1。"""
b = -1
for i, lv in enumerate(levels):
if lv <= price:
b = i
else:
break
return b
def _eval_grid(st, pos, day, now, ctx):
"""网格(逐档穿越): 价每向下跌破一个新档买一手(只买中枢下方), 向上涨破一个档就把下面对应
档买的那手卖掉(从 avail)。一跳只走一档; 跌破下界=继续持有不再买; 买入暂停只停买、卖出照常;
filled_levels 每跳与真实 ADD 持仓对账, 被对账冲销后收敛 —— 绝不卖幻影档(误卖底仓)。"""
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 = levels[0]
mlot = _min_lot(st.get("ts_code"))
per_lot = _round_lot(params.get("per_lot")) or LOT
if per_lot < mlot:
per_lot = mlot # 科创板 200 股起: 手工挂的 100 股档在这里抬成合法数量
max_capital = _f(params.get("max_capital"))
filled = {int(k): dict(v) for k, v in (state.get("filled_levels") or {}).items()}
invested = _f(state.get("invested"))
avail = int(pos.get("avail_qty") or 0)
actual_add = int(pos.get("add_qty") or 0) # 网格买入记 ADD 批次, 这是真实网格持仓
# —— 对账收敛: filled 声称的网格股 > 真实(被 RECON 冲销) → 收敛; 真实为0则清空重来 ——
claimed = sum(int(v.get("qty") or 0) for v in filled.values())
if claimed > actual_add:
if actual_add <= 0:
if filled:
ctx["notes"].append(f"{st['ts_code']} 网格持仓已被对账冲销(真实网格股0), 清空网格档位重来")
filled, invested = {}, 0.0
state["last_band"] = None
else:
for k in sorted(filled.keys()):
if claimed <= actual_add:
break
q = int(filled[k].get("qty") or 0)
invested = max(0.0, invested - _f(filled[k].get("price")) * q)
claimed -= q
del filled[k]
ctx["notes"].append(f"{st['ts_code']} 网格档位与真实持仓对齐(真实网格股 {actual_add})")
state["filled_levels"] = {str(k): v for k, v in filled.items()}
state["invested"] = invested
cur = _band(levels, price)
last = state.get("last_band")
# 跌破下界: 停买、保留已买、告警; 记基准档但不交易
if price < lo:
if not state.get("below_floor"):
state["below_floor"] = True
ctx["notes"].append(f"{st['ts_code']} 跌破网格下界 {lo}, 已停止网格买入(继续持有已买档)")
state["last_band"] = cur
return None
state["below_floor"] = False
# 首跳: 只记基准档, 不交易 (等价格真正穿越档位才动)
if last is None:
state["last_band"] = cur
return None
# —— 上行: 价涨破 → 卖掉离开的这一档买的那手 (从 avail, 且确有网格股, 才卖) ——
if cur > last:
k = last
if k in filled and avail >= LOT and actual_add >= LOT:
q = min(per_lot, _round_lot(avail))
if q >= mlot: # 科创板部分卖低于 200 股不合法, 量不足先不推进
state["last_band"] = last + 1
info = filled[k]
return {"side": "sell", "action": A_SELL, "qty": q, "leg": f"grid_sell:{k}",
"grid_sell_idx": k,
"reason": f"网格卖: 现价 {price} 涨破档{k}(买价 {info.get('price')}), 卖 {q}"}
return None # 有档可卖但量不足, 先不推进, 下跳再试
state["last_band"] = last + 1 # 该档无网格持仓可卖, 只随价上移
return None
# —— 下行: 价跌破新档 → 买这一档 (只买中枢下方; 受下界/暂停/上限约束) ——
if cur < last:
k = last - 1 # 刚跌破的这一档
if k < 0:
state["last_band"] = cur
return None
# **只买中枢下方** (2026-08-28 审查修): 档位表覆盖整个区间, 原实现对跌破的任何档
# 都买 —— 价格先涨向上界再回落一档, 就在中枢上方接了一手; 上半区震荡还会把预算
# (按中枢以下档数摊的 per_lot) 先吃光, 真跌回吸筹区反而触顶停买。恰好高买低不买。
center = _f(params.get("center")) or ((lo + _f(params.get("upper"))) / 2
if _f(params.get("upper")) > 0 else 0.0)
if center > 0 and k < len(levels) and levels[k] >= center:
state["last_band"] = last - 1 # 中枢上方: 只随价下移, 不买
return None
if k in filled:
state["last_band"] = last - 1 # 已买, 只推进
return None
if ctx.get("buy_paused"):
return 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
state["last_band"] = last - 1
return {"side": "buy", "action": A_GRID_BUY, "qty": per_lot, "leg": f"grid_buy:{k}",
"grid_buy_idx": k, "grid_buy_price": price,
"reason": f"网格买: 现价 {price} 跌破档{k}(档价 {levels[k]}), 买 {per_lot}"}
return None # cur == last, 同档不动
# ================================================================ 评估器: 跟踪止盈 (设计 §七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%})")
total = int(pos.get("total_qty") or 0)
mlot = _min_lot(st.get("ts_code"))
if avail < LOT:
# 零股尾巴的出口 (2026-08-28 审查修): 主板整百取整后剩的 <100 股, 原来在这里
# 永远 return None, 挂着策略又被动作引擎排除 —— 尾巴永久滞留。A股规则允许
# 零股**一次性全部卖出**, 所以当尾巴就是全部持仓、且硬目标或回落条件仍成立时,
# 按全清把尾巴一次清掉; 其余情形照旧只更新高水位。
tail_hit = (hard_target > 0 and profit >= hard_target) or (
state.get("armed") and hw > 0 and giveback > 0 and price <= hw * (1 - giveback))
if avail > 0 and avail == total and tail_hit:
return {"side": "sell", "action": A_EXIT, "qty": avail, "leg": "trail_hard",
"reason": f"跟踪止盈-零股收尾: 余 {avail} 股为全部持仓, 一次性清出"}
return None # 无 T+1 可卖 (或尾巴暂不能清), 只更新高水位
def _all_out():
# 全清数量: 科创板持仓不足 200 股时按交易所例外一次性全卖, 否则整百
return avail if avail < mlot else (_round_lot(avail) or avail)
# 盘中 SAR 止损线 (2026-09-11 工作包三 part 3): 09:45 后现价跌破昨日 SAR 值的缓冲即全清, 当日只触发一次。
# SAR 值由 09:40 的 strategy_advisor.scan 刷进 params["sar_line"] (工作包三 part 3 的刷新腿);
# 没刷进 (技术面无读数、首日、映射停更) 就跳过这道线 —— 加不改, 拿不到 SAR 绝不当"跌破"。
# 与 giveback/hard_target 并列的一道硬止损, 不看 armed (SAR 翻空是趋势破位, 不必先武装)。
sar_line = _f(params.get("sar_line"))
if sar_line > 0 and now is not None and param_store.get_bool("PMS_TECH_SAR_STOP_ON_TRAIL", True):
hm = now.hour * 100 + now.minute
if hm >= 945 and int(state.get("sar_stop_day") or 0) != td.ymd(now):
buf = param_store.get_float("PMS_TECH_SAR_STOP_BUFFER", 0.003)
if price <= sar_line * (1 - buf):
state["sar_stop_day"] = td.ymd(now) # 当日只触发一次
return {"side": "sell", "action": A_EXIT, "qty": _all_out(), "leg": "trail_sar",
"reason": f"跟踪止盈-SAR 止损: 现价 {price} 跌破昨日 SAR {sar_line}"
f"×(1-{buf:.1%}), 全清 {_all_out()}"}
# 硬止盈目标: 直接全清
if hard_target > 0 and profit >= hard_target:
return {"side": "sell", "action": A_EXIT, "qty": _all_out(), "leg": "trail_hard",
"reason": f"跟踪止盈-硬目标: 浮盈 {profit:.1%}{hard_target:.1%}, 全清 avail {avail}"}
# 已武装且从高点回落到设定比例 → 卖。
# **部分卖有一次性闩锁** (2026-08-28 审查修): "回撤 5% 卖 50%" 的本意是这轮回撤卖一次。
# 原来高水位只抬不降、armed 永不复位, 卖完上一单后条件仍成立, 每隔一单再卖剩余的一半,
# 几何级联直到卖光。闩锁记"已按哪个高水位卖过" (trail_fired_hw), 只有高水位**再创新高**
# 之后的下一轮回撤才允许再卖; 全清路径 (sell_ratio≥1 与硬目标) 不上锁 —— 清仓意图
# 失败了就该重试。
fired_hw = _f(state.get("trail_fired_hw"))
if state.get("armed") and hw > 0 and price <= hw * (1 - giveback) and giveback > 0 \
and (sell_ratio >= 1 or hw > fired_hw + 1e-9):
if sell_ratio < 1:
q = _round_lot(avail * sell_ratio)
if mlot > LOT:
# 科创板: 部分卖低于 200 股不合法 —— 量够就抬到 200 (方向是保利润, 多卖
# 一点偏保守), 可卖的本来就不足 200 则这条部分卖路径放弃, 等硬目标或人工
q = 0 if avail < mlot else min(max(q, mlot), _round_lot(avail))
else:
q = _all_out()
if q >= mlot or (sell_ratio >= 1 and q > 0):
act = A_EXIT if sell_ratio >= 1 else A_SELL
if sell_ratio < 1:
state["trail_fired_hw"] = round(hw, 3)
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()
paused_codes = _buypause_codes() # 决策系统风控预警暂停买入的票 (本模块只读)
macro_gated = _macro_gate_active() # 宏观偏热闸: 全体策略买开腿暂停, 闸解除自动恢复
if macro_gated:
out["notes"].append("宏观偏热闸生效: 策略买入开腿本轮暂停 (卖出与平回照常)")
# 全局当日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"],
"buy_paused": (code in paused_codes) or macro_gated}
dec = fn(st, pos, day, now, ctx)
if dec:
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"),
"grid_sell_idx": dec.get("grid_sell_idx"),
"grid_restore": dec.get("_grid_restore")}
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 (买入占用一档, 卖出释放一档)。
卖出弹掉的档位信息随手塞回 dec (_grid_restore): 委托当日一股没成交时,
_reconcile 要把这档**放回去** —— 原来发单即弹档、未成交不回滚, 股票还在档位
记录没了, 下次跌破同档会再买一手 (同档双份), invested 也被提前扣掉导致上限
实际被突破一档 (2026-08-28 审查修)。"""
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:
dec["_grid_restore"] = dict(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仓打平, 绝不过夜。已在途的不重复 —— 这里只补「还没平」的。
平回不受买入暂停影响: 暂停挡的是开新仓, 平回是把已开的打平, 必须放行 (反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 已买 → 卖平
avail_now = int(pos.get("avail_qty") or 0)
q_close = min(q, avail_now)
if q_close % LOT and q_close != int(pos.get("total_qty") or 0):
q_close = _round_lot(q_close) # 零股平回会被拒单, 整百部分先平 (2026-08-28)
dec = {"side": "sell", "action": A_T0, "qty": q_close,
"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