lzf_0626
ff681df143
一、第十五条豁免规则（docs/25 第七节 #2，业务裁定：实现） 政策原文：C3→R4 签署风险揭示书后**可买**，但单只 R4 持仓不超过总资产 20%； C4→R5 同理，上限 10%。越级购买本身不是违规，超出额度才是 —— 原先扫描侧只看 "留痕是否齐全"，于是"签了字但买超额度"这种明确违规没有预警；研判侧也把豁免的 前提条件（留痕齐全）当成了结论，直接判"疑似误报"。 - 扫描侧：新增 EXEMPTION_LIMITS 与 RiskRuleEngine._exemption_state，核算 "单只持仓 / 总资产"并写进证据快照；触发条件改为 gap > 0 and (missing_trace or 超出额度)。 - 研判侧：_assess_rw007 先判额度再判留痕。超限 → 证据支持风险；留痕齐全且在额度 内 → 疑似误报；留痕齐全但快照缺总资产/持仓 → 继续复核。 数据前提（tools/probe_exemption_data.py，证据见 docs/evidence/exemption-data-probe.json）： 库内 fin_customer_profile 仅 1 行且 total_asset = 0.00、fin_holding 0 行、无任何申购 交易 —— 这条规则当前不会被触发，与 behavior_score 同源（画像与持仓由本项目之外的 流程写入）。因此刻意不把"算不出来"当成"超限"：拿 0 去算会让每一笔 C3→R4 都变成违规， 豁免规则反倒成了误报源。上游把数据写入后无需再改代码即可生效。 二、三处业务裁定（此前挂在"待裁定"） - 模型网关 chat + tools 入口：本轮不补，按基座能力缺口记录。它要贯穿 ModelGateway → … → BaseAgent 整条链路，属公共契约变更，演示联调期影响面大于收益。 - exclude（关闭误报）是否必须先"调查中"：保持现状，不加门禁。 - 政策冲突：以第十四条 C ≥ R 为准；客服侧 check_suitability 复核后确认本来就按 C ≥ R 实现，无需改动。 三、其他 - 新增 tests/unit/service/test_risk_judgement_rw007.py（6 例）与扫描侧 4 例。 - 风控文档 03/05 同步 RW-007 的豁免额度条件与研判口径。
102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""只读探查：第十五条豁免规则（持仓占比）所需的数据是否齐备。
|
||
|
||
只做 SELECT。结果写 `docs/evidence/exemption-data-probe.json`：
|
||
|
||
python tools/probe_exemption_data.py
|
||
|
||
要回答的问题：
|
||
1. `fin_customer_profile.total_asset` 有没有值、是否为正；
|
||
2. `fin_holding` 有没有行、`market_value` / `current_value` 是否可用；
|
||
3. 库内是否存在 C3→R4 / C4→R5 的申购交易 —— 即这条豁免规则是否真的会被触发。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.infrastructure.db import SessionFactory
|
||
|
||
OUTPUT = Path("docs/evidence/exemption-data-probe.json")
|
||
|
||
QUERIES: dict[str, str] = {
|
||
"profile_total_asset": """
|
||
SELECT COUNT(*) AS rows_count,
|
||
SUM(total_asset > 0) AS positive_assets,
|
||
SUM(total_asset = 0) AS zero_assets,
|
||
MIN(total_asset) AS min_asset,
|
||
MAX(total_asset) AS max_asset
|
||
FROM fin_customer_profile
|
||
""",
|
||
"profile_investor_type": """
|
||
SELECT investor_type, COUNT(*) AS rows_count
|
||
FROM fin_customer_profile GROUP BY investor_type ORDER BY investor_type
|
||
""",
|
||
"holding_shape": """
|
||
SELECT COUNT(*) AS rows_count,
|
||
SUM(market_value IS NULL) AS null_market_value,
|
||
SUM(current_value > 0) AS positive_current_value,
|
||
MIN(current_value) AS min_current_value,
|
||
MAX(current_value) AS max_current_value
|
||
FROM fin_holding
|
||
""",
|
||
"subscription_pairs": """
|
||
SELECT c.investor_type AS customer_level,
|
||
p.risk_level AS product_level,
|
||
COUNT(*) AS transactions
|
||
FROM fin_transaction t
|
||
JOIN fin_customer_profile c ON c.customer_id = t.customer_id
|
||
JOIN fin_product p ON p.id = t.product_id
|
||
WHERE t.transaction_type = '申购'
|
||
GROUP BY c.investor_type, p.risk_level
|
||
ORDER BY c.investor_type, p.risk_level
|
||
""",
|
||
"exemptible_pairs_detail": """
|
||
SELECT t.transaction_no,
|
||
c.investor_type AS customer_level,
|
||
p.risk_level AS product_level,
|
||
c.total_asset,
|
||
(
|
||
SELECT h.current_value FROM fin_holding h
|
||
WHERE h.customer_id = t.customer_id AND h.product_id = t.product_id
|
||
ORDER BY h.id DESC LIMIT 1
|
||
) AS holding_current_value
|
||
FROM fin_transaction t
|
||
JOIN fin_customer_profile c ON c.customer_id = t.customer_id
|
||
JOIN fin_product p ON p.id = t.product_id
|
||
WHERE t.transaction_type = '申购'
|
||
AND (
|
||
(c.investor_type = 'C3' AND p.risk_level = 'R4')
|
||
OR (c.investor_type = 'C4' AND p.risk_level = 'R5')
|
||
)
|
||
ORDER BY t.id
|
||
""",
|
||
}
|
||
|
||
|
||
async def collect() -> dict[str, Any]:
|
||
report: dict[str, Any] = {}
|
||
async with SessionFactory() as session:
|
||
for name, sql in QUERIES.items():
|
||
rows = (await session.execute(text(sql))).mappings().all()
|
||
report[name] = [dict(row) for row in rows]
|
||
return report
|
||
|
||
|
||
async def main() -> None:
|
||
report = await collect()
|
||
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
||
OUTPUT.write_text(
|
||
json.dumps(report, ensure_ascii=False, indent=2, default=str),
|
||
encoding="utf-8",
|
||
)
|
||
print(f"wrote {OUTPUT}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|