diff --git a/DEVLOG.md b/DEVLOG.md index fae2f15..b8c01b8 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -1803,3 +1803,17 @@ docs/交接_2026-09-20.md(新)、docs/dev_preview/mock_preview.py 与 sample ## YYYY-MM-DD · 一句话标题 **做了什么** / **动了哪些文件** / **部署方式** / **真机判收** / **还欠着什么** --> + +## 2026-10-15 · 每日计划版本贯通、消费顺序和提议重新核验 + +**做了什么。** 按用户批准的跨仓方案落本地四批实现;旧阈值与运行开关未改。候选消费记录分三类,执行保留消费顺序与信号调整原因;提议核验与人工再次确认通过条件更新保护并发;报告链接和两端研究展示贯通。 + +**动了哪些文件。** plan_feed、action_engine、plan_revision、plan_recheck、proposal_service、command_service、planner、executor、pms_repo、company_report、logic_state_service、Web 入口及页面;新控制 PMS_PLAN_VERSION_RECHECK 默认 False;新增第 38 批 15 例并登记总入口。 + +**本地验证。** 配置隔离且禁止联网的全套入口 ALL SUITES PASS,包含新增第 38 批 15 例;三个页面守卫通过。浏览器只连接 127.0.0.1 假数据页面,看到研究信息与过期内容重新确认提示,没有连接交易通道。 + +**部署方式。** 先兼容读与留痕,再桥发布,最后模拟实例启用修订与核验。实际机器、表前缀和容器需部署前确认,容器操作需用户同意。 + +**真机判收。** 未判收;本轮未连接生产机器,未改运行参数,未生成任何真实指令。 + +**还欠着什么。** 获准后的部署、真实参数及实例身份核对、十二分钟传播时延和连续五个交易日模拟观察。详见四仓同步规格书及交接。 diff --git a/UPSTREAM_PLAN_API.md b/UPSTREAM_PLAN_API.md index 2cb85fb..f8298d1 100644 --- a/UPSTREAM_PLAN_API.md +++ b/UPSTREAM_PLAN_API.md @@ -1,5 +1,7 @@ # 上游选股计划接口 · 接入记录与待确认口径 +当前开发契约见本文末尾“2026-10-15 补充:完整决策版本契约”;下文早期接入实测保留历史语义。 + > 2026-07-30 建。上游「选股系统」提供 HTTP 计划接口, PMS 的升仓/建仓候选池自此改吃它, > 不再读 `trading_buy_plan`。本文是接口现状 + PMS 侧接法 + **需要上游回答的问题**。 > 代码入口: `app/services/plan_feed.py`; 单测: `scripts/test_batch7_units.py` (23 例); @@ -672,3 +674,17 @@ make changes # 页面同款结论 **候选按上游名次排 (自门控)。** 计划顶层 `rank_axis` 为 quality 时, `select_candidates` 按上游名次升序排、关掉 PMS 自己的基本面立场分桶 (参数 `PMS_PLAN_RANK_BY_UPSTREAM`, 默认开); 上游没带这个键时无作用, 逐字回到按立场分桶排。这样选股序、入池、PMS 候选三处同一个序。本次用哪种序记在候选返回与名册快照元数据的 `rank_mode` 里 (upstream / quality_bucket / score)。 **前缀效应 —— 强传导票变少不是故障。** 质地序把好公司排到前面, 质地差但强传导的票会往后掉, 可能落到我们要的三百只主榜前缀之外, 于是这批票在候选里变少甚至消失。这与"质地差降关注"同向, 是预期结果, 不要当成上游漏发或接口出错。要看全量强传导票, 去上游的逐票对账接口 `/plan/verdict` (它的"主榜第 N 名"仍按原始分序, 应答带 `rank_axis: transmission` 说明), 或把 `PMS_PLAN_RANK_BY_UPSTREAM` 关掉临时回到按分桶排。 +# 2026-10-15 补充:完整决策版本契约 + +本节为当前开发实现;下文历史实测里“实时重装配、没有版本戳”的描述不再用于新版。 +部署状态见 `docs/交接_每日选股计划完善_2026-10-15.md`,本地实现不等于已经上线。 + +- 顶层新增 schema_version=2、decision、traceability;plan_version 仍为代码版本。 +- decision 包括 id、parent_id、cutoff_at、input_hash、params_hash、reason;时间带时区。 +- 计划行新增全量 research_rank、research、review_ref。展示裁剪不改变全量名次。 +- GET /plan、/plan/verdict、/logic_state 可指定 decision_id;指定日期或版本只读发布内容。缺档不现算。 +- 最新查询带 refresh_status;pending_codes 为已知待修订或来源核验失败的股票,暂停新增自主提议。 +- 评析报告 report_version_url 引用不可变对象;旧数据缺引用明确标示,不能用最新报告替代历史。 +- PMS meta.kind 区分 received、consumed、rechecked;消费记录包括真实顺序、有效参数和过滤原因。 +- 采纳核验由 PMS_PLAN_VERSION_RECHECK 控制,默认关闭;内容更新时接口返回 ok=false、原因及更新后的 proposal,页面必须重新确认其 proposal_revision。 +- 旧消费者可忽略新增字段,新消费者接受旧字段为空;无版本旧数据不能通过已启用的自主采纳核验。 diff --git a/app/core/action_engine.py b/app/core/action_engine.py index 4cd7790..e33c620 100644 --- a/app/core/action_engine.py +++ b/app/core/action_engine.py @@ -858,6 +858,8 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float): "tier": c.get("tier"), "upside": c.get("upside"), "heat": c.get("heat"), "plan_rank": c.get("rank"), "plan_bucket": c.get("bucket"), "plan_src": c.get("src"), "sector": c.get("sector"), + **{k: c.get(k) for k in ('decision', 'consumption_id', 'consumption_error', + 'plan_basis', 'selection_rank', 'research_rank', 'research', 'review_ref')}, # 决策系统今天盘中判过这只票转多没有。这是**定性材料**不是仓位数字, 所以送研判时 # 也带过去 (见 judge.OPEN_JUDGE_KEYS) —— 等于当面告诉它「你自己今天判过这只票转多」, # 它可以拿自己的结论去对照; 没有信号时这两项是 None, 不误导。 @@ -972,13 +974,18 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float, # 候选池的产物), 候选层那一整套过滤一道都不绕。 # PMS_OPEN_SIGNAL_PRIORITY 关掉即退回纯分数排序, 页面可改, 不用改码。 sig_first = bool(params.get("open_signal_priority", True)) - for c in sorted(candidates or [], + for execution_rank, c in enumerate(sorted(candidates or [], key=lambda x: (0 if (sig_first and x.get("sig_buy")) else 1, - -_f(x.get("score")), _f(x.get("rank"), 10 ** 9), - str(x.get("ts_code") or ""))): + (0, x['selection_rank']) if x.get('selection_rank') is not None + else (1, -_f(x.get('score'))), + _f(x.get("rank"), 10 ** 9), + str(x.get("ts_code") or ""))), 1): code = c.get("ts_code") if not code: continue + if code in ((c.get('refresh_status') or {}).get('pending_codes') or []) or c.get('consumption_error'): + skipped.append({'ts_code': code, 'action': A_OPEN, 'why': '计划修订或消费留痕尚未完成,暂停新的自主建仓'}) + continue if route and c.get("verdict") == VERDICT_DISPLAY: # 仅展示的行不占名额、不占金额, 也不进后面任何一道闸; 放在名额判断之前, # 是为了让它的跳过原因永远是「仅展示」而不是碰巧的「名额用完」。 @@ -1043,6 +1050,9 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float, continue try: cand.setdefault("hard_numbers", {})["advice"] = adv + cand['hard_numbers']['execution_rank'] = execution_rank + cand['hard_numbers']['priority_reason'] = ('盘中转多信号优先' if sig_first and c.get('sig_buy') + else '继承候选消费顺序') cand["hard_numbers"]["advice_text"] = adv.get("text") if isinstance(adv, dict) else None cand["hard_numbers"]["plan_by_advice"] = bool(params.get("plan_by_advice")) if isinstance(adv, dict) and adv.get("tier") == "试探仓": diff --git a/app/core/plan_revision.py b/app/core/plan_revision.py new file mode 100644 index 0000000..d866032 --- /dev/null +++ b/app/core/plan_revision.py @@ -0,0 +1,45 @@ +"""计划消费与提议版本核验的纯函数;不改变准入、仓位或自然语言判断。""" +import hashlib +import json + + +def fingerprint(value): + return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, + separators=(',', ':'), default=str).encode()).hexdigest() + + +def basis(row): + # 排名与审计时间不是资格/风险/数量依据。完整评析引用变化需要重新研判。 + return {k: row.get(k) for k in ('score', 'tier', 'upside', 'heat', 'theme', + 'verdict', 'risk', 'missing', 'reasons', 'logic_state', 'company_gate', + 'company_review', 'company_review_text', 'events', 'pricing_state', 'news')} + + +def portfolio_context(view, commitments, stock=None): + """持仓/资金/在途承诺变化使旧数量依据失效;价格漂移仍由执行终检管理。""" + totals = view.get('totals') or {} + return fingerprint({'held': sorted((r['ts_code'], r.get('total_qty'), r.get('sector')) + for r in view.get('held') or []), + 'cash_source': totals.get('cash_source'), 'cash_avail': totals.get('cash_avail'), + 'sector_ready': view.get('sector_ready'), 'stock': stock or {}, + 'inflight': {'names': sorted(commitments.get('names') or []), + 'amount': commitments.get('amount')}}) + + +def compare(old, current, code): + """返回 unavailable/material/unchanged。资格失效由调用方通过缺候选表达。""" + if not current or not (current.get('decision') or {}).get('id'): + return 'unavailable' + state = current.get('refresh_status') or {} + if code in (state.get('pending_codes') or []) or current.get('consumption_error'): + return 'unavailable' + if old is None or fingerprint(old) != fingerprint(basis(current)): + return 'material' + return 'unchanged' + + +def needs_new_judgement(hard, previous): + """决策系统缓存按股票/动作,缺同依据的既有研判就必须要求重新回答。""" + keys = ('plan_basis', 'scan_params_hash', 'portfolio_context_hash') + return (not previous or any(previous.get(k) is None for k in keys) + or any(fingerprint(hard.get(k)) != fingerprint(previous.get(k)) for k in keys)) diff --git a/app/core/planner.py b/app/core/planner.py index efe159a..1c9a9aa 100644 --- a/app/core/planner.py +++ b/app/core/planner.py @@ -303,7 +303,7 @@ def plan_increase_exposure(*, add_amount: float, positions: list, candidates=Non ctx = _ctx_after(ctx, amt, is_new_name=False, sector=p.get("sector")) # ---- ② 候选池新票建仓 ---- - for c in sorted(candidates or [], key=lambda x: _sortable(x, "score", reverse=True)): + for execution_rank, c in enumerate(sorted(candidates or [], key=lambda x: _sortable(x, "score", reverse=True)), 1): if acc >= target: break price = float(c.get("price") or 0) @@ -328,8 +328,11 @@ def plan_increase_exposure(*, add_amount: float, positions: list, candidates=Non if bad: rejects.append({"ts_code": c.get("ts_code"), "reasons": bad}) continue - items.extend(_open_batch_items(c.get("ts_code"), sp, price, - reason_prefix="升仓建新仓")) + batches = _open_batch_items(c.get("ts_code"), sp, price, reason_prefix="升仓建新仓") + for item in batches: + item['plan_provenance'] = dict(c.get('plan_provenance') or {}, execution_rank=execution_rank, + priority_reason='用户升仓命令,沿用命令调配次序') + items.extend(batches) intended += want_amt acc += actual ctx = _ctx_after(ctx, actual, is_new_name=True, sector=c.get("sector")) diff --git a/app/repo/pms_repo.py b/app/repo/pms_repo.py index 862b9b1..be597a7 100644 --- a/app/repo/pms_repo.py +++ b/app/repo/pms_repo.py @@ -513,10 +513,20 @@ def get_proposal(proposal_id: str): return r -def decide_proposal(proposal_id: str, status: str) -> int: +def revise_pending_proposal(old, candidate, verdict): + return execute('UPDATE pms_proposal SET qty=:qty, hard_numbers_json=:hn, ' + 'judge_verdict=:jv, judge_reason=:jr WHERE proposal_id=:pid ' + "AND status='WAIT_USER' AND hard_numbers_json=:old", + {'qty': candidate['qty'], 'hn': _dumps(candidate['hard_numbers']), + 'jv': verdict.get('verdict'), 'jr': (verdict.get('reason') or '')[:500], + 'pid': old['proposal_id'], 'old': old['hard_numbers_json']}) + + +def decide_proposal(proposal_id: str, status: str, expected_hard=None) -> int: return execute("UPDATE pms_proposal SET status = :st, decided_at = :ts " - "WHERE proposal_id = :pid AND status = 'WAIT_USER'", - {"st": status, "pid": proposal_id, "ts": _NOW()}) + "WHERE proposal_id = :pid AND status = 'WAIT_USER'" + + (' AND hard_numbers_json = :expected' if expected_hard is not None else ''), + {"st": status, "pid": proposal_id, "ts": _NOW(), 'expected': expected_hard}) def expire_proposals(now=None) -> int: @@ -811,9 +821,14 @@ def upsert_industry(rows: list) -> int: # ================================================================ pms_plan_snapshot -def latest_plan_digest(): +def latest_plan_digest(kind=None): """最新一份快照的 (plan_date, digest); 没有则 None。判"这份榜变没变"只该拿它比。""" - r = fetch_one("SELECT plan_date, digest FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1") + if kind: + r = fetch_one("SELECT plan_date, digest FROM pms_plan_snapshot " + "WHERE JSON_UNQUOTE(JSON_EXTRACT(meta_json, '$.kind')) = :kind " + "ORDER BY id DESC LIMIT 1", {'kind': kind}) + else: + r = fetch_one("SELECT plan_date, digest FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1") return (r["plan_date"], r["digest"]) if r else None @@ -837,7 +852,8 @@ def insert_plan_snapshot(*, plan_date, digest, roster, meta=None, n_main=0, n_ob 不产生任何假信号), 换掉的是一整类方向错误, 这笔交换是划算的。 """ now = fetched_at or _NOW() - cur = latest_plan_digest() + kind = (meta or {}).get('kind') + cur = latest_plan_digest(kind) if kind else latest_plan_digest() if cur and cur[0] == str(plan_date) and cur[1] == str(digest): return 0 execute( @@ -857,7 +873,8 @@ def latest_plan_snapshots(limit: int = 2) -> list: rows = fetch_all( "SELECT id, plan_date, digest, fetched_ymd, fetched_at, n_main, n_observe, " " capped_main, capped_obs, roster_json, meta_json " - "FROM pms_plan_snapshot ORDER BY id DESC LIMIT :n", {"n": max(1, int(limit))}) + "FROM pms_plan_snapshot WHERE COALESCE(JSON_UNQUOTE(JSON_EXTRACT(meta_json, '$.kind')), '') " + "NOT IN ('consumed', 'rechecked') ORDER BY id DESC LIMIT :n", {"n": max(1, int(limit))}) return [_snapshot_row(r) for r in rows] @@ -881,20 +898,12 @@ def list_plan_snapshots(*, plan_date=None, limit: int = 50) -> list: def prune_plan_snapshots(keep: int = 200) -> int: - """只留最近 keep 份。 + """停止按行数删除。保留全部记录,满足至少 120 交易日与未结束引用保留。 - 单表守卫不许子查询跨表, 所以先查出保留水位的 id 再按 id 删 —— 两条单表语句, - 比一条 `DELETE ... WHERE id NOT IN (SELECT ...)` 干净, 也不会被守卫拦下。 - - 下限是 **2 份不是 1 份**: 比对天生要两份, 留一份等于把榜单变化永久锁死在"首次"上, - 而且页面上看起来一切正常。参数填了 0 或 1 也按 2 算。 + 未来若加清理器,必须同时核验交易日水位、未结束提议及持仓引用;不能恢复行数上限。 + keep 仅为旧调用兼容,当前不删除任何历史。 """ - keep = max(2, int(keep or 0)) - r = fetch_one("SELECT id FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1 OFFSET :k", - {"k": keep}) - if not r: - return 0 - return execute("DELETE FROM pms_plan_snapshot WHERE id <= :id", {"id": int(r["id"])}) + return 0 def _snapshot_row(r: dict) -> dict: diff --git a/app/services/command_service.py b/app/services/command_service.py index 0cb988b..7724582 100644 --- a/app/services/command_service.py +++ b/app/services/command_service.py @@ -349,6 +349,9 @@ def plan_command(cmd: dict) -> dict: notes.insert(0, f"⚠ 停买入侧有 {len(stop_r['errors'])} 处未成: " + "; ".join(str(x) for x in stop_r["errors"][:5])) progress = {"target_amount": result.get("target_amount", 0.0), + 'plan_provenance': [{'ts_code': it['ts_code'], 'plan_id': cs.make_plan_id(command_id, i), + **it['plan_provenance']} + for i, it in enumerate(items, 1) if it.get('plan_provenance')], "planned_amount": result.get("planned_amount", 0.0), "done_amount": 0.0, "gap": result.get("gap", 0.0), "plan_count": len(rows), "deadline": str(deadline), @@ -768,7 +771,7 @@ def _candidates(view: dict) -> list: or plan_feed.SRC_PLAN_API).strip() out, seen, noprice, fallback = [], set(), [], [] - def _push(c, score, tag, *, price=None, theme=None): + def _push(c, score, tag, *, price=None, theme=None, plan_provenance=None): if not c or c in held or c in black or c in seen: return px, psrc = float(price or 0), "given" @@ -785,13 +788,16 @@ def _candidates(view: dict) -> list: # 不参与约束 —— theme 是"传导主题"(事件驱动、天天变、覆盖率约三成), 拿它当行业标签 # 会让集中度约束跟着漂, 已于 2026-07-31 明确不再灌行业表 (见 UPSTREAM_PLAN_API.md §6)。 out.append({"ts_code": c, "price": px, "price_source": psrc, "score": float(score or 0), - "sector": industry.get(c), "theme": theme, "src": tag}) + "sector": industry.get(c), "theme": theme, "src": tag, + 'plan_provenance': plan_provenance}) if src in (plan_feed.SRC_PLAN_API, plan_feed.SRC_BOTH): try: sel = plan_feed.candidates(held=held, black=black) for r in sel["items"]: - _push(r["ts_code"], r["score"], "plan_api", theme=r.get("theme")) + _push(r["ts_code"], r["score"], "plan_api", theme=r.get("theme"), + plan_provenance={k: r.get(k) for k in ('decision', 'consumption_id', + 'selection_rank', 'research_rank', 'review_ref', 'research')}) logger.info("[候选池] 上游计划 %s: 排序池 %s → 合格 %s → 取 %d 只 (丢弃 %s)", sel.get("date"), sel.get("considered"), sel.get("eligible"), len(sel["items"]), sel.get("dropped")) diff --git a/app/services/company_report.py b/app/services/company_report.py index 0765386..3efcdbe 100644 --- a/app/services/company_report.py +++ b/app/services/company_report.py @@ -34,7 +34,7 @@ _cache: dict = {} # url -> {"at": 时刻, "text": 正文} def _url_of(cr) -> str | None: if not isinstance(cr, dict): return None - u = cr.get("report_url") + u = cr.get('report_version_url') or cr.get("report_url") u = str(u).strip() if u else "" return u if u.startswith(("http://", "https://")) else None @@ -106,9 +106,26 @@ def _fail_why(e: Exception, timeout: int) -> str: return f"{type(e).__name__}: {e}" -def get(code: str, *, now=None, timeout: int = TIMEOUT_SEC, getter=None, state_map=None, plan=None) -> dict: +def get(code: str, *, now=None, timeout: int = TIMEOUT_SEC, getter=None, state_map=None, plan=None, ref=None) -> dict: """页面用的一站式取法。返回固定键: ok / ts_code / url / source / markdown / cached / fetched_at / error。""" url, src = report_url_for(code, state_map=state_map, plan=plan) + if ref: + import re + from urllib.parse import urlsplit, urlunsplit, urlencode + if not re.fullmatch('[0-9a-f]{64}', ref): + return {'ok': False, 'error': '报告版本标识无效'} + if not url: + from app.repo import pms_repo + for row in pms_repo.list_ledger(ts_code=code, limit=200): + cr = (row.get('hard_numbers') or {}).get('company_review') or {} + if cr.get('review_ref') == ref: + url = _url_of(cr) + if url: + src = 'ledger' + break + if url: + parts = urlsplit(url) + url = urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode({'ref': ref}), '')) out = {"ok": False, "ts_code": code, "url": url, "source": src, "markdown": "", "cached": False, "fetched_at": None, "error": None} if not url: diff --git a/app/services/executor.py b/app/services/executor.py index 0c9fe7b..6f5da5f 100644 --- a/app/services/executor.py +++ b/app/services/executor.py @@ -103,6 +103,10 @@ def materialize_plans(limit: int = 100) -> dict: or param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3)) prog0 = {"deadline": str(p.get("deadline") or ""), "command_id": cid, "is_command": True, "children": []} + provenance = next((x for x in (cmd.get('progress') or {}).get('plan_provenance', []) + if x.get('plan_id') == p['plan_id']), None) + if provenance: + prog0['plan'] = provenance if cmd.get("cmd_type") == "LIQUIDATE_ALL": # 一键清仓是 danger 级紧急命令, 方案注记写着"紧急, 不做择时优化" —— # 这个语义从前没传到执行层, 紧急清仓照样等开盘半小时、等均价。 @@ -234,7 +238,7 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict: pms_repo.insert_ledger( ts_code=code, action=ins.get("action"), arbiter="rule", verdict="WARN", price_at=day_ctx.get("price") or 0, - hard_numbers={"ref_lock": prog.get("ref_lock"), + hard_numbers={**(prog.get('plan') or {}), "ref_lock": prog.get("ref_lock"), "advice": (prog.get("exec_advice") or {})}, ref_id=ins["instruction_id"], reason=d["ref_drift"]) except Exception as e: @@ -304,7 +308,7 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict: pms_repo.insert_ledger( ts_code=code, action=ins.get("action"), arbiter="rule", verdict="REJECT", price_at=day_ctx.get("price") or 0, - hard_numbers=gate["hard_numbers"], + hard_numbers={**gate["hard_numbers"], **(prog.get('plan') or {})}, failed_checks=gate["failed"], ref_id=ins["instruction_id"], reason="规则闸终检未通过 (宁可不动; 同因当日只记一条)") prog["gate_reject_logged"] = {"ymd": ymd_today, "key": rej_key} @@ -361,7 +365,7 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict: f"出口表与本端不一致") pms_repo.insert_ledger(ts_code=code, action=ins.get("action"), arbiter="rule", verdict="PASS", price_at=day_ctx.get("price") or 0, - hard_numbers={**gate["hard_numbers"], "limit": d["limit_price"], + hard_numbers={**gate["hard_numbers"], **(prog.get('plan') or {}), "limit": d["limit_price"], "mode": res["mode"]}, ref_id=ins["instruction_id"], reason=d["reason"]) out["fired"].append({"instruction_id": ins["instruction_id"], "code": code, @@ -450,7 +454,8 @@ def cancel_instruction(instruction_id: str, reason: str = "页面人工撤销") logger.warning("撤销指令进度回写失败 %s: %s", instruction_id, e) pms_repo.update_instruction(instruction_id, status=ST_CANCELLED) pms_repo.insert_ledger(ts_code=ins["ts_code"], action=ins.get("action"), arbiter="user", - verdict="REJECT", price_at=0, ref_id=instruction_id, reason=reason) + verdict="REJECT", price_at=0, ref_id=instruction_id, reason=reason, + hard_numbers=(ins.get('progress') or {}).get('plan') or {}) return {"ok": True, "downstream": r, "message": f"指令 {instruction_id} 已撤销"} diff --git a/app/services/logic_state_service.py b/app/services/logic_state_service.py index 54c7dd8..4f30f73 100644 --- a/app/services/logic_state_service.py +++ b/app/services/logic_state_service.py @@ -186,7 +186,9 @@ def company_view(st): "line": f"质地{cr.get('overall')}(回报{g.get('回报与护城河') or '缺'}、盈余{g.get('盈余质量与财务安全') or '缺'}、成长{g.get('成长与含金量') or '缺'})" f",估值{cr.get('valuation') or '无'}" + (f",置信{cr['confidence']}" if cr.get("confidence") else ""), "invalidation": cr.get("invalidation"), "doubt_hard": bool(cr.get("doubt_hard")), - "report_url": cr.get("report_url"), "text": (st or {}).get("company_review_text")} + "report_url": cr.get("report_version_url") or cr.get("report_url"), + "text": (st or {}).get("company_review_text"), 'company_review': cr, + **{k: st.get(k) for k in ('decision', 'research', 'review_ref', 'observation_at', 'observation_scope')}} def ref_target_view(st): @@ -237,6 +239,8 @@ def entry_view(ts_code: str) -> dict: 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], + **{k: hn.get(k) for k in ('decision', 'research', 'review_ref', 'company_review', + 'consumption_id', 'execution_rank', 'priority_reason')}, "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], diff --git a/app/services/param_store.py b/app/services/param_store.py index dbb949d..b47543c 100644 --- a/app/services/param_store.py +++ b/app/services/param_store.py @@ -40,6 +40,7 @@ SECRET_KEYS = ("PMS_QMT_SIGN_SEED_HEX", "PMS_QMT_PEER_PUBKEY_B64", "PMS_SESSION_ # 运行态开关: key -> (默认值, 类型, 说明) RUNTIME_EXTRA = { + 'PMS_PLAN_VERSION_RECHECK': (False, bool, '自主建仓采纳前核验计划版本;新版发布与留痕验收后开启,当前不可核验时暂停采纳'), "PMS_GLOBAL_BUY_HALT": (False, bool, "全局暂停买入 (HALT_BUY 命令置位)"), "PMS_GLOBAL_EXEC_HALT": (False, bool, "全局暂停执行"), "PMS_BRAKE_UNTIL": (0, int, "组合刹车解除日 YYYYMMDD, 0=未刹车"), diff --git a/app/services/plan_feed.py b/app/services/plan_feed.py index 912b544..7f9cec9 100644 --- a/app/services/plan_feed.py +++ b/app/services/plan_feed.py @@ -204,7 +204,10 @@ def _company_review_or_none(v): "invalidation": _text_or_none(v.get("invalidation")), "antithesis": _text_or_none(v.get("antithesis")), "catalyst": _text_or_none(v.get("catalyst")), "brain_status": _text_or_none(v.get("brain_status")), "ran_at": _text_or_none(v.get("ran_at")), "age_days": _int_or_none(v.get("age_days")), - "report_url": _text_or_none(v.get("report_url"))} + "report_url": _text_or_none(v.get("report_url")), + **{k: v.get(k) for k in ('review_ref', 'report_version_url', 'report_ran_at', + 'period_fi', 'period_rf', 'qual', 'n_materials', 'unknown_n', + 'numbers_unmatched', 'expectation', 'cycle', 'thesis_lens_refs')}} def _industry_catalyst_or_none(v, limit=3): @@ -337,6 +340,9 @@ def _rows(raw, bucket: str) -> list: "pointed": (bool(it.get("pointed")) if it.get("pointed") is not None else None), "rank_axis": _text_or_none(it.get("rank_axis")), "rank_old": _int_or_none(it.get("rank_old")), + "research_rank": _int_or_none(it.get('research_rank')), + "research": it.get('research') if isinstance(it.get('research'), dict) else None, + "review_ref": _text_or_none(it.get('review_ref')), }) return out @@ -356,6 +362,8 @@ def parse_plan(payload, *, requested=None) -> dict: observe = _rows(payload.get("observe"), BUCKET_OBSERVE) if not main and not observe: raise PlanFeedError(f"计划 {date} 主榜与观察档都是空的") + state = dict(payload.get('refresh_status') or {}) + state['pending_codes'] = [normalize_code(c) for c in state.get('pending_codes') or []] counts = payload.get("counts") if isinstance(payload.get("counts"), dict) else {} req = dict(requested or {}) req_top = _int_or_none(req.get("top")) @@ -368,6 +376,8 @@ def parse_plan(payload, *, requested=None) -> dict: themes[r["ts_code"]] = r["theme"] return { "date": date, + **{k: payload.get(k) for k in ('schema_version', 'decision', 'traceability')}, + 'refresh_status': state, # 计划版本与生成时刻 (2026-09-02 起上游下发)。同一个 date 的计划会被重算多版, # 这两项是上游自己的版本戳; 旧版应答没有, 缺就是 None (名册快照那套指纹照旧管用)。 "plan_version": _text_or_none(payload.get("plan_version")), @@ -505,6 +515,10 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N dropped = {"held": 0, "black": 0, "tier": 0, "score": 0, "sources": 0, "upside": 0, "st": 0, "theme": 0, "dup": 0, "capped": 0, "display": 0} passed, seen, per_theme, st_unknown, display_only = [], set(), {}, [], [] + filtered = [] + def drop(reason, code): + dropped[reason] += 1 + filtered.append({'ts_code': code, 'reason': reason}) # 排序: 默认按分数降序、同分按 rank 升序 (与 2026-09-11 之前逐字相同)。开了按质地排序后, 先按 # 基本面立场分桶 (质地看多的桶排在前) 再按分数, 在主题限额与截断之前生效; 技术面不参与排序。 # 2026-09-17 工作包乙: 上游是质地轴且开了按上游名次排时, 直接按上游名次升序 (同名次原始分降序), @@ -513,28 +527,28 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N mode = _rank_mode(plan, rank_by_upstream, rank_by_quality) if mode == "upstream": def _sort_key(x): - return (x.get("rank") or 10 ** 9, -(x.get("score") or 0.0)) + return (x.get('research_rank') or x.get("rank") or 10 ** 9, -(x.get("score") or 0.0), x['ts_code']) elif rank_by_quality: def _sort_key(x): return (0 if _fund_stance_of(x) == "看多" else 1, - -(x.get("score") or 0.0), x.get("rank") or 10 ** 9) + -(x.get("score") or 0.0), x.get("rank") or 10 ** 9, x['ts_code']) else: def _sort_key(x): - return (-(x.get("score") or 0.0), x.get("rank") or 10 ** 9) + return (-(x.get("score") or 0.0), x.get("rank") or 10 ** 9, x['ts_code']) for r in sorted(pool, key=_sort_key): c = r["ts_code"] if c in seen: - dropped["dup"] += 1 + drop('dup', c) continue seen.add(c) if c in held: - dropped["held"] += 1 + drop('held', c) continue if c in black: - dropped["black"] += 1 + drop('black', c) continue if route_by_verdict and r.get("verdict") == VERDICT_DISPLAY: - dropped["display"] += 1 # 上游判为仅展示: 不进池, 单独列出让人看得见 + drop('display', c) # 带上判决依据: 仅展示至少有三种来路 (差一个当日涨幅、吸筹没到确认线、 # 被昨夜坏信号一票否决), 三者处置完全不同。只记代码的话, 下游只能对所有 # 仅展示的票写同一句话, 人看不出哪只该继续盯。 @@ -545,30 +559,31 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N if not nm: st_unknown.append(c) # 没名字判不了, 留着但要让人知道 elif is_st(nm): - dropped["st"] += 1 + drop('st', c) continue if tiers and r.get("tier") is not None and r["tier"] not in tiers: - dropped["tier"] += 1 + drop('tier', c) continue if min_score is not None and (r.get("score") or 0.0) < min_score: - dropped["score"] += 1 + drop('score', c) continue if min_sources and (r.get("n_sources") or 0) < min_sources: - dropped["sources"] += 1 + drop('sources', c) continue if min_upside is not None and (r.get("upside") or 0.0) < min_upside: - dropped["upside"] += 1 + drop('upside', c) continue - if theme_cap > 0: - t = r.get("theme") or "(无主题)" + if theme_cap > 0 and r.get('theme'): + t = r['theme'] if per_theme.get(t, 0) >= theme_cap: - dropped["theme"] += 1 + drop('theme', c) continue per_theme[t] = per_theme.get(t, 0) + 1 passed.append(r) n = max(0, int(top_n or 0)) or len(passed) dropped["capped"] = max(0, len(passed) - n) + filtered.extend({'ts_code': r['ts_code'], 'reason': 'capped'} for r in passed[n:]) items = [{"ts_code": r["ts_code"], "name": r["name"], "score": r.get("score") or 0.0, "sector": r.get("theme"), "theme": r.get("theme"), "tier": r.get("tier"), "heat": r.get("heat"), "upside": r.get("upside"), "rank": r.get("rank"), @@ -593,9 +608,14 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N "company_gate": r.get("company_gate"), # 排序轴四键 (2026-09-17 工作包乙): 原样透传给动作引擎与提议卡, 缺就是 None。 "quality_band": r.get("quality_band"), "pointed": r.get("pointed"), - "rank_axis": r.get("rank_axis"), "rank_old": r.get("rank_old")} + "rank_axis": r.get("rank_axis"), "rank_old": r.get("rank_old"), + **{k: r.get(k) for k in ('research_rank', 'research', 'review_ref')}} for r in passed[:n]] + for i, item in enumerate(items, 1): + item.update(selection_rank=i, decision=plan.get('decision'), + refresh_status=plan.get('refresh_status')) return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed), + 'filtered': filtered, 'decision': plan.get('decision'), "items": items, "dropped": dropped, "st_unknown": st_unknown, "display_only": display_only, "rank_mode": mode, "tier_complete": _tier_complete(pool, tiers)} @@ -806,6 +826,11 @@ def parse_logic_states(payload) -> dict: # 公司深度两键随状态回 (2026-09-09 接入方案): 持仓页"证据还在不在"旁显示质地与失效条件。 st["company_review"] = _company_review_or_none(it.get("company_review")) st["company_review_text"] = _text_or_none(it.get("company_review_text")) + st['decision'] = it.get('decision_version') or (payload or {}).get('decision_version') + st['observation_at'] = it.get('observation_at') + st['observation_scope'] = it.get('observation_scope') + st['research'] = it.get('research') + st['review_ref'] = it.get('review_ref') out[code] = st return out @@ -844,7 +869,7 @@ def get_plan(*, force: bool = False, date=None) -> dict: now = time.time() with _lock: fresh_hit = (not force and _cache["key"] == key and _cache["plan"] is not None - and now - _cache["at"] < max(1, p["cache_sec"])) + and now - _cache["at"] < min(300, max(1, p["cache_sec"]))) if fresh_hit: return _cache["plan"] if (not force and _cache["key"] == key and _cache["error"] @@ -914,10 +939,15 @@ def snapshot(plan: dict, *, keep: int = 200, rank_mode=None) -> dict: from app.repo import pms_repo roster = pdf.roster_of(plan) digest = pdf.digest_of(roster) + if plan.get('decision'): + from app.core.plan_revision import fingerprint + digest = fingerprint({'received': plan['decision'], 'requested': plan.get('requested')})[:40] meta = {k: plan.get(k) for k in ("counts", "returned", "requested", "funnel", "theme_cap", "heat_date", "market_snapshot_days", "age_tdays", "url", - "plan_source", "snapshot_generated_at", "gate_on")} + "plan_source", "snapshot_generated_at", "gate_on", + "decision", "traceability")} + meta['kind'] = 'received' # 排序模式 (2026-09-17 工作包乙): 让选股系统周报按它还原当天交付名单的序。旧调用没传即不写。 if rank_mode is not None: meta["rank_mode"] = rank_mode @@ -929,7 +959,8 @@ def snapshot(plan: dict, *, keep: int = 200, rank_mode=None) -> dict: capped_main=bool(truncated.get("main")), capped_obs=bool(truncated.get("observe"))) out = {"digest": digest, "rows": len(roster), "stored": bool(n)} if n: - out["pruned"] = pms_repo.prune_plan_snapshots(keep=keep) + # 停用按行数清理:盘中修订会把 200 行很快用尽。归档前保留全部历史及引用。 + out["pruned"] = 0 logger.info("[上游计划] 名册快照已落: %s digest=%s (%d 只)", plan.get("date"), digest[:10], len(roster)) return out @@ -1109,7 +1140,8 @@ def snapshot_log(*, plan_date=None, limit: int = 50) -> dict: # ================================================================ 对外: 候选与状态 -def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict: +def candidates(*, held=(), black=(), route_by_verdict: bool = False, force=False, + record=True, recheck_code=None) -> dict: """参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。 route_by_verdict 默认 False: 命令驱动建仓 (command_service._candidates) 不传, 于是用户 @@ -1119,14 +1151,14 @@ def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict: 显式传 True。要让命令驱动也按判决过滤, 改这里的调用方, 不改默认值。 """ p = _params() - plan = get_plan() + plan = get_plan(force=force) if force else get_plan() route = bool(route_by_verdict) - out = select_candidates(plan, held=held, black=black, top_n=p["top_n"], tiers=p["tiers"], + out = select_candidates(plan, held=held, black=black, top_n=0 if recheck_code else p["top_n"], tiers=p["tiers"], include_observe=p["include_observe"], min_score=(p["min_score"] or None), min_sources=p["min_sources"], min_upside=(p["min_upside"] or None), - theme_cap=p["theme_cap_local"], exclude_st=p["exclude_st"], + theme_cap=0 if recheck_code else p["theme_cap_local"], exclude_st=p["exclude_st"], route_by_verdict=route, rank_by_quality=p["rank_by_quality"], rank_by_upstream=p["rank_by_upstream"], @@ -1134,6 +1166,36 @@ def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict: out["age_tdays"] = plan.get("age_tdays") out["theme_cap"] = plan.get("theme_cap") out["route_by_verdict"] = route + from app.core.plan_revision import fingerprint, basis + # 实际参数与组合上下文参与消费指纹;来源地址等接入信息不进入业务参数。 + effective = {k: p[k] for k in ('top_n', 'tiers', 'include_observe', 'min_score', + 'min_sources', 'min_upside', 'theme_cap_local', 'exclude_st', 'rank_by_quality', + 'rank_by_upstream', 'observe_if_fund_bull')} + context = {'held': sorted(held), 'black': sorted(black), 'route_by_verdict': route, + 'recheck_code': recheck_code} + if recheck_code: + out['items'] = [r for r in out['items'] if r['ts_code'] == recheck_code] + consumption = {'kind': 'rechecked' if recheck_code else 'consumed', 'decision': plan.get('decision'), + 'params': effective, 'context': context, 'rank_mode': out['rank_mode'], + 'filtered': out['filtered'], 'order': [r['ts_code'] for r in out['items']]} + cid = fingerprint(consumption) + for row in out['items']: + row['consumption_id'] = cid + row['plan_basis'] = basis(row) + out['consumption'] = dict(consumption, id=cid) + if record: + from app.repo import pms_repo + try: + pms_repo.insert_plan_snapshot(plan_date=plan['date'], digest=cid[:40], + roster=out['items'], meta=dict(consumption, id=cid), + n_main=sum(r.get('bucket') == BUCKET_MAIN for r in out['items']), + n_observe=sum(r.get('bucket') != BUCKET_MAIN for r in out['items']), + capped_main=False, capped_obs=False) + except Exception as e: + # 记录失败可展示,不能把无凭据候选变成自主指令。 + out['consumption']['error'] = type(e).__name__ + for row in out['items']: + row['consumption_error'] = type(e).__name__ return out diff --git a/app/services/plan_recheck.py b/app/services/plan_recheck.py new file mode 100644 index 0000000..95519e9 --- /dev/null +++ b/app/services/plan_recheck.py @@ -0,0 +1,78 @@ +"""采纳前核验;复用原扫描、合议、研判与规则闸,不解释自然语言。""" +from datetime import datetime +from app.core.plan_revision import compare, fingerprint, portfolio_context + + +def current_basis(hard, code, proposal_id=None): + from app.services import plan_feed, portfolio, command_service, proposal_service as ps + view = portfolio.positions_view() + stock = command_service.effective_stock_params() + params = ps._scan_params(view) + selected = plan_feed.candidates(held=[x['ts_code'] for x in view['held']], + black=[c for c, v in stock.items() if v.get('black')], force=True, + route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code) + current = next((r for r in selected['items'] if r['ts_code'] == code), None) + if not (selected.get('decision') or {}).get('id'): + return {'ok': False, 'error': '当前计划没有可核验版本'} + if current is None: + return {'ok': False, 'error': '该票已不满足当前候选资格', 'material': True} + result = compare(hard.get('plan_basis'), current, code) + if result == 'unavailable': + return {'ok': False, 'error': '当前计划修订或消费留痕尚未完成'} + ph = fingerprint({k: v for k, v in params.items() if not k.startswith('_')}) + commitments = ps._inflight_open_commitments(view, exclude_proposal=proposal_id) + context = portfolio_context(view, commitments, stock.get(code)) + material = (result == 'material' or hard.get('scan_params_hash') != ph + or hard.get('portfolio_context_hash') != context) + return {'ok': not material, 'material': material, + 'error': '研究、资格或仓位参数已变化,须重新评估' if material else None, + 'view': view, 'stock': stock, 'params': params, + 'provenance': {'decision': current['decision'], 'consumption_id': current['consumption_id'], + 'version_check': {'status': 'passed', 'at': datetime.now().astimezone().isoformat(), + 'reason': '依据相同,名次变化不撤销'}}} + + +def before_accept(p, expected_revision): + from app.services import proposal_service as ps + from app.repo import pms_repo + hard = p.get('hard_numbers') or {} + if p.get('action') != 'OPEN' or hard.get('kind') == 'strategy': + return {'ok': True, 'proposal': p} + try: + check = current_basis(hard, p['ts_code'], proposal_id=p.get('proposal_id')) + if check.get('ok'): + if hard.get('proposal_revision') and expected_revision != hard['proposal_revision']: + return {'ok': False, 'error': '提议内容已更新,请刷新后再次确认', 'proposal': p} + return {'ok': True, 'proposal': dict(p, hard_numbers=dict(hard, **check['provenance']))} + if not check.get('material') or not check.get('view'): + return check + now = datetime.now() + out = {'skipped': [], 'rejected': [], 'queued': [], 'executed': []} + params, view = check['params'], check['view'] + mkt = {} + params['_mkt'] = mkt + fresh = ps._scan_open(view, params, check['stock'], {}, mkt, out, now=now, + review_code=p['ts_code'], exclude_proposal=p['proposal_id']) + if not fresh: + return {'ok': False, 'error': '重新评估未产出可采纳提议', 'details': out} + c = fresh[0] + c['_plan_recheck'] = True + from app.services import param_store + from app.core import tradedays as td + brake = td.ymd() < param_store.get_int('PMS_BRAKE_UNTIL', 0) + result = ps._route_one(c, view, params, check['stock'], brake, now, True, out, review_only=True) + if not result: + return {'ok': False, 'error': '新版重新评估未通过', 'details': out} + hn = c['hard_numbers'] + hn['version_check'] = {'status': 'reconfirmed_required', 'at': now.astimezone().isoformat(), + 'reason': '依据变化,已重跑合议、研判、数量及规则闸'} + hn['proposal_revision'] = fingerprint({'qty': c['qty'], 'hard_numbers': hn}) + if not pms_repo.revise_pending_proposal(p, c, result['verdict']): + return {'ok': False, 'error': '提议被并发更新,请刷新'} + pms_repo.insert_ledger(ts_code=p['ts_code'], action=p['action'], arbiter='rule', + verdict='WARN', price_at=hn.get('price') or 0, hard_numbers=hn, + ref_id=p['proposal_id'], reason='新版依据变化,更新提议并要求重新确认') + return {'ok': False, 'error': '提议已按新版重新评估,请查看更新内容后再次确认', + 'proposal': dict(p, qty=c['qty'], hard_numbers=hn)} + except Exception as e: + return {'ok': False, 'error': f'版本核验暂不可用,保留提议:{type(e).__name__}'} diff --git a/app/services/proposal_service.py b/app/services/proposal_service.py index 3850821..fddfd34 100644 --- a/app/services/proposal_service.py +++ b/app/services/proposal_service.py @@ -212,7 +212,8 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict: out["skipped"].append({"action": ae.A_OPEN, "why": "新建仓档位 off, 不扫描候选池"}) else: try: - open_cands = _scan_open(view, params, stock_params, skip, mkt, out, now=now) + open_cands = _scan_open(view, params, stock_params, skip, mkt, out, now=now, + record_consumption=not dry_run) except Exception as e: logger.exception("新建仓扫描失败") out["errors"].append(f"新建仓扫描失败: {type(e).__name__}: {e}") @@ -307,7 +308,8 @@ def _save_tech_exit_done(codes) -> None: # ================================================================ 新建仓的取数与筛选 -def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list: +def _scan_open(view, params, stock_params, skip, mkt, out, now=None, + review_code=None, exclude_proposal=None, record_consumption=True) -> list: """候选池 → 新建仓候选。产出的候选数天生不超过剩余名额 (名额与金额在纯逻辑里边走边扣)。 与命令驱动那条路 (command_service._candidates) 有意不同的两点: @@ -339,7 +341,9 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list: # 按判决分流开着时, 上游判为「仅展示」的票在候选阶段就剔掉 (理由见 # plan_feed.select_candidates 的说明: 不剔的话它们会白占 top_n 的名额)。 sel = plan_feed.candidates(held=held, black=black, - route_by_verdict=bool(params.get("open_route_by_verdict"))) + route_by_verdict=bool(params.get("open_route_by_verdict")), + **({'recheck_code': review_code} if review_code else {}), + **({'record': False} if not record_consumption else {})) except plan_feed.PlanFeedError as e: # 拿不到 ≠ 今天没票可买。显式留痕, 本轮不产新建仓候选, 已有持仓的四类照常。 logger.error("[新建仓] 候选池取数失败, 本轮不建仓: %s", e) @@ -353,6 +357,8 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list: out["skipped"].append({"ts_code": row.get("ts_code"), "action": ae.A_OPEN, "why": ae.display_only_why(row)}) items = list(sel.get("items") or []) + if review_code: + items = [x for x in items if x['ts_code'] == review_code] if not items: out["skipped"].append({"action": ae.A_OPEN, "why": _no_candidate_why(sel)}) return [] @@ -381,7 +387,8 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list: # 名额与金额都要扣掉**在途的建仓承诺** (2026-08-28 审查修): names_count 只数已成交 # 持仓, 在途 OPEN 指令与等拍板的 OPEN 提议两头都不占 —— 候选没进买入区间时指令滞留 # 在途, 下一跳名额照旧又放两只, 几分钟就把持仓数上限击穿。同票去重挡不住换一只票。 - inflight = _inflight_open_commitments(view) + inflight = (_inflight_open_commitments(view, exclude_proposal=exclude_proposal) + if exclude_proposal else _inflight_open_commitments(view)) slots = int(p["max_names"] or 0) - int(t["names_count"] or 0) - len(inflight["names"]) room = float(p["portfolio_cap"] or 0) * float(p["scale"] or 0) \ - float(t["portfolio_mv"] or 0) - inflight["amount"] @@ -416,7 +423,11 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list: # 当日行情快照只对**真的产出了候选**的票取 —— 候选数已被名额与金额扣到很小, # 不必为整池几十只票各拉一遍分钟线。规则闸的「不追高」靠的就是这一份。 for c in res["candidates"]: + from app.core.plan_revision import fingerprint, portfolio_context + c.setdefault('hard_numbers', {})['scan_params_hash'] = fingerprint( + {k: v for k, v in _scan_params(view).items() if not k.startswith('_')}) code = c["ts_code"] + c['hard_numbers']['portfolio_context_hash'] = portfolio_context(view, inflight, stock_params.get(code)) d = {"ma5": market.get_ma5(code)} try: d["day"] = market.day_snapshot(code) or {} @@ -585,7 +596,7 @@ def _no_candidate_why(sel) -> str: return head + ":" + ",".join(parts) + "。" -def _inflight_open_commitments(view) -> dict: +def _inflight_open_commitments(view, exclude_proposal=None) -> dict: """在途的新建仓承诺: {names: {代码}, amount: 预估占用金额}。 口径: 在途 (LIVE) 的 OPEN 指令 + 等拍板 (WAIT_USER) 的 OPEN 提议, 只算**尚未持有** @@ -603,6 +614,8 @@ def _inflight_open_commitments(view) -> dict: logger.warning("读在途 OPEN 指令失败 (按无在途): %s", e) try: for pr in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=200): + if exclude_proposal and pr.get('proposal_id') == exclude_proposal: + continue if pr.get("action") == ae.A_OPEN and pr["ts_code"] not in held: hn = pr.get("hard_numbers") or {} rows.append((pr["ts_code"], int(pr.get("qty") or 0), @@ -719,7 +732,7 @@ def disposition_snapshot(now=None) -> dict: for _k in _inflight_keys(): # 在途; 前端多半已按指令/提议标成「已下单/待你确认」 skip[_k] = "系统正在处理这只票(见『待确认提议』或『当日指令』)" sink = {"skipped": []} # 丢弃用: _scan_open 只往里 append/extend, 不读它 - would = _scan_open(view, params, stock_params, skip, {}, sink, now=now) + would = _scan_open(view, params, stock_params, skip, {}, sink, now=now, record_consumption=False) except Exception as e: logger.exception("候选处置快照失败") return {"ok": False, "ready": False, "error": f"{type(e).__name__}: {e}", @@ -754,7 +767,7 @@ def disposition_snapshot(now=None) -> dict: def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, - *, deadline=None): + *, deadline=None, review_only=False): code, action, side = c["ts_code"], c["action"], c["side"] is_open = (action == ae.A_OPEN) pos = _pos_of(view, code) @@ -829,9 +842,18 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, _rtext = reask_service.reask_text(_rk) else: c.pop("reask", None) # 名额没落上就不算重问, 这一跳按普通候选走 + if is_open and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False): + from app.core.plan_revision import needs_new_judgement + try: + previous = next((r.get('hard_numbers') for r in pms_repo.list_ledger(ts_code=code, limit=20) + if r.get('action') == ae.A_OPEN and r.get('arbiter') in ('judge', 'system')), None) + except Exception: + previous = None + # 即使上一跳因版本变化放弃自动采纳,下一跳也不能复用该票旧批准的缓存。 + _force = _force or needs_new_judgement(c.get('hard_numbers') or {}, previous) verdict = judge.request(c, context={"position": _judge_ctx(pos), "recent_ledger": _recent_ledger(code)}, - force=_force, reask_text=_rtext) + force=_force or bool(c.get('_plan_recheck')), reask_text=_rtext) # 新建仓的低把握驳回改交人 (2026-09-07, 第二道保险)。择时决策系统的提示词已改成 # 「证据不足以判断 → 不可用」, 但模型未必每次守得住; 它给的把握度是现成的读数, # 低于阈值的驳回按「不可用」处理 —— 进人的待确认队列, 不记驳回、不杀提议。 @@ -967,6 +989,14 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, if auto_why: auto_exec = True + if review_only: + _apply_advice_tier(c, c.setdefault('hard_numbers', {}), verdict, + force_trial=(sd.get('tier') == 'trial')) + c['hard_numbers'].update(judge_verdict=verdict.get('verdict'), + judge_conf=verdict.get('confidence'), + price=price, reason=c['reason']) + return {'candidate': c, 'verdict': verdict} + if dry_run: (out["executed"] if auto_exec else out["queued"]).append( {**_brief(c), "route": "auto" if auto_exec else "queue", @@ -975,6 +1005,13 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out, return if auto_exec: + if is_open and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False): + check = verify_open_basis(c.get('hard_numbers') or {}, code) + if not check.get('ok'): + out['skipped'].append({'ts_code': code, 'action': action, + 'why': check['error'] + ';下一跳按新输入重新评估'}) + return + c['hard_numbers'].update(check['provenance']) sd_accept = sd_decision == "accept" strat_cancelled, buys_cancelled = [], [] if sd_accept and side == "sell": @@ -1089,6 +1126,14 @@ def _judge_budget_left(deadline) -> bool: # ================================================================ 落表 +def verify_open_basis(hard, code): + from app.services.plan_recheck import current_basis + try: + return current_basis(hard, code) + except Exception as e: + return {'ok': False, 'error': f'当前计划版本无法核验:{type(e).__name__}'} + + def _make_instruction(c, price, now) -> str: ymd = td.ymd(now) # 序号用完整时分秒 (2026-08-28 修): 原来 %1000 只留「分钟个位+秒」, 同日同票同动作 @@ -1107,7 +1152,9 @@ def _make_instruction(c, price, now) -> str: window_tdays=window, status=executor.ST_PROPOSED, progress={"deadline": str(td.window_deadline(now.date(), window)), "is_command": False, "children": [], "auto": True, - "reason": c["reason"]}) + "reason": c["reason"], + 'plan': {k: (c.get('hard_numbers') or {}).get(k) for k in + ('decision', 'consumption_id', 'review_ref', 'execution_rank', 'priority_reason')}}) iid = try_iid break except Exception as e: @@ -1376,6 +1423,8 @@ def _make_proposal(c, price, verdict) -> str: # 采纳按钮那边直接抄提议里的股数原样落单 (只有卖出侧会按可卖量重算), 所以不在这里 # 改, 后面就没有第二次机会了。 _apply_advice_tier(c, hn, verdict) + from app.core.plan_revision import fingerprint + hn['proposal_revision'] = fingerprint({'qty': c['qty'], 'hard_numbers': hn}) try: pms_repo.insert_proposal( proposal_id=pid, ts_code=c["ts_code"], action=c["action"], qty=c["qty"], diff --git a/app/web/main.py b/app/web/main.py index 7735b08..589fdb9 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -517,7 +517,18 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default= except Exception as e: # noqa: BLE001 logger.warning("[提议] 过期清扫失败: %s", e) return {"ok": False, "error": "提议已过失效时刻, 不能再裁决; 该票由系统重新评估"} - if not pms_repo.decide_proposal(proposal_id, decision): + expected_hard = None + if (decision == 'ACCEPTED' and p['action'] == ae.A_OPEN + and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False)): + from app.services.plan_recheck import before_accept + checked = before_accept(p, payload.get('proposal_revision')) + if not checked.get('ok'): + return checked + expected_hard = p.get('hard_numbers_json') + p = checked['proposal'] + changed = (pms_repo.decide_proposal(proposal_id, decision, expected_hard=expected_hard) + if expected_hard is not None else pms_repo.decide_proposal(proposal_id, decision)) + if not changed: return {"ok": False, "error": "提议状态已变更, 请刷新"} hn = p.get("hard_numbers") or {} # 用到的预设理由码进硬数字备查 (复盘要能按码聚合"人常用哪条理由驳回")。 @@ -556,6 +567,8 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default= window_tdays=_win, status="PROPOSED", progress={"from_proposal": proposal_id, "is_command": False, "children": [], + 'plan': {k: hn.get(k) for k in ('decision', 'consumption_id', 'review_ref', + 'execution_rank', 'priority_reason', 'version_check')}, "deadline": str(td.window_deadline(None, _win))}) # 与自主执行同一口径: 采纳即算「做过一次」, 计数器要跟着走 # (否则页面采纳的那条动作绕开了 §6 的一次性约束) @@ -805,12 +818,12 @@ def _research_st(code): @app.get("/api/research/{ts_code}/report") -def api_research_report(ts_code: str): +def api_research_report(ts_code: str, ref: str = None): """个股深度评析全文 (2026-09-18): 上游给的 markdown 由后端代取, 页面自己排版成同款样式。 点击时调, 不进轮询; 按链接缓存十分钟。链接只认上游给的, 不收页面传入的地址。""" from app.services import company_report try: - return company_report.get(cs.normalize_code(ts_code)) + return company_report.get(cs.normalize_code(ts_code), **({'ref': ref} if ref else {})) except Exception as e: # noqa: BLE001 return {"ok": False, "error": f"评析报告读取失败: {type(e).__name__}: {e}"} @@ -907,7 +920,8 @@ def api_upstream_plan(limit: int = Query(30), date: str = Query(None), plan = plan_feed.get_plan(date=(date or None)) rows = plan.get("observe" if bucket == "observe" else "main") or [] out["rows"] = rows[:max(1, int(limit or 30))] - out["candidates_raw"] = plan_feed.candidates() + out['rows'] = [dict(r, decision=plan.get('decision')) for r in out['rows']] + out["candidates_raw"] = plan_feed.candidates(record=False) except Exception as e: out["status"] = {**st, "ok": False, "hint": f"{type(e).__name__}: {e}"} return out diff --git a/app/web/static/index.html b/app/web/static/index.html index 2335c78..f83d414 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -1262,6 +1262,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
+
计划版本与研究变化
{{ line }}
采纳 驳回
@@ -1838,6 +1839,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;} 查回的逻辑状态、第一条出处与证据截止日。状态值是上游给的中文原文, 不是枚举码。 -->