262 lines
15 KiB
SQL
262 lines
15 KiB
SQL
-- 数据基座因果论断与研判结论只读视图：两张视图，供选股系统的候选卡读"证据线"。
|
||
-- 出处：docs/主观量化系统方案_2026-09-03.md 第三节 3.3"数据基座"表的前两项。
|
||
--
|
||
-- 应用（在运行选股系统容器的机器上执行，容器内的 AKG_PG_* 指向数据基座的 PostgreSQL）：
|
||
-- docker compose exec -T akg-factor-bridge python run.py apply-views --file sql/astock_kg_logic_views.sql --dry-run
|
||
-- docker compose exec -T -e AKG_PG_USER=<属主账号> -e AKG_PG_PASSWORD=<口令> \
|
||
-- akg-factor-bridge python run.py apply-views --file sql/astock_kg_logic_views.sql
|
||
-- 本文件只用 CREATE OR REPLACE VIEW：不建表、不改表、不 DROP，重复应用无害。
|
||
-- 应用时避开每天 06:10 的投影表重建窗口（第一张视图要读投影表）。
|
||
-- 注意 CREATE OR REPLACE VIEW 不允许改动已有列的名字与顺序，只能在末尾追加列；
|
||
-- 将来若要调整列，需要另起视图名或由属主手工 DROP 后重建。
|
||
--
|
||
-- 一、v_factor_logic —— 因果论断，一行一条（客体是环节时展开到成员公司，一条论断多行）
|
||
--
|
||
-- 用途：回答"谁利好或利空这家公司、机制是什么、有什么前提、多长时效、出处在哪"。
|
||
-- 数据来源：断言表 claims 里谓词为 DRIVES 的行（研报因果论断的结构化，只追加、带原文与披露日），
|
||
-- 关联文档表 documents 取标题、来源类型与文档锚定公司。
|
||
--
|
||
-- 口径与数据基座后端接口 /graph/logic/{ts_code} 保持一致的部分：
|
||
-- 1. 那个接口在图上取指向公司节点的 DRIVES 边，边的方向、机制、条件、时效、强度、可信层级、
|
||
-- 置信度都来自断言的 qualifiers 与元数据；本视图直接从断言层取同一批字段，不经过图。
|
||
-- 2. 公司身份的判定仿照数据基座的实体解析器（backend/app/linking/resolver.py）：
|
||
-- 客体原始名经 akg_norm 归一后查别名表 entity_links；曾用名类别名（method 为 namechange）
|
||
-- 要求披露日落在有效期内，缺有效期的曾用名不采用；恰好一个候选才链接；多个候选时若
|
||
-- 文档锚定公司在候选之中则取它，否则视为歧义不链接；客体是自指词（公司、本公司、本集团、
|
||
-- 本行、发行人、上市公司）时取文档锚定公司 documents.meta->>'company_ts_code'。
|
||
-- 与接口不同的部分：
|
||
-- 3. 客体是环节（object_type = 'Segment'）时，接口不展开；本视图按投影表
|
||
-- segment_members_projection 展开到该环节的全部上市成员公司，每个成员一行，
|
||
-- via_segment 列写环节名，link_method 列写 segment。客体是公司时 via_segment 为空。
|
||
-- 4. 接口读的是图上的边，被后续披露取代或封口的边带有 status 与 valid_to；本视图逐条断言出行，
|
||
-- 没有这两个字段（belief 图只在 Neo4j，PostgreSQL 没有 beliefs 表）。作为替代，
|
||
-- disputed 列标出该断言是否处于未结的多空分歧复核（disputes 表 status = 'open'），
|
||
-- review_flag 列带出逻辑评析在 logic_reviews.merged->'flags' 里对该断言的"疑似误抽"标记。
|
||
--
|
||
-- 连接键：
|
||
-- claims.doc_id = documents.doc_id；
|
||
-- akg_norm(claims.object_id) = entity_links.alias_norm，entity_links.canonical_id 即 ts_code；
|
||
-- btrim(claims.object_id) = segment_aliases.alias（环节同义归一，未命中用原名）；
|
||
-- 环节名 = COALESCE(segment_members_projection.orig_segment_name, segment_name)，
|
||
-- 即投影表里的图上原名，与 sql/astock_kg_card_views.sql 的用法相同；
|
||
-- ts_code = company_master.ts_code 取简称（只做展示，不参与过滤）。
|
||
-- 代码格式：ts_code 一律是点后缀式，如 600000.SH。
|
||
--
|
||
-- 已知局限：
|
||
-- a. 投影表每天 06:10 删后插，只存当前态：环节展开得到的成员集合是"今天的"，历史日的论断
|
||
-- 按今天的成员展开，复盘时存在成员前视；复盘应按冻结快照重建。
|
||
-- b. 视图里的公司链接只做了解析器的精确层（T1）与自指、文档破歧两条规则，没有做缩减形态
|
||
-- （剥公司后缀、剥括注，即 T2）与主语亲和规则；akg_norm 也不剥行情尾标。因此视图链接到的
|
||
-- 公司是图上链接结果的子集，未链接的论断不出现在视图里。要看覆盖率，用下面的诊断语句。
|
||
-- c. 客体是概念、行业、未上市实体或未能链接的名字时，没有 ts_code，不出现在视图里。
|
||
-- d. 环节同义归一只查库表 segment_aliases，解析器代码里那几条兜底字典没有复制到 SQL。
|
||
-- e. 一条论断的客体是环节时会展开成多行，claim_id 在视图里不唯一；按论断计数要 DISTINCT claim_id。
|
||
-- f. 不做去重与归并：同一研报同一句话对同一公司只会有一条（断言层有 dedup_key 唯一约束），
|
||
-- 但不同研报的相同论断各算一条，归并留给逻辑评析层。
|
||
--
|
||
-- 诊断（只读，看覆盖）：
|
||
-- SELECT object_type, count(*) FROM claims WHERE predicate = 'DRIVES' GROUP BY 1 ORDER BY 2 DESC;
|
||
-- SELECT link_method, count(DISTINCT claim_id), count(*) FROM v_factor_logic GROUP BY 1;
|
||
-- SELECT max(disclosure_date), count(*) FROM v_factor_logic;
|
||
|
||
CREATE OR REPLACE VIEW v_factor_logic AS
|
||
WITH drives AS (
|
||
SELECT c.claim_id,
|
||
c.subject_type,
|
||
c.subject_id,
|
||
c.object_type,
|
||
c.object_id,
|
||
c.qualifiers,
|
||
c.source_span,
|
||
c.disclosure_date,
|
||
c.confidence,
|
||
c.tier,
|
||
c.doc_id,
|
||
d.title AS doc_title,
|
||
d.source_type,
|
||
d.meta->>'company_ts_code' AS doc_company,
|
||
akg_norm(c.object_id) AS object_norm
|
||
FROM claims c
|
||
JOIN documents d ON d.doc_id = c.doc_id
|
||
WHERE c.predicate = 'DRIVES'
|
||
),
|
||
-- 公司类客体的候选身份：仿解析器的精确层，曾用名过披露日时点闸门。
|
||
company_cands AS (
|
||
SELECT x.claim_id, l.canonical_id
|
||
FROM drives x
|
||
JOIN entity_links l ON l.alias_norm = x.object_norm
|
||
WHERE x.object_type IN ('Company', 'Entity')
|
||
AND (l.method <> 'namechange'
|
||
OR ((l.valid_from IS NOT NULL OR l.valid_to IS NOT NULL)
|
||
AND (l.valid_from IS NULL OR l.valid_from <= x.disclosure_date)
|
||
AND (l.valid_to IS NULL OR x.disclosure_date <= l.valid_to)))
|
||
GROUP BY x.claim_id, l.canonical_id
|
||
),
|
||
company_cand_agg AS (
|
||
SELECT claim_id,
|
||
count(*) AS n_cands,
|
||
min(canonical_id) AS only_id,
|
||
array_agg(canonical_id) AS cand_ids
|
||
FROM company_cands
|
||
GROUP BY claim_id
|
||
),
|
||
company_link AS (
|
||
SELECT x.claim_id,
|
||
CASE
|
||
WHEN x.object_norm IN ('公司', '本公司', '本集团', '本行', '发行人', '上市公司')
|
||
THEN x.doc_company
|
||
WHEN a.n_cands = 1 THEN a.only_id
|
||
WHEN a.n_cands > 1 AND x.doc_company = ANY(a.cand_ids) THEN x.doc_company
|
||
ELSE NULL
|
||
END AS ts_code,
|
||
CASE
|
||
WHEN x.object_norm IN ('公司', '本公司', '本集团', '本行', '发行人', '上市公司')
|
||
THEN 'self_ref'
|
||
WHEN a.n_cands = 1 THEN 'dict'
|
||
WHEN a.n_cands > 1 AND x.doc_company = ANY(a.cand_ids) THEN 'tiebreak_doc'
|
||
WHEN a.n_cands > 1 THEN 'ambiguous'
|
||
ELSE 'unresolved'
|
||
END AS link_method
|
||
FROM drives x
|
||
LEFT JOIN company_cand_agg a ON a.claim_id = x.claim_id
|
||
WHERE x.object_type IN ('Company', 'Entity')
|
||
),
|
||
-- 环节类客体：同义归一后按投影表展开到上市成员。
|
||
segment_rows AS (
|
||
SELECT x.claim_id,
|
||
COALESCE(sa.canonical, btrim(x.object_id)) AS segment_canon
|
||
FROM drives x
|
||
LEFT JOIN segment_aliases sa ON sa.alias = btrim(x.object_id)
|
||
WHERE x.object_type = 'Segment'
|
||
),
|
||
segment_members AS (
|
||
SELECT DISTINCT COALESCE(orig_segment_name, segment_name) AS seg, ts_code
|
||
FROM segment_members_projection
|
||
WHERE ts_code IS NOT NULL AND ts_code <> ''
|
||
),
|
||
linked AS (
|
||
SELECT cl.claim_id, cl.ts_code, NULL::text AS via_segment, cl.link_method
|
||
FROM company_link cl
|
||
WHERE cl.ts_code IS NOT NULL
|
||
UNION ALL
|
||
SELECT sr.claim_id, m.ts_code, sr.segment_canon AS via_segment, 'segment' AS link_method
|
||
FROM segment_rows sr
|
||
JOIN segment_members m ON m.seg = sr.segment_canon
|
||
),
|
||
-- 逻辑评析对单条论断的"疑似误抽"标记（编号已在落库前映射回 claim_id）。
|
||
review_flags AS (
|
||
SELECT f->>'claim_id' AS claim_id_text,
|
||
min(f->>'issue') AS issue
|
||
FROM logic_reviews r
|
||
CROSS JOIN LATERAL jsonb_array_elements(
|
||
CASE WHEN jsonb_typeof(r.merged->'flags') = 'array'
|
||
THEN r.merged->'flags' ELSE '[]'::jsonb END) f
|
||
WHERE f->>'claim_id' IS NOT NULL
|
||
GROUP BY f->>'claim_id'
|
||
)
|
||
SELECT t.ts_code,
|
||
cm.short_name AS company_name,
|
||
t.via_segment,
|
||
t.link_method,
|
||
x.subject_type,
|
||
x.subject_id AS subject_name,
|
||
x.object_type,
|
||
x.object_id AS object_name,
|
||
x.qualifiers->>'direction' AS direction,
|
||
x.qualifiers->>'mechanism' AS mechanism,
|
||
x.qualifiers->>'condition' AS condition,
|
||
x.qualifiers->>'horizon' AS horizon,
|
||
x.qualifiers->>'strength' AS strength,
|
||
x.tier,
|
||
x.confidence,
|
||
x.disclosure_date,
|
||
x.source_type,
|
||
x.doc_id,
|
||
x.doc_title,
|
||
left(x.source_span, 200) AS source_span,
|
||
x.claim_id,
|
||
EXISTS (SELECT 1 FROM disputes dp
|
||
WHERE dp.status = 'open' AND x.claim_id = ANY(dp.claim_ids)) AS disputed,
|
||
rf.issue AS review_flag
|
||
FROM linked t
|
||
JOIN drives x ON x.claim_id = t.claim_id
|
||
LEFT JOIN company_master cm ON cm.ts_code = t.ts_code
|
||
LEFT JOIN review_flags rf ON rf.claim_id_text = x.claim_id::text;
|
||
|
||
|
||
-- 二、v_factor_judgement —— 研判结论，一行一个评析簇
|
||
--
|
||
-- 研判结论已经落表：产业研判面板生成的结论存在 logic_reviews 表，簇键 cluster_key 以
|
||
-- "Industry::主题" 开头（后端 /admin/review_industry 任务写入，/atlas/{theme} 与主题页读取）；
|
||
-- 同一张表还存按受益方聚簇的逻辑评析（"Company:代码"、"Segment:环节名"、"Concept:概念名"，
|
||
-- 由 /admin/review_logic 任务写入，个股逻辑面接口的 review 段读取）。两类同表同结构，
|
||
-- 本视图一并暴露，用 scope 列区分：industry 产业研判、company 个股评析、segment 环节评析、
|
||
-- concept 概念评析，其余 other。
|
||
--
|
||
-- 列的含义：
|
||
-- subject_name 是主题名或受益方原名；ts_code 只在 scope = company 时有值（从簇键截取）；
|
||
-- segment_name 只在 scope = segment 时有值。
|
||
-- leaning 是方向（模型四选一：偏多、偏空、中性、证据不足；产业研判与个股评析同一取值集合）；
|
||
-- core_contradiction、supply_view、demand_view、expectation_shift 只有产业研判有，个股评析为空；
|
||
-- reasons、bull_points、bear_points 是依据，JSON 数组，每项带 text 或 statement 与引用的 claim_ids，
|
||
-- claim_ids 可回到 v_factor_logic 或 claims 表看原文；
|
||
-- verified 是第三轮自我校验结果，false 时 verify_problems 列出校验指出的问题，读的一方应降权；
|
||
-- review_date 与 reviewed_at 是本版结论的生成时间；input_version 是输入材料集合的指纹，
|
||
-- 材料变了指纹就变，可当版本号用；n_materials 是这一版评析读了多少条断言（论断与事件之和）。
|
||
--
|
||
-- 已知局限：
|
||
-- a. logic_reviews 以 cluster_key 唯一，重评时整行覆盖并把 created_at 改为当前时间，
|
||
-- 库里只有每个簇的最新一版，没有历史版本；要保留版本史需要新表，本次不做。
|
||
-- b. 产业研判的簇键是股票池主题名（industry_pools.theme），不是环节名；主题与环节的对应
|
||
-- 关系不在库表里（池成员的 segments 字段是投影的派生），本视图不做主题到环节的映射。
|
||
-- c. 只出有第二轮结果（comparison 非空）的簇；只跑完第一轮归并的簇没有方向，不出行。
|
||
-- d. 结论是模型综合生成的旁批，不是研报原文；数据基座前端对它的说明是"系统综合生成，非研报原文"。
|
||
|
||
CREATE OR REPLACE VIEW v_factor_judgement AS
|
||
SELECT CASE
|
||
WHEN r.cluster_key LIKE 'Industry::%' THEN 'industry'
|
||
WHEN r.cluster_key LIKE 'Company:%' THEN 'company'
|
||
WHEN r.cluster_key LIKE 'Segment:%' THEN 'segment'
|
||
WHEN r.cluster_key LIKE 'Concept:%' THEN 'concept'
|
||
ELSE 'other'
|
||
END AS scope,
|
||
r.beneficiary AS subject_name,
|
||
CASE WHEN r.cluster_key LIKE 'Company:%'
|
||
THEN substr(r.cluster_key, length('Company:') + 1) END AS ts_code,
|
||
-- 双冒号那一支必须写在前面：双冒号的键同样满足单冒号的前缀匹配，
|
||
-- 顺序反了会截出一个多余的冒号。
|
||
-- 双冒号 Segment:: 是 2026-09-04 起按投影环节做的评析，环节名与传导目标同一命名空间；
|
||
-- 单冒号 Segment: 是逐条论断聚簇出来的研报自由短语，在投影里未必查得到这个环节。
|
||
-- scope 的判定不用改：双冒号同样满足 LIKE 'Segment:%'，两者都归 segment。
|
||
CASE WHEN r.cluster_key LIKE 'Segment::%'
|
||
THEN substr(r.cluster_key, length('Segment::') + 1)
|
||
WHEN r.cluster_key LIKE 'Segment:%'
|
||
THEN substr(r.cluster_key, length('Segment:') + 1) END AS segment_name,
|
||
r.comparison->>'leaning' AS leaning,
|
||
r.comparison->'core_contradiction'->>'text' AS core_contradiction,
|
||
r.comparison->'supply_demand'->>'supply' AS supply_view,
|
||
r.comparison->'supply_demand'->>'demand' AS demand_view,
|
||
r.comparison->'expectation_shift'->>'text' AS expectation_shift,
|
||
r.comparison->'reasons' AS reasons,
|
||
r.comparison->'bull' AS bull_points,
|
||
r.comparison->'bear' AS bear_points,
|
||
CASE WHEN jsonb_typeof(r.comparison->'bull') = 'array'
|
||
THEN jsonb_array_length(r.comparison->'bull') ELSE 0 END AS n_bull,
|
||
CASE WHEN jsonb_typeof(r.comparison->'bear') = 'array'
|
||
THEN jsonb_array_length(r.comparison->'bear') ELSE 0 END AS n_bear,
|
||
CASE WHEN jsonb_typeof(r.merged->'flags') = 'array'
|
||
THEN jsonb_array_length(r.merged->'flags') ELSE 0 END AS n_flags,
|
||
r.verified,
|
||
r.comparison->'verify_problems' AS verify_problems,
|
||
r.model,
|
||
r.rounds,
|
||
r.created_at::date AS review_date,
|
||
r.created_at AS reviewed_at,
|
||
r.fingerprint AS input_version,
|
||
COALESCE(array_length(r.claim_ids, 1), 0) AS n_materials,
|
||
r.claim_ids,
|
||
r.review_id,
|
||
r.cluster_key
|
||
FROM logic_reviews r
|
||
WHERE r.comparison IS NOT NULL;
|