验证和联调

sql/astock_kg_slot_views.sql

@@ -1,56 +1,39 @@
--- ============================================================================
+-- astock-kg 因子插槽接口：只读视图（供 akg-factor-bridge 消费）
--- astock-kg 因子插槽接口：只读视图 v2（供 akg-factor-bridge 消费）
+-- 应用：docker compose exec -T akg-factor-bridge python run.py apply-views
--- ----------------------------------------------------------------------------
--- 相对 v1 的变更（2026-07-26 评审）：
---   ① v_factor_transmission：新增 n_sources（distinct 源数）；n_paths 原名原义保留，修「变长边每种
---      长度各返回一条 + 三桶间不去重」导致的路径重复计数。
---   ② v_factor_transmission：暴露 target / target_type / members_total / moved /
---      mkt_trade_date，让桥能自检上游截断与快照新鲜度。
---   ③ v_factor_events：补 doc_id / source_type / tier（供年报污染防护）。
---   ④ v_factor_universe / v_factor_consensus 逻辑未变，只补注释。
 --
--- 应用：docker exec -i akg-postgres psql -U akg -d akg < sql/astock_kg_slot_views.sql
+-- 用 DROP + CREATE 而非 CREATE OR REPLACE：后者不允许改变已有列的名字与顺序
--- 幂等：全部 CREATE OR REPLACE。
+-- （只能在末尾追加），而本次调整了列。已确认这四个视图在 astock-kg 内无任何
---
+-- 消费方（grep v_factor_ 零命中），DROP 安全。
--- ★ 数据安全性：本文件**只创建视图 + 加列 + 加索引**，不写、不改、不删任何一行数据。
+-- 顺序要求：ALTER 必须在引用新列的视图之前。
---   claims / documents 一个字节都不动，之前抽取的成果全部保留，无需 replay、无需重抽。
---   唯一的 DDL 变更是 transmission_candidates 加一个可空列（见文件末尾 ALTER 段），
---   已存行取值 NULL，桥侧按 NULL 容忍（见 factors.build_transmission）。
--- ============================================================================
 
--- 0) 覆盖池 universe = 全部 industry_pools 成员 ts_code 并集
+ALTER TABLE transmission_candidates ADD COLUMN IF NOT EXISTS mkt_trade_date DATE;
---    ⚠️ 注意（评审 §4）：industry_pools 只存最新态、每周一 refresh_pools 自动长大。
+CREATE INDEX IF NOT EXISTS idx_tc_target ON transmission_candidates (target, scan_date);
---    本视图是「当前态」，用它过滤历史子因子会引入成员性前视。
+
---    子因子表建议不过此过滤（桥侧 SUBFACTOR_UNIVERSE=market），只在 akg_score 侧用。
+
-CREATE OR REPLACE VIEW v_factor_universe AS
+-- 覆盖池 universe = 全部 industry_pools 成员并集。
+-- 注意它是当前态：池只存最新快照、每周一 refresh_pools 自动长大。
+DROP VIEW IF EXISTS v_factor_universe;
+CREATE VIEW v_factor_universe AS
 SELECT DISTINCT m->>'ts_code' AS ts_code
 FROM industry_pools p
 CROSS JOIN LATERAL jsonb_array_elements(COALESCE(p.members, '[]'::jsonb)) AS m
 WHERE m->>'ts_code' IS NOT NULL AND m->>'ts_code' <> '';
 
 
--- 1) 一致预期（供 upside）：consensus_daily 全 asof 行投影，桥按 trade_date 做 as-of。
+-- 一致预期（供 upside）：桥按 trade_date 做 as-of backward。
---    n_orgs / n_reports_90d 一并给出——可作 upside 的可信度权重（当前桥未用）。
+DROP VIEW IF EXISTS v_factor_consensus;
-CREATE OR REPLACE VIEW v_factor_consensus AS
+CREATE VIEW v_factor_consensus AS
-SELECT ts_code,
+SELECT ts_code, asof_date, target_mid_avg, eps_med, np_med, n_orgs, n_reports_90d
-       asof_date,
-       target_mid_avg,
-       eps_med,
-       np_med,
-       n_orgs,
-       n_reports_90d
 FROM consensus_daily
 WHERE target_mid_avg IS NOT NULL;
 
 
--- 2) 利好利空事件（供 event）：EVENT 断言投影。
+-- 利好利空事件（供 event）。
---    评审 §6.5：补 doc_id + source_type。年报也带 company_ts_code 锚，而一份年报能抽
+-- source_type 供桥做年报污染防护：年报同样带 company_ts_code 锚，而一份年报能抽
---    十几条 EVENT（且含**历史**诉讼/处罚），桥侧必须按来源过滤或封顶，
+-- 十几条 EVENT 且含历史诉讼/处罚，会在披露日形成负值尖峰。
---    否则 S2 的负面否决闸会因一份年报否决掉一只好股。
+-- doc_id 供单文档封顶。取值：annual_report/research_report/announcement/news/interactive_qa
---    ✅ 列名已核实（db/postgres/init.sql:12）：documents.source_type，取值
+DROP VIEW IF EXISTS v_factor_events;
---       annual_report / research_report / announcement / news / interactive_qa
+CREATE VIEW v_factor_events AS
---    另补 tier（fact/opinion/market）—— 桥侧若想只信事实层事件，这里就够用。
-CREATE OR REPLACE VIEW v_factor_events AS
 SELECT d.meta->>'company_ts_code'   AS ts_code,
        c.disclosure_date,
        c.qualifiers->>'event_type'  AS event_type,
@@ -59,52 +42,34 @@ SELECT d.meta->>'company_ts_code'   AS ts_code,
        c.confidence,
        c.tier,
        c.dedup_key,
-       c.doc_id,                    -- 供「单文档封顶」用
+       c.doc_id,
-       d.source_type                -- 供「只取公告」用
+       d.source_type
 FROM claims c
 JOIN documents d ON d.doc_id = c.doc_id
 WHERE c.predicate = 'EVENT'
   AND d.meta->>'company_ts_code' IS NOT NULL;
 
 
--- 3) 板块传导（供 transmission）：transmission_candidates 的未动成员(quiet)摊平。
+-- 板块传导（供 transmission）：quiet 未动成员摊平成每股一行。
---    ★ 核心修复：n_sources = 指向该环节的 distinct 源数。
+-- n_sources = 指向该环节的 distinct 源数（新）；n_paths = 原路径条数（保留原义）。
---      原 n_paths = jsonb_array_length(paths) 是「被提及次数」——
+-- 二者的比值就是路径重复计数的倍数：cascade 的变长边 *1..N 每种长度各返回一条，
---      cascade() 的变长边 *1..N 会把 A→B 与 A→X→B 各返回一条，
+-- transmission_targets 的 updown/supply/drives 三桶之间也不去重。
---      transmission_targets() 的 updown/supply/drives 三桶之间也不去重，
+-- members_total / n_quiet_stored / mkt_trade_date 供桥自检上游截断与快照新鲜度。
---      于是同一 source 对同一 target 重复计入，而 n_paths 是 factor_value 的主量级。
+DROP VIEW IF EXISTS v_factor_transmission;
---      改成 distinct source 数，同时也更贴合传导模块自述的语义「多源汇聚=逻辑更硬」。
+CREATE VIEW v_factor_transmission AS
-CREATE OR REPLACE VIEW v_factor_transmission AS
 SELECT t.scan_date,
+       q->>'ts_code'                                       AS ts_code,
+       jsonb_array_length(COALESCE(t.paths, '[]'::jsonb))  AS n_paths,
+       t.moved_ratio,
+       (SELECT count(DISTINCT p->>'source')
+          FROM jsonb_array_elements(COALESCE(t.paths, '[]'::jsonb)) p) AS n_sources,
        t.rank,
        t.target,
        t.target_type,
-       q->>'ts_code'  AS ts_code,
+       t.members_total,
-       (SELECT count(DISTINCT p->>'source')
-          FROM jsonb_array_elements(COALESCE(t.paths, '[]'::jsonb)) p)  AS n_sources,
-       jsonb_array_length(COALESCE(t.paths, '[]'::jsonb))               AS n_paths,
-       -- ↑ 原列名原语义保留不动：让 v1 桥代码在 v2 视图上照常工作（联调四象限全兼容），
-       --   同时它本身也是有用的观察列（n_paths / n_sources 的比值 = 路径重复计数的倍数）
-       t.moved_ratio,
-       t.members_total,        -- 桥用：== 上游 topic_context cap 时 moved_ratio 不可信
        t.moved,
-       jsonb_array_length(COALESCE(t.quiet, '[]'::jsonb))              AS n_quiet_stored,
+       jsonb_array_length(COALESCE(t.quiet, '[]'::jsonb))  AS n_quiet_stored,
-       t.mkt_trade_date        -- 桥用：≠ scan_date 说明用了陈旧 movers（可能为 NULL）
+       t.mkt_trade_date
 FROM transmission_candidates t
 CROSS JOIN LATERAL jsonb_array_elements(COALESCE(t.quiet, '[]'::jsonb)) AS q
 WHERE q->>'ts_code' IS NOT NULL AND q->>'ts_code' <> '';
-
-
--- ============================================================================
--- 前置 ALTER（幂等，可反复跑）
--- ----------------------------------------------------------------------------
--- mkt_trade_date：记录本次扫描实际用到的 mkt_daily stock 快照日。
--- 动因（评审 §6.2）：hotspot._latest_mkt 用 `max(trade_date) FROM mkt_daily`，
--- 不是 scan_date 的快照。17:30 sync_market 晚点时，18:15 的传导扫描会用 T−1 的
--- movers 写成 scan_date=T，桥完全无从察觉。落这一列后，桥可以自检并拒收。
--- ============================================================================
-ALTER TABLE transmission_candidates
-    ADD COLUMN IF NOT EXISTS mkt_trade_date DATE;
-
--- 便于按环节回查与审计
-CREATE INDEX IF NOT EXISTS idx_tc_target ON transmission_candidates (target, scan_date);