From f2f144454bab4c8359546a7cc9252c3d3fe3c0e9 Mon Sep 17 00:00:00 2001 From: zlt Date: Tue, 11 Aug 2026 16:19:26 +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/services/exec_advisor.py | 30 ++++++++ app/services/strategy_runner.py | 123 ++++++++++++++++++++++---------- 2 files changed, 115 insertions(+), 38 deletions(-) diff --git a/app/services/exec_advisor.py b/app/services/exec_advisor.py index 87667a4..80622e3 100644 --- a/app/services/exec_advisor.py +++ b/app/services/exec_advisor.py @@ -72,6 +72,31 @@ def _post(url: str, payload: dict, timeout: int) -> dict: return r.json() or {} +def _strategy_immediate(*, side, now, day, quota, fired_today) -> dict: + """网格/跟踪止盈的到价即成交决策: 只做硬事实检查(配额/时段/停牌/一字板), 不做 vwap 择时, + 不做 14:45 强制。真正的合规(单股上限/T+1/暂停买入等)由 run_tick 随后的 rule_gate 终检兜底。 + 这样网格是"到档位价才成交、当日没成交则由窗口收口作废", 不会像命令那样被收盘强平成市价单。""" + left = max(0, int(quota) - int(fired_today)) + now_min = et.hm_to_min(now) + + def _o(a, r, limit=None): + return {"action": a, "qty_hint": left, "limit_price": limit, "reason": r, "source": "策略即时"} + + if left <= 0: + return _o(et.ACT_WAIT, "当日配额已出完") + if not et.in_session(now_min): + return _o(et.ACT_WAIT, "非交易时段") + price = _f(day.get("price")) + if price <= 0 or day.get("halted"): + return _o(et.ACT_SKIP, "停牌或无实时价, 当日顺延") + if str(side).lower() == "buy" and day.get("limit_up"): + return _o(et.ACT_WAIT, "涨停封板, 不追买") + if str(side).lower() == "sell" and day.get("limit_down"): + return _o(et.ACT_WAIT, "跌停封板, 挂单等回封") + disc = 1.002 if str(side).lower() == "buy" else 0.998 + return _o(et.ACT_FIRE, f"按档触发即时成交, 现价 {price}", round(price * disc, 2)) + + def decide(*, side: str, action: str, ts_code: str, now, day: dict, params: dict, is_last_day: bool, is_command: bool = True, fired_today: int = 0, quota: int = 0, pos: dict = None, tdays_left=None, prog: dict = None) -> dict: @@ -85,6 +110,11 @@ def decide(*, side: str, action: str, ts_code: str, now, day: dict, params: dict prog 由调用方传入指令的 progress dict, 咨询结果/失败冷却会写进 prog["exec_advice"], 随调用方既有的落表动作持久化; 传 None 则本轮结论不缓存 (dry_run 语义)。 """ + # 网格/跟踪止盈: 策略层已按档位触发, 这里不做择时博弈, 到价即时成交、收盘不强制。 + # 只识别策略来源且类型为 GRID/TRAIL 的指令; 做T(T0)与其余来源一律走原有逻辑, 不受影响。 + if (prog or {}).get("origin") == "strategy" and str((prog or {}).get("stype")) in ("GRID", "TRAIL"): + return _strategy_immediate(side=side, now=now, day=day, quota=quota, fired_today=fired_today) + if not available(): d = et.decide(side=side, now=now, day=day, params=params, is_last_day=is_last_day, is_command=is_command, fired_today=fired_today, quota=quota) diff --git a/app/services/strategy_runner.py b/app/services/strategy_runner.py index 8237b91..105f442 100644 --- a/app/services/strategy_runner.py +++ b/app/services/strategy_runner.py @@ -113,7 +113,9 @@ def _emit_instruction(st: dict, dec: dict, *, forced: bool = False) -> str: 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": [], + 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( @@ -314,9 +316,21 @@ def _grid_levels(params: dict) -> list: 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): - """网格: 现价跌破未买档 → 买 1 份; 现价涨破已买档 → 卖 1 份(从 avail); 越上界停做; - 跌破下界 = 继续持有、不再买 (设计四点拍板①)。买入暂停时只停买入、卖出照常。""" + """网格(逐档穿越): 价每向下跌破一个新档买一手(只买中枢下方), 向上涨破一个档就把下面对应 + 档买的那手卖掉(从 avail)。一跳只走一档; 跌破下界=继续持有不再买; 买入暂停只停买、卖出照常; + filled_levels 每跳与真实 ADD 持仓对账, 被对账冲销后收敛 —— 绝不卖幻影档(误卖底仓)。""" state = ctx["state"] params = st.get("params") or {} price = _f(day.get("price")) @@ -325,57 +339,90 @@ def _eval_grid(st, pos, day, now, ctx): 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 + lo = levels[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()} + 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}, 已停止网格买入(继续持有已买档)") - 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 or ctx.get("buy_paused"): + state["last_band"] = cur 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: + 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 >= LOT: + 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 + 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 + 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 + 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)