171 lines
7.0 KiB
Python
171 lines
7.0 KiB
Python
"""产品推介材料确定性合规检查。"""
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from app.core.promotion_material_contracts import MANDATORY_RISK_DISCLOSURE
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ComplianceFinding:
|
|
rule_code: str
|
|
rule_name: str
|
|
scope: str
|
|
severity: str
|
|
hit_text: str | None
|
|
suggestion: str
|
|
|
|
|
|
BLOCKED_PHRASES = (
|
|
"保本",
|
|
"稳赚",
|
|
"无风险",
|
|
"保证收益",
|
|
"快速募集",
|
|
"抢购",
|
|
"稳赚不赔",
|
|
"本金无忧",
|
|
)
|
|
|
|
|
|
class PromotionComplianceChecker:
|
|
"""不依赖模型的输入和文案规则检查器。"""
|
|
|
|
def check_inputs(self, inputs: dict[str, Any]) -> list[ComplianceFinding]:
|
|
findings: list[ComplianceFinding] = []
|
|
required_paths = (
|
|
("product_info", "fund_type"),
|
|
("product_info", "operation_mode"),
|
|
("product_info", "investment_objective"),
|
|
("manager_info", "management_company"),
|
|
("manager_info", "registration_code"),
|
|
("manager_info", "manager_name"),
|
|
("team_info", "team_description"),
|
|
("strategy_info", "investment_scope"),
|
|
("strategy_info", "strategy"),
|
|
("strategy_info", "restrictions"),
|
|
)
|
|
for group, field in required_paths:
|
|
value = inputs.get(group, {}).get(field)
|
|
if not str(value or "").strip():
|
|
findings.append(ComplianceFinding(
|
|
f"required.{group}.{field}", "必填信息缺失", "input", "block",
|
|
None, f"请补充 {group}.{field}",
|
|
))
|
|
fees = inputs.get("fee_structure", {})
|
|
fee_fields = (
|
|
"subscription_fee",
|
|
"purchase_fee",
|
|
"redemption_fee",
|
|
"sales_service_fee",
|
|
"management_fee",
|
|
"custody_fee",
|
|
"client_maintenance_fee",
|
|
)
|
|
missing_fees = [
|
|
field for field in fee_fields if not str(fees.get(field) or "").strip()
|
|
]
|
|
if missing_fees:
|
|
findings.append(ComplianceFinding(
|
|
"fee_structure.incomplete", "费用结构未完整揭示", "input", "block",
|
|
"、".join(missing_fees), "请逐项填写费率，暂不适用的项目填写“不适用”",
|
|
))
|
|
performance = inputs.get("performance_info", {})
|
|
if performance.get("show_product_performance"):
|
|
months = performance.get("history_months")
|
|
if not isinstance(months, int) or months <= 6:
|
|
findings.append(ComplianceFinding(
|
|
"performance.history_short", "业绩展示区间不超过六个月",
|
|
"input", "block", str(months), "不得展示为本基金历史业绩",
|
|
))
|
|
ranking = performance.get("ranking", {})
|
|
if ranking.get("enabled"):
|
|
years = self._years(ranking.get("evaluation_period_years"))
|
|
if years < 3 or not ranking.get("institution_name") or not ranking.get(
|
|
"public_source"
|
|
):
|
|
findings.append(ComplianceFinding(
|
|
"performance.ranking_source_invalid", "业绩排名来源不满足要求",
|
|
"input", "block", ranking.get("ranking_text"), "补充三年期以上公开评价数据来源",
|
|
))
|
|
shows_performance = (
|
|
performance.get("show_product_performance")
|
|
or performance.get("show_manager_performance")
|
|
)
|
|
if shows_performance and not performance.get("performance_attachment_id"):
|
|
findings.append(ComplianceFinding(
|
|
"performance.data_attachment_missing", "业绩曲线数据附件缺失",
|
|
"input", "block", None, "上传包含产品、基准或代表产品曲线的 CSV/XLSX 文件",
|
|
))
|
|
return findings
|
|
|
|
def check_draft(
|
|
self, draft: dict[str, Any], inputs: dict[str, Any]
|
|
) -> list[ComplianceFinding]:
|
|
text = self._flatten_text(draft)
|
|
findings: list[ComplianceFinding] = []
|
|
for phrase in BLOCKED_PHRASES:
|
|
if phrase in text:
|
|
findings.append(ComplianceFinding(
|
|
"copy.blocked_phrase", "宣传禁用表达", "generated_text", "block",
|
|
phrase, "删除收益承诺、无风险或快速募集导向的表达",
|
|
))
|
|
if MANDATORY_RISK_DISCLOSURE not in text:
|
|
findings.append(ComplianceFinding(
|
|
"disclosure.mandatory", "法定风险声明缺失", "generated_text", "block",
|
|
None, "必须保留完整风险声明",
|
|
))
|
|
manager = inputs.get("manager_info", {})
|
|
if manager.get("employment_years") and manager.get("investment_management_experience"):
|
|
if re.search(r"(从业年限|投资管理经验).{0,8}(等同|即|就是|等于)", text):
|
|
findings.append(ComplianceFinding(
|
|
"manager.experience_distinction", "从业年限与投资管理经验混淆",
|
|
"generated_text", "block", "从业年限/投资管理经验",
|
|
"分别说明两个口径，不得互相替代",
|
|
))
|
|
performance = inputs.get("performance_info", {})
|
|
if (
|
|
performance.get("show_product_performance")
|
|
or performance.get("show_manager_performance")
|
|
):
|
|
months = performance.get("history_months")
|
|
if not isinstance(months, int) or months <= 6:
|
|
findings.append(ComplianceFinding(
|
|
"performance.history_claim", "业绩展示区间不超过六个月",
|
|
"generated_text", "block", "本基金历史业绩", "改为不展示本基金业绩",
|
|
))
|
|
has_return = any(
|
|
performance.get(key)
|
|
for key in ("product_return", "max_drawdown", "volatility", "sharpe_ratio")
|
|
)
|
|
has_risk = all(
|
|
performance.get(key)
|
|
for key in ("max_drawdown", "volatility", "sharpe_ratio")
|
|
)
|
|
shows_performance = (
|
|
performance.get("show_product_performance")
|
|
or performance.get("show_manager_performance")
|
|
)
|
|
if (has_return or shows_performance) and not has_risk:
|
|
findings.append(ComplianceFinding(
|
|
"performance.risk_metrics_missing", "业绩展示缺少风险指标",
|
|
"generated_text", "block", None, "同时展示最大回撤、波动率和夏普比率",
|
|
))
|
|
return findings
|
|
|
|
@staticmethod
|
|
def _years(value: object) -> int:
|
|
match = re.search(r"\d+", str(value or ""))
|
|
return int(match.group()) if match else 0
|
|
|
|
@staticmethod
|
|
def _flatten_text(value: object) -> str:
|
|
if isinstance(value, dict):
|
|
return " ".join(
|
|
PromotionComplianceChecker._flatten_text(item) for item in value.values()
|
|
)
|
|
if isinstance(value, list):
|
|
return " ".join(PromotionComplianceChecker._flatten_text(item) for item in value)
|
|
return str(value or "")
|