Download study.py from alidenewade/lapse-surrender-study-scelo: direct link, hf CLI and curl.
- Browser
- Download file 14.2 kB
-
https://huggingface.co/spaces/alidenewade/lapse-surrender-study-scelo/resolve/main/study.py
- Command line
-
hf download hf://spaces/alidenewade/lapse-surrender-study-scelo/study.py
-
curl -L -o study.py https://huggingface.co/spaces/alidenewade/lapse-surrender-study-scelo/resolve/main/study.py
14.2 kB
| """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") | |
| 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) | |
| 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) | |
| 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 | |
| # ───────────────────────────────────────────────────────────── | |
| 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) | |
| 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 | |
| # ───────────────────────────────────────────────────────────── | |
| 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 | |
| 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()) | |
| 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) | |