290 lines
15 KiB
Python
290 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
择时实现A客户端 · 委托决策系统 (设计 §8, 待办 #9)
|
|
==================================================
|
|
接口契约见 BIONIC_PMS_INTERFACE.md。决策系统按它凌晨算好的支撑/压力推出买入区间与
|
|
卖出区间, 盘中只回答「现价在不在区间内 + 限价挂哪里」, 不做任何新的盘中判断
|
|
(2026-08-03 用户定的原则: 提前计算为主、盘中监控为辅, 可以接受买不上)。
|
|
|
|
分工与降级 (三条, 都是纪律不是实现细节):
|
|
1. **本地检查先行**: 配额/兜底/停牌/一字板/不追高(当日涨幅) 由 core.exec_timing 的
|
|
hard_gate 先判, 命中就不咨询 —— 这些检查始终留在 PMS 本地。尤其 14:45 兜底:
|
|
到点必须完成, 决策系统说什么都不算。
|
|
2. **拿不到不等于有答案**: 咨询失败/超时/对端回 UNAVAILABLE/答复无法识别 → 本轮
|
|
整体退实现B (设计 §13「决策系统择时不可用 → 择时退实现B」), 并进入冷却期
|
|
(冷却内不再咨询, 免得每分钟 tick 都白等一次超时)。绝不把「拿不到」当成 FIRE 或 WAIT。
|
|
3. **应答带有效期**: 结果缓存在指令 progress.exec_advice 里 (随既有落表持久化,
|
|
页面/t-ins 可见), 有效期内不重复咨询。对端可用 valid_min 缩短有效期, 只缩不放。
|
|
|
|
默认 PMS_EXEC_IMPL=B —— 本模块整个短路, run_tick 行为与接通前一字不差。
|
|
页面把 PMS_EXEC_IMPL 改成 A (并保证 PMS_EXEC_API_BASE 或 PMS_JUDGE_API_BASE 已填)
|
|
即切实现A, 随时可改回 B, 不需要重启。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from app.core import exec_timing as et
|
|
from app.core import tradedays as td
|
|
from app.services import judge, param_store
|
|
|
|
logger = logging.getLogger("pms.exec_advisor")
|
|
|
|
IMPL_A, IMPL_B = "A", "B"
|
|
FIRE, WAIT = "FIRE", "WAIT"
|
|
|
|
|
|
def impl() -> str:
|
|
v = str(param_store.get("PMS_EXEC_IMPL", IMPL_B) or IMPL_B).strip().upper()
|
|
return v if v in (IMPL_A, IMPL_B) else IMPL_B
|
|
|
|
|
|
def base_url() -> str:
|
|
"""实现A的接口根地址; PMS_EXEC_API_BASE 留空时沿用研判闸的 PMS_JUDGE_API_BASE
|
|
(两者是同一个 bionic 服务, 不逼着用户填两遍)。"""
|
|
b = (param_store.get("PMS_EXEC_API_BASE", "") or "").strip().rstrip("/")
|
|
return b or judge.base_url()
|
|
|
|
|
|
def available() -> bool:
|
|
return impl() == IMPL_A and bool(base_url())
|
|
|
|
|
|
def status() -> dict:
|
|
"""页面/排查用的一句话状态。"""
|
|
if impl() != IMPL_A:
|
|
return {"impl": IMPL_B, "available": False,
|
|
"note": "内置保守择时 (PMS_EXEC_IMPL=B)。切实现A: 页面改 PMS_EXEC_IMPL=A"}
|
|
if not base_url():
|
|
return {"impl": IMPL_A, "available": False,
|
|
"note": "PMS_EXEC_IMPL=A 但接口地址为空 (PMS_EXEC_API_BASE 与 "
|
|
"PMS_JUDGE_API_BASE 都没填) —— 实际全程退实现B"}
|
|
return {"impl": IMPL_A, "available": True, "base": base_url(),
|
|
"path": param_store.get("PMS_EXEC_PATH", "/api/intraday/pms_exec"),
|
|
"ttl_min": param_store.get_int("PMS_EXEC_ADVICE_TTL_MIN", 10)}
|
|
|
|
|
|
def _post(url: str, payload: dict, timeout: int) -> dict:
|
|
"""HTTP 一跳, 单测在这里打桩。"""
|
|
import requests
|
|
r = requests.post(url, json=payload, timeout=timeout)
|
|
r.raise_for_status()
|
|
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:
|
|
"""择时判定统一入口 (executor.run_tick 的唯一调用点)。
|
|
|
|
返回结构与 exec_timing.decide 相同, 另带 source 字段标明这条决定是谁做的:
|
|
B 实现B (默认档位, 或实现A未配置)
|
|
guard 本地事实性检查 (配额/兜底/停牌/一字板/不追高) —— 与实现无关
|
|
A / A缓存 决策系统应答 (新咨询 / 有效期内复用)
|
|
B(实现A不可用: ...) 咨询失败退实现B, 括号里是原因
|
|
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)
|
|
d["source"] = IMPL_B
|
|
return d
|
|
|
|
left = max(0, int(quota) - int(fired_today))
|
|
h = et.hard_gate(side=side, now=now, day=day, params=params, is_last_day=is_last_day,
|
|
is_command=is_command, fired_today=fired_today, quota=quota)
|
|
if h is not None:
|
|
h["source"] = "guard" # 本地事实性检查 (配额/兜底等), 与实现无关
|
|
return h
|
|
|
|
advice, note = _advice(ts_code=ts_code, side=side, action=action, now=now, day=day,
|
|
pos=pos or {}, left=left, is_last_day=is_last_day,
|
|
tdays_left=tdays_left, prog=prog)
|
|
if advice is not None:
|
|
d = et.apply_advice(side=side, day=day, params=params, advice=advice, left=left)
|
|
if d is not None:
|
|
d["source"] = advice.get("_source") or "A"
|
|
# 把昨夜定性顺着决策一起带回去, 规则闸终检要用 (见 rule_gate.BAD_Y_SIGNALS)。
|
|
# 缓存命中的那条 advice 里也存着它, 所以十个交易分钟的缓存期内一样有值。
|
|
d["y_signal"] = advice.get("y_signal")
|
|
# 参考位盘中被改写的说明 (只有新建仓会有)。executor 看到它就在评审账本落一条
|
|
# WARN —— 这一路的留痕不能只写在指令的 progress 里, 账本才是判分事实源。
|
|
if advice.get("ref_drift"):
|
|
d["ref_drift"] = advice["ref_drift"]
|
|
return d
|
|
note = f"研判动作无法识别: {advice.get('verdict')!r}"
|
|
|
|
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)
|
|
d["source"] = f"B(实现A不可用: {note})"
|
|
return d
|
|
|
|
|
|
def _advice(*, ts_code, side, action, now, day, pos, left, is_last_day, tdays_left, prog):
|
|
"""取一份有效研判: 缓存命中 → 直接用; 冷却中 → (None, 原因); 否则咨询一次。
|
|
返回 (advice|None, 不可用原因)。advice 带 _source 标明 A / A缓存。"""
|
|
now_min = et.hm_to_min(now)
|
|
today = td.ymd()
|
|
ttl = max(1, param_store.get_int("PMS_EXEC_ADVICE_TTL_MIN", 10))
|
|
cached = dict((prog or {}).get("exec_advice") or {})
|
|
|
|
if int(cached.get("ymd") or 0) == today:
|
|
if (cached.get("verdict") in (FIRE, WAIT)
|
|
and now_min <= int(cached.get("valid_until_min") or -1)):
|
|
c = dict(cached)
|
|
c["_source"] = "A缓存"
|
|
return c, ""
|
|
if cached.get("fail_until_min") and now_min <= int(cached["fail_until_min"]):
|
|
return None, (f"冷却至 {et._fmt(int(cached['fail_until_min']))}: "
|
|
f"{cached.get('error') or '上次咨询失败'}")
|
|
|
|
payload = {
|
|
"direction": "PMS_EXEC", "ts_code": ts_code, "side": side, "action": action,
|
|
"qty_left": left, "is_last_day": bool(is_last_day), "tdays_left": tdays_left,
|
|
"now": et._fmt(now_min),
|
|
"day": {k: day.get(k) for k in ("price", "vwap", "open", "high", "low",
|
|
"day_chg_from_open", "bars")},
|
|
"refs": {"support": pos.get("support_ref"), "pressure": pos.get("pressure_ref"),
|
|
"stop": pos.get("stop_ref"), "source": pos.get("ref_source")},
|
|
"position": {"total_qty": pos.get("total_qty"), "avail_qty": pos.get("avail_qty"),
|
|
"avg_cost": pos.get("avg_cost"), "cushion_pct": pos.get("cushion_pct")},
|
|
}
|
|
to = max(1, param_store.get_int("PMS_EXEC_TIMEOUT_SEC", 8))
|
|
url = base_url() + (param_store.get("PMS_EXEC_PATH", "/api/intraday/pms_exec") or "")
|
|
|
|
def _cool(err: str):
|
|
cool = max(1, param_store.get_int("PMS_EXEC_FAIL_COOLDOWN_MIN", 5))
|
|
if prog is not None:
|
|
prog["exec_advice"] = {"ymd": today, "error": err[:200],
|
|
"fail_until_min": et.add_trade_minutes(now_min, cool),
|
|
"consulted_at": et._fmt(now_min)}
|
|
|
|
try:
|
|
data = _post(url, payload, to)
|
|
except Exception as e:
|
|
err = f"{type(e).__name__}: {e}"
|
|
logger.error("[择时A] 咨询失败, 本轮退实现B (%s %s): %s", ts_code, side, err)
|
|
_cool(err)
|
|
return None, err
|
|
|
|
verdict = str(data.get("verdict") or "").strip().upper()
|
|
if verdict not in (FIRE, WAIT):
|
|
# 对端明说给不出结论 (UNAVAILABLE), 或答复不认识 —— 都按拿不到处理, 退实现B
|
|
reason = str(data.get("reason") or f"verdict={verdict or '空'}")[:200]
|
|
logger.warning("[择时A] 决策系统给不出结论 (%s %s): %s —— 本轮退实现B",
|
|
ts_code, side, reason)
|
|
_cool(f"UNAVAILABLE: {reason}")
|
|
return None, f"UNAVAILABLE: {reason}"
|
|
|
|
try:
|
|
valid_min = int(data.get("valid_min") or ttl)
|
|
except (TypeError, ValueError):
|
|
valid_min = ttl
|
|
valid_min = max(1, min(valid_min, ttl)) # 对端只能缩短有效期, 不能放长
|
|
obs = data.get("observed") or {}
|
|
adv = {"ymd": today, "verdict": verdict,
|
|
"limit_price": data.get("limit_price"),
|
|
"reason": str(data.get("reason") or "")[:200],
|
|
"confidence": data.get("confidence"),
|
|
# 应答 observed 里的昨夜定性: 存进缓存, 让规则闸在缓存期内也拿得到
|
|
"y_signal": obs.get("y_signal"),
|
|
# 支撑/压力/买入区间也一并存下来 —— 新建仓的漂移比对要用 (见 _check_ref_drift)
|
|
"support": obs.get("support"), "pressure": obs.get("pressure"),
|
|
"buy_band": obs.get("buy_band"),
|
|
"valid_until_min": et.add_trade_minutes(now_min, valid_min),
|
|
"consulted_at": et._fmt(now_min)}
|
|
drift = _check_ref_drift(action=action, prog=prog, adv=adv, today=today)
|
|
if drift:
|
|
# 输入在盘中被改写了 —— 本轮改判等待, 并把原因写在明面上。
|
|
# 不是拒绝、也不是退实现B: 这是一次「输入不可信, 先不动」的显式等待。
|
|
adv = {**adv, "verdict": WAIT, "limit_price": None, "reason": drift[:200],
|
|
"ref_drift": drift}
|
|
if prog is not None:
|
|
prog["exec_advice"] = adv
|
|
a = dict(adv)
|
|
a["_source"] = "A"
|
|
return a, ""
|
|
|
|
|
|
def _f(v):
|
|
try:
|
|
x = float(v)
|
|
return x if x > 0 else 0.0
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def _check_ref_drift(*, action, prog, adv, today) -> str:
|
|
"""新建仓的参考位「当日首答锁定, 之后偏离即停」。返回漂移说明, 没漂就返回空串。
|
|
|
|
要防的是这件事: 择时读的 `strategy_daily_results` 会被盘中跑 push-pool 触发的补扫
|
|
**就地改写**。实证过两只 —— 000035 的压力 5.2 变 5.15, 002335 的支撑 31.27 变 29.00
|
|
(差 7.3%)。已有持仓有摊薄成本与安全垫做锚, 支撑压力漂一点不会让判断翻转; 而新建仓的
|
|
买入区间**完全由支撑压力推出来**, 支撑一变, 区间整体平移 —— 上午判「现价高于上沿、
|
|
不追」的票, 下午可能变成「在区间内、出手」, 而这时候没有人在看。
|
|
|
|
所以只对 OPEN 生效: 当天第一次拿到应答时把支撑/压力/区间锁进指令的 progress.ref_lock,
|
|
之后每次应答都跟它比, 偏离超过 PMS_OPEN_REF_DRIFT_MAX 就改判等待。
|
|
锁按「日」重置 —— 次日的昨夜结论本来就该是新的一份, 那不叫漂移。
|
|
|
|
顺带一个副作用是特意要的: 漂移一旦发生就在账本里留下痕迹, 漂了几次、每次多少都能统计。
|
|
将来真要决定「给 fetch_yesterday_strategy 加日期过滤」那件事时, 手里有实证而不是印象。
|
|
"""
|
|
if str(action or "").upper() != "OPEN" or prog is None:
|
|
return ""
|
|
sup, pre = _f(adv.get("support")), _f(adv.get("pressure"))
|
|
lock = dict(prog.get("ref_lock") or {})
|
|
if int(lock.get("ymd") or 0) != int(today):
|
|
if sup or pre: # 当日首答: 锁定, 不比对
|
|
prog["ref_lock"] = {"ymd": int(today), "support": sup, "pressure": pre,
|
|
"buy_band": adv.get("buy_band"),
|
|
"at": adv.get("consulted_at")}
|
|
return ""
|
|
max_drift = param_store.get_float("PMS_OPEN_REF_DRIFT_MAX", 0.03)
|
|
if max_drift <= 0:
|
|
return ""
|
|
moved = []
|
|
for label, key, now_v in (("支撑", "support", sup), ("压力", "pressure", pre)):
|
|
old = _f(lock.get(key))
|
|
if old <= 0 or now_v <= 0:
|
|
continue
|
|
gap = abs(now_v / old - 1)
|
|
if gap > max_drift:
|
|
moved.append(f"{label} {old} → {now_v} (偏离 {gap:.1%})")
|
|
if not moved:
|
|
return ""
|
|
return (f"昨夜结论盘中被改写: {'; '.join(moved)}, 超过容忍幅度 {max_drift:.0%} —— "
|
|
f"该票新建仓当日暂停。无人值守的建仓不能建在会漂移的输入上 "
|
|
f"(锁定于 {lock.get('at') or '当日首答'})")
|