From be7884da46d36b449c520f26006e1315cb095310 Mon Sep 17 00:00:00 2001 From: zlt Date: Fri, 11 Sep 2026 14:33:58 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=89=E6=BA=90=E5=90=88=E8=AE=AE=E5=B7=A5?= =?UTF-8?q?=E4=BD=9C=E5=8C=85=E4=BA=8C(=E4=BA=8C)=EF=BC=9A=E8=A3=85?= =?UTF-8?q?=E9=85=8D=E6=9C=8D=E5=8A=A1=E4=B8=8E=E6=89=B9=E9=87=8F=E8=AF=BB?= =?UTF-8?q?=E6=98=A8=E5=A4=9C=E5=AE=9A=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - downstream_repo.fetch_nightly_verdicts: 批量读 strategy_daily_results.signal_type (择时那票的源) - consensus_service: 给候选/持仓行凑齐基本面/技术面/择时四块意见, hard_keys 折成六个硬数字键 - 参数登记 PMS_FUND_REQUIRED/STALE_DAYS/CONSENSUS_GOOD_MIN、PMS_CONSENSUS_ROUTE - batch28 加装配三例 (脱库桩参数与映射), 现 29 例 装配层不改下单链路。advise_v2 仓位矩阵、扫描接入路由、增持门、页面合议行随后。 Co-Authored-By: Claude Opus 4.8 --- app/repo/downstream_repo.py | 25 ++++++++ app/services/consensus_service.py | 101 ++++++++++++++++++++++++++++++ app/services/param_store.py | 7 +++ scripts/run_tests.py | 4 +- scripts/test_batch28_units.py | 32 ++++++++++ 5 files changed, 167 insertions(+), 2 deletions(-) create mode 100644 app/services/consensus_service.py diff --git a/app/repo/downstream_repo.py b/app/repo/downstream_repo.py index d6f986f..f93aab3 100644 --- a/app/repo/downstream_repo.py +++ b/app/repo/downstream_repo.py @@ -241,6 +241,31 @@ def fetch_refs(ts_code: str): "trade_date": r.get("trade_date")} +def fetch_nightly_verdicts(codes): + """决策系统昨夜定性 (strategy_daily_results.signal_type) 批量: {ts_code: {verdict, trade_date}}。 + 单表访问; 同票多日取最新 trade_date。代码口径与 fetch_refs 一致 (点式转 SH600000)。 + 择时那一票的数据源 (2026-09-11 三源合议)。取不到某票就不在结果里, 由上层按无读数处理。""" + uniq = [c for c in dict.fromkeys(to_prefix(x) for x in (codes or []) if x) if c] + if not uniq: + return {} + out = {} + for i in range(0, len(uniq), 800): + chunk = uniq[i:i + 800] + keys, p = [], {} + for j, c in enumerate(chunk): + keys.append(f":c{j}") + p[f"c{j}"] = c + rows = fetch_all(f"SELECT stock_code, signal_type, trade_date FROM {T_DAILY} " + f"WHERE stock_code IN ({', '.join(keys)})", p) + for r in rows: + code = to_dot(str(r.get("stock_code") or "")) + td_ = r.get("trade_date") + cur = out.get(code) + if cur is None or (td_ is not None and str(td_) > str(cur.get("trade_date") or "")): + out[code] = {"verdict": r.get("signal_type"), "trade_date": td_} + return out + + INDUSTRY_COL_CANDIDATES = ("industry", "category", "sector", "industry_name", "sw_industry", "sw_l1", "sw_l2", "industry_l1", "board_name", "plate_name") CODE_COL_CANDIDATES = ("ts_code", "stock_code", "code", "symbol", "sec_code", "secu_code", diff --git a/app/services/consensus_service.py b/app/services/consensus_service.py new file mode 100644 index 0000000..f6d64ca --- /dev/null +++ b/app/services/consensus_service.py @@ -0,0 +1,101 @@ +# -*- coding: utf-8 -*- +"""三源合议装配 (2026-09-11 工作包二)。把基本面、技术面、择时三票凑齐, 合成方向与路由。 + +三个纯逻辑判定在 core (fund_rules / tech_rules / timing_rules / consensus); 本模块负责取数与装配: + 基本面 —— 候选行 / 持仓行里带的公司深度评析 (company_review), 过 fund_rules。 + 技术面 —— tech_service 早上写的映射 (PMS_TECH_STATE_MAP) 里该票的紧凑立场。 + 择时 —— 决策系统昨夜定性 (downstream.fetch_nightly_verdicts 读 strategy_daily_results.signal_type), + 加当天盘中转多留痕 (flip_at, 由调用方传入, 暂缺就只按昨夜定性)。 +装配后给每行凑出四块意见, 再由 hard_keys 折成提议的六个硬数字键 (方案附录丁)。 +立场词一律中文 (看多/看空/中性/无读数); 任何一路无读数都弃权, 绝不折成看空 (设计原则二)。 +""" +from __future__ import annotations + +import logging + +from app.core import consensus, fund_rules, timing_rules, tradedays +from app.core.command_spec import normalize_code +from app.services import param_store, tech_service + +logger = logging.getLogger("pms.consensus") + + +def _params() -> dict: + gi, gb = param_store.get_int, param_store.get_bool + return { + "fund_required": gb("PMS_FUND_REQUIRED", True), + "fund_stale_days": gi("PMS_FUND_STALE_DAYS", 120), + "fund_consensus_good_min": gi("PMS_FUND_CONSENSUS_GOOD_MIN", 8), + "consensus_route": gb("PMS_CONSENSUS_ROUTE", True), + "timing_stale_tdays": gi("PMS_REF_STALE_TDAYS", 3), + } + + +def _timing_stance(entry, *, today_ymd, stale_tdays, flip_at=None): + """昨夜定性 entry {verdict, trade_date} → timing_rules 立场; trade_date 超期传 fresh=False。""" + if not entry: + return timing_rules.synthesize(None, intraday_flip_at=flip_at) + fresh = True + td_ = entry.get("trade_date") + if td_ is not None: + try: + ymd = int(str(td_).replace("-", "")[:8]) + elapsed = max(0, tradedays.trade_days_left(today_ymd, ymd) - 1) + fresh = elapsed <= int(stale_tdays) + except (ValueError, TypeError): + fresh = True + return timing_rules.synthesize(entry.get("verdict"), intraday_flip_at=flip_at, fresh=fresh) + + +def nightly_map(codes) -> dict: + """批量读昨夜定性 {ts_code: {verdict, trade_date}}。取不到返回空, 不抛 (无读数弃权)。""" + try: + from app.repo import downstream_repo + return downstream_repo.fetch_nightly_verdicts(codes) + except Exception as e: # noqa: BLE001 + logger.warning("[合议] 昨夜定性取不到: %s", e) + return {} + + +def assemble(row, *, nightly=None, tech_states=None, flip_at=None, params=None) -> dict: + """给一行 (候选 / 持仓) 装配四块意见。row 需带 ts_code 与 company_review。不改 row。 + nightly: 该票昨夜定性 entry; tech_states: 技术面映射 (state_map); flip_at: 盘中转多时刻。""" + p = params or _params() + code = normalize_code(row.get("ts_code") or "") + cr = row.get("company_review") if isinstance(row.get("company_review"), dict) else None + + fund = fund_rules.synthesize(cr, params={"stale_days": p["fund_stale_days"], + "consensus_good_min": p["fund_consensus_good_min"]}) + tech = (tech_states or {}).get(code) or {"stance": "无读数", "no_read_why": "没有技术面读数"} + tm = _timing_stance(nightly, today_ymd=tradedays.ymd(), + stale_tdays=p["timing_stale_tdays"], flip_at=flip_at) + # 合议参与分流的开关关掉时, 基本面无读数不由合议这层拦 (回到只按判决分流) + con = consensus.decide(fund["stance"], tech.get("stance") or "无读数", tm["stance"], + tech_phase=tech.get("phase"), + fund_required=(p["fund_required"] and p["consensus_route"])) + return {"fund": fund, "tech": tech, "timing": tm, "consensus": con} + + +def hard_keys(blocks) -> dict: + """四块意见 → 提议硬数字六键 (方案附录丁)。这六键不进 judge.OPEN_JUDGE_KEYS。""" + fund, tech, tm, con = blocks["fund"], blocks["tech"], blocks["timing"], blocks["consensus"] + return { + "fund_stance": {"stance": fund["stance"], "mark": fund.get("mark"), "fact": fund.get("fact")}, + "tech": tech, + "tech_text": tech.get("reason") or tech.get("no_read_why"), + "timing": {"stance": tm["stance"], "nightly": tm.get("nightly"), + "flip_at": tm.get("intraday_flip_at")}, + "consensus": {"direction": con["direction"], "votes": con["votes"], + "strength": con.get("strength"), "reason": con.get("reason"), + "route": con.get("route"), "route_reason": con.get("route_reason")}, + "consensus_text": con.get("reason"), + } + + +def state_map(): + """技术面映射, 透传 tech_service.state_map (装配时批量取一次给所有行用)。""" + try: + return tech_service.state_map() + except Exception as e: # noqa: BLE001 + logger.warning("[合议] 技术面映射取不到: %s", e) + return {} diff --git a/app/services/param_store.py b/app/services/param_store.py index d25ce41..7234c8a 100644 --- a/app/services/param_store.py +++ b/app/services/param_store.py @@ -107,6 +107,11 @@ RUNTIME_EXTRA = { "PMS_TECH_CHOPPY_FLIPS": (4, int, "震荡市判定: 最近 20 个交易日 SAR 翻向达到这么多次即震荡, 翻向不当信号"), "PMS_TECH_FLIP_FRESH_DAYS": (2, int, "SAR 翻向后这么多个交易日内算「刚转向」(转多/转空相位用)"), "PMS_TECH_STATE_MAP": ("", str, "技术面读数映射 (tech_pull 与 08:40 补拉写入的 JSON: 时刻/读数日/逐票紧凑立场), 扫描层与页面只读, 勿手改"), + # ── 2026-09-11 三源合议工作包二 (台账 007/009): 基本面立场与合议分流的开关。 + "PMS_FUND_REQUIRED": (True, bool, "无买方评析不建仓 (拍板: 没有读数或超 120 天的候选一律不建仓)"), + "PMS_FUND_STALE_DAYS": (120, int, "买方评析报告超过这么多天算无读数"), + "PMS_FUND_CONSENSUS_GOOD_MIN": (8, int, "十视角好达这么多个且估值贵 → 打共识已定价标记, 让首批推迟"), + "PMS_CONSENSUS_ROUTE": (True, bool, "三源合议参与新建仓分流 (拍板: 上线即开; 关掉即回到只按选股判决分流)"), } # **读不到时必须按"已暂停"处理的键 (fail-closed)。** @@ -483,6 +488,8 @@ _RANGES = { "PMS_TECH_STALE_TDAYS": (0, 10), "PMS_TECH_SQUEEZE_LOOKBACK": (1, 20), "PMS_TECH_OPEN_BW_GROWTH": (0, 2), "PMS_TECH_CHOPPY_FLIPS": (1, 20), "PMS_TECH_FLIP_FRESH_DAYS": (1, 10), + # 三源合议 (2026-09-11) + "PMS_FUND_STALE_DAYS": (0, 3650), "PMS_FUND_CONSENSUS_GOOD_MIN": (0, 10), # 宏观择时层 "PMS_MACRO_HOT_TH": (0, 100), "PMS_MACRO_COLD_TH": (-100, 0), "PMS_MACRO_EXIT_BAND": (0, 100), "PMS_MACRO_CONFIRM_DAYS": (1, 10), diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 4bfbef6..6d9fe5f 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -74,14 +74,14 @@ 取数解析与代码归一/分页拼接与状态非 OK 不落表/映射只覆盖有当天读数的相关票/ 读数陈旧按无读数/参数登记与交易日差/upsert 用 VALUES 与按 matched 分页两个真机哨兵 (29 例) test_batch28_units.py 三源合议·工作包二纯逻辑 (2026-09-11): 基本面质地档映射与共识已定价/择时定性 - 映射与盘中转多升级/合议多数决与强弱/附录丙路由表逐条 (26 例) + 映射与盘中转多升级/合议多数决与强弱/附录丙路由表逐条/装配服务凑齐四块与六键 (29 例) test_page_enum_guard.py 页面文案守卫 (静态扫描, 不连库不起浏览器): 枚举字段不许 直接印到页面上 / 判据码显示前必须剥前缀 / 不许把整个对象 打给交易员看 / 翻译兜底不许让英文码单独当句子 (1 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) + 目标价到价必定入队 (档位 full 也不自动卖) + 用户设的止损价与目标价单独成列显示 (70 例) - 共 752 例 + 共 755 例 (总数按实跑逐批相加校正过两次: 曾写 649 是笔误, 实为 650; 09-03 先后加了同轮只发一条 减持与研究理由两键各一例, 到 652; 09-04 加了仅展示跳过原因与空候选说明各一例, 到 654; 又加了页面文案守卫一例, 到 655; 09-07 审查修复加了跨轮减持等五例, 到 660; 第二件低把握驳回交人一例, 到 661; 第三件逻辑状态接入第二十二批十五例, 到 676; 第四件安全边际整句透传一例, 到 677; 参考目标价一例, 到 678; 催化事件与定价状态透传一例, 到 679; 两个期限的头一例, 现为 680; 09-10 建议档位对齐第二十六批十七例到 697; 09-11 技术面接入工作包一第二十七批二十七例到 724) 任一子集失败即整体失败 (退出码 1)。 diff --git a/scripts/test_batch28_units.py b/scripts/test_batch28_units.py index 0c6c16f..bfb50df 100644 --- a/scripts/test_batch28_units.py +++ b/scripts/test_batch28_units.py @@ -192,6 +192,38 @@ def _(): assert r["route"] == "放行" and "机械" in r["route_reason"] +# ================================================================ D 装配服务 (脱库, 桩参数与映射) +_ASM_P = {"fund_required": True, "fund_stale_days": 120, "fund_consensus_good_min": 8, + "consensus_route": True, "timing_stale_tdays": 3} + + +@case("D 装配: 三方看多 → 合议看多强放行, 六键齐全") +def _(): + from app.services import consensus_service as csv + row = {"ts_code": "600000.SH", "company_review": _cr(overall="好")} + r = csv.assemble(row, nightly={"verdict": "BUY", "trade_date": 20260910}, + tech_states={"600000.SH": {"stance": "看多", "phase": "趋势多"}}, params=_ASM_P) + assert r["consensus"]["direction"] == "看多" and r["consensus"]["route"] == "放行" + assert set(csv.hard_keys(r).keys()) == {"fund_stance", "tech", "tech_text", + "timing", "consensus", "consensus_text"} + + +@case("D 装配: 无评析 → 基本面无读数, 合议跳过没有买方评析") +def _(): + from app.services import consensus_service as csv + r = csv.assemble({"ts_code": "000001.SZ"}, nightly=None, tech_states={}, params=_ASM_P) + assert r["fund"]["stance"] == "无读数" and r["consensus"]["route"] == "跳过" + + +@case("D 装配: 昨夜定性超期 → 择时无读数, 绝不折看空") +def _(): + from app.services import consensus_service as csv + row = {"ts_code": "600000.SH", "company_review": _cr(overall="好")} + r = csv.assemble(row, nightly={"verdict": "BUY", "trade_date": 20260101}, + tech_states={}, params=_ASM_P) + assert r["timing"]["stance"] == "无读数" + + def main(): ok = 0 for name, fn in RESULTS: