Blame for study.py

"""The study's data and the scelo calls behind every tab of the Space.

Every calculation is a `scelo` call (``sc.load``, ``sc.tab``, ``sc.freq_sev``,
``sc.credibility``, ``sc.glm``, ``sc.lift``, ``sc.gini``, ``sc.basicterm``,
``sc.scr_life``, ``sc.csm``, ``sc.profile``, ``sc.suggest``, ``sc.hard``,
``sc.report``) on the Hub dataset
``intelligentactuaries/synthetic-lapse-surrender-study-2023-2025``, downloaded
once per process through ``huggingface_hub``. Nothing here needs SciPy or
statsmodels: without statsmodels ``sc.glm`` fits with scelo's own IRLS engine,
which gives the published model to machine precision.
"""

from __future__ import annotations

import hashlib
import json
import math
import tempfile
from dataclasses import asdict
from functools import lru_cache
from pathlib import Path

import numpy as np
import pandas as pd
import scelo as sc
from huggingface_hub import hf_hub_download
from scelo import BasicTermAssumptions

DATASET = "intelligentactuaries/synthetic-lapse-surrender-study-2023-2025"
DATASET_URL = f"https://huggingface.co/datasets/{DATASET}"
SPACE_URL = "https://huggingface.co/spaces/intelligentactuaries/lapse-surrender-study-scelo"
SOA_REQUEST_URL = (
    "https://www.soa.org/research/opportunities/2026/"
    "2023-2025-term-whole-life-lapse-sur-exp-study-data-request/"
)
SOA_2015_22_URL = "https://www.soa.org/resources/experience-studies/2024/15-22-twlls/"
SCELO_DOCS_URL = "https://docs.intelligentactuaries.com/scelo/libraries/"

FILES = {
    "vm51_policy_years": "data/vm51_policy_years.csv",
    "policies": "data/policies.csv",
    "agents": "data/agents.csv",
    "term_book": "data/term_book_2023_2025.csv",
    "term_book_inforce": "data/term_book_inforce_2025.csv",
}
ROWS = {"vm51_policy_years": 27_692, "policies": 10_000, "agents": 10_000,
        "term_book": 4_254, "term_book_inforce": 2_698}

MARKETS = {
    "US": "United States", "ZA": "South Africa", "CD": "DR Congo", "PH": "Philippines",
    "SG": "Singapore", "GB": "England", "FR": "France", "PL": "Poland", "IN": "India",
    "NG": "Nigeria",
}
PRODUCTS = ["term_level", "term_decreasing", "term_art", "whole_life_par", "whole_life_nonpar"]
YEARS = ["2023", "2024", "2025"]

# Decrements as the VM-51 policy-year status column records them.
DECREMENTS = {
    "Lapse": ("lapsed",),
    "Surrender": ("surrendered",),
    "Lapse or surrender": ("lapsed", "surrendered"),
    "Reduced paid-up": ("reduced_paid_up",),
    "Expiry": ("expired",),
    "Death (imposed overlay)": ("died",),
}

# Grouping columns offered on the experience tab: label -> column.
EXPERIENCE_BY = {
    "Market": "country",
    "Attained-age band": "age_band",
    "MBTI type": "mbti",
    "Product": "product",
    "Policy duration": "duration_band",
    "Observation year": "observation_year",
    "Sex": "sex",
    "Distribution channel": "channel",
    "Income band": "income_band",
    "Education": "education",
    "Employment": "employment",
    "Urban / rural": "region",
    "Marital status": "marital",
}

# Covariates offered to the lapse GLM: label -> column.
GLM_COVARIATES = {
    "Market": "country",
    "Product": "product",
    "MBTI type": "mbti",
    "Attained-age band": "age_band",
    "Sex": "sex",
    "Distribution channel": "channel",
    "Income band": "income_band",
    "Education": "education",
    "Employment": "employment",
    "Urban / rural": "region",
    "Premium mode": "premium_mode",
    "Smoker status": "smoker",
    "Underwriting": "underwriting",
}
PUBLISHED_COVARIATES = ("country", "product", "mbti")

# Only the VM-51 columns the experience tables need: parsing 16 of 71 columns
# keeps the in-browser load short (the long free-text fields are skipped).
VM51_COLUMNS = [
    "agent_id", "country", "product", "status", "exposure_years", "policy_year",
    "item3_observation_year", "age_band", "mbti", "sex", "channel", "income_band",
    "education", "employment", "region", "marital",
]
DURATION_BANDS = ([0, 1, 2, 3, 5, 10, 15, 20, 10_000], ["1", "2", "3", "4–5", "6–10", "11–15", "16–20", "21+"])


# ─────────────────────────────────────────────────────────────
# Loading
# ─────────────────────────────────────────────────────────────

def fetch(path: str) -> str:
    """Download one file of the dataset (cached on disk by huggingface_hub) and return its path."""
    return hf_hub_download(DATASET, path, repo_type="dataset")


@lru_cache(maxsize=16)
def load(name: str, columns: tuple[str, ...] | None = None) -> pd.DataFrame:
    """A dataset file typed the way Scelo types an import (``sc.load``)."""
    kwargs = {"usecols": list(columns)} if columns else {}
    return sc.load(fetch(FILES[name]), **kwargs)


@lru_cache(maxsize=1)
def facts() -> dict:
    """Every figure the study's reports quote, including the IDE and lifelib results."""
    with open(fetch("analysis/facts.json"), encoding="utf-8") as f:
        return json.load(f)


@lru_cache(maxsize=1)
def experience() -> pd.DataFrame:
    """The VM-51 policy-year records with the grouping columns the experience tab offers."""
    d = load("vm51_policy_years", tuple(VM51_COLUMNS)).copy()
    edges, labels = DURATION_BANDS
    d["duration_band"] = pd.cut(d["policy_year"], edges, labels=labels).astype(str)
    d["observation_year"] = d["item3_observation_year"].astype(int).astype(str)
    return d


# ─────────────────────────────────────────────────────────────
# Experience: occurrence / exposure
# ─────────────────────────────────────────────────────────────

def _poisson_cdf(k: int, lam: float) -> float:
    """P(X ≤ k) for X ~ Poisson(lam), summed in log space."""
    if k < 0:
        return 0.0
    if lam <= 0:
        return 1.0
    j = np.arange(k + 1)
    log_fact = np.concatenate([[0.0], np.cumsum(np.log(np.arange(1, k + 1)))])
    log_p = -lam + j * math.log(lam) - log_fact
    top = log_p.max()
    return float(min(1.0, math.exp(top) * np.exp(log_p - top).sum()))


def _invert_cdf(k: int, target: float) -> float:
    """The λ at which P(X ≤ k; λ) = target (the CDF falls as λ grows); bisection."""
    lo, hi = 0.0, k + 10.0 * math.sqrt(k + 1.0) + 10.0
    for _ in range(80):
        mid = 0.5 * (lo + hi)
        if _poisson_cdf(k, mid) > target:
            lo = mid
        else:
            hi = mid
    return 0.5 * (lo + hi)


def poisson_interval(events, exposure, level: float = 0.95):
    """Exact (Garwood) interval for a Poisson rate, events / exposure, without SciPy."""
    a = (1.0 - level) / 2.0
    lo, hi = [], []
    for d, e in zip(events, exposure):
        d = int(round(d))
        lo.append(0.0 if d == 0 else _invert_cdf(d - 1, 1.0 - a) / e)
        hi.append(_invert_cdf(d, a) / e)
    return np.array(lo), np.array(hi)


def rates(df: pd.DataFrame, by: str, statuses: tuple[str, ...]):
    """Rate per exposure-year by group via ``sc.freq_sev``, with exact 95 % intervals."""
    d = df.assign(event=df["status"].isin(statuses).astype(int))
    t = sc.freq_sev(d, by, count="event", exposure="exposure_years")
    notes = list(getattr(t, "notes", []))
    out = pd.DataFrame({
        by: t[by].astype(str).to_numpy(),
        "exposure": t["exposure"].to_numpy(dtype=float),
        "events": t["claims"].to_numpy(dtype=float),
        "rate": t["frequency"].to_numpy(dtype=float),
    })
    out["lo_95"], out["hi_95"] = poisson_interval(out["events"], out["exposure"])
    return out, notes


def credibility(df: pd.DataFrame, by: str, statuses: tuple[str, ...]):
    """Bühlmann–Straub credibility of each group's rate toward the book (``sc.credibility``)."""
    d = df[df["exposure_years"] > 0].copy()
    d["rate"] = d["status"].isin(statuses).astype(float) / d["exposure_years"]
    t = sc.credibility(d, group=by, value="rate", weight="exposure_years")
    return pd.DataFrame(t).reset_index(), getattr(t, "basis", None), list(getattr(t, "notes", []))


# ─────────────────────────────────────────────────────────────
# Lapse GLM
# ─────────────────────────────────────────────────────────────

@lru_cache(maxsize=2)
def glm_population(name: str) -> pd.DataFrame:
    """One row per policy with ``lapse_claim_count`` and ``exposure`` (policy-years 2023–25).

    ``term`` is the published analysis population (4,254 term policies). ``all``
    is every policy, all products, with the exposure summed from the VM-51
    policy-years and the socio-economic attributes joined from the design.
    """
    if name == "term":
        d = load("term_book").copy()
    else:
        vm = experience()
        expo = vm.groupby("agent_id", as_index=False)["exposure_years"].sum().rename(
            columns={"exposure_years": "exposure"})
        pol = load("policies", ("agent_id", "country", "age_band", "sex", "mbti", "product", "premium_mode",
                                "smoker", "underwriting", "channel", "final_status"))
        attrs = load("agents", ("agent_id", "income_band", "education", "employment", "region", "marital"))
        d = pol.merge(expo, on="agent_id", how="left").merge(attrs, on="agent_id", how="left")
        d = d[d["exposure"] > 0].copy()
    d["lapse_claim_count"] = (d["final_status"] == "lapsed").astype(int)
    return d.reset_index(drop=True)


@lru_cache(maxsize=32)
def fit_glm(population: str, covariates: tuple[str, ...], base: str):
    """Poisson GLM, log link, offset log(exposure) (the published lapse model's form), and its lift."""
    df = glm_population(population)
    formula = "lapse_claim_count ~ " + " + ".join(f"C({c})" for c in covariates)
    model = sc.glm(df, formula, "poisson", offset="exposure", base=base)
    actual = df["lapse_claim_count"] / df["exposure"]
    predicted = np.asarray(model.fitted) / df["exposure"].to_numpy()
    lift = pd.DataFrame(sc.lift(actual, predicted, bins=10, exposure=df["exposure"]))
    gini = float(sc.gini(actual, predicted, exposure=df["exposure"]))
    return model, lift, gini


# ─────────────────────────────────────────────────────────────
# Pricing and valuation of the in-force term book
# ─────────────────────────────────────────────────────────────

@lru_cache(maxsize=1)
def model_points() -> pd.DataFrame:
    """The in-force book as model points, with the IDE project's derived columns."""
    d = load("term_book_inforce").copy()
    d["policy_count"] = 1
    d["sum_assured"] = d["face_amount_usd"]
    d["premium_pp"] = d["annual_premium_usd"] / 12
    lt = d["level_term_years"].fillna(0)
    d["policy_term"] = np.where(lt > 0, lt, 20)
    d["duration_mth"] = (2025 - d["issue_year"]) * 12 + (12 - d["issue_month"])
    return d


@lru_cache(maxsize=1)
def crude_lapse_rate() -> float:
    """Lapses per exposure-year on the term book, 2023–25."""
    tb = load("term_book")
    return float((tb["final_status"] == "lapsed").sum() / tb["exposure"].sum())


@lru_cache(maxsize=64)
def value(assumptions: tuple, ra: float) -> dict:
    """BasicTerm projection, Solvency II life SCR and IFRS 17 CSM on one basis."""
    a = BasicTermAssumptions(**dict(assumptions))
    mp = model_points()
    return {"bt": sc.basicterm(mp, a), "scr": sc.scr_life(mp, a), "csm": sc.csm(mp, a, ra=ra)}


def board_pack(res: dict, a: BasicTermAssumptions, ra: float, basis_label: str) -> str:
    """A stamped board pack (``sc.hard`` + ``sc.report``) of the current valuation; returns its path."""
    pv = res["bt"].attrs["pv"]
    ack = (
        "Acknowledgement and status of the data. The book valued here is synthetic: 10,000 "
        "synthetic policyholders generated by Intelligent Actuaries in response to the Society of "
        "Actuaries (SOA) Research Institute and LIMRA 2023-2025 Term and Whole Life Lapse and "
        "Surrender Experience Study data request, whose scope and VM-51 record layout it imitates. "
        "We gratefully acknowledge the SOA Research Institute and LIMRA. This is not SOA or LIMRA "
        "data, and the work is not affiliated with or endorsed by either organisation. "
        f"Data: {DATASET_URL}"
    )
    summary = (
        f"In-force term book at 31 December 2025, {ROWS['term_book_inforce']:,} model points, lapse basis: "
        f"{basis_label}. PV net cash flow {pv['net']:,.0f} (premiums {pv['premiums']:,.0f}, claims "
        f"{pv['claims']:,.0f}, expenses {pv['expenses']:,.0f}); Solvency II life SCR "
        f"{res['scr'].attrs['scr']:,.0f}; IFRS 17 CSM at initial recognition {res['csm'].attrs['csm0']:,.0f} "
        f"(risk adjustment {ra:.0%} of PV claims). Illustrative basis: not a priced or reserving basis."
    )
    tables = [
        sc.hard(res["bt"], title="BasicTerm projection (scelo port)", assumptions=asdict(a)),
        sc.hard(res["scr"], title="Solvency II life SCR (standard-formula shocks on the projection)"),
        sc.hard(res["csm"], title="IFRS 17 CSM roll-forward (general model)"),
        ack,
    ]
    key = hashlib.sha1(repr((sorted(asdict(a).items()), ra, basis_label)).encode()).hexdigest()[:12]
    path = Path(tempfile.gettempdir()) / "lapse-study-board-packs" / key / "lapse_study_inforce_board_pack.html"
    path.parent.mkdir(parents=True, exist_ok=True)
    sc.report(*tables, title="In-force term book @ 31 Dec 2025 · scelo", to=path, summary=summary,
              author="Intelligent Actuaries · lapse-surrender-study-scelo")
    return str(path)