baodan/api/insurance/ppt/comparison.py

"""多份保险计划书的确定性对比计算。"""
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")
        kinds = {str(p.get("kind") or "").lower() for p in products}
        if len(kinds) > 1:
            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