From 7471b00995cf84ca2deaf2ca48e50f60451bbcba Mon Sep 17 00:00:00 2001 From: Shay Palachy Date: Thu, 11 Jun 2026 23:49:12 +0300 Subject: [PATCH 1/3] feat(lifecycle): motif families + mechanism policies [LTV-Pi] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the lifecycle mechanism policy layer — the parameter tables and dispatch function the weekly simulation engine will call to know how each world behaves. leadforge/schemes/lifecycle/mechanisms.py: - Three frozen dataclasses: ChurnHazardParams, ExpansionPropensityParams, PaymentFailureParams — covering the three mechanism types from design.md §6. - Per-motif parameter tables for all five retention motif families: product_led_retention, relationship_led_retention, expansion_led_growth, payment_fragile, churner_dominated. Each table is calibrated to be structurally coherent with the latent-bias design in population.py: churner_dominated has the highest base churn rate; expansion_led_growth has the lowest churn + highest expansion; payment_fragile has the highest payment-failure rate and lowest recovery rate. - assign_lifecycle_mechanisms(motif_family) → LifecycleMechanismAssignment: the single public entry point that constructs all three param objects. Unknown motif families fall back to defaults rather than raising, so a new family can be prototyped before its tables are calibrated. - mechanism_params_for_motif(): inspection helper returning a plain dict. tests/schemes/lifecycle/test_mechanisms.py (74 tests): - Dispatch: assignment returned for every motif; all three params present; unknown motif falls back gracefully. - Value ranges: rates in (0, 1); renewal multiplier > 1; MRR frac range valid; dunning_weeks >= 1; recovery_rate in [0, 1]. - Structural ordering: churner_dominated has highest churn; expansion_led_growth has highest expansion and lowest churn; payment_fragile has highest payment failure and lowest recovery. - Latent weights non-empty; reference valid lifecycle trait names only. - Frozen dataclasses: mutation raises. - mechanism_params_for_motif: covers all keys and is consistent with assignment. Full suite 1645 passed / 51 skipped; ruff + mypy clean. Co-Authored-By: Claude Sonnet 4.6 --- .agent-plan.md | 7 +- docs/ltv/roadmap.md | 4 +- leadforge/schemes/lifecycle/mechanisms.py | 415 +++++++++++++++++++++ tests/schemes/lifecycle/test_mechanisms.py | 235 ++++++++++++ 4 files changed, 656 insertions(+), 5 deletions(-) create mode 100644 leadforge/schemes/lifecycle/mechanisms.py create mode 100644 tests/schemes/lifecycle/test_mechanisms.py diff --git a/.agent-plan.md b/.agent-plan.md index a9ac0cb..f6f40ce 100644 --- a/.agent-plan.md +++ b/.agent-plan.md @@ -52,9 +52,10 @@ merged (#111); Pg.2 (split lead-scoring schema: entity rows/ALL_ROW_TYPES/ ALL_CONSTRAINTS/LEAD_SNAPSHOT_FEATURES/CONVERTED_WITHIN_90_DAYS moved to `schemes/lead_scoring/`; shared primitives stay in `schema/`) opened as **#112**. All M2 moves byte-identical. Sibling `leadforge-datasets-private` consumes bundle files, not internals — no lockstep update needed (heads-up -issue #8). Next: `LTV-Pg.2` merged (#112). **LTV-M3** started: `LTV-Ph` (lifecycle customer -population builder) opened as **#113**. Next: `LTV-Pi` (lifecycle motif -families + mechanism policies). +issue #8). Next: `LTV-Pg.2` merged (#112). **LTV-M3**: `LTV-Ph` merged (#113); `LTV-Pi` +(lifecycle motif families + mechanism policy — ChurnHazardParams, +ExpansionPropensityParams, PaymentFailureParams, assign_lifecycle_mechanisms() ++ 74 tests) opened as **#NNN**. Next: `LTV-Pj`/`Pk` (simulation engine). --- diff --git a/docs/ltv/roadmap.md b/docs/ltv/roadmap.md index a9eb887..fefc529 100644 --- a/docs/ltv/roadmap.md +++ b/docs/ltv/roadmap.md @@ -157,13 +157,13 @@ Total: ~19 PRs across 9 milestones. > Built directly under `schemes/lifecycle/`. -- [ ] **`LTV-Ph`** — `feat(lifecycle): customer population builder` (**PR #113**). Customer +- [x] **`LTV-Ph`** — `feat(lifecycle): customer population builder` (**PR #113**). Customer entities, 5 new latent traits, **staggered start dates** ending at the absolute `observation_date` (D4); seam for future chained generation (D3). - Tests: determinism, latent distributions, staggered-start spread, FK integrity, acquisition-window boundary. - Labels: `type: feature`, `layer: simulation` -- [ ] **`LTV-Pi`** — `feat(lifecycle): motif families + mechanism policies`. 5 +- [ ] **`LTV-Pi`** — `feat(lifecycle): motif families + mechanism policies` (**PR #NNN**). 5 retention motif families; `assign_lifecycle_mechanisms()` mapping motif → churn/expansion/payment params. - Tests: per-motif param tables, dispatch, determinism. diff --git a/leadforge/schemes/lifecycle/mechanisms.py b/leadforge/schemes/lifecycle/mechanisms.py new file mode 100644 index 0000000..4a4af1e --- /dev/null +++ b/leadforge/schemes/lifecycle/mechanisms.py @@ -0,0 +1,415 @@ +"""Lifecycle mechanism policies — parameter tables and motif dispatch. + +:func:`assign_lifecycle_mechanisms` is the single public entry point. It maps +a retention motif family to a :class:`LifecycleMechanismAssignment` carrying the +concrete parameter values the simulation engine uses on each weekly step. + +The three mechanism types +------------------------- +**Churn hazard** — weekly probability a customer churns. Two-component: + +- *Background rate*: low constant hazard driven by ``latent_product_fit`` + (poor fit → higher churn) and ``latent_champion_strength``. +- *Renewal spike*: at contract-anniversary weeks the hazard multiplies by + ``renewal_hazard_multiplier``; the exact spike is reduced by + ``latent_champion_strength`` (a strong champion fights hard at renewal). + +**Expansion propensity** — weekly probability of an upsell/seat-add event, +driven by ``latent_adoption_velocity`` and ``feature_depth_score`` health +signals. The resulting MRR delta is drawn from +``expansion_mrr_frac_range = (lo, hi)`` × current MRR. + +**Payment failure** — monthly billing event; probability of a failed invoice +driven by ``latent_budget_stability`` (low stability → higher failure rate). +Failed invoices enter a dunning window; unrecovered invoices escalate to churn. + +Motif-family tuning +------------------- +Each of the five retention motif families tilts the base parameters so the DGP +is consistent with the population biases sampled in +:mod:`leadforge.schemes.lifecycle.population`: + +- ``product_led_retention`` — low churn (strong product fit), moderate expansion. +- ``relationship_led_retention`` — moderate churn driven by champion strength at renewal. +- ``expansion_led_growth`` — very low churn, high expansion; pLTV variance from upsell. +- ``payment_fragile`` — moderate-to-high churn triggered by payment failure. +- ``churner_dominated`` — high background churn; strong early-warning signals. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +# --------------------------------------------------------------------------- +# Mechanism assignment dataclasses +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class ChurnHazardParams: + """Parameters for the weekly churn hazard. + + Attributes: + base_weekly_rate: Unconditional weekly churn probability before + latent-score modulation. + latent_weights: ``{trait: weight}`` — positive weights increase churn, + negative weights decrease it. Applied as a sigmoid-adjusted scalar + on top of ``base_weekly_rate``. + renewal_hazard_multiplier: Factor by which the hazard is amplified at a + contract-anniversary week (e.g. ``10.0`` → 10× background). + renewal_latent_weights: Trait weights used *only* at the renewal spike to + model the champion fighting-for-renewal effect (typically only + ``latent_champion_strength``). + """ + + base_weekly_rate: float + latent_weights: dict[str, float] + renewal_hazard_multiplier: float + renewal_latent_weights: dict[str, float] + + +@dataclass(frozen=True) +class ExpansionPropensityParams: + """Parameters for the weekly expansion (upsell / seat-add) propensity. + + Attributes: + base_weekly_rate: Unconditional weekly probability of an expansion event. + latent_weights: Trait weights that scale the base rate. + expansion_mrr_frac_range: ``(lo, hi)`` — expansion MRR delta drawn + uniformly from ``[lo * current_mrr, hi * current_mrr]``. + """ + + base_weekly_rate: float + latent_weights: dict[str, float] + expansion_mrr_frac_range: tuple[float, float] + + +@dataclass(frozen=True) +class PaymentFailureParams: + """Parameters for the monthly payment-failure event. + + Attributes: + base_monthly_rate: Unconditional monthly probability of a payment failure. + latent_weights: Trait weights (negative ``latent_budget_stability`` + increases failure probability). + dunning_weeks: Weeks before a failed invoice is escalated — either + recovered (``payment_recovered``) or written off and triggers churn. + recovery_rate: Probability a failed invoice is recovered within the + dunning window (vs. written off → churn). + """ + + base_monthly_rate: float + latent_weights: dict[str, float] + dunning_weeks: int + recovery_rate: float + + +@dataclass(frozen=True) +class LifecycleMechanismAssignment: + """All mechanism parameters for one lifecycle simulation run. + + Produced by :func:`assign_lifecycle_mechanisms` and consumed by the + weekly simulation engine. + """ + + motif_family: str + churn_hazard: ChurnHazardParams + expansion_propensity: ExpansionPropensityParams + payment_failure: PaymentFailureParams + + +# --------------------------------------------------------------------------- +# Per-motif parameter tables +# --------------------------------------------------------------------------- + +# Churn hazard base weekly rates. Calibrated so annual churn rates at a +# neutral latent score (0.50) sit inside the difficulty-profile bands: +# intro [0.10, 0.20] / intermediate [0.20, 0.35] / advanced [0.30, 0.50]. +# These base rates target the intermediate tier; difficulty scaling is applied +# by the engine on top of them. +_CHURN_BASE_WEEKLY: dict[str, float] = { + "product_led_retention": 0.0042, # ~20% annual + "relationship_led_retention": 0.0055, # ~25% annual + "expansion_led_growth": 0.0028, # ~14% annual (lowest — high-fit customers) + "payment_fragile": 0.0060, # ~27% annual (high base; mostly payment-driven) + "churner_dominated": 0.0090, # ~38% annual +} + +# Latent-trait weights for the background churn hazard. +# Positive weights *increase* churn probability (bad signal); negative *decrease* it. +_CHURN_LATENT_WEIGHTS: dict[str, dict[str, float]] = { + "product_led_retention": { + "latent_product_fit": -2.0, # strong fit → low churn + "latent_adoption_velocity": -0.8, + "latent_champion_strength": -0.5, + }, + "relationship_led_retention": { + "latent_champion_strength": -2.0, # champion quality dominates + "latent_product_fit": -0.8, + "latent_organizational_stability": -0.6, + }, + "expansion_led_growth": { + "latent_adoption_velocity": -1.5, + "latent_product_fit": -1.5, + "latent_budget_stability": -0.5, + }, + "payment_fragile": { + "latent_budget_stability": -2.5, # budget dominates + "latent_organizational_stability": -1.0, + "latent_product_fit": -0.5, + }, + "churner_dominated": { + "latent_product_fit": -1.8, + "latent_champion_strength": -1.2, + "latent_adoption_velocity": -0.8, + }, +} + +# Renewal-date hazard spike multiplier. +_RENEWAL_HAZARD_MULTIPLIER: dict[str, float] = { + "product_led_retention": 6.0, + "relationship_led_retention": 12.0, # renewal is the key decision point + "expansion_led_growth": 4.0, + "payment_fragile": 8.0, + "churner_dominated": 10.0, +} + +# Trait weights at the renewal spike (champion fighting for renewal). +_RENEWAL_LATENT_WEIGHTS: dict[str, dict[str, float]] = { + "product_led_retention": { + "latent_champion_strength": -1.5, + "latent_product_fit": -1.0, + }, + "relationship_led_retention": { + "latent_champion_strength": -2.5, # champion strength matters most here + "latent_organizational_stability": -1.0, + }, + "expansion_led_growth": { + "latent_adoption_velocity": -1.5, + "latent_champion_strength": -1.0, + }, + "payment_fragile": { + "latent_budget_stability": -2.0, + "latent_champion_strength": -1.0, + }, + "churner_dominated": { + "latent_champion_strength": -1.5, + "latent_product_fit": -1.0, + }, +} + +# Expansion propensity base weekly rates. +# Calibrated to yield ~10–30% annual expansion rates at neutral latents. +_EXPANSION_BASE_WEEKLY: dict[str, float] = { + "product_led_retention": 0.0045, # ~21% annual + "relationship_led_retention": 0.0030, # ~15% annual + "expansion_led_growth": 0.0075, # ~32% annual — the pLTV-variance driver + "payment_fragile": 0.0020, # ~10% annual (budget-constrained) + "churner_dominated": 0.0018, # ~9% annual (churners don't expand) +} + +# Latent weights for expansion propensity. +_EXPANSION_LATENT_WEIGHTS: dict[str, dict[str, float]] = { + "product_led_retention": { + "latent_adoption_velocity": 1.5, + "latent_product_fit": 1.0, + }, + "relationship_led_retention": { + "latent_champion_strength": 1.5, + "latent_adoption_velocity": 0.8, + }, + "expansion_led_growth": { + "latent_adoption_velocity": 2.0, + "latent_product_fit": 1.0, + "latent_budget_stability": 0.5, + }, + "payment_fragile": { + "latent_adoption_velocity": 1.0, + "latent_budget_stability": 1.5, # only expands when budget is stable + }, + "churner_dominated": { + "latent_adoption_velocity": 1.0, + "latent_product_fit": 0.8, + }, +} + +# MRR delta fraction range (lo, hi) for expansion events. +# Expansion MRR = randint(lo * current_mrr, hi * current_mrr). +_EXPANSION_MRR_FRAC: dict[str, tuple[float, float]] = { + "product_led_retention": (0.20, 0.60), + "relationship_led_retention": (0.15, 0.50), + "expansion_led_growth": (0.30, 1.00), # large expansions drive the tail + "payment_fragile": (0.10, 0.30), + "churner_dominated": (0.10, 0.25), +} + +# Payment-failure base monthly rates. +_PAYMENT_FAILURE_BASE_MONTHLY: dict[str, float] = { + "product_led_retention": 0.015, + "relationship_led_retention": 0.020, + "expansion_led_growth": 0.012, + "payment_fragile": 0.080, # high — financial fragility is the defining trait + "churner_dominated": 0.030, +} + +# Latent weights for payment failure. +# Negative latent_budget_stability increases failure probability. +_PAYMENT_FAILURE_LATENT_WEIGHTS: dict[str, dict[str, float]] = { + "product_led_retention": { + "latent_budget_stability": -1.5, + }, + "relationship_led_retention": { + "latent_budget_stability": -1.5, + "latent_organizational_stability": -0.5, + }, + "expansion_led_growth": { + "latent_budget_stability": -1.2, + }, + "payment_fragile": { + "latent_budget_stability": -3.0, # dominant driver + "latent_organizational_stability": -1.0, + }, + "churner_dominated": { + "latent_budget_stability": -2.0, + }, +} + +# Dunning period (weeks) before a failed invoice is escalated. +_DUNNING_WEEKS: dict[str, int] = { + "product_led_retention": 4, + "relationship_led_retention": 4, + "expansion_led_growth": 4, + "payment_fragile": 3, # shorter grace — fragile accounts have less runway + "churner_dominated": 3, +} + +# Probability a failed invoice is recovered within the dunning window +# (vs. written off → forced churn). +_RECOVERY_RATE: dict[str, float] = { + "product_led_retention": 0.70, + "relationship_led_retention": 0.65, + "expansion_led_growth": 0.75, + "payment_fragile": 0.40, # low — these accounts are genuinely fragile + "churner_dominated": 0.50, +} + +# Fallback values for unknown motif families. +_DEFAULT_CHURN_BASE_WEEKLY: float = 0.0055 +_DEFAULT_CHURN_LATENT_WEIGHTS: dict[str, float] = { + "latent_product_fit": -1.5, + "latent_champion_strength": -0.8, +} +_DEFAULT_RENEWAL_MULTIPLIER: float = 8.0 +_DEFAULT_RENEWAL_LATENT_WEIGHTS: dict[str, float] = { + "latent_champion_strength": -1.5, +} +_DEFAULT_EXPANSION_BASE_WEEKLY: float = 0.0035 +_DEFAULT_EXPANSION_LATENT_WEIGHTS: dict[str, float] = { + "latent_adoption_velocity": 1.2, +} +_DEFAULT_EXPANSION_MRR_FRAC: tuple[float, float] = (0.20, 0.60) +_DEFAULT_PAYMENT_FAILURE_BASE_MONTHLY: float = 0.025 +_DEFAULT_PAYMENT_FAILURE_LATENT_WEIGHTS: dict[str, float] = { + "latent_budget_stability": -1.5, +} +_DEFAULT_DUNNING_WEEKS: int = 4 +_DEFAULT_RECOVERY_RATE: float = 0.60 + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + + +def assign_lifecycle_mechanisms(motif_family: str) -> LifecycleMechanismAssignment: + """Return a :class:`LifecycleMechanismAssignment` for *motif_family*. + + Looks up pre-calibrated parameter tables and constructs the three mechanism + param objects consumed by the weekly simulation engine. Unrecognised motif + families fall back to sensible intermediate-tier defaults rather than + raising, so a new motif family can be prototyped before its tables are + calibrated. + + Args: + motif_family: One of the five registered lifecycle retention motif + families (see :data:`~leadforge.schemes.lifecycle.population.LIFECYCLE_MOTIF_FAMILIES`). + + Returns: + A fully populated :class:`LifecycleMechanismAssignment`. + """ + churn = ChurnHazardParams( + base_weekly_rate=_CHURN_BASE_WEEKLY.get(motif_family, _DEFAULT_CHURN_BASE_WEEKLY), + latent_weights=dict(_CHURN_LATENT_WEIGHTS.get(motif_family, _DEFAULT_CHURN_LATENT_WEIGHTS)), + renewal_hazard_multiplier=_RENEWAL_HAZARD_MULTIPLIER.get( + motif_family, _DEFAULT_RENEWAL_MULTIPLIER + ), + renewal_latent_weights=dict( + _RENEWAL_LATENT_WEIGHTS.get(motif_family, _DEFAULT_RENEWAL_LATENT_WEIGHTS) + ), + ) + + expansion = ExpansionPropensityParams( + base_weekly_rate=_EXPANSION_BASE_WEEKLY.get(motif_family, _DEFAULT_EXPANSION_BASE_WEEKLY), + latent_weights=dict( + _EXPANSION_LATENT_WEIGHTS.get(motif_family, _DEFAULT_EXPANSION_LATENT_WEIGHTS) + ), + expansion_mrr_frac_range=_EXPANSION_MRR_FRAC.get(motif_family, _DEFAULT_EXPANSION_MRR_FRAC), + ) + + payment = PaymentFailureParams( + base_monthly_rate=_PAYMENT_FAILURE_BASE_MONTHLY.get( + motif_family, _DEFAULT_PAYMENT_FAILURE_BASE_MONTHLY + ), + latent_weights=dict( + _PAYMENT_FAILURE_LATENT_WEIGHTS.get( + motif_family, _DEFAULT_PAYMENT_FAILURE_LATENT_WEIGHTS + ) + ), + dunning_weeks=_DUNNING_WEEKS.get(motif_family, _DEFAULT_DUNNING_WEEKS), + recovery_rate=_RECOVERY_RATE.get(motif_family, _DEFAULT_RECOVERY_RATE), + ) + + return LifecycleMechanismAssignment( + motif_family=motif_family, + churn_hazard=churn, + expansion_propensity=expansion, + payment_failure=payment, + ) + + +def mechanism_params_for_motif(motif_family: str) -> dict[str, Any]: + """Return a plain dict of the mechanism parameter tables for *motif_family*. + + Useful for inspection and testing without constructing mechanism objects. + """ + return { + "motif_family": motif_family, + "churn_base_weekly_rate": _CHURN_BASE_WEEKLY.get(motif_family, _DEFAULT_CHURN_BASE_WEEKLY), + "churn_latent_weights": _CHURN_LATENT_WEIGHTS.get( + motif_family, _DEFAULT_CHURN_LATENT_WEIGHTS + ), + "renewal_hazard_multiplier": _RENEWAL_HAZARD_MULTIPLIER.get( + motif_family, _DEFAULT_RENEWAL_MULTIPLIER + ), + "renewal_latent_weights": _RENEWAL_LATENT_WEIGHTS.get( + motif_family, _DEFAULT_RENEWAL_LATENT_WEIGHTS + ), + "expansion_base_weekly_rate": _EXPANSION_BASE_WEEKLY.get( + motif_family, _DEFAULT_EXPANSION_BASE_WEEKLY + ), + "expansion_latent_weights": _EXPANSION_LATENT_WEIGHTS.get( + motif_family, _DEFAULT_EXPANSION_LATENT_WEIGHTS + ), + "expansion_mrr_frac_range": _EXPANSION_MRR_FRAC.get( + motif_family, _DEFAULT_EXPANSION_MRR_FRAC + ), + "payment_failure_base_monthly_rate": _PAYMENT_FAILURE_BASE_MONTHLY.get( + motif_family, _DEFAULT_PAYMENT_FAILURE_BASE_MONTHLY + ), + "payment_failure_latent_weights": _PAYMENT_FAILURE_LATENT_WEIGHTS.get( + motif_family, _DEFAULT_PAYMENT_FAILURE_LATENT_WEIGHTS + ), + "dunning_weeks": _DUNNING_WEEKS.get(motif_family, _DEFAULT_DUNNING_WEEKS), + "recovery_rate": _RECOVERY_RATE.get(motif_family, _DEFAULT_RECOVERY_RATE), + } diff --git a/tests/schemes/lifecycle/test_mechanisms.py b/tests/schemes/lifecycle/test_mechanisms.py new file mode 100644 index 0000000..c0169ad --- /dev/null +++ b/tests/schemes/lifecycle/test_mechanisms.py @@ -0,0 +1,235 @@ +"""Tests for the lifecycle mechanism policies (LTV-Pi).""" + +import pytest + +from leadforge.schemes.lifecycle.mechanisms import ( + ChurnHazardParams, + ExpansionPropensityParams, + LifecycleMechanismAssignment, + PaymentFailureParams, + assign_lifecycle_mechanisms, + mechanism_params_for_motif, +) +from leadforge.schemes.lifecycle.population import LIFECYCLE_MOTIF_FAMILIES + +# --------------------------------------------------------------------------- +# Basic dispatch +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_assignment_returned_for_each_motif(motif: str) -> None: + assignment = assign_lifecycle_mechanisms(motif) + assert isinstance(assignment, LifecycleMechanismAssignment) + assert assignment.motif_family == motif + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_all_three_mechanisms_present(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + assert isinstance(a.churn_hazard, ChurnHazardParams) + assert isinstance(a.expansion_propensity, ExpansionPropensityParams) + assert isinstance(a.payment_failure, PaymentFailureParams) + + +def test_unknown_motif_falls_back_to_defaults() -> None: + # Unknown families must not raise — they fall back to defaults. + a = assign_lifecycle_mechanisms("nonexistent_motif") + assert a.motif_family == "nonexistent_motif" + assert a.churn_hazard.base_weekly_rate > 0 + assert a.expansion_propensity.base_weekly_rate > 0 + assert a.payment_failure.base_monthly_rate > 0 + + +# --------------------------------------------------------------------------- +# Parameter value ranges +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_churn_base_rate_is_positive_and_subunit(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + r = a.churn_hazard.base_weekly_rate + assert 0.0 < r < 1.0, f"{motif}: base_weekly_rate={r} not in (0, 1)" + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_renewal_multiplier_gt_one(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + m = a.churn_hazard.renewal_hazard_multiplier + assert m > 1.0, f"{motif}: renewal_hazard_multiplier={m} must be > 1" + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_expansion_base_rate_is_positive_and_subunit(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + r = a.expansion_propensity.base_weekly_rate + assert 0.0 < r < 1.0, f"{motif}: expansion base_weekly_rate={r} not in (0, 1)" + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_expansion_mrr_frac_range_is_valid(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + lo, hi = a.expansion_propensity.expansion_mrr_frac_range + assert 0.0 < lo < hi, f"{motif}: expansion_mrr_frac_range ({lo}, {hi}) invalid" + assert hi <= 2.0, f"{motif}: expansion hi={hi} unrealistically large" + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_payment_failure_base_rate_is_positive_and_subunit(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + r = a.payment_failure.base_monthly_rate + assert 0.0 < r < 1.0, f"{motif}: payment base_monthly_rate={r} not in (0, 1)" + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_dunning_weeks_is_positive(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + assert a.payment_failure.dunning_weeks >= 1 + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_recovery_rate_in_unit_interval(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + r = a.payment_failure.recovery_rate + assert 0.0 <= r <= 1.0, f"{motif}: recovery_rate={r}" + + +# --------------------------------------------------------------------------- +# Motif-family structural ordering +# --------------------------------------------------------------------------- + + +def test_expansion_led_growth_has_highest_expansion_rate() -> None: + rates = { + m: assign_lifecycle_mechanisms(m).expansion_propensity.base_weekly_rate + for m in LIFECYCLE_MOTIF_FAMILIES + } + assert rates["expansion_led_growth"] == max(rates.values()), ( + f"expansion_led_growth should have the highest expansion rate; got {rates}" + ) + + +def test_churner_dominated_has_highest_churn_rate() -> None: + rates = { + m: assign_lifecycle_mechanisms(m).churn_hazard.base_weekly_rate + for m in LIFECYCLE_MOTIF_FAMILIES + } + assert rates["churner_dominated"] == max(rates.values()), ( + f"churner_dominated should have the highest base churn rate; got {rates}" + ) + + +def test_expansion_led_growth_has_lowest_churn_rate() -> None: + # expansion_led_growth is calibrated as the lowest-churn world — customers + # that are growing fast are least likely to churn. + rates = { + m: assign_lifecycle_mechanisms(m).churn_hazard.base_weekly_rate + for m in LIFECYCLE_MOTIF_FAMILIES + } + assert rates["expansion_led_growth"] == min(rates.values()), ( + f"expansion_led_growth should have the lowest base churn rate; got {rates}" + ) + + +def test_payment_fragile_has_highest_payment_failure_rate() -> None: + rates = { + m: assign_lifecycle_mechanisms(m).payment_failure.base_monthly_rate + for m in LIFECYCLE_MOTIF_FAMILIES + } + assert rates["payment_fragile"] == max(rates.values()), ( + f"payment_fragile should have the highest payment failure rate; got {rates}" + ) + + +def test_payment_fragile_has_lowest_recovery_rate() -> None: + rates = { + m: assign_lifecycle_mechanisms(m).payment_failure.recovery_rate + for m in LIFECYCLE_MOTIF_FAMILIES + } + assert rates["payment_fragile"] == min(rates.values()), ( + f"payment_fragile should have the lowest recovery rate; got {rates}" + ) + + +# --------------------------------------------------------------------------- +# Latent weights structure +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_churn_latent_weights_non_empty(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + assert len(a.churn_hazard.latent_weights) >= 1 + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_expansion_latent_weights_non_empty(motif: str) -> None: + a = assign_lifecycle_mechanisms(motif) + assert len(a.expansion_propensity.latent_weights) >= 1 + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_churn_weights_reference_valid_lifecycle_traits(motif: str) -> None: + valid_traits = { + "latent_product_fit", + "latent_adoption_velocity", + "latent_budget_stability", + "latent_champion_strength", + "latent_organizational_stability", + } + a = assign_lifecycle_mechanisms(motif) + for trait in a.churn_hazard.latent_weights: + assert trait in valid_traits, f"{motif}: churn weight references unknown trait {trait!r}" + + +# --------------------------------------------------------------------------- +# Frozen dataclasses (immutability) +# --------------------------------------------------------------------------- + + +def test_assignment_is_frozen() -> None: + a = assign_lifecycle_mechanisms("product_led_retention") + with pytest.raises((AttributeError, TypeError)): + a.motif_family = "other" # type: ignore[misc] + + +def test_churn_params_are_frozen() -> None: + a = assign_lifecycle_mechanisms("product_led_retention") + with pytest.raises((AttributeError, TypeError)): + a.churn_hazard.base_weekly_rate = 0.99 # type: ignore[misc] + + +# --------------------------------------------------------------------------- +# mechanism_params_for_motif inspection helper +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_params_dict_covers_all_keys(motif: str) -> None: + params = mechanism_params_for_motif(motif) + expected_keys = { + "motif_family", + "churn_base_weekly_rate", + "churn_latent_weights", + "renewal_hazard_multiplier", + "renewal_latent_weights", + "expansion_base_weekly_rate", + "expansion_latent_weights", + "expansion_mrr_frac_range", + "payment_failure_base_monthly_rate", + "payment_failure_latent_weights", + "dunning_weeks", + "recovery_rate", + } + assert set(params.keys()) == expected_keys + + +def test_params_dict_consistent_with_assignment() -> None: + motif = "expansion_led_growth" + params = mechanism_params_for_motif(motif) + a = assign_lifecycle_mechanisms(motif) + assert params["churn_base_weekly_rate"] == a.churn_hazard.base_weekly_rate + assert params["expansion_base_weekly_rate"] == a.expansion_propensity.base_weekly_rate + assert params["payment_failure_base_monthly_rate"] == a.payment_failure.base_monthly_rate + assert params["recovery_rate"] == a.payment_failure.recovery_rate From fc0b45a931a1f0fae86d7e51c0b271a8736a5b43 Mon Sep 17 00:00:00 2001 From: Shay Palachy Date: Thu, 11 Jun 2026 23:49:47 +0300 Subject: [PATCH 2/3] docs(ltv): record LTV-Pi (#116) in roadmap + agent-plan [LTV-Pi] Co-Authored-By: Claude Sonnet 4.6 --- .agent-plan.md | 2 +- docs/ltv/roadmap.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.agent-plan.md b/.agent-plan.md index f6f40ce..06725bf 100644 --- a/.agent-plan.md +++ b/.agent-plan.md @@ -55,7 +55,7 @@ consumes bundle files, not internals — no lockstep update needed (heads-up issue #8). Next: `LTV-Pg.2` merged (#112). **LTV-M3**: `LTV-Ph` merged (#113); `LTV-Pi` (lifecycle motif families + mechanism policy — ChurnHazardParams, ExpansionPropensityParams, PaymentFailureParams, assign_lifecycle_mechanisms() -+ 74 tests) opened as **#NNN**. Next: `LTV-Pj`/`Pk` (simulation engine). ++ 74 tests) opened as **#116**. Next: `LTV-Pj`/`Pk` (simulation engine). --- diff --git a/docs/ltv/roadmap.md b/docs/ltv/roadmap.md index fefc529..7771447 100644 --- a/docs/ltv/roadmap.md +++ b/docs/ltv/roadmap.md @@ -163,7 +163,7 @@ Total: ~19 PRs across 9 milestones. - Tests: determinism, latent distributions, staggered-start spread, FK integrity, acquisition-window boundary. - Labels: `type: feature`, `layer: simulation` -- [ ] **`LTV-Pi`** — `feat(lifecycle): motif families + mechanism policies` (**PR #NNN**). 5 +- [ ] **`LTV-Pi`** — `feat(lifecycle): motif families + mechanism policies` (**PR #116**). 5 retention motif families; `assign_lifecycle_mechanisms()` mapping motif → churn/expansion/payment params. - Tests: per-motif param tables, dispatch, determinism. From 43e0e376756759b87b7950e82a6878958fc11ae1 Mon Sep 17 00:00:00 2001 From: Shay Palachy Date: Thu, 11 Jun 2026 23:57:31 +0300 Subject: [PATCH 3/3] fix(lifecycle): address self-review findings on mechanism policies [LTV-Pi] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four issues found by hostile self-review of the initial commit: 1. latent_weights dicts were NOT actually immutable (BUG). frozen=True on dataclasses prevents attribute reassignment but not dict mutation. A simulation engine receiving the LifecycleMechanismAssignment could silently corrupt the shared module-level dicts. Fix: wrap every latent_weights field in MappingProxyType so mutation raises TypeError. Added test_latent_weights_dicts_are_truly_immutable. 2. mechanism_params_for_motif duplicated all 12 .get() lookups from assign_lifecycle_mechanisms. Any table change required two edits. Fix: rewrite to call assign_lifecycle_mechanisms() and reflect its fields, guaranteeing consistency by construction. 3. Annual churn rate annotations were systematically inaccurate (off by 0.3–2.5pp). Corrected to 1-(1-r)^52 exact values with a formula note. 4. Structural ordering tests (highest/lowest churn, expansion, failure) pinned absolute rankings across all 5 families and would block valid recalibration. Replaced with directional pair comparisons (e.g. churner_dominated > product_led_retention) and a 2× threshold test for payment_fragile. These express the semantically load-bearing invariants without over-specifying the calibration. Also: added __all__ to declare the public surface. Full suite 1646 passed / 51 skipped; ruff + mypy clean. Co-Authored-By: Claude Sonnet 4.6 --- leadforge/schemes/lifecycle/mechanisms.py | 98 +++++++++++----------- tests/schemes/lifecycle/test_mechanisms.py | 86 ++++++++++--------- 2 files changed, 97 insertions(+), 87 deletions(-) diff --git a/leadforge/schemes/lifecycle/mechanisms.py b/leadforge/schemes/lifecycle/mechanisms.py index 4a4af1e..e8f057c 100644 --- a/leadforge/schemes/lifecycle/mechanisms.py +++ b/leadforge/schemes/lifecycle/mechanisms.py @@ -39,6 +39,7 @@ from __future__ import annotations from dataclasses import dataclass +from types import MappingProxyType from typing import Any # --------------------------------------------------------------------------- @@ -46,6 +47,16 @@ # --------------------------------------------------------------------------- +__all__ = [ + "ChurnHazardParams", + "ExpansionPropensityParams", + "LifecycleMechanismAssignment", + "PaymentFailureParams", + "assign_lifecycle_mechanisms", + "mechanism_params_for_motif", +] + + @dataclass(frozen=True) class ChurnHazardParams: """Parameters for the weekly churn hazard. @@ -53,20 +64,19 @@ class ChurnHazardParams: Attributes: base_weekly_rate: Unconditional weekly churn probability before latent-score modulation. - latent_weights: ``{trait: weight}`` — positive weights increase churn, - negative weights decrease it. Applied as a sigmoid-adjusted scalar - on top of ``base_weekly_rate``. + latent_weights: Read-only ``{trait: weight}`` mapping — positive weights + increase churn, negative decrease it. Wrapped in ``MappingProxyType`` + so the simulation engine cannot accidentally mutate the shared table. renewal_hazard_multiplier: Factor by which the hazard is amplified at a contract-anniversary week (e.g. ``10.0`` → 10× background). renewal_latent_weights: Trait weights used *only* at the renewal spike to - model the champion fighting-for-renewal effect (typically only - ``latent_champion_strength``). + model the champion fighting-for-renewal effect. Also read-only. """ base_weekly_rate: float - latent_weights: dict[str, float] + latent_weights: MappingProxyType # type: ignore[type-arg] renewal_hazard_multiplier: float - renewal_latent_weights: dict[str, float] + renewal_latent_weights: MappingProxyType # type: ignore[type-arg] @dataclass(frozen=True) @@ -75,13 +85,13 @@ class ExpansionPropensityParams: Attributes: base_weekly_rate: Unconditional weekly probability of an expansion event. - latent_weights: Trait weights that scale the base rate. + latent_weights: Read-only trait-weight mapping. expansion_mrr_frac_range: ``(lo, hi)`` — expansion MRR delta drawn uniformly from ``[lo * current_mrr, hi * current_mrr]``. """ base_weekly_rate: float - latent_weights: dict[str, float] + latent_weights: MappingProxyType # type: ignore[type-arg] expansion_mrr_frac_range: tuple[float, float] @@ -91,8 +101,8 @@ class PaymentFailureParams: Attributes: base_monthly_rate: Unconditional monthly probability of a payment failure. - latent_weights: Trait weights (negative ``latent_budget_stability`` - increases failure probability). + latent_weights: Read-only trait-weight mapping (negative + ``latent_budget_stability`` increases failure probability). dunning_weeks: Weeks before a failed invoice is escalated — either recovered (``payment_recovered``) or written off and triggers churn. recovery_rate: Probability a failed invoice is recovered within the @@ -100,7 +110,7 @@ class PaymentFailureParams: """ base_monthly_rate: float - latent_weights: dict[str, float] + latent_weights: MappingProxyType # type: ignore[type-arg] dunning_weeks: int recovery_rate: float @@ -129,11 +139,12 @@ class LifecycleMechanismAssignment: # These base rates target the intermediate tier; difficulty scaling is applied # by the engine on top of them. _CHURN_BASE_WEEKLY: dict[str, float] = { - "product_led_retention": 0.0042, # ~20% annual - "relationship_led_retention": 0.0055, # ~25% annual - "expansion_led_growth": 0.0028, # ~14% annual (lowest — high-fit customers) - "payment_fragile": 0.0060, # ~27% annual (high base; mostly payment-driven) - "churner_dominated": 0.0090, # ~38% annual + # Exact annual equivalent: 1 - (1-r)^52. + "product_led_retention": 0.0042, # 19.7% annual + "relationship_led_retention": 0.0055, # 24.9% annual + "expansion_led_growth": 0.0028, # 13.6% annual (lowest — high-fit customers) + "payment_fragile": 0.0060, # 26.9% annual (high base; mostly payment-driven) + "churner_dominated": 0.0090, # 37.5% annual } # Latent-trait weights for the background churn hazard. @@ -340,18 +351,20 @@ def assign_lifecycle_mechanisms(motif_family: str) -> LifecycleMechanismAssignme """ churn = ChurnHazardParams( base_weekly_rate=_CHURN_BASE_WEEKLY.get(motif_family, _DEFAULT_CHURN_BASE_WEEKLY), - latent_weights=dict(_CHURN_LATENT_WEIGHTS.get(motif_family, _DEFAULT_CHURN_LATENT_WEIGHTS)), + latent_weights=MappingProxyType( + _CHURN_LATENT_WEIGHTS.get(motif_family, _DEFAULT_CHURN_LATENT_WEIGHTS) + ), renewal_hazard_multiplier=_RENEWAL_HAZARD_MULTIPLIER.get( motif_family, _DEFAULT_RENEWAL_MULTIPLIER ), - renewal_latent_weights=dict( + renewal_latent_weights=MappingProxyType( _RENEWAL_LATENT_WEIGHTS.get(motif_family, _DEFAULT_RENEWAL_LATENT_WEIGHTS) ), ) expansion = ExpansionPropensityParams( base_weekly_rate=_EXPANSION_BASE_WEEKLY.get(motif_family, _DEFAULT_EXPANSION_BASE_WEEKLY), - latent_weights=dict( + latent_weights=MappingProxyType( _EXPANSION_LATENT_WEIGHTS.get(motif_family, _DEFAULT_EXPANSION_LATENT_WEIGHTS) ), expansion_mrr_frac_range=_EXPANSION_MRR_FRAC.get(motif_family, _DEFAULT_EXPANSION_MRR_FRAC), @@ -361,7 +374,7 @@ def assign_lifecycle_mechanisms(motif_family: str) -> LifecycleMechanismAssignme base_monthly_rate=_PAYMENT_FAILURE_BASE_MONTHLY.get( motif_family, _DEFAULT_PAYMENT_FAILURE_BASE_MONTHLY ), - latent_weights=dict( + latent_weights=MappingProxyType( _PAYMENT_FAILURE_LATENT_WEIGHTS.get( motif_family, _DEFAULT_PAYMENT_FAILURE_LATENT_WEIGHTS ) @@ -382,34 +395,21 @@ def mechanism_params_for_motif(motif_family: str) -> dict[str, Any]: """Return a plain dict of the mechanism parameter tables for *motif_family*. Useful for inspection and testing without constructing mechanism objects. + Derives directly from :func:`assign_lifecycle_mechanisms` so it is always + consistent with the actual assignment — no duplicated lookup logic. """ + a = assign_lifecycle_mechanisms(motif_family) return { - "motif_family": motif_family, - "churn_base_weekly_rate": _CHURN_BASE_WEEKLY.get(motif_family, _DEFAULT_CHURN_BASE_WEEKLY), - "churn_latent_weights": _CHURN_LATENT_WEIGHTS.get( - motif_family, _DEFAULT_CHURN_LATENT_WEIGHTS - ), - "renewal_hazard_multiplier": _RENEWAL_HAZARD_MULTIPLIER.get( - motif_family, _DEFAULT_RENEWAL_MULTIPLIER - ), - "renewal_latent_weights": _RENEWAL_LATENT_WEIGHTS.get( - motif_family, _DEFAULT_RENEWAL_LATENT_WEIGHTS - ), - "expansion_base_weekly_rate": _EXPANSION_BASE_WEEKLY.get( - motif_family, _DEFAULT_EXPANSION_BASE_WEEKLY - ), - "expansion_latent_weights": _EXPANSION_LATENT_WEIGHTS.get( - motif_family, _DEFAULT_EXPANSION_LATENT_WEIGHTS - ), - "expansion_mrr_frac_range": _EXPANSION_MRR_FRAC.get( - motif_family, _DEFAULT_EXPANSION_MRR_FRAC - ), - "payment_failure_base_monthly_rate": _PAYMENT_FAILURE_BASE_MONTHLY.get( - motif_family, _DEFAULT_PAYMENT_FAILURE_BASE_MONTHLY - ), - "payment_failure_latent_weights": _PAYMENT_FAILURE_LATENT_WEIGHTS.get( - motif_family, _DEFAULT_PAYMENT_FAILURE_LATENT_WEIGHTS - ), - "dunning_weeks": _DUNNING_WEEKS.get(motif_family, _DEFAULT_DUNNING_WEEKS), - "recovery_rate": _RECOVERY_RATE.get(motif_family, _DEFAULT_RECOVERY_RATE), + "motif_family": a.motif_family, + "churn_base_weekly_rate": a.churn_hazard.base_weekly_rate, + "churn_latent_weights": dict(a.churn_hazard.latent_weights), + "renewal_hazard_multiplier": a.churn_hazard.renewal_hazard_multiplier, + "renewal_latent_weights": dict(a.churn_hazard.renewal_latent_weights), + "expansion_base_weekly_rate": a.expansion_propensity.base_weekly_rate, + "expansion_latent_weights": dict(a.expansion_propensity.latent_weights), + "expansion_mrr_frac_range": a.expansion_propensity.expansion_mrr_frac_range, + "payment_failure_base_monthly_rate": a.payment_failure.base_monthly_rate, + "payment_failure_latent_weights": dict(a.payment_failure.latent_weights), + "dunning_weeks": a.payment_failure.dunning_weeks, + "recovery_rate": a.payment_failure.recovery_rate, } diff --git a/tests/schemes/lifecycle/test_mechanisms.py b/tests/schemes/lifecycle/test_mechanisms.py index c0169ad..391c236 100644 --- a/tests/schemes/lifecycle/test_mechanisms.py +++ b/tests/schemes/lifecycle/test_mechanisms.py @@ -100,55 +100,52 @@ def test_recovery_rate_in_unit_interval(motif: str) -> None: # --------------------------------------------------------------------------- -def test_expansion_led_growth_has_highest_expansion_rate() -> None: - rates = { - m: assign_lifecycle_mechanisms(m).expansion_propensity.base_weekly_rate - for m in LIFECYCLE_MOTIF_FAMILIES - } - assert rates["expansion_led_growth"] == max(rates.values()), ( - f"expansion_led_growth should have the highest expansion rate; got {rates}" +def test_expansion_led_growth_expansion_rate_above_churner_dominated() -> None: + # Directional invariant: a growth-led world expands more than a churn-dominated one. + # Uses a pair comparison rather than max() so future recalibration that changes + # which other family has the highest rate won't break this test. + elg = assign_lifecycle_mechanisms("expansion_led_growth").expansion_propensity.base_weekly_rate + cd = assign_lifecycle_mechanisms("churner_dominated").expansion_propensity.base_weekly_rate + assert elg > cd, ( + f"expansion_led_growth expansion rate ({elg}) should exceed churner_dominated ({cd})" ) -def test_churner_dominated_has_highest_churn_rate() -> None: - rates = { - m: assign_lifecycle_mechanisms(m).churn_hazard.base_weekly_rate - for m in LIFECYCLE_MOTIF_FAMILIES - } - assert rates["churner_dominated"] == max(rates.values()), ( - f"churner_dominated should have the highest base churn rate; got {rates}" - ) +def test_churner_dominated_churn_rate_above_product_led_retention() -> None: + # Directional invariant: churner-dominated worlds churn more than product-led ones. + cd = assign_lifecycle_mechanisms("churner_dominated").churn_hazard.base_weekly_rate + plr = assign_lifecycle_mechanisms("product_led_retention").churn_hazard.base_weekly_rate + assert cd > plr, f"churner_dominated churn ({cd}) should exceed product_led_retention ({plr})" -def test_expansion_led_growth_has_lowest_churn_rate() -> None: - # expansion_led_growth is calibrated as the lowest-churn world — customers - # that are growing fast are least likely to churn. - rates = { - m: assign_lifecycle_mechanisms(m).churn_hazard.base_weekly_rate - for m in LIFECYCLE_MOTIF_FAMILIES - } - assert rates["expansion_led_growth"] == min(rates.values()), ( - f"expansion_led_growth should have the lowest base churn rate; got {rates}" - ) +def test_expansion_led_growth_churn_rate_below_churner_dominated() -> None: + # Directional invariant: fast-growing worlds churn less than churn-dominated ones. + elg = assign_lifecycle_mechanisms("expansion_led_growth").churn_hazard.base_weekly_rate + cd = assign_lifecycle_mechanisms("churner_dominated").churn_hazard.base_weekly_rate + assert elg < cd, f"expansion_led_growth churn ({elg}) should be below churner_dominated ({cd})" -def test_payment_fragile_has_highest_payment_failure_rate() -> None: - rates = { - m: assign_lifecycle_mechanisms(m).payment_failure.base_monthly_rate +def test_payment_fragile_failure_rate_substantially_above_others() -> None: + # Directional invariant: payment_fragile failure rate is materially higher + # than any non-fragile world's. Uses 2× threshold rather than max() so + # a recalibration that raises another family's rate modestly won't fail. + pf = assign_lifecycle_mechanisms("payment_fragile").payment_failure.base_monthly_rate + others = [ + assign_lifecycle_mechanisms(m).payment_failure.base_monthly_rate for m in LIFECYCLE_MOTIF_FAMILIES - } - assert rates["payment_fragile"] == max(rates.values()), ( - f"payment_fragile should have the highest payment failure rate; got {rates}" + if m != "payment_fragile" + ] + assert all(pf > 2 * r for r in others), ( + f"payment_fragile ({pf:.4f}) should be >2× all other families' rates: {others}" ) -def test_payment_fragile_has_lowest_recovery_rate() -> None: - rates = { - m: assign_lifecycle_mechanisms(m).payment_failure.recovery_rate - for m in LIFECYCLE_MOTIF_FAMILIES - } - assert rates["payment_fragile"] == min(rates.values()), ( - f"payment_fragile should have the lowest recovery rate; got {rates}" +def test_payment_fragile_recovery_rate_below_product_led_retention() -> None: + # Directional invariant: fragile accounts recover failed payments less often. + pf = assign_lifecycle_mechanisms("payment_fragile").payment_failure.recovery_rate + plr = assign_lifecycle_mechanisms("product_led_retention").payment_failure.recovery_rate + assert pf < plr, ( + f"payment_fragile recovery ({pf}) should be below product_led_retention ({plr})" ) @@ -200,6 +197,19 @@ def test_churn_params_are_frozen() -> None: a.churn_hazard.base_weekly_rate = 0.99 # type: ignore[misc] +def test_latent_weights_dicts_are_truly_immutable() -> None: + # Regression: frozen=True on a dataclass prevents attribute reassignment + # but NOT mutation of a plain dict field. latent_weights are wrapped in + # MappingProxyType so the simulation engine cannot corrupt shared state. + a = assign_lifecycle_mechanisms("product_led_retention") + with pytest.raises(TypeError): + a.churn_hazard.latent_weights["latent_product_fit"] = 999.0 # type: ignore[index] + with pytest.raises(TypeError): + a.expansion_propensity.latent_weights["latent_adoption_velocity"] = 999.0 # type: ignore[index] + with pytest.raises(TypeError): + a.payment_failure.latent_weights["latent_budget_stability"] = 999.0 # type: ignore[index] + + # --------------------------------------------------------------------------- # mechanism_params_for_motif inspection helper # ---------------------------------------------------------------------------