tradingSystem/app/services/signal_service.py

# -*- 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), 日切自动作废

# 买入信号留痕在评审账本里用的动作名 (2026-08-06)。
# **刻意不叫 OPEN**: 账本里 action=OPEN 的行是真的建仓评审 (放行或拒绝), 两者混在一起,
# `make t-gate` 就分不清「决策系统说这只票转多了」和「PMS 决定建这只票」。
# 动作引擎那条路读的就是这个动作名 —— 见 pms_repo.buy_signals_today。
SIGNAL_BUY_ACTION = "SIGNAL_BUY"

_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()
    try:
        strat_codes = pms_repo.active_strategy_codes()
    except Exception:
        strat_codes = set()

    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, strat_codes)
                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, strat_codes=frozenset()):
    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

    if act == sr.ACT_NOTE_BUY:
        # 买入信号留痕: **持没持仓都写**。未持仓的票正是新建仓关心的那一批, 从前它们
        # 连一行账本都没有 —— 「决策系统今天看多了哪几只」查不到, 事后没法复盘。
        #
        # 三条防冲垮的口径 (与上面 RECORD 那句「否则全市场广播会把评审账本冲垮」同一个顾虑):
        #   1. 只认 BUY, HOLD 等仍走 RECORD 的老口径;
        #   2. 按 (日期, 来源, 股票, 动作) 当日去重 —— 同一只票一天最多一行;
        #   3. 上游那侧本来就有节流: watcher 对同股看多研判有两小时防抖锁, 昨夜已看多的票
        #      直接免疫不派单, 所以 REVERSAL_BUY 是几十条的量级, 不是全市场广播。
        # verdict 用 NOTE 不用 PASS —— 这是「记下来」不是「放行」, 账本里必须分得开
        # (watch.py 对不认识的 verdict 有兜底符号, 不会显示异常)。
        key = sr.dedup_key(sig, ymd)
        if key in seen:
            out["ignored"] += 1
            return
        out.setdefault("buy_notes", []).append({"ts_code": code, "reason": d["reason"]})
        if dry_run:
            return
        px = float((pos or {}).get("price") or d["hard_numbers"].get("suggested_price") or 0)
        pms_repo.insert_ledger(
            ts_code=code, action=SIGNAL_BUY_ACTION, arbiter="rule", verdict="NOTE",
            price_at=px, hard_numbers={**d["hard_numbers"], "msg_id": sig.get("msg_id")},
            reason=d["reason"][:500])
        seen.add(key)
        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) 再落库, 一旦 _make_exit /
    # _make_proposal 抛异常 (DB 抖一下、下游拒一次), 上层 catch 住记进 out["errors"],
    # 但这一天的去重键已经烧掉了 —— 同一条风控卖出信号后面再来多少次都被当成重复丢弃,
    # 指令一条都不会落。失败长得像成功: 页面只多一行 error, 而该卖的票就那么留着了。
    # 2026-07-31 修。
    on_strategy = bool(code) and code in strat_codes
    if act == sr.ACT_EXIT and not on_strategy:
        iid = _make_exit(code, d, pos)
        seen.add(key)
        out["exits"].append({**brief, "instruction_id": iid})
    else:
        # 挂了策略的票: 决策系统的风控卖出只提示、不自动清仓 (强制离场会推翻你特意设的策略);
        # 一律落提议进「等我拍板」由你定"维持 / 采纳即撤策略并清仓", 采纳的是全清 (as_exit)。
        pid = _make_proposal(code, d, pos, sig, on_strategy=on_strategy,
                             as_exit=(on_strategy and act == sr.ACT_EXIT))
        seen.add(key)
        out["proposals"].append({**brief, "proposal_id": pid, "on_strategy": on_strategy})
    if on_strategy and not dry_run:
        # 同时暂停该票策略的买入这一侧 (不平仓、不动卖出、页面可恢复) —— 挡住"资金在流出、
        # 网格还在逢跌买入", 又不替你做清仓这种不可逆的事。
        try:
            from app.services import strategy_service
            affected = strategy_service.pause_buy(code, reason=d["reason"],
                                                  source=(sig.get("source") or "signal"))
            if affected:
                out.setdefault("strategy_buy_paused", []).extend(affected)
                logger.warning("[信号消化] %s 挂着策略, 决策系统卖出只提示不自动清仓; "
                               "已暂停该票策略买入(页面可恢复): %s", code, affected)
        except Exception as e:
            logger.error("[信号消化] 暂停策略买入失败 %s: %s", code, e)


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, *, on_strategy=False, as_exit=False) -> str:
    ttl = param_store.get_int("PMS_PROPOSAL_TTL_HOURS", 24)
    pid = f"PRP_{td.ymd()}_{code.replace('.', '')}_SIGSELL"
    action = "EXIT" if as_exit else "TRIM"
    reason = d["reason"]
    if on_strategy:
        reason = ("这只票挂着交易方案(策略): 决策系统卖出信号只提示、未自动清仓; 该票策略买入已暂停。"
                  "你定: 维持观察 / 采纳即撤策略并清仓。 —— " + reason)
    hn = {**d["hard_numbers"], "price": float((pos or {}).get("price") or 0),
          "reason": reason, "signal_source": sig.get("source"), "on_strategy": on_strategy}
    pms_repo.insert_proposal(proposal_id=pid, ts_code=code, action=action, qty=d["qty"],
                             hard_numbers=hn,
                             expire_at=datetime.now() + timedelta(hours=ttl),
                             judge_verdict=("STRATEGY_RISK" if on_strategy else None),
                             judge_reason=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