From 99ea218f2f48be439640f8f2a3bd1e78eaa5d395 Mon Sep 17 00:00:00 2001 From: zlt Date: Mon, 7 Sep 2026 14:37:35 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AC=AC=E4=B8=89=E4=BB=B6=20PMS=20=E4=BE=A7?= =?UTF-8?q?=EF=BC=9A=E6=8C=89=E9=80=BB=E8=BE=91=E7=8A=B6=E6=80=81=E5=88=86?= =?UTF-8?q?=E6=B5=81=EF=BC=8C=E5=85=A5=E5=9C=BA=E8=AE=BA=E7=82=B9=E9=9A=8F?= =?UTF-8?q?=E6=8C=81=E4=BB=93=E8=B5=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新模块 logic_state_service：早上拉完计划后对在持票查一次逻辑状态（桥的 /logic_state）， 写进运行参数 PMS_LOGIC_STATE_MAP，取不到留空绝不折成存疑；按结果暂停或恢复策略买入腿 （来源 logic，不动风控与定性停的）；把当日态挂到持仓行；持仓视图两栏（当初为什么买 / 现在证据还在不在），三种情形都说得出话。 plan_feed：解析层收 logic_state（出处最多三条），候选 items 带它；加按代码查询与应答解析。 action_engine：硬数字带 logic_state（研判白名单不收）；判决候选而逻辑存疑的新建仓强制交人； 持仓存疑停增持侧；新求值器 eval_weak_research（默认关，开了必交人，按可卖量夹紧）排在 保垫减仓前；同轮减持优先级改为到价清仓 > 研究走弱 > 保垫减仓。 proposal_service：扫描参数加三个旋钮，扫描前挂逻辑状态。scheduler.plan_pull 顺带取回。 param_store：四个运行参数（分流开关默认开、减持提议默认关、比例三分之一、映射只读）。 pms_repo.ledger_by_ref 与 FakeRepo 同签名；api_positions 加两块；页面加两列。 测试：新建 test_batch22_units.py 十四例并登记；第十二批求值器名单钉子同步。 Co-Authored-By: Claude Fable 5.1 --- app/core/action_engine.py | 96 +++++- app/repo/pms_repo.py | 17 ++ app/scheduler.py | 13 +- app/services/logic_state_service.py | 239 +++++++++++++++ app/services/param_store.py | 6 + app/services/plan_feed.py | 74 ++++- app/services/proposal_service.py | 13 + app/web/main.py | 9 +- app/web/static/index.html | 30 +- scripts/run_tests.py | 10 +- scripts/test_batch12_units.py | 6 +- scripts/test_batch22_units.py | 455 ++++++++++++++++++++++++++++ scripts/test_wiring.py | 4 + 13 files changed, 961 insertions(+), 11 deletions(-) create mode 100644 app/services/logic_state_service.py create mode 100644 scripts/test_batch22_units.py diff --git a/app/core/action_engine.py b/app/core/action_engine.py index 736c727..20b6eef 100644 --- a/app/core/action_engine.py +++ b/app/core/action_engine.py @@ -53,6 +53,16 @@ BUY, SELL = "buy", "sell" # 只在同一只票的同一轮里比较, 不影响不同票, 也不影响下一轮。 _SELL_PRIORITY = {A_EXIT: 0, A_TRIM: 1} + +def _sell_priority(c: dict) -> int: + """同一轮多条减持留哪一条 (数字小的优先; 2026-09-07 第三件起按动作加来源): + 到价清仓 (人的意思) → 研究证据走弱的减持 (必交人) → 保垫减仓 (规则自动)。""" + if c.get("action") == A_EXIT: + return 0 + if c.get("source") == SRC_RESEARCH_WEAK: + return 1 + return 2 + # 候选来源 (2026-09-03)。每条候选都带一个来源, 提议分流那一步拿它决定「这条减持能不能不等人」。 # # 为什么需要这一项: 分流从前写的是「卖出方向一律自动执行」, 只看方向、不看来源。那条规矩 @@ -96,6 +106,38 @@ VERDICT_CANDIDATE, VERDICT_WATCH, VERDICT_DISPLAY = "候选", "关注", "仅展 WHY_WATCH_CONFIRM = "选股系统判为关注（无法判断），交人裁决" WHY_DISPLAY_ONLY = "选股系统判为仅展示" +# 逻辑状态四态里唯一会触发动作的值 (2026-09-07 第三件)。四态是选股系统按「支撑这只票的研究证据 +# 还在不在」算出来的跨日跟踪, 与候选卡判决是正交的两维; 只有「逻辑存疑」会改分流通道 —— 持仓停 +# 增持侧自主动作、按开关出必交人的减持; 判决候选而逻辑存疑的新建仓强制人工确认。其余三个值 +# (逻辑强化 / 逻辑成立 / 无法判断) 不产生任何动作。中文原文与上游逐字对应, 不做映射。 +LOGIC_DOUBT = "逻辑存疑" +WHY_LOGIC_DOUBT_CONFIRM = "可交易口径成立，但研究证据在走弱（逻辑存疑），交人裁决" +WHY_LOGIC_DOUBT_PAUSE = "研究证据在走弱（逻辑存疑），停掉增持侧的自主动作" + + +def is_logic_doubt(logic_state) -> bool: + return isinstance(logic_state, dict) and str(logic_state.get("state") or "").strip() == LOGIC_DOUBT + + +def logic_doubt_text(logic_state, head: str) -> str: + """把存疑的出处接在一句话后面: 第一条说明与证据截止日, 让人看得出是哪一路在走弱、证据是哪天的。""" + st = logic_state if isinstance(logic_state, dict) else {} + parts = [head] + reasons = [str(x) for x in (st.get("reasons") or []) if x] + if reasons: + parts.append(reasons[0][:120]) + if st.get("as_of"): + parts.append(f"证据截止 {st.get('as_of')}") + return "；".join(parts) + + +def logic_confirm_why(logic_state): + """按逻辑状态要不要强制人工确认: 逻辑存疑回原因 (带出处), 其余一律 None。 + 与 verdict_confirm_why 并列: 判决管「今天要不要买」, 逻辑状态管「研究证据还在不在」。""" + if not is_logic_doubt(logic_state): + return None + return logic_doubt_text(logic_state, WHY_LOGIC_DOUBT_CONFIRM) + # 冻结的三种来路，对交易员是三件不同的事：自己下的命令、自己手设的、系统刹车期间。 # 原先一句「该股 COMMAND_HALT, 禁增持」三种都不分，还把枚举值印在了页面上。 @@ -367,6 +409,36 @@ def eval_target(p: dict, params: dict = None, mkt: dict = None): return c +def eval_weak_research(p: dict, params: dict, mkt: dict = None): + """研究证据走弱的减持 (2026-09-07 第三件): 持仓行上的逻辑状态是「逻辑存疑」时, 提议减掉一部分。 + + 这是 reduce_on_weak_research 那条预留通道的第一个上游。三条口径一次定死: + 开关 params["logic_doubt_trim"] (PMS_LOGIC_DOUBT_TRIM), **默认关** —— 先跑停增持一个月看误伤; + 比例 params["logic_doubt_trim_ratio"] (PMS_LOGIC_DOUBT_TRIM_RATIO), 默认与保垫减仓相同的三分之一; + 数量 按 T+1 可卖量夹紧, 不足最小申报数量不产出 (与保垫减仓、到价清仓同口径)。 + 产出的候选带 research_weak 来源与强制确认标记, 分流那一步一票否决入人工队列, 绝不自动卖。 + 没有逻辑状态 (早上没查到) 不产出 —— 「没有读数」不是「证据走弱」。 + """ + if not params.get("logic_doubt_trim"): + return None + st = p.get("logic_state") + if not is_logic_doubt(st): + return None + total = int(p.get("total_qty") or 0) + ratio = _f(params.get("logic_doubt_trim_ratio"), 1.0 / 3) + if ratio <= 0 or ratio > 1: + ratio = 1.0 / 3 + qty = clamp_sell_qty(int(total * ratio // LOT) * LOT, p) + if qty < lot_of(p.get("ts_code")): + return None + why = logic_doubt_text(st, f"研究证据走弱: 拟减 {qty} 股 (持仓 {total} 股的 {ratio:.0%}), 卖不卖由你拍板") + return reduce_on_weak_research( + p["ts_code"], qty=qty, reason=why, + hard_numbers={"logic_state": st, "price": p.get("price"), + "cushion_pct": p.get("cushion_pct"), "total_qty": total, + "avail_qty": p.get("avail_qty"), "trim_ratio": round(ratio, 4)}) + + def clamp_sell_qty(qty: int, pos: dict) -> int: """卖出数量按 T+1 可卖量夹紧。可卖量缺失 (None) 时按总持仓, 不当成零。 @@ -383,7 +455,9 @@ def clamp_sell_qty(qty: int, pos: dict) -> int: return max(0, min(q, int(avail or 0))) -EVALUATORS = ((A_TRIM, eval_trim), (A_EXIT, eval_target), +# 研究走弱的减持排在保垫减仓前面 (2026-09-07 第三件): 两者动作名都是 TRIM, 同轮同现时按 +# _sell_priority 留研究走弱那条 (它必交人), 保垫减仓让路 —— 证据在走弱的票, 不该由规则自动先卖一刀。 +EVALUATORS = ((A_TRIM, eval_weak_research), (A_TRIM, eval_trim), (A_EXIT, eval_target), (A_ADD, eval_add), (A_FILL, eval_fill), (A_DCA, eval_dca)) # 减持方向的动作 —— 冻结只禁增持, 这几类照评 (与规则闸 _check_sell 同一口径)。 @@ -461,6 +535,9 @@ def scan(*, positions: list, params: dict, market: dict, skip=None, if sp: p = {**p, "target_price": sp.get("target_price"), "stop_price": sp.get("stop_price")} frozen = (p.get("frozen_reason") or "NONE") != "NONE" + # 逻辑存疑停增持侧 (2026-09-07 第三件): 持仓行上挂着早上查回的逻辑状态 (logic_state_service.attach), + # 存疑的票补足 / 加仓 / 补仓一律不评, 减持侧照评。没挂这个键 = 没有读数, 不拦。 + doubt = bool(params.get("logic_state_route")) and is_logic_doubt(p.get("logic_state")) # 跨轮的减持让路 (2026-09-07 审查修)。同轮只发一条减持那条规矩 (见下) 只管一次扫描; # 下一分钟再扫时, 等人拍板的到价清仓已成在途、按 (代码, EXIT) 被跳过, 而保垫减仓 # 单独产出、不需确认、卖出方向又不走强制入队 —— 系统刚说「卖不卖由你定」, 一分钟后 @@ -486,6 +563,10 @@ def scan(*, positions: list, params: dict, market: dict, skip=None, skipped.append({"ts_code": code, "action": action, "why": frozen_why(p.get("frozen_reason"))}) continue + if doubt and action not in SELL_SIDE_ACTIONS: + skipped.append({"ts_code": code, "action": action, + "why": logic_doubt_text(p.get("logic_state"), WHY_LOGIC_DOUBT_PAUSE)}) + continue try: c = fn(p, params, mkt) except Exception as e: # 单票异常不能拖垮整轮扫描 @@ -516,7 +597,7 @@ def scan(*, positions: list, params: dict, market: dict, skip=None, # 算出来的减仓前面。被让路的那条记进跳过原因, 下一轮条件仍成立时自然会再来。 sells_now = [c for c in cands_this if c["side"] == SELL] if len(sells_now) > 1: - keep = min(sells_now, key=lambda c: _SELL_PRIORITY.get(c["action"], 99)) + keep = min(sells_now, key=_sell_priority) for c in sells_now: if c is not keep: skipped.append({"ts_code": code, "action": c["action"], @@ -603,6 +684,9 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float): # 研报里说这家公司好在哪、出处是哪份研报哪一天。两者只进评审账本与提议卡给人看, # 不参与任何判定, 也不送研判 —— 产业逻辑归数据基座与选股系统, 不归择时决策系统。 "basis": c.get("basis"), "logic": c.get("logic"), + # 逻辑状态四态 (2026-09-07 第三件): 入场那一刻「研究证据还在不在」的读数, 进账本与提议卡, + # 持仓视图的「当初为什么买」从这里回溯; 不送研判 (judge.OPEN_JUDGE_KEYS 不收它)。 + "logic_state": c.get("logic_state"), # ---- 仓位口径: 只进评审账本做判分锚。judge.py 送研判时会把这几项过滤掉, # 理由见那边的 OPEN_JUDGE_KEYS —— 决策系统本来就不管仓位, 别送过去带偏它。 "target_pct": target_pct, "target_amount": round(full_amt, 2), @@ -721,6 +805,14 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float, # 只是最后一步不许自动落指令 —— 与深档补仓强制确认走的是同一个字段。 cand["needs_user_confirm"] = True cand["confirm_why"] = confirm_why + # 判决候选而逻辑存疑 (2026-09-07 第三件): 可交易口径成立而研究证据在走弱, 这正是杀逻辑与 + # 杀估值那条边界, 强制交人; 复用同一个强制确认字段, 与关注判决走同一条队列。已经要确认的 + # (关注 / 认不出的判决) 保留原因, 不覆盖。开关 PMS_LOGIC_STATE_ROUTE 关着时一行都不执行。 + if params.get("open_route_by_logic") and not cand.get("needs_user_confirm"): + lw = logic_confirm_why(c.get("logic_state")) + if lw: + cand["needs_user_confirm"] = True + cand["confirm_why"] = lw out.append(cand) used = _f(cand.get("target_amount")) left -= used diff --git a/app/repo/pms_repo.py b/app/repo/pms_repo.py index 2380dfc..e1f2318 100644 --- a/app/repo/pms_repo.py +++ b/app/repo/pms_repo.py @@ -673,6 +673,23 @@ def list_ledger(*, ts_code=None, limit: int = 200) -> list: return rows +def ledger_by_ref(ref_ids, limit: int = 50) -> list: + """按 ref_id (指令号或提议号) 取账本行, 按写入顺序 (2026-09-07 第三件: 持仓视图回溯入场论点用)。""" + ids = [str(x) for x in (ref_ids or []) if x] + if not ids: + return [] + keys, p = [], {"n": int(limit)} + for i, x in enumerate(ids): + keys.append(f":r{i}") + p[f"r{i}"] = x + rows = fetch_all(f"SELECT * FROM pms_action_ledger WHERE ref_id IN ({', '.join(keys)}) " + f"ORDER BY id ASC LIMIT :n", p) + for r in rows: + r["hard_numbers"] = _loads(r.get("hard_numbers_json"), {}) + r["failed_checks"] = _loads(r.get("failed_checks_json"), []) + return rows + + # ================================================================ pms_daily_report def upsert_report(ymd: int, report: dict) -> int: return execute( diff --git a/app/scheduler.py b/app/scheduler.py index 020ef07..f17356a 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -159,8 +159,17 @@ def plan_pull(): from app.services import plan_feed plan_feed.invalidate() plan = plan_feed.get_plan(force=True) - return {"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"), - "returned": plan["returned"], "theme_sync": plan.get("theme_sync")} + out = {"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"), + "returned": plan["returned"], "theme_sync": plan.get("theme_sync")} + # 在持票的逻辑状态 (2026-09-07 第三件): 拉完计划顺带查一次, 写进运行参数并按结果停或恢复策略 + # 买入腿。失败只记录, 不影响拉计划 —— 取不到就留空, 扫描层按没有读数处理。 + try: + from app.services import logic_state_service + out["logic_state"] = logic_state_service.pull_for_held() + except Exception as e: # noqa: BLE001 + logger.error("[plan_pull] 逻辑状态取回失败: %s", e) + out["logic_state"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} + return out @celery_app.task(name="pms.premarket") diff --git a/app/services/logic_state_service.py b/app/services/logic_state_service.py new file mode 100644 index 0000000..1afcf70 --- /dev/null +++ b/app/services/logic_state_service.py @@ -0,0 +1,239 @@ +# -*- 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 diff --git a/app/services/param_store.py b/app/services/param_store.py index df8c824..6c9c27d 100644 --- a/app/services/param_store.py +++ b/app/services/param_store.py @@ -67,6 +67,12 @@ RUNTIME_EXTRA = { "PMS_PUBLISH_ACCOUNT": ("天盟实业", str, "公示表·开单账户列的账户名"), "PMS_PUBLISH_STRUCTURE": ("二级（平层）", str, "公示表·结构列的产品结构标签"), "PMS_PUBLISH_NAV_SCALE": (0.0, float, "公示净值规模 (元): 净值=1+累计盈亏/此数; 0=用 PMS_TOTAL_SCALE"), + # 2026-09-07 第三件: 选股系统的逻辑状态四态在 PMS 的三个开关与一份运行态映射 (模块见 + # logic_state_service.py)。映射与 PMS_MACRO_GATE_STATE 同一手法: 早上写一次, 扫描层与页面只读。 + "PMS_LOGIC_STATE_ROUTE": (True, bool, "按选股系统的逻辑状态分流: 逻辑存疑的持仓停增持侧自主动作并暂停策略买入腿, 判决候选而逻辑存疑的新建仓强制人工确认 (默认开; 关掉即回到只按判决)"), + "PMS_LOGIC_DOUBT_TRIM": (False, bool, "逻辑存疑的持仓出减持提议 (必定交人裁决, 绝不自动卖; 默认关, 先跑停增持一个月看误伤再开)"), + "PMS_LOGIC_DOUBT_TRIM_RATIO": (0.3333, float, "逻辑存疑减持提议的比例 (占总持仓; 默认与保垫减仓相同的三分之一; 一次定死, 不按复盘读数回调)"), + "PMS_LOGIC_STATE_MAP": ("", str, "在持票的逻辑状态映射 (plan_pull 每早写入的 JSON: 时刻/日期/逐票状态), 扫描层与页面只读, 勿手改"), } # **读不到时必须按"已暂停"处理的键 (fail-closed)。** diff --git a/app/services/plan_feed.py b/app/services/plan_feed.py index 2ccd402..e3d7ab6 100644 --- a/app/services/plan_feed.py +++ b/app/services/plan_feed.py @@ -145,6 +145,27 @@ REASONS_KEEP = 4 # 第一条, 其余折叠, 3 条正好是「折开也还看得完」的量。 LOGIC_KEEP = 3 +# 逻辑状态四态 (2026-09-07 第三件): 上游每行带一个字典 —— state 是落定态 (下游按它分流), raw_state +# 是当天原始态, why 是无法判断的子因 (证据不足 / 证据矛盾), settle_note 是抗抖动的说明, as_of 是 +# 证据截止日, usable / missing 是四路里哪几路在、哪几路缺, reasons 是每路的一句话出处。 +# 只收这几个键; 出处最多带三条 (与因果论断同一个量级, 提议卡与账本放得下)。 +LOGIC_STATE_REASONS_KEEP = 3 + + +def _logic_state_or_none(v): + """逻辑状态归一: 不是字典或没有 state 一律 None (旧版计划没有这个字段)。""" + if not isinstance(v, dict): + return None + state = _text_or_none(v.get("state")) + if not state: + return None + return {"state": state, "raw_state": _text_or_none(v.get("raw_state")) or state, + "why": _text_or_none(v.get("why")), "settle_note": _text_or_none(v.get("settle_note")), + "as_of": _text_or_none(v.get("as_of")), + "usable": _list_or_none(v.get("usable")) or [], + "missing": _list_or_none(v.get("missing")) or [], + "reasons": _list_or_none(v.get("reasons"), limit=LOGIC_STATE_REASONS_KEEP) or []} + def _list_or_none(v, limit: int = 0): """字符串列表归一: 列表逐项去空白、丢空项, 单个字符串当一项; 缺失或类型不对一律 None。 @@ -199,6 +220,8 @@ def _rows(raw, bucket: str) -> list: # 研究理由 (2026-09-03): 只给人看, 不参与任何判定, 也不送研判。见模块说明。 "basis": _text_or_none(it.get("basis")), "logic": _list_or_none(it.get("logic"), limit=LOGIC_KEEP), + # 逻辑状态四态 (2026-09-07 第三件): 跟着候选进硬数字, 不送研判 (judge.OPEN_JUDGE_KEYS 不收)。 + "logic_state": _logic_state_or_none(it.get("logic_state")), }) return out @@ -387,7 +410,10 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N "card_rank": r.get("card_rank"), # 研究理由两键: 跟着候选一路带到提议卡与评审账本, 让人看得到研报说了什么。 # 它们不进送研判的那份名单 (judge.OPEN_JUDGE_KEYS), 产业逻辑不归择时决策系统判。 - "basis": r.get("basis"), "logic": r.get("logic")} + "basis": r.get("basis"), "logic": r.get("logic"), + # 逻辑状态 (2026-09-07 第三件): 动作引擎放进硬数字; 判决候选而逻辑存疑的在 scan_open + # 里强制交人。它不改判决、不改排序 —— 收敛规则是单调的 (选股系统 logic_state.apply_to_card)。 + "logic_state": r.get("logic_state")} for r in passed[:n]] return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed), "items": items, "dropped": dropped, "st_unknown": st_unknown, @@ -567,6 +593,52 @@ def fetch(*, date=None, base=None, path=None, timeout=None, extra_params=None) - return plan +def parse_logic_states(payload) -> dict: + """按代码查询接口的应答 → {ts_code: 逻辑状态}。带 error 的行跳过 (代码形态认不出), 查不到的不在结果里。 + + 上游按我们送去的原样 (input) 回, 代码用它映射回 PMS 的形态; 每条另带 source (daily = 早上落定的 + 当日行, computed = 此刻现算) 与日期, 页面要区分「早上的态」与「现算的态」。""" + out = {} + day = _text_or_none((payload or {}).get("date")) if isinstance(payload, dict) else None + for it in ((payload or {}).get("states") or []) if isinstance(payload, dict) else []: + if not isinstance(it, dict) or it.get("error"): + continue + code = normalize_code(str(it.get("input") or it.get("code") or "")) + st = _logic_state_or_none(it) + if not code or not st: + continue + st["source"] = _text_or_none(it.get("source")) + st["date"] = day + out[code] = st + return out + + +def fetch_logic_states(codes, *, base=None, timeout=None, path: str = "/logic_state") -> dict: + """按代码向选股系统查逐票逻辑状态 (2026-09-07 第三件, 给在持的票用)。任何失败抛 PlanFeedError, + 由调用方决定留空 —— 这里不吞错, 因为「查不到」与「判为成立」在扫描层是两件事。""" + codes = [normalize_code(str(c)) for c in (codes or []) if c] + if not codes: + return {} + if base is None or timeout is None: + p = _params() + base = p["base"] if base is None else base + timeout = p["timeout"] if timeout is None else timeout + base = (base or "").strip().rstrip("/") + if not base: + raise PlanFeedError(_NO_BASE_WHY) + url = base + (path if path.startswith("/") else "/" + path) + to = int(timeout or 10) + try: + import requests + r = requests.get(url, params={"codes": ",".join(codes)}, timeout=to) + r.raise_for_status() + payload = r.json() + except Exception as e: + logger.warning("查逻辑状态失败 %s: %s: %s", url, type(e).__name__, e) + raise PlanFeedError(_fetch_fail_why(e, to)) from e + return parse_logic_states(payload) + + def get_plan(*, force: bool = False, date=None) -> dict: """带缓存的当前计划。失败同样缓存 FAIL_CACHE_SEC, 但每次调用都照样抛。""" p = _params() diff --git a/app/services/proposal_service.py b/app/services/proposal_service.py index c802def..9d55211 100644 --- a/app/services/proposal_service.py +++ b/app/services/proposal_service.py @@ -104,6 +104,13 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict: strategy_codes = pms_repo.active_strategy_codes() except Exception: strategy_codes = set() + # 逻辑状态挂到持仓行上 (2026-09-07 第三件): 早上拉计划时查回的当日态。读不到就没有这个键, + # 动作引擎按「没有读数」处理, 绝不折成存疑; 挂不上只记日志, 不拖垮扫描。 + try: + from app.services import logic_state_service + logic_state_service.attach(view["held"]) + except Exception as e: # noqa: BLE001 + logger.warning("[逻辑状态] 挂到持仓行失败, 本轮按没有读数: %s", e) scanned = ae.scan(positions=view["held"], params=params, market=mkt, skip=skip, strategy_codes=strategy_codes, stock_params=stock_params) except Exception as e: @@ -766,6 +773,12 @@ def _scan_params(view: dict) -> dict: "fill_max_loss": param_store.get_float("PMS_FILL_MAX_LOSS", -0.03), "open_signal_priority": param_store.get_bool("PMS_OPEN_SIGNAL_PRIORITY", True), "open_route_by_verdict": param_store.get_bool("PMS_PLAN_ROUTE_BY_VERDICT", True), + # 逻辑状态四态的三个旋钮 (2026-09-07 第三件): 分流开关 (持仓停增持 + 新建仓强制确认)、 + # 减持提议开关 (默认关)、减持比例 (默认与保垫减仓相同的三分之一)。 + "logic_state_route": param_store.get_bool("PMS_LOGIC_STATE_ROUTE", True), + "open_route_by_logic": param_store.get_bool("PMS_LOGIC_STATE_ROUTE", True), + "logic_doubt_trim": param_store.get_bool("PMS_LOGIC_DOUBT_TRIM", False), + "logic_doubt_trim_ratio": param_store.get_float("PMS_LOGIC_DOUBT_TRIM_RATIO", 1.0 / 3), }) return p diff --git a/app/web/main.py b/app/web/main.py index dfe1c10..be95f68 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -25,7 +25,8 @@ from app.core import command_spec as cs from app.core import tradedays as td from app.db import session as dbs from app.repo import downstream_repo, pms_repo -from app.services import command_service, industry, ledger_service, param_store, portfolio +from app.services import (command_service, industry, ledger_service, logic_state_service, + param_store, portfolio) from app.web import auth as authmod logging.basicConfig(level=logging.INFO, @@ -374,6 +375,12 @@ def api_positions(): # 把两个数分列显示 —— 只是显示, 判断那一侧照旧各读各的事实源, 见 # command_service.attach_user_prices 的说明。 command_service.attach_user_prices(v["positions"], sp) + # 入场论点随持仓走 (2026-09-07 第三件): 每行加「当初为什么买」(entry) 与「现在证据还在不在」 + # (logic_now)。只是显示; 任何一行取不到都写原因, 页面不能因此塌掉。 + try: + logic_state_service.decorate_positions(v["positions"]) + except Exception as e: # noqa: BLE001 + logger.warning("[持仓] 入场论点与逻辑状态两栏取不到: %s", e) return {"ok": True, **v, "stock_params": sp} return ok(_view) diff --git a/app/web/static/index.html b/app/web/static/index.html index 8d74e77..8264ba3 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -1567,6 +1567,32 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;} 未设 + + + + + + +