"""多份保险计划书的确定性对比计算。""" from __future__ import annotations from typing import Iterable from insurance.ppt.irr import compute_irr_ma, ia_irr_cap DEFAULT_COMPARISON_YEARS = [5, 10, 20, 30] MAX_COMPARISON_PRODUCTS = 10 SCENARIO_SINGLE_SAVINGS = "single_savings" SCENARIO_MULTI_SAVINGS = "multi_savings_comparison" SCENARIO_SAVINGS_IUL = "savings_iul_comprehensive" SCENARIO_GENERIC_SINGLE = "generic_single" SCENARIO_GENERIC_COMPARE = "generic_compare" SCENARIO_GENERIC_PORTFOLIO = "generic_portfolio" def detect_generation_scenario(products: list[dict]) -> str: """根据正式计划书险种组合确定生成场景，不依赖文件上传顺序。""" kinds = [str(product.get("kind") or "").lower() for product in products] savings_count = kinds.count("savings") iul_count = kinds.count("iul") if len(kinds) == 1 and savings_count == 1: return SCENARIO_SINGLE_SAVINGS if len(kinds) >= 2 and savings_count == len(kinds): return SCENARIO_MULTI_SAVINGS if savings_count >= 1 and iul_count >= 1 and savings_count + iul_count == len(kinds): return SCENARIO_SAVINGS_IUL if len(kinds) == 1: return SCENARIO_GENERIC_SINGLE if len(set(kinds)) == 1: return SCENARIO_GENERIC_COMPARE return SCENARIO_GENERIC_PORTFOLIO def generation_mode_for_scenario(scenario: str) -> str: if scenario in {SCENARIO_MULTI_SAVINGS, SCENARIO_GENERIC_COMPARE}: return "compare" if scenario in {SCENARIO_SAVINGS_IUL, SCENARIO_GENERIC_PORTFOLIO}: return "portfolio" return "single" def build_scenario_slides(products: list[dict], scenario: str) -> list[dict]: """生成场景对应的硬性页面清单；模板只负责视觉，不得删减这些页面。""" if scenario == SCENARIO_SINGLE_SAVINGS: product = products[0] slides = [ {"pageType": "cover", "title": "单份储蓄计划分析"}, { "pageType": "guidance", "title": "先确认客户任务，再讨论产品数字", "bullets": [ "确认可持续投入、资金使用年份与年度现金流目标", "确认退休、教育或传承任务，以及短期流动性要求", "所有结论均以正式计划书数据和客户已确认资料为基础", ], }, {"pageType": "company", "title": "长期承诺需要长期经营能力支撑"}, { "pageType": "guidance", "title": "这份保单承担长期储蓄与计划提取任务", "bullets": [ "前期按计划完成缴费，中期保留积累窗口", "到正式计划书约定年度后再按计划提取", "提取后继续核对剩余退保价值与身故保障", ], }, {"pageType": "timeline", "title": "缴费、积累与提取分成三个阶段"}, {"pageType": "chart", "chartType": "growth", "title": "不提取路径建立长期价值基准"}, {"pageType": "table", "tableType": "no_withdraw", "title": "不提取情景关键年度"}, {"pageType": "compare", "title": "保证与非保证利益必须分开理解"}, { "pageType": "policy_summary", "productIndex": 0, "summaryMode": "no_withdraw", "title": "保单摘要｜不提取", }, ] if product.get("withdrawalRows"): slides.extend([ {"pageType": "table", "tableType": "withdraw", "title": "计划提取情景关键年度"}, { "pageType": "policy_summary", "productIndex": 0, "summaryMode": "withdraw", "title": "保单摘要｜计划提取", }, ]) slides.extend([ { "pageType": "guidance", "title": "流动性来自选择，也伴随成本", "bullets": [ "早期退保价值可能低于累计已缴保费", "提取、贷款或货币转换可能降低后续价值与身故利益", "非保证利益可能高于或低于演示，不能表述为确定收益", ], }, { "pageType": "guidance", "title": "先判断单份计划是否独立适合客户", "bullets": [ "确认缴费能力与独立应急资金", "确认计划使用年份与非保证利益接受程度", "单份计划成立后，再讨论是否需要其他保障或资产工具", ], }, {"pageType": "closing", "title": "下一步：确认目标年份与正式数据"}, ]) return slides if scenario == SCENARIO_MULTI_SAVINGS: slides = [ {"pageType": "cover", "title": "储蓄计划专业对比"}, { "pageType": "guidance", "title": "先建立统一口径，再比较产品差异", "bullets": [ "核对年龄、性别、吸烟状态、币种、年缴保费和缴费年限", "按相同保单年度比较保证价值、非保证利益和总退保价值", "条件不一致时只展示差异，不做产品高低排名", ], }, {"pageType": "comparison_chart", "title": "同一保单年度下的总退保价值走势"}, {"pageType": "comparison_table", "title": "关键年度同时拆分保证与非保证价值"}, ] for index, _product in enumerate(products): slides.append({ "pageType": "policy_summary", "productIndex": index, "summaryMode": "no_withdraw", "title": f"产品 {index + 1} 保单摘要", }) slides.extend([ { "pageType": "guidance", "title": "目标年份决定比较重点", "bullets": [ "缴费完成年度关注早期退出损失", "第10、20年关注中期使用与保证底层", "第25、30年及以后关注退休、长期增长和传承", ], }, { "pageType": "guidance", "title": "提取情景必须取得相同条件的正式计划书", "bullets": [ "开始年度、结束年度、年度金额、频率与币种必须一致", "不得用不提取数据推算提取后的价值", "缺失数据统一标记“待正式计划书确认”", ], }, {"pageType": "conclusion", "title": "结论取决于目标年份、确定性与功能需求"}, {"pageType": "closing", "title": "下一步：统一条件后完成最终选择"}, ]) return slides if scenario == SCENARIO_SAVINGS_IUL: slides = [ {"pageType": "cover", "title": "储蓄现金流与 IUL 保障综合方案"}, { "pageType": "guidance", "title": "先讲家庭任务，再安排两份保单", "bullets": [ "储蓄计划负责建立可持续提取的现金流仓", "IUL 负责放大身故保障并承接长期家庭责任", "两份保单按现金流发生年度对应，不按年龄强行对齐", ], }, {"pageType": "timeline", "title": "先看缴费、提取与保障启动时间"}, {"pageType": "synergy", "title": "组合不是并列持有，而是现金流接力"}, ] for index, product in enumerate(products): slides.extend([ { "pageType": "chart", "productIndex": index, "chartType": "growth", "title": f"{product.get('productName') or f'方案 {index + 1}'}价值演示", }, { "pageType": "policy_summary", "productIndex": index, "summaryMode": "no_withdraw", "title": f"{product.get('productName') or f'方案 {index + 1}'}保单摘要", }, ]) slides.extend([ {"pageType": "cashflow_bridge", "title": "储蓄提取如何承接 IUL 计划保费"}, {"pageType": "launch_paths", "title": "组合方案必须同时呈现两种启动路径"}, {"pageType": "combined_summary", "title": "组合保单摘要"}, {"pageType": "alignment_table", "title": "两份计划书按保单年度对应"}, { "pageType": "guidance", "title": "风险与执行条件必须说清楚", "bullets": [ "储蓄保单未到正式提取年度前，IUL 保费属于客户自有资金", "保证价值与非保证演示价值必须分开呈现", "所有金额来自正式计划书，缺失数字不自行推算", ], }, {"pageType": "conclusion", "title": "选择保障先行或现金流同步"}, {"pageType": "closing", "title": "下一步：确认启动路径与前期现金流"}, ]) return slides page_types = ["cover", "company", "narrative", "chart"] if len(products) > 1: page_types.append("compare") if scenario != SCENARIO_GENERIC_COMPARE: page_types.append("synergy") page_types.extend(["table", "conclusion", "closing"]) return [{"pageType": page_type} for page_type in page_types] def build_comparison_contract( products: list[dict], mode: str = "single", comparison_years: Iterable[int] | None = None, ) -> dict: """将归一化产品转换为渲染器可直接使用的对比契约。""" normalized_mode = (mode or "single").lower() if normalized_mode not in {"single", "compare", "portfolio"}: raise ValueError("generationMode 必须是 single、compare 或 portfolio") if normalized_mode == "single": return {"mode": "single", "products": [], "years": [], "warnings": []} if len(products) < 2: raise ValueError("产品对比或组合方案至少需要 2 份有效计划书") if len(products) > MAX_COMPARISON_PRODUCTS: raise ValueError(f"一次最多对比 {MAX_COMPARISON_PRODUCTS} 份计划书") years = _normalize_years(comparison_years) kinds = {str(p.get("kind") or "").lower() for p in products} currencies = { str((p.get("policy") or {}).get("currency") or "").upper() for p in products if (p.get("policy") or {}).get("currency") } warnings: list[str] = [] if normalized_mode == "compare": if len(kinds) != 1: raise ValueError("不同险种不能直接横向排名，请选择“组合方案”模式") if len(currencies) > 1: raise ValueError("不同币种不能直接比较，请上传相同币种的计划书") elif len(currencies) > 1: warnings.append("组合中包含不同币种，金额仅分别展示，不计算合计") insured_signatures = { ( (p.get("insured") or {}).get("age"), str((p.get("insured") or {}).get("gender") or "").lower(), ) for p in products } if len(insured_signatures) > 1: warnings.append("计划书的受保人年龄或性别不一致，比较结论需谨慎使用") return { "mode": normalized_mode, "years": years, "currency": next(iter(currencies), ""), "productKinds": sorted(kinds), "products": [_build_product_metrics(product, years) for product in products], "warnings": warnings, } def _normalize_years(years: Iterable[int] | None) -> list[int]: values = [] for value in years or DEFAULT_COMPARISON_YEARS: try: year = int(value) except (TypeError, ValueError): continue if 1 <= year <= 100 and year not in values: values.append(year) return sorted(values) or list(DEFAULT_COMPARISON_YEARS) def _build_product_metrics(product: dict, years: list[int]) -> dict: policy = product.get("policy") or {} rows = product.get("benefitRows") or [] annual_premium = _number(policy.get("annualPremium")) pay_years = int(_number(policy.get("payYears"))) total_premium = _number( policy.get("contractualTotalPremium") or policy.get("totalPremium") or annual_premium * pay_years ) currency = str(policy.get("currency") or "").upper() year_values = {} for year in years: row = _find_year(rows, year) if not row: year_values[str(year)] = None continue total_value = _number(row.get("totalSurrenderValue")) guaranteed_value = _number(row.get("guaranteedCashValue")) death_benefit = _number( row.get("nonGuaranteedDeathBenefit") or row.get("deathBenefit") or row.get("guaranteedDeathBenefit") ) irr = compute_irr_ma( annual_premium, pay_years, total_value, year, ia_irr_cap(currency or "HKD"), ) year_values[str(year)] = { "guaranteedValue": guaranteed_value, "totalValue": total_value, "deathBenefit": death_benefit, "irr": round(irr * 100, 2) if irr is not None else None, "sourcePage": row.get("sourcePage"), } return { "fileId": product.get("fileId", ""), "pdfName": product.get("pdfName", ""), "companyId": product.get("companyId", ""), "productName": product.get("productName", ""), "kind": product.get("kind", "savings"), "currency": currency, "annualPremium": annual_premium, "payYears": pay_years, "totalPremium": total_premium, "sumInsured": _number(policy.get("sumInsured")), "breakevenYear": _find_breakeven_year(rows), "yearValues": year_values, } def _find_year(rows: list[dict], year: int) -> dict | None: for row in rows: if int(_number(row.get("policyYear"))) == year: return row return None def _find_breakeven_year(rows: list[dict]) -> int | None: for row in rows: premium = _number(row.get("totalPremiumPaid")) value = _number(row.get("totalSurrenderValue")) if premium > 0 and value >= premium: return int(_number(row.get("policyYear"))) return None def _number(value) -> float: try: return float(value or 0) except (TypeError, ValueError): return 0.0