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;} 未设 + + + + + {{ s.row.entry.basis }} + {{ s.row.entry.logic[0] }} + {{ s.row.entry.decided_at || '' }} + · 判决 {{ s.row.entry.verdict }} + · 入场时 {{ s.row.entry.logic_state_at_entry }} + + {{ (s.row.entry && s.row.entry.why) || '—' }} + + + + + + {{ s.row.logic_now.state }} + {{ s.row.logic_now.text }} + 证据截止 {{ s.row.logic_now.as_of || '—' }};{{ s.row.logic_now.settle_note }} + + {{ (s.row.logic_now && s.row.logic_now.text) || '今天没有读数' }} + + @@ -2194,6 +2220,8 @@ createApp({ const stTag = s => ({ EXECUTING: 'primary', DONE: 'success', PARTIAL: 'warning', CANCELLED: 'info', EFFECTIVE: 'success', SUPERSEDED: 'info' }[s] || 'warning'); const cuTag = s => ({ SOLID: 'info', THIN: 'warning', NONE: 'warning' }[s] || 'info'); + // 逻辑状态四态的标签色 (上游给的中文原文, 直接当键): 存疑红、强化与成立绿、无法判断灰。 + const logicTag = s => ({ '逻辑存疑': 'danger', '逻辑强化': 'success', '逻辑成立': 'success', '无法判断': 'info' }[s] || 'info'); const canCancel = r => r.cmd_class === 'task' && ['PENDING', 'PLANNING', 'EXECUTING', 'PARTIAL'].includes(r.status); const progPct = p => { @@ -3577,7 +3605,7 @@ createApp({ return { tab, loading, err, health, ov, params, catalog, commands, plans, plansOf, positions, lots, lotsOf, instructions, ledger, ledgerToday, proposals, report, reportDrawer, opsDrawer, opsResult, opsLoading, issuing, form, curSpec, dirtyCount, dm, - money, pct, groupLabel, fieldLabel, stTag, cuTag, canCancel, progPct, insPct, + money, pct, groupLabel, fieldLabel, stTag, cuTag, logicTag, canCancel, progPct, insPct, scaleGap, wsRaw, wsAuto, wsHidePong, wsc, wst, wsMode, wsOrders, wsInbox, wsWarnings, loadWs, loadAll, loadParams, saveParams, loadPlans, loadLots, onCmdChange, issue, diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 82cb390..beda445 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -56,15 +56,19 @@ 强制入队对卖出同样有效/研判不可用的减持入队/研究证据走弱 来源必定交人裁决/风控高置信卖出与命令清仓不经提议分流/ 到价提议挂着时止损照落并作废该提议 (17 例) + test_batch22_units.py 逻辑状态四态接入 (2026-09-07 第三件): 解析层收逻辑状态/硬数字带它而研判 + 白名单不收/判决候选而逻辑存疑强制交人/持仓存疑停增持侧/研究走弱 + 减持默认关且开了必交人/同轮减持优先级/早上取回与映射新鲜度/策略 + 买入腿按来源暂停恢复/持仓视图两栏三情形/假仓库签名 (14 例) test_page_enum_guard.py 页面文案守卫 (静态扫描, 不连库不起浏览器): 枚举字段不许 直接印到页面上 / 判据码显示前必须剥前缀 / 不许把整个对象 打给交易员看 / 翻译兜底不许让英文码单独当句子 (1 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) + 目标价到价必定入队 (档位 full 也不自动卖) + 用户设的止损价与目标价单独成列显示 (70 例) - 共 661 例 + 共 675 例 (总数按实跑逐批相加校正过两次: 曾写 649 是笔误, 实为 650; 09-03 先后加了同轮只发一条 - 减持与研究理由两键各一例, 到 652; 09-04 加了仅展示跳过原因与空候选说明各一例, 到 654; 又加了页面文案守卫一例, 到 655; 09-07 审查修复加了跨轮减持等五例, 到 660; 第二件低把握驳回交人一例, 现为 661) + 减持与研究理由两键各一例, 到 652; 09-04 加了仅展示跳过原因与空候选说明各一例, 到 654; 又加了页面文案守卫一例, 到 655; 09-07 审查修复加了跨轮减持等五例, 到 660; 第二件低把握驳回交人一例, 到 661; 第三件逻辑状态接入第二十二批十四例, 现为 675) 任一子集失败即整体失败 (退出码 1)。 哨兵位置清单 (2026-09-03 抄录; 改了对应的东西就得来这些地方改断言, 断言不动就是漏了): @@ -106,7 +110,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", "test_batch13_units.py", "test_batch14_units.py", "test_batch15_units.py", "test_batch16_units.py", "test_batch17_units.py", "test_batch18_units.py", "test_batch19_units.py", "test_batch20_units.py", - "test_batch21_units.py", + "test_batch21_units.py", "test_batch22_units.py", "test_page_enum_guard.py", "test_wiring.py"] diff --git a/scripts/test_batch12_units.py b/scripts/test_batch12_units.py index 322d71b..9992e55 100644 --- a/scripts/test_batch12_units.py +++ b/scripts/test_batch12_units.py @@ -367,7 +367,11 @@ def _(): # 2026-09-03 接通「设定某股目标价」后多了一条 EXIT (到价产出清仓提议)。它排在 TRIM 之后、 # 三类买入之前 —— 顺序有意义: 同轮买卖互斥要先看到减持才让买入让路。四类老动作的相对 # 次序与判据一个字没动, 减持照旧不送研判。 - assert [a for a, _ in ae.EVALUATORS] == ["TRIM", "EXIT", "ADD", "FILL", "DCA"] + # 2026-09-07 第三件: 研究走弱的减持 (eval_weak_research, 动作名同为 TRIM) 排在保垫减仓前面, + # 其余五个的名单与次序一个字不动。 + assert [a for a, _ in ae.EVALUATORS] == ["TRIM", "TRIM", "EXIT", "ADD", "FILL", "DCA"] + assert [f.__name__ for _, f in ae.EVALUATORS] == [ + "eval_weak_research", "eval_trim", "eval_target", "eval_add", "eval_fill", "eval_dca"] assert "EXIT" not in ae.JUDGE_ACTIONS, "减持永远不送研判 (目标价到价也是减持)" # 持仓那条扫描的输入与产出一个字没改: 空持仓进去, 空候选出来, 不会去碰候选池 r = ae.scan(positions=[], params=params(), market={}) diff --git a/scripts/test_batch22_units.py b/scripts/test_batch22_units.py new file mode 100644 index 0000000..03da209 --- /dev/null +++ b/scripts/test_batch22_units.py @@ -0,0 +1,455 @@ +# -*- coding: utf-8 -*- +""" +第二十二批模块单测 (逻辑状态四态接入 PMS, 零外部依赖, 不连库不触网) +================================================================== +运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch22_units.py + +背景 (2026-09-07 下一阶段方案第三件): 选股系统每早给每只票算一个「支撑它的研究证据还在不在」的 +状态 (逻辑强化 / 逻辑成立 / 无法判断 / 逻辑存疑), 随计划每行下发, 另有按代码查询的接口给持仓票用。 +此前 PMS 收到了却没人读; 持仓不绑入场论点; 「逻辑不证伪不退出」没有判据。本批钉住: + + * 解析层收逻辑状态, 硬数字带它, 但送研判的白名单不收 (一致性检查表第九行: 择时层不判产业逻辑); + * 判决候选而逻辑存疑的新建仓强制人工确认 (复用强制确认字段, 与关注判决同一条队列); + * 逻辑存疑的持仓停增持侧自主动作, 减持侧照评; 开关关着一行都不执行; 没有读数不拦; + * 研究走弱的减持: 默认关, 开了也必定交人 (research_weak 来源一票否决), 数量按可卖量夹紧; + * 同轮只发一条减持的优先级: 到价清仓 > 研究走弱 > 保垫减仓; + * 早上取回: 取不到写空映射带原因、绝不折成存疑、不动已有暂停; 映射超龄按没有读数; + * 策略买入腿按来源暂停与恢复, 不动风控与定性停的; + * 持仓视图两栏: 有指令链 / 外部成交并入 / 账本无行 三种情形都说得出话, 仓库读失败页面不塌。 +约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。 +""" +import os +import sys +import traceback +from datetime import date, datetime, timedelta + +_HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.dirname(_HERE)) +sys.path.insert(0, _HERE) + +from app.core import action_engine as ae # noqa: E402 +from app.services import judge as jd # noqa: E402 +from app.services import plan_feed as pf # noqa: E402 +from app.services import logic_state_service as lss # noqa: E402 + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +NOW = datetime(2026, 9, 8, 10, 30) +CODE = "600000.SH" +TODAY = date.today().isoformat() + + +# ================================================================ 夹具 +class _Patch: + """临时替换若干模块属性, 退出时原样还回去 —— 单测不连库不触网。""" + + def __init__(self): + self.saved = [] + + def __call__(self, mod, name, value): + self.saved.append((mod, name, getattr(mod, name))) + setattr(mod, name, value) + + def __enter__(self): + return self + + def __exit__(self, *a): + for mod, name, old in reversed(self.saved): + setattr(mod, name, old) + return False + + +def _patch_params(p, mod, values): + v = values + + def _get_list(k, d=None, sep=","): + x = v.get(k) + if x in (None, ""): + return list(d or []) + return [s.strip() for s in str(x).split(sep) if s.strip()] + p(mod, "get", lambda k, d=None: v.get(k, d)) + p(mod, "get_int", lambda k, d=0: int(v.get(k, d))) + p(mod, "get_float", lambda k, d=0.0: float(v.get(k, d))) + p(mod, "get_bool", lambda k, d=False: bool(v.get(k, d))) + p(mod, "get_list", _get_list) + + +def doubt(as_of="2026-09-04"): + return {"state": "逻辑存疑", "raw_state": "逻辑存疑", "why": None, "settle_note": "进入逻辑存疑即刻成立", + "as_of": as_of, "usable": ["券商行动"], "missing": ["研报论断", "产业研判", "公司事件"], + "reasons": ["券商行动的转弱达到进入条件:每股收益预测下修且覆盖机构收缩", + "券商行动(2026-09-04):每股收益预测中位数下修 40%,同时覆盖机构从 5 家收缩到 3 家"]} + + +def hold(): + return {"state": "逻辑成立", "raw_state": "逻辑成立", "why": None, "settle_note": "维持", + "as_of": "2026-09-04", "usable": ["研报论断"], "missing": ["产业研判", "券商行动", "公司事件"], + "reasons": ["研报论断(2026-08-25):最近一条利好"]} + + +def pos(code=CODE, **kw): + p = {"ts_code": code, "total_qty": 3000, "avail_qty": 3000, "price": 10.0, "price_ok": True, + "frozen_reason": "NONE", "cushion_pct": 0.02, "cushion_peak": 0.03, "avg_cost": 9.8} + p.update(kw) + return p + + +SCAN_PARAMS = {"trim_peak": 0.06, "trim_giveback": 0.5, "logic_state_route": True, + "logic_doubt_trim": False, "logic_doubt_trim_ratio": 1.0 / 3, "scale": 1_000_000} + +PARAMS_BASE = {"PMS_EXEC_WINDOW_TDAYS": 3, "PMS_PROPOSAL_TTL_HOURS": 24, + "PMS_JUDGE_TICK_BUDGET_SEC": 150} +JUDGE_PASS = {"verdict": "PASS", "reason": "理由仍成立", "degraded": False, + "raw": {"verdict": "PASS"}, "confidence": 70.0} + + +def _route(c, *, autonomy="propose_only", judge_resp=JUDGE_PASS, dry_run=False, values=None, + price=10.0): + """把一条候选送进 proposal_service._route_one, 规则闸/研判/落表全换成桩 (口径同第二十一批)。""" + from app.core import rule_gate + from app.repo import pms_repo + from app.services import judge, param_store, portfolio, proposal_service as psvc + got = {"ledger": [], "instructions": [], "proposals": [], "judge_calls": []} + with _Patch() as p: + p(rule_gate, "check", lambda **kw: {"passed": True, "failed": [], "warnings": []}) + p(judge, "request", lambda cand_, context=None, **kw: ( + got["judge_calls"].append(cand_.get("action")) or dict(judge_resp))) + p(portfolio, "caps_ctx", lambda *a, **kw: {}) + p(pms_repo, "insert_ledger", lambda **kw: got["ledger"].append(kw) or 1) + p(pms_repo, "insert_instruction", lambda **kw: got["instructions"].append(kw) or 1) + p(pms_repo, "insert_proposal", lambda **kw: got["proposals"].append(kw) or 1) + p(pms_repo, "list_ledger", lambda **kw: []) + p(pms_repo, "update_position", lambda code, **kw: 1) + _patch_params(p, param_store, {**PARAMS_BASE, **(values or {})}) + out = {"autonomy": autonomy, "open_autonomy": autonomy, "executed": [], "queued": [], + "rejected": [], "skipped": [], "errors": [], "degraded": False} + held = [{"ts_code": c["ts_code"], "total_qty": 3000, "avail_qty": 3000, + "price": price, "frozen_reason": "NONE"}] + view = {"positions": held, "held": held, "sector_ready": False, + "params": {}, "totals": {}} + psvc._route_one(c, view, {"_mkt": {}}, {}, False, NOW, dry_run, out) + return out, got + + +# ================================================================ 一, 解析层与硬数字 +@case("解析层·逻辑状态按约定的键归一, 出处最多三条; 缺键 / 非字典 / 没有 state 一律 None") +def _(): + raw = {"state": "逻辑存疑", "raw_state": "逻辑存疑", "why": None, "settle_note": "进入逻辑存疑即刻成立", + "as_of": "2026-09-04", "usable": ["券商行动"], "missing": ["研报论断", "公司事件"], + "reasons": ["a", "b", "c", "d"], "paths": [{"path": "x"}], "prev_state": "逻辑成立"} + st = pf._logic_state_or_none(raw) + assert st["state"] == "逻辑存疑" and st["as_of"] == "2026-09-04", st + assert st["reasons"] == ["a", "b", "c"] and st["usable"] == ["券商行动"], st + assert "paths" not in st and "prev_state" not in st, st # 只收约定的键 + assert pf._logic_state_or_none(None) is None and pf._logic_state_or_none("逻辑存疑") is None + assert pf._logic_state_or_none({"why": "x"}) is None + # 没有原始态的旧形状: 原始态退回落定态 + assert pf._logic_state_or_none({"state": "逻辑成立"})["raw_state"] == "逻辑成立" + row = pf._rows([{"code": "600000.SH", "rank": 1, "score": 210, "logic_state": raw}], "main")[0] + assert row["logic_state"]["state"] == "逻辑存疑", row + row2 = pf._rows([{"code": "600000.SH", "rank": 1, "score": 210}], "main")[0] + assert row2["logic_state"] is None, row2 + + +@case("解析层·按代码查询接口的应答映射回 PMS 的代码形态, 带 error 的行跳过, source 与日期带上") +def _(): + payload = {"date": "2026-09-04", "count": 3, "states": [ + {"input": "600000.SH", "code": "SH600000", "source": "daily", **doubt()}, + {"input": "300750", "code": "SZ300750", "source": "computed", **hold()}, + {"input": "abc", "error": "认不出的代码形态"}]} + m = pf.parse_logic_states(payload) + assert set(m) == {"600000.SH", "300750.SZ"}, m + assert m["600000.SH"]["state"] == "逻辑存疑" and m["600000.SH"]["source"] == "daily" + assert m["300750.SZ"]["source"] == "computed" and m["300750.SZ"]["date"] == "2026-09-04" + assert pf.parse_logic_states(None) == {} and pf.parse_logic_states({"states": "x"}) == {} + assert pf.fetch_logic_states([]) == {} # 没票不触网 + + +@case("候选筛选·items 带 logic_state (缺就是 None), 不改资格也不改次序") +def _(): + plan = {"date": "2026-09-04", "main": pf._rows([ + {"code": "600000.SH", "rank": 1, "score": 220, "verdict": "候选", "logic_state": doubt()}, + {"code": "300750.SZ", "rank": 2, "score": 210, "verdict": "候选"}], "main"), "observe": []} + sel = pf.select_candidates(plan, top_n=10, route_by_verdict=True) + items = sel["items"] + assert [x["ts_code"] for x in items] == ["600000.SH", "300750.SZ"], items + assert items[0]["logic_state"]["state"] == "逻辑存疑" and items[1]["logic_state"] is None + + +@case("硬数字带逻辑状态, 而送研判的白名单不收它 (择时层不判产业逻辑, 检查表第九行)") +def _(): + c = {"ts_code": CODE, "price": 10.0, "score": 220, "rank": 1, "bucket": "main", "src": "plan_api", + "verdict": "候选", "basis": "三门槛全过", "logic": ["研报说好"], "logic_state": doubt()} + params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25)} + with _Patch() as p: + p(ae, "check_all_caps", lambda **kw: []) + p(ae, "_new_name_ctx", lambda caps, c: caps) + cand, why = ae.eval_open(c, params, {"names_count": 0, "max_names": 10}, 500_000) + assert cand and not why, why + hn = cand["hard_numbers"] + assert hn["logic_state"]["state"] == "逻辑存疑" and hn["basis"] == "三门槛全过", hn + sent = jd._judge_hard_numbers("OPEN", hn) + assert "logic_state" not in sent and "basis" not in sent and "verdict" in sent, sent + assert "logic_state" not in jd.OPEN_JUDGE_KEYS + + +# ================================================================ 二, 新建仓与持仓的分流 +def _scan_open(cands, *, route_logic=True): + params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25), + "open_route_by_verdict": True, "open_route_by_logic": route_logic, + "open_signal_priority": False} + with _Patch() as p: + p(ae, "check_all_caps", lambda **kw: []) + p(ae, "_new_name_ctx", lambda caps, c: caps) + p(ae, "_ctx_after", lambda ctx, *a, **kw: ctx) + return ae.scan_open(candidates=cands, params=params, + caps={"names_count": 0, "max_names": 10}, room_amt=900_000, slots=5) + + +@case("新建仓·判决候选而逻辑存疑 → 强制人工确认并写明出处; 关注保留自己的原因; 成立不强制; 开关关不强制") +def _(): + base = {"price": 10.0, "score": 220, "rank": 1, "bucket": "main", "src": "plan_api"} + cands = [{**base, "ts_code": "600000.SH", "verdict": "候选", "logic_state": doubt()}, + {**base, "ts_code": "600001.SH", "verdict": "关注", "logic_state": doubt()}, + {**base, "ts_code": "600002.SH", "verdict": "候选", "logic_state": hold()}, + {**base, "ts_code": "600003.SH", "verdict": "候选"}] + r = _scan_open(cands) + by = {c["ts_code"]: c for c in r["candidates"]} + assert set(by) == {"600000.SH", "600001.SH", "600002.SH", "600003.SH"}, r["skipped"] + a = by["600000.SH"] + assert a["needs_user_confirm"] and "逻辑存疑" in a["confirm_why"] and "证据截止 2026-09-04" in a["confirm_why"], a + assert "每股收益预测下修" in a["confirm_why"], a["confirm_why"] + b = by["600001.SH"] + assert b["needs_user_confirm"] and b["confirm_why"] == ae.WHY_WATCH_CONFIRM, b + assert not by["600002.SH"].get("needs_user_confirm") and not by["600003.SH"].get("needs_user_confirm") + r2 = _scan_open(cands, route_logic=False) + by2 = {c["ts_code"]: c for c in r2["candidates"]} + assert not by2["600000.SH"].get("needs_user_confirm"), by2["600000.SH"] + assert ae.logic_confirm_why(hold()) is None and ae.logic_confirm_why(None) is None + + +@case("持仓·逻辑存疑停增持侧 (补足/加仓/补仓写明原因), 减持侧照评; 开关关不停; 没有读数不停") +def _(): + p = pos(logic_state=doubt(), cushion_peak=0.10, cushion_pct=0.04) # 保垫回吐过半 → TRIM 照评 + r = ae.scan(positions=[p], params=SCAN_PARAMS, market={CODE: {}}, skip={}) + acts = {c["action"] for c in r["candidates"]} + assert acts == {"TRIM"}, r["candidates"] + stopped = {s["action"]: s["why"] for s in r["skipped"] if "逻辑存疑" in s["why"]} + assert set(stopped) == {"ADD", "FILL", "DCA"}, r["skipped"] + assert all("停掉增持侧" in w and "证据截止 2026-09-04" in w for w in stopped.values()), stopped + r2 = ae.scan(positions=[pos(logic_state=doubt())], params={**SCAN_PARAMS, "logic_state_route": False}, + market={CODE: {}}, skip={}) + assert not [s for s in r2["skipped"] if "逻辑存疑" in s["why"]], r2["skipped"] + r3 = ae.scan(positions=[pos()], params=SCAN_PARAMS, market={CODE: {}}, skip={}) + assert not [s for s in r3["skipped"] if "逻辑存疑" in s["why"]], r3["skipped"] + + +@case("研究走弱的减持·默认关不产出; 开了按三分之一取整并按可卖量夹紧, 来源 research_weak 且强制确认; 不足一手不产出") +def _(): + p = pos(logic_state=doubt()) + assert ae.eval_weak_research(p, SCAN_PARAMS) is None # 开关关 + on = {**SCAN_PARAMS, "logic_doubt_trim": True} + c = ae.eval_weak_research(p, on) + assert c and c["action"] == "TRIM" and c["side"] == "sell" and c["qty"] == 1000, c + assert c["source"] == ae.SRC_RESEARCH_WEAK and c["needs_user_confirm"] and c["confirm_why"], c + assert "拟减 1000 股" in c["reason"] and "证据截止 2026-09-04" in c["reason"], c["reason"] + assert c["hard_numbers"]["logic_state"]["state"] == "逻辑存疑" and c["hard_numbers"]["trim_ratio"] == 0.3333 + c2 = ae.eval_weak_research(pos(logic_state=doubt(), avail_qty=400), on) + assert c2["qty"] == 400, c2 # 夹到可卖量 + assert ae.eval_weak_research(pos(logic_state=doubt(), total_qty=200, avail_qty=200), on) is None + assert ae.eval_weak_research(pos(logic_state=hold()), on) is None + assert ae.eval_weak_research(pos(), on) is None # 没有读数 + c3 = ae.eval_weak_research(p, {**on, "logic_doubt_trim_ratio": 0.5}) + assert c3["qty"] == 1500, c3 + + +@case("同轮只发一条减持·优先级 到价清仓 > 研究走弱 > 保垫减仓, 让路的记进跳过原因") +def _(): + on = {**SCAN_PARAMS, "logic_doubt_trim": True} + p = pos(logic_state=doubt(), cushion_peak=0.10, cushion_pct=0.04) + r = ae.scan(positions=[p], params=on, market={CODE: {}}, skip={}, + stock_params={CODE: {"target_price": 9.5}}) + assert [c["action"] for c in r["candidates"]] == ["EXIT"], r["candidates"] + yielded = [s for s in r["skipped"] if "一轮只发一条减持" in s["why"]] + assert len(yielded) == 2 and all(s["action"] == "TRIM" for s in yielded), yielded + r2 = ae.scan(positions=[p], params=on, market={CODE: {}}, skip={}) + assert len(r2["candidates"]) == 1 and r2["candidates"][0]["source"] == ae.SRC_RESEARCH_WEAK, r2 + assert ae._sell_priority({"action": "EXIT"}) < ae._sell_priority({"action": "TRIM", "source": "research_weak"}) \ + < ae._sell_priority({"action": "TRIM", "source": "engine"}) + + +@case("分流·研究走弱的减持在 full 档也入人工队列, 不落指令, 提议带来源与逻辑状态") +def _(): + c = ae.eval_weak_research(pos(logic_state=doubt()), {**SCAN_PARAMS, "logic_doubt_trim": True}) + for autonomy in ("propose_only", "full"): + out, got = _route(c, autonomy=autonomy) + assert not out["executed"] and len(out["queued"]) == 1, (autonomy, out) + assert "研究证据走弱" in out["queued"][0]["why"], out["queued"] + assert not got["instructions"] and len(got["proposals"]) == 1, (autonomy, got) + hn = got["proposals"][0]["hard_numbers"] + assert hn["source"] == ae.SRC_RESEARCH_WEAK and hn["logic_state"]["state"] == "逻辑存疑", hn + + +# ================================================================ 三, 早上取回与策略买入腿 +@case("映射·超过三个自然日没刷新按没有读数; 时刻缺失或坏了也按没有读数; attach 只给有读数的行挂键") +def _(): + fresh = {"at": f"{TODAY} 08:41:00", "states": {CODE: doubt()}} + assert lss.state_map(fresh) == {CODE: doubt()} + old = {"at": (date.today() - timedelta(days=5)).isoformat() + " 08:41:00", "states": {CODE: doubt()}} + assert lss.state_map(old) == {} + assert lss.state_map({"states": {CODE: doubt()}}) == {} and lss.state_map({}) == {} + rows = [pos(), pos("300750.SZ", logic_state="陈旧的")] + lss.attach(rows, {CODE: doubt()}) + assert rows[0]["logic_state"]["state"] == "逻辑存疑" and "logic_state" not in rows[1], rows + + +@case("策略买入腿·存疑按来源 logic 暂停; 明确不存疑只清本来源; 没读数不动; 开关关整段跳过") +def _(): + from app.services import param_store, strategy_service + calls = {"pause": [], "clear": []} + with _Patch() as p: + p(strategy_service, "pause_buy", lambda code, *, reason="", source="signal": ( + calls["pause"].append((code, source, reason)) or ["S1"])) + p(strategy_service, "clear_buypause", lambda code, only_source=None: ( + calls["clear"].append((code, only_source)) or {"ok": True, "cleared": code == "600001.SH"})) + _patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": True}) + r = lss.apply_pauses({CODE: doubt(), "600001.SH": hold(), "600002.SH": hold()}, + [CODE, "600001.SH", "600002.SH", "600003.SH"]) + assert r["paused"] == [CODE] and r["resumed"] == ["600001.SH"] and not r["errors"], r + assert calls["pause"] == [(CODE, "logic", calls["pause"][0][2])] and "停掉增持侧" in calls["pause"][0][2] + assert [c for c, _ in calls["clear"]] == ["600001.SH", "600002.SH"], calls # 没读数的 600003 不动 + assert all(s == "logic" for _, s in calls["clear"]), calls + with _Patch() as p: + p(strategy_service, "pause_buy", lambda *a, **k: (_ for _ in ()).throw(AssertionError("不该调"))) + p(strategy_service, "clear_buypause", lambda *a, **k: (_ for _ in ()).throw(AssertionError("不该调"))) + _patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": False}) + r2 = lss.apply_pauses({CODE: doubt()}, [CODE]) + assert r2["paused"] == [] and r2.get("skipped"), r2 + + +@case("早上取回·查不到写空映射带原因且不动暂停 (绝不折成存疑); 查到写映射并按结果暂停") +def _(): + from app.repo import pms_repo + from app.services import param_store, strategy_service + saved, calls = {}, {"pause": [], "clear": []} + with _Patch() as p: + p(pms_repo, "list_positions", lambda *, only_open=False: [{"ts_code": CODE}, {"ts_code": "600001.SH"}]) + p(param_store, "set_param", lambda k, v, by="user": saved.__setitem__(k, v) or {"ok": True}) + _patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": True}) + p(strategy_service, "pause_buy", lambda code, *, reason="", source="signal": calls["pause"].append(code) or []) + p(strategy_service, "clear_buypause", lambda code, only_source=None: calls["clear"].append(code) or {"ok": True, "cleared": False}) + + def _boom(codes): + raise pf.PlanFeedError("选股系统连不上") + r = lss.pull_for_held(now=NOW, fetch=_boom) + import json + m = json.loads(saved[lss.MAP_KEY]) + assert r["ok"] is False and m["states"] == {} and "连不上" in m["error"], (r, m) + assert not calls["pause"] and not calls["clear"], calls + + r2 = lss.pull_for_held(now=NOW, fetch=lambda codes: {CODE: {**doubt(), "date": "2026-09-04"}}) + m2 = json.loads(saved[lss.MAP_KEY]) + assert r2["ok"] and r2["got"] == 1 and r2["by_state"] == {"逻辑存疑": 1}, r2 + assert m2["date"] == "2026-09-04" and m2["states"][CODE]["state"] == "逻辑存疑", m2 + assert r2["paused"] == [CODE] and r2["missing"] == ["600001.SH"], r2 # 没查到的票记明, 不动 + + +# ================================================================ 四, 持仓视图两栏 +def _fake_repo(): + from test_wiring import FakeRepo + f = FakeRepo() + f.insert_instruction(instruction_id="I1", origin_type="proposal", origin_id="P1", ts_code=CODE, + action="OPEN", side="buy", qty=1000) + f.insert_ledger(ts_code=CODE, action="OPEN", arbiter="user", verdict="PASS", price_at=10.0, + hard_numbers={"basis": "三门槛全过、无硬风险", "logic": ["研报说好 —— 出处 A", "B", "C", "D"], + "verdict": "候选", "logic_state": hold()}, + ref_id="P1", reason="人工采纳: 看好") + f.insert_lot(ts_code=CODE, lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-01", + instruction_id="I1") + f.insert_lot(ts_code="600001.SH", lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-01") + f.insert_instruction(instruction_id="I2", origin_type="command", origin_id="C1", ts_code="600002.SH", + action="OPEN", side="buy", qty=1000) + f.insert_lot(ts_code="600002.SH", lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-02", + instruction_id="I2") + return f + + +@case("入场论点·有指令链 (批次→指令→提议号→账本) / 外部成交并入 / 账本无行 三种情形都说得出话") +def _(): + from app.repo import pms_repo + f = _fake_repo() + with _Patch() as p: + p(pms_repo, "list_lots", f.list_lots) + p(pms_repo, "get_instruction", f.get_instruction) + p(pms_repo, "ledger_by_ref", f.ledger_by_ref) + a = lss.entry_view(CODE) + b = lss.entry_view("600001.SH") + c = lss.entry_view("600002.SH") + d = lss.entry_view("600009.SH") + assert a["basis"] == "三门槛全过、无硬风险" and a["logic"] == ["研报说好 —— 出处 A", "B", "C"], a + assert a["verdict"] == "候选" and a["logic_state_at_entry"] == "逻辑成立" and a["open_date"] == "2026-09-01", a + assert a["instruction_id"] == "I1" and a["arbiter"] == "user", a + assert "外部成交并入" in b["why"] and b["open_date"] == "2026-09-01", b + assert "账本里没有" in c["why"] and c["instruction_id"] == "I2", c + assert "没有未平的批次" in d["why"], d + assert f.ledger_by_ref(["P1", "X"]) and not f.ledger_by_ref([]), "假仓库的按引用取账本" + + +@case("持仓接口两栏·仓库读失败也不抛, 每行都有 entry 与 logic_now; 没有读数写明; 空仓行 entry 为空") +def _(): + from app.repo import pms_repo + rows = [pos(), pos("600001.SH"), pos("600002.SH", total_qty=0)] + with _Patch() as p: + p(pms_repo, "list_lots", lambda *a, **k: (_ for _ in ()).throw(OSError("db down"))) + lss.decorate_positions(rows, {CODE: doubt()}) + assert rows[0]["logic_now"]["state"] == "逻辑存疑" and "每股收益预测下修" in rows[0]["logic_now"]["text"] + assert rows[0]["logic_now"]["as_of"] == "2026-09-04" and "批次读取失败" in rows[0]["entry"]["why"], rows[0] + assert rows[1]["logic_now"]["state"] is None and "没有读数" in rows[1]["logic_now"]["text"], rows[1] + assert rows[2]["entry"] is None, rows[2] + nv = lss.now_view(hold()) + assert nv["state"] == "逻辑成立" and nv["settle_note"] == "维持" and "研报论断" in nv["text"], nv + + +@case("假仓库·ledger_by_ref 与真 repo 同名同签名 (第十批 [M1] 也会扫, 这里先钉一次)") +def _(): + import ast + import inspect + from test_wiring import FakeRepo + path = os.path.join(os.path.dirname(_HERE), "app", "repo", "pms_repo.py") + with open(path, encoding="utf-8") as fh: + names = {n.name: [a.arg for a in n.args.args] for n in ast.parse(fh.read()).body + if isinstance(n, ast.FunctionDef)} + assert names["ledger_by_ref"] == ["ref_ids", "limit"], names["ledger_by_ref"] + fp = list(inspect.signature(FakeRepo.ledger_by_ref).parameters) + assert fp == ["self", "ref_ids", "limit"], fp + + +# ================================================================ 入口 +def main(): + ok = 0 + for name, fn in RESULTS: + try: + fn() + ok += 1 + print(f" ok {name}") + except Exception: + print(f" FAIL {name}") + traceback.print_exc() + print("-" * 60) + if ok == len(RESULTS): + print(f"ALL PASS ({ok} cases)") + return 0 + print(f"FAILED {len(RESULTS) - ok}/{len(RESULTS)}") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index 0f19b65..36a2275 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -314,6 +314,10 @@ class FakeRepo: def list_ledger(self, *, ts_code=None, limit=200): return self.ledger[-limit:] + def ledger_by_ref(self, ref_ids, limit=50): + ids = {str(x) for x in (ref_ids or []) if x} + return [r for r in self.ledger if str(r.get("ref_id")) in ids][:limit] + def rule_rejected_today(self, since): # 内存桩里所有留痕都算"今天"; 只认规则闸的 REJECT (研判结论会变, 不参与当日去重) return {(r["ts_code"], r["action"]) for r in self.ledger