240 lines
11 KiB
Python
240 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""逻辑状态四态在 PMS 的落点 (2026-09-07 下一阶段方案第三件 PMS 侧)。
|
|
|
|
选股系统每天早上给每只票算一个「支撑它的研究证据还在不在」的状态 —— 逻辑强化 / 逻辑成立 /
|
|
无法判断 / 逻辑存疑 (它那边的 logic_state.py), 随计划每行下发, 另有一个按代码查询的接口给持仓票用
|
|
(持仓票在候选筛选第一步就被整行剔掉, 计划里读不到它)。PMS 只读它, 不算它, 也不送研判 —— 产业逻辑
|
|
不归择时决策系统判, judge.OPEN_JUDGE_KEYS 不收它, 那是一致性检查表第九行要继续成立的东西。
|
|
|
|
四态里只有「逻辑存疑」会改分流通道, 其余三个值不产生任何动作 (收敛规则是单调的: 强化不升判决,
|
|
存疑不把仅展示变成可执行)。本模块做四件事:
|
|
|
|
一, 早上拉完计划后 (scheduler.plan_pull), 对在持的票查一次状态, 写进运行参数 PMS_LOGIC_STATE_MAP。
|
|
取不到就留空 (写一份空映射并记原因), **绝不折成存疑**。
|
|
二, 按状态暂停或恢复策略腿的买入: 存疑 → strategy_service.pause_buy(source="logic");
|
|
已明确不存疑 → 只清本来源的暂停, 不动风控 (signal) 与定性 (accum) 停的; 没查到的票不动。
|
|
三, 把当日态挂到持仓行上 (attach), 动作引擎据此停增持侧的自主动作、按开关出研究走弱的减持
|
|
(那条减持带 research_weak 来源, 分流那一步一票否决强制交人, 绝不自动卖)。
|
|
四, 持仓视图的两栏: 「当初为什么买」(账本里首批入场那条记录的判决依据与论断) 与
|
|
「现在证据还在不在」(今天的状态与出处)。
|
|
|
|
安全方向: 读不到状态 = 没有读数, 什么都不额外拦, 也不解除已有的暂停。总开关
|
|
PMS_LOGIC_STATE_ROUTE 关掉即回到只按判决分流; 减持提议另有 PMS_LOGIC_DOUBT_TRIM, 默认关。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from collections import Counter
|
|
from datetime import date, datetime
|
|
|
|
from app.core import action_engine as ae
|
|
from app.repo import pms_repo
|
|
from app.services import param_store, plan_feed, strategy_service
|
|
|
|
logger = logging.getLogger("pms.logic")
|
|
|
|
MAP_KEY = "PMS_LOGIC_STATE_MAP"
|
|
PAUSE_SOURCE = "logic" # 策略买入暂停表里本模块用的来源名, 与 signal / accum 并列
|
|
FRESH_DAYS = 3 # 映射超过这么多个自然日没刷新就当没有读数 (调度断了不能拿旧态拦人)
|
|
|
|
# 入场那条账本记录里, 只要带了这几个键之一就算「有研究理由」
|
|
_ENTRY_KEYS = ("basis", "logic", "verdict", "logic_state")
|
|
|
|
|
|
# ================================================================ 运行参数里的映射
|
|
def load_map() -> dict:
|
|
"""整份映射 (含时刻与错误), 页面与运维看。读不到或坏了返回空字典。"""
|
|
raw = param_store.get(MAP_KEY, "") or ""
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
m = json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
return {}
|
|
return m if isinstance(m, dict) else {}
|
|
|
|
|
|
def state_map(m: dict | None = None) -> dict:
|
|
"""{ts_code: 逻辑状态} —— 只在映射新鲜时给; 超过 FRESH_DAYS 个自然日没刷新一律按没有读数。"""
|
|
m = load_map() if m is None else (m or {})
|
|
at = str(m.get("at") or "")[:10]
|
|
try:
|
|
age = (date.today() - date.fromisoformat(at)).days
|
|
except ValueError:
|
|
return {}
|
|
if age > FRESH_DAYS:
|
|
return {}
|
|
states = m.get("states")
|
|
return dict(states) if isinstance(states, dict) else {}
|
|
|
|
|
|
def _save_map(m: dict) -> dict:
|
|
r = param_store.set_param(MAP_KEY, json.dumps(m, ensure_ascii=False), "system")
|
|
if not r.get("ok"):
|
|
# 写不进要吭声: 映射停在旧值, 明天早上以前扫描层看到的都是昨天的态
|
|
logger.error("[逻辑状态] 映射写入失败: %s —— 扫描层沿用上一份 (最多 %d 天后自动失效)",
|
|
r.get("error"), FRESH_DAYS)
|
|
return r
|
|
|
|
|
|
def route_enabled() -> bool:
|
|
return param_store.get_bool("PMS_LOGIC_STATE_ROUTE", True)
|
|
|
|
|
|
# ================================================================ 早上那一步
|
|
def pull_for_held(now=None, fetch=None) -> dict:
|
|
"""在持票的逻辑状态取回一次并落映射, 再按结果暂停或恢复策略买入腿。
|
|
|
|
fetch 可注入 (单测)。任何一步失败都不抛: 取数失败写空映射带原因 (扫描层按没有读数处理),
|
|
暂停那一步失败只记进 errors。返回一份读数给调度任务写日志。
|
|
"""
|
|
now = now or datetime.now()
|
|
stamp = now.strftime("%Y-%m-%d %H:%M:%S")
|
|
out = {"ok": True, "at": stamp, "held": 0, "got": 0, "by_state": {}, "paused": [],
|
|
"resumed": [], "errors": []}
|
|
try:
|
|
held = [r["ts_code"] for r in pms_repo.list_positions(only_open=True)]
|
|
except Exception as e: # noqa: BLE001
|
|
held = []
|
|
out["errors"].append(f"读持仓失败: {type(e).__name__}: {e}")
|
|
out["held"] = len(held)
|
|
if not held:
|
|
_save_map({"at": stamp, "date": None, "states": {}, "note": "没有在持的票"})
|
|
return out
|
|
try:
|
|
states = (fetch or plan_feed.fetch_logic_states)(held)
|
|
except Exception as e: # noqa: BLE001
|
|
# 取不到就留空, 绝不折成存疑; 也不动已有的暂停 (没有新证据不解除)
|
|
_save_map({"at": stamp, "date": None, "states": {}, "error": str(e)[:300]})
|
|
out.update(ok=False)
|
|
out["errors"].append(f"查逻辑状态失败: {e}")
|
|
return out
|
|
day = next((s.get("date") for s in states.values() if s.get("date")), None)
|
|
r = _save_map({"at": stamp, "date": day, "states": states})
|
|
if not r.get("ok"):
|
|
out["errors"].append(f"映射写入失败: {r.get('error')}")
|
|
out["got"] = len(states)
|
|
out["by_state"] = dict(Counter(s.get("state") or "空" for s in states.values()))
|
|
missing = [c for c in held if c not in states]
|
|
if missing:
|
|
out["missing"] = missing
|
|
p = apply_pauses(states, held)
|
|
out["paused"], out["resumed"] = p["paused"], p["resumed"]
|
|
out["errors"].extend(p["errors"])
|
|
if p.get("skipped"):
|
|
out["pause_skipped"] = p["skipped"]
|
|
return out
|
|
|
|
|
|
def apply_pauses(states: dict, held) -> dict:
|
|
"""存疑 → 暂停策略买入腿 (来源 logic); 明确不存疑 → 只清本来源的暂停; 没查到的票不动。"""
|
|
out = {"paused": [], "resumed": [], "errors": []}
|
|
if not route_enabled():
|
|
out["skipped"] = "PMS_LOGIC_STATE_ROUTE 关着, 不动策略买入腿"
|
|
return out
|
|
for code in held or []:
|
|
st = (states or {}).get(code)
|
|
if not st:
|
|
continue # 没有读数: 不拦也不放
|
|
try:
|
|
if ae.is_logic_doubt(st):
|
|
strategy_service.pause_buy(
|
|
code, reason=ae.logic_doubt_text(st, ae.WHY_LOGIC_DOUBT_PAUSE),
|
|
source=PAUSE_SOURCE)
|
|
out["paused"].append(code)
|
|
else:
|
|
r = strategy_service.clear_buypause(code, only_source=PAUSE_SOURCE) or {}
|
|
if r.get("cleared"):
|
|
out["resumed"].append(code)
|
|
except Exception as e: # noqa: BLE001
|
|
out["errors"].append(f"{code} 策略买入腿处理失败: {type(e).__name__}: {e}")
|
|
return out
|
|
|
|
|
|
# ================================================================ 扫描层
|
|
def attach(rows: list, states: dict | None = None) -> list:
|
|
"""把当日逻辑状态挂到持仓行上 (原地改)。没有读数的票不挂键, 动作引擎按「没有读数」处理。"""
|
|
states = state_map() if states is None else (states or {})
|
|
for r in rows or []:
|
|
st = states.get(r.get("ts_code"))
|
|
if st:
|
|
r["logic_state"] = st
|
|
else:
|
|
r.pop("logic_state", None)
|
|
return rows
|
|
|
|
|
|
# ================================================================ 持仓视图两栏
|
|
def now_view(st) -> dict:
|
|
"""「现在证据还在不在」: 今天的状态、截止日、落定说明与第一条出处; 没有读数就说没有。"""
|
|
if not isinstance(st, dict) or not st.get("state"):
|
|
return {"state": None, "text": "今天没有读数（早上拉计划时没查到这只票）"}
|
|
reasons = [str(x) for x in (st.get("reasons") or []) if x]
|
|
return {"state": st.get("state"), "raw_state": st.get("raw_state"),
|
|
"as_of": st.get("as_of"), "settle_note": st.get("settle_note"),
|
|
"text": (reasons[0] if reasons else (st.get("why") or ""))[:160]}
|
|
|
|
|
|
def entry_view(ts_code: str) -> dict:
|
|
"""「当初为什么买」: 首批未平批次 → 它的指令 → 该指令或它来自的提议在账本上的放行记录 → 硬数字里的
|
|
判决依据、论断、判决与入场时的逻辑状态。三种情形都要说得出话: 有指令链、外部成交并入、账本无行。"""
|
|
try:
|
|
lots = pms_repo.list_lots(ts_code, status="OPEN", limit=50)
|
|
except Exception as e: # noqa: BLE001
|
|
return {"why": f"批次读取失败: {type(e).__name__}"}
|
|
if not lots:
|
|
return {"why": "没有未平的批次记录"}
|
|
first = lots[0]
|
|
open_date = str(first.get("open_date") or "")[:10]
|
|
iid = first.get("instruction_id")
|
|
if not iid:
|
|
return {"why": "外部成交并入的持仓，没有系统的入场记录", "open_date": open_date}
|
|
refs = [str(iid)]
|
|
try:
|
|
ins = pms_repo.get_instruction(str(iid)) or {}
|
|
except Exception: # noqa: BLE001
|
|
ins = {}
|
|
for k in (ins.get("origin_id"), (ins.get("progress") or {}).get("from_proposal")):
|
|
if k and str(k) not in refs:
|
|
refs.append(str(k))
|
|
try:
|
|
rows = pms_repo.ledger_by_ref(refs)
|
|
except Exception as e: # noqa: BLE001
|
|
return {"why": f"账本读取失败: {type(e).__name__}", "open_date": open_date}
|
|
hit = None
|
|
for r in rows:
|
|
hn = r.get("hard_numbers") or {}
|
|
if r.get("verdict") == "PASS" and any(hn.get(k) for k in _ENTRY_KEYS):
|
|
hit = r
|
|
break
|
|
if not hit:
|
|
return {"why": "账本里没有这次入场的研究理由（命令建仓或策略买入不带候选卡）",
|
|
"open_date": open_date, "instruction_id": str(iid)}
|
|
hn = hit.get("hard_numbers") or {}
|
|
st = hn.get("logic_state") if isinstance(hn.get("logic_state"), dict) else {}
|
|
return {"basis": hn.get("basis"), "logic": list(hn.get("logic") or [])[:3],
|
|
"verdict": hn.get("verdict"), "logic_state_at_entry": st.get("state"),
|
|
"decided_at": str(hit.get("decided_at") or "")[:16],
|
|
"arbiter": hit.get("arbiter"), "reason": str(hit.get("reason") or "")[:200],
|
|
"open_date": open_date, "instruction_id": str(iid)}
|
|
|
|
|
|
def decorate_positions(rows: list, states: dict | None = None) -> list:
|
|
"""给持仓接口的每一行加 entry 与 logic_now 两块 (原地改)。任何一行失败只写原因, 页面不能塌。"""
|
|
states = state_map() if states is None else (states or {})
|
|
for r in rows or []:
|
|
code = r.get("ts_code")
|
|
try:
|
|
r["logic_now"] = now_view(states.get(code))
|
|
except Exception as e: # noqa: BLE001
|
|
r["logic_now"] = {"state": None, "text": f"读数处理失败: {type(e).__name__}"}
|
|
if int(r.get("total_qty") or 0) <= 0:
|
|
r["entry"] = None
|
|
continue
|
|
try:
|
|
r["entry"] = entry_view(code)
|
|
except Exception as e: # noqa: BLE001
|
|
r["entry"] = {"why": f"入场记录处理失败: {type(e).__name__}"}
|
|
return rows
|