267 lines
10 KiB
Python
267 lines
10 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
决策系统盘中信号消化 (设计 §10「信号消化」)
|
||
|
|
=============================================
|
||
|
|
订阅两条流, 转成 PMS 自己的卖出动作或提议:
|
||
|
|
|
||
|
|
db2 `intraday_signals:{YYYY-MM-DD}` 盘中 BUY/SELL/HOLD 广播 (每日一条流)
|
||
|
|
db3 `bionic:signals:llm_sell_actions` 风控 LLM 卖出动作 (固定 key)
|
||
|
|
|
||
|
|
**用独立消费组** (`pms_signal_consumer`), 与 trading_service 的 `qmt_main_activator` /
|
||
|
|
`qmt_sell_activator` 互不抢消息 —— Redis Stream 的消费组之间各自看到全量消息,
|
||
|
|
所以 PMS 可以和现有消费者并行订阅, 迁移期两边都能跑。
|
||
|
|
|
||
|
|
不做常驻进程: 由调度器 `signal_digest` 每分钟拉一批, 与其余任务同一套守卫和降级口径。
|
||
|
|
解析与消化规则在 core/signal_rules.py (纯逻辑), 本模块只管连 Redis、落表、留痕。
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
|
||
|
|
from config.settings import settings
|
||
|
|
from app.core import command_spec as cs
|
||
|
|
from app.core import signal_rules as sr
|
||
|
|
from app.core import tradedays as td
|
||
|
|
from app.repo import pms_repo
|
||
|
|
from app.services import executor, param_store, portfolio
|
||
|
|
|
||
|
|
logger = logging.getLogger("pms.signal")
|
||
|
|
|
||
|
|
SEEN_KEY = "PMS_SIGNAL_SEEN" # 当日去重集合 (JSON), 日切自动作废
|
||
|
|
_clients = {}
|
||
|
|
|
||
|
|
|
||
|
|
def _client(db: int):
|
||
|
|
"""Redis 客户端。强制 RESP2 —— 与行情库同一个坑 (服务端 <6.0 不认 HELLO)。"""
|
||
|
|
if db in _clients:
|
||
|
|
return _clients[db]
|
||
|
|
import redis
|
||
|
|
kw = dict(host=settings.SIGNAL_REDIS_HOST, port=settings.SIGNAL_REDIS_PORT,
|
||
|
|
password=settings.SIGNAL_REDIS_PASSWORD or None, db=db,
|
||
|
|
decode_responses=True, socket_timeout=settings.SIGNAL_REDIS_SOCKET_TIMEOUT)
|
||
|
|
try:
|
||
|
|
c = redis.Redis(protocol=2, **kw)
|
||
|
|
except TypeError:
|
||
|
|
c = redis.Redis(**kw)
|
||
|
|
_clients[db] = c
|
||
|
|
return c
|
||
|
|
|
||
|
|
|
||
|
|
def group_name() -> str:
|
||
|
|
return param_store.get("PMS_SIGNAL_GROUP", "pms_signal_consumer") or "pms_signal_consumer"
|
||
|
|
|
||
|
|
|
||
|
|
def streams() -> list:
|
||
|
|
"""[(db, key, parser)] —— 盘中流按日期拼 key, 风控流是固定 key。"""
|
||
|
|
ymd = datetime.now().strftime("%Y-%m-%d")
|
||
|
|
tpl = param_store.get("PMS_SIGNAL_STREAM_INTRADAY", "intraday_signals:{ymd}")
|
||
|
|
sell_key = param_store.get("PMS_SIGNAL_STREAM_SELL", "bionic:signals:llm_sell_actions")
|
||
|
|
return [(settings.SIGNAL_REDIS_DB_INTRADAY, tpl.format(ymd=ymd), sr.parse_intraday),
|
||
|
|
(settings.SIGNAL_REDIS_DB_ACTIONS, sell_key, sr.parse_risk_sell)]
|
||
|
|
|
||
|
|
|
||
|
|
def status() -> dict:
|
||
|
|
"""页面用: 两条流的连通性与积压情况。"""
|
||
|
|
out = {"enabled": param_store.get_bool("PMS_SIGNAL_ENABLED", True),
|
||
|
|
"group": group_name(), "streams": []}
|
||
|
|
for db, key, _ in streams():
|
||
|
|
item = {"db": db, "key": key}
|
||
|
|
try:
|
||
|
|
c = _client(db)
|
||
|
|
item["length"] = c.xlen(key)
|
||
|
|
groups = c.xinfo_groups(key)
|
||
|
|
mine = [g for g in groups if g.get("name") == group_name()]
|
||
|
|
item["pending"] = mine[0].get("pending") if mine else None
|
||
|
|
item["group_ready"] = bool(mine)
|
||
|
|
item["other_groups"] = [g.get("name") for g in groups
|
||
|
|
if g.get("name") != group_name()]
|
||
|
|
except Exception as e:
|
||
|
|
item["error"] = f"{type(e).__name__}: {e}"
|
||
|
|
out["streams"].append(item)
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
# ================================================================ 消费
|
||
|
|
def consume(*, batch: int = 50, dry_run: bool = False) -> dict:
|
||
|
|
"""拉一批信号并消化。每分钟一跳, 幂等 (消费组 ACK + 当日去重)。"""
|
||
|
|
out = {"ok": True, "read": 0, "exits": [], "proposals": [], "recorded": 0,
|
||
|
|
"ignored": 0, "errors": [], "dry_run": dry_run}
|
||
|
|
if not param_store.get_bool("PMS_SIGNAL_ENABLED", True):
|
||
|
|
out["skipped"] = "信号消化已关闭 (PMS_SIGNAL_ENABLED=False)"
|
||
|
|
return out
|
||
|
|
|
||
|
|
try:
|
||
|
|
view = portfolio.positions_view()
|
||
|
|
except Exception as e:
|
||
|
|
return {**out, "ok": False, "errors": [f"读账本失败: {type(e).__name__}: {e}"]}
|
||
|
|
|
||
|
|
prm = {"sell_conf_min": param_store.get_float("PMS_SIGNAL_SELL_CONF_MIN", 0.75),
|
||
|
|
"auto_exit_conf": param_store.get_float("PMS_SIGNAL_AUTO_EXIT_CONF", 0.85),
|
||
|
|
"trim_ratio": param_store.get_float("PMS_SIGNAL_TRIM_RATIO", 1 / 3)}
|
||
|
|
seen = _load_seen()
|
||
|
|
ymd = td.ymd()
|
||
|
|
|
||
|
|
for db, key, parser in streams():
|
||
|
|
try:
|
||
|
|
msgs = _read(db, key, batch)
|
||
|
|
except Exception as e:
|
||
|
|
out["errors"].append(f"{key} 读取失败: {type(e).__name__}: {e}")
|
||
|
|
continue
|
||
|
|
out["read"] += len(msgs)
|
||
|
|
for msg_id, fields in msgs:
|
||
|
|
try:
|
||
|
|
sig = parser(fields, msg_id=msg_id)
|
||
|
|
_handle(sig, view, prm, seen, ymd, dry_run, out)
|
||
|
|
if not dry_run:
|
||
|
|
_ack(db, key, msg_id)
|
||
|
|
except Exception as e:
|
||
|
|
logger.exception("信号处理失败 %s", msg_id)
|
||
|
|
out["errors"].append(f"{msg_id}: {type(e).__name__}: {e}")
|
||
|
|
|
||
|
|
if not dry_run:
|
||
|
|
_save_seen(seen, ymd)
|
||
|
|
out["ok"] = not out["errors"]
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def _handle(sig, view, prm, seen, ymd, dry_run, out):
|
||
|
|
code = sig.get("ts_code")
|
||
|
|
pos = _pos_of(view, code) if code else None
|
||
|
|
d = sr.digest(sig, pos, prm)
|
||
|
|
act = d["action"]
|
||
|
|
|
||
|
|
if act == sr.ACT_IGNORE:
|
||
|
|
out["ignored"] += 1
|
||
|
|
return
|
||
|
|
if act == sr.ACT_RECORD:
|
||
|
|
# 只给持有的票留痕, 否则全市场广播会把评审账本冲垮
|
||
|
|
if pos and int(pos.get("total_qty") or 0) > 0 and not dry_run:
|
||
|
|
pms_repo.insert_ledger(ts_code=code, action="SIGNAL", arbiter="rule",
|
||
|
|
verdict="PASS", price_at=float(pos.get("price") or 0),
|
||
|
|
hard_numbers={**d["hard_numbers"], "msg_id": sig.get("msg_id")},
|
||
|
|
reason=d["reason"][:500])
|
||
|
|
out["recorded"] += 1
|
||
|
|
return
|
||
|
|
|
||
|
|
key = sr.dedup_key(sig, ymd)
|
||
|
|
if key in seen:
|
||
|
|
out["ignored"] += 1
|
||
|
|
return
|
||
|
|
if _has_inflight(code):
|
||
|
|
out["ignored"] += 1
|
||
|
|
out.setdefault("skipped_inflight", []).append(code)
|
||
|
|
return
|
||
|
|
|
||
|
|
brief = {"ts_code": code, "qty": d["qty"], "confidence": d["hard_numbers"]["confidence"],
|
||
|
|
"reason": d["reason"]}
|
||
|
|
if dry_run:
|
||
|
|
(out["exits"] if act == sr.ACT_EXIT else out["proposals"]).append(
|
||
|
|
{**brief, "dry_run": True})
|
||
|
|
return
|
||
|
|
|
||
|
|
seen.add(key)
|
||
|
|
if act == sr.ACT_EXIT:
|
||
|
|
iid = _make_exit(code, d, pos)
|
||
|
|
out["exits"].append({**brief, "instruction_id": iid})
|
||
|
|
else:
|
||
|
|
pid = _make_proposal(code, d, pos, sig)
|
||
|
|
out["proposals"].append({**brief, "proposal_id": pid})
|
||
|
|
|
||
|
|
|
||
|
|
def _make_exit(code, d, pos) -> str:
|
||
|
|
"""高置信风控卖出 → 直接落卖出指令 (减持方向不设确认门槛)。"""
|
||
|
|
now = datetime.now()
|
||
|
|
iid = cs.make_instruction_id(td.ymd(now), code, "EXIT", int(now.strftime("%H%M%S")) % 1000)
|
||
|
|
window = param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3)
|
||
|
|
pms_repo.insert_instruction(
|
||
|
|
instruction_id=iid, origin_type="system", origin_id=d["hard_numbers"].get("source"),
|
||
|
|
ts_code=code, action="EXIT", side="sell", qty=d["qty"], limit_price=None,
|
||
|
|
window_tdays=window, status=executor.ST_PROPOSED,
|
||
|
|
progress={"deadline": str(td.window_deadline(now.date(), window)),
|
||
|
|
"is_command": False, "children": [], "from_signal": True,
|
||
|
|
"reason": d["reason"]})
|
||
|
|
pms_repo.insert_ledger(ts_code=code, action="EXIT", arbiter="rule", verdict="PASS",
|
||
|
|
price_at=float((pos or {}).get("price") or 0),
|
||
|
|
hard_numbers=d["hard_numbers"], ref_id=iid,
|
||
|
|
reason=d["reason"][:500])
|
||
|
|
logger.warning("[信号消化] %s 转清仓指令 %s —— %s", code, iid, d["reason"])
|
||
|
|
return iid
|
||
|
|
|
||
|
|
|
||
|
|
def _make_proposal(code, d, pos, sig) -> str:
|
||
|
|
ttl = param_store.get_int("PMS_PROPOSAL_TTL_HOURS", 24)
|
||
|
|
pid = f"PRP_{td.ymd()}_{code.replace('.', '')}_SIGSELL"
|
||
|
|
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"],
|
||
|
|
hard_numbers=hn,
|
||
|
|
expire_at=datetime.now() + timedelta(hours=ttl),
|
||
|
|
judge_verdict=None, judge_reason=d["reason"][:500])
|
||
|
|
return pid
|
||
|
|
|
||
|
|
|
||
|
|
# ================================================================ Redis 细节
|
||
|
|
def _read(db: int, key: str, batch: int) -> list:
|
||
|
|
c = _client(db)
|
||
|
|
g, consumer = group_name(), param_store.get("PMS_SIGNAL_CONSUMER", "pms_1")
|
||
|
|
try:
|
||
|
|
c.xgroup_create(key, g, id="$", mkstream=True) # 只消化新消息, 不回溯历史
|
||
|
|
logger.info("[信号消化] 建消费组 %s @ %s", g, key)
|
||
|
|
except Exception as e:
|
||
|
|
if "BUSYGROUP" not in str(e):
|
||
|
|
raise
|
||
|
|
resp = c.xreadgroup(g, consumer, {key: ">"}, count=int(batch), block=100)
|
||
|
|
out = []
|
||
|
|
for _stream, messages in (resp or []):
|
||
|
|
out.extend(messages)
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def _ack(db: int, key: str, msg_id: str):
|
||
|
|
try:
|
||
|
|
_client(db).xack(key, group_name(), msg_id)
|
||
|
|
except Exception as e:
|
||
|
|
logger.warning("[信号消化] ACK 失败 %s: %s", msg_id, e)
|
||
|
|
|
||
|
|
|
||
|
|
# ================================================================ 去重与助手
|
||
|
|
def _load_seen() -> set:
|
||
|
|
try:
|
||
|
|
raw = pms_repo.get_param(SEEN_KEY)
|
||
|
|
d = json.loads(raw) if raw else {}
|
||
|
|
if str(d.get("ymd")) != str(td.ymd()):
|
||
|
|
return set()
|
||
|
|
return set(d.get("keys") or [])
|
||
|
|
except Exception:
|
||
|
|
return set()
|
||
|
|
|
||
|
|
|
||
|
|
def _save_seen(seen: set, ymd):
|
||
|
|
try:
|
||
|
|
pms_repo.set_param(SEEN_KEY, json.dumps({"ymd": ymd, "keys": sorted(seen)[-500:]}),
|
||
|
|
"system")
|
||
|
|
except Exception as e:
|
||
|
|
logger.warning("[信号消化] 去重集合写入失败: %s", e)
|
||
|
|
|
||
|
|
|
||
|
|
def _has_inflight(code: str) -> bool:
|
||
|
|
try:
|
||
|
|
for i in pms_repo.list_instructions(statuses=list(executor.LIVE), ts_code=code, limit=20):
|
||
|
|
if str(i.get("side")).lower() == "sell":
|
||
|
|
return True
|
||
|
|
for p in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=200):
|
||
|
|
if p["ts_code"] == code and p["action"] in ("TRIM", "EXIT"):
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
logger.warning("[信号消化] 在途检查失败(按无在途继续): %s", e)
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def _pos_of(view: dict, ts_code: str):
|
||
|
|
for x in view["positions"]:
|
||
|
|
if x["ts_code"] == ts_code:
|
||
|
|
return x
|
||
|
|
return None
|