diff --git a/.agent-plan.md b/.agent-plan.md index 06725bf..384839e 100644 --- a/.agent-plan.md +++ b/.agent-plan.md @@ -53,9 +53,11 @@ 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**: `LTV-Ph` merged (#113); `LTV-Pi` -(lifecycle motif families + mechanism policy — ChurnHazardParams, -ExpansionPropensityParams, PaymentFailureParams, assign_lifecycle_mechanisms() -+ 74 tests) opened as **#116**. Next: `LTV-Pj`/`Pk` (simulation engine). +(mechanism policies) merged (#116) — **LTV-M3 complete**. **LTV-M4** +started: `LTV-Pj` (hazard functions — churn_probability with onboarding +elevation + renewal spike, expansion_probability with health modulation, +payment_failure_probability; pure/deterministic, Cox-style latent multipliers; +40 tests) opened as **#117**. Next: `LTV-Pk` (weekly simulation engine). --- diff --git a/docs/ltv/roadmap.md b/docs/ltv/roadmap.md index 7771447..a68327c 100644 --- a/docs/ltv/roadmap.md +++ b/docs/ltv/roadmap.md @@ -44,7 +44,7 @@ protocol + registry, with the package physically reorganized into | `LTV-M1` | Lifecycle schema foundation | `LTV-Pb`, `LTV-Pc` | #104 (Pb) | | `LTV-M2` | Generation-scheme architecture + physical reorg | `LTV-Pd`, `LTV-Pe`, `LTV-Pf`, `LTV-Pg` | #107 (Pd), #108 (Pe), #109 (Pf.1), #110 (Pf.2), #111 (Pg.1), #112 (Pg.2) | | `LTV-M3` | Customer population + lifecycle world | `LTV-Ph`, `LTV-Pi` | #113 (Ph) | -| `LTV-M4` | Lifecycle simulation engine | `LTV-Pj`, `LTV-Pk` | | +| `LTV-M4` | Lifecycle simulation engine | `LTV-Pj`, `LTV-Pk` | #117 (Pj) | | `LTV-M5` | Customer snapshots + pLTV targets (both regimes) | `LTV-Pl`, `LTV-Pm` | | | `LTV-M6` | Register LifecycleScheme + recipe + manifest/version | `LTV-Pn`, `LTV-Po` | | | `LTV-M7` | Validation + regression-metric calibration | `LTV-Pp` | | @@ -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 #116**). 5 +- [x] **`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. @@ -173,7 +173,7 @@ Total: ~19 PRs across 9 milestones. ## `LTV-M4` — Lifecycle simulation engine -- [ ] **`LTV-Pj`** — `feat(lifecycle): churn / expansion / payment hazards`. +- [ ] **`LTV-Pj`** — `feat(lifecycle): churn / expansion / payment hazards` (**PR #117**). Weibull churn hazard with renewal-date spike, expansion propensity (the heavy-tail generator for pLTV), payment failure + dunning. - Tests: hazard shape over tenure, renewal spike, dunning escalation, diff --git a/leadforge/schemes/lifecycle/hazards.py b/leadforge/schemes/lifecycle/hazards.py new file mode 100644 index 0000000..a805863 --- /dev/null +++ b/leadforge/schemes/lifecycle/hazards.py @@ -0,0 +1,219 @@ +"""Lifecycle hazard functions — latent state + mechanism params → probabilities. + +Three pure functions convert a customer's latent traits and the motif-family +mechanism parameters (from :mod:`leadforge.schemes.lifecycle.mechanisms`) into +per-step event probabilities: + +- :func:`churn_probability` — weekly churn hazard with onboarding elevation and + a contract-anniversary renewal spike. +- :func:`expansion_probability` — weekly upsell/seat-add propensity, optionally + modulated by the current feature-depth health signal. +- :func:`payment_failure_probability` — monthly invoice-failure probability. + +All three are **deterministic** — they take no RNG and perform no draws. The +weekly simulation engine owns every Bernoulli sample; these functions only +compute the probability for the draw. This keeps the hazard math directly +testable (exact values, monotonicity, spike shape) without seeding. + +Latent modulation (proportional-hazards style) +---------------------------------------------- +Each mechanism's ``latent_weights`` are applied as a Cox-style multiplicative +factor on the base rate:: + + multiplier = exp( Σ_i w_i · (latent_i − 0.5) ) + +Latents are centred at the neutral 0.5, so a customer with all-neutral traits +gets multiplier 1.0 (base rate unchanged). Per the sign convention set in +``mechanisms.py``: a **negative** weight on a trait means a *high* trait value +*reduces* the probability (e.g. ``latent_product_fit: -2.0`` on churn). +A trait missing from the latent dict is treated as neutral (0.5) — it +contributes nothing, rather than raising or silently zeroing the hazard. + +Tenure shape +------------ +The churn hazard is elevated during onboarding (decreasing-hazard Weibull +behaviour, approximated by an exponential decay from +``_ONBOARDING_PEAK_MULTIPLIER`` toward 1.0 with time-constant +``_ONBOARDING_DECAY_WEEKS``) and spikes at each contract anniversary +(:func:`is_renewal_week`), where the ``renewal_hazard_multiplier`` and the +renewal-specific latent weights (champion-fights-for-renewal) apply. +""" + +from __future__ import annotations + +import math +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Mapping + + from leadforge.schemes.lifecycle.mechanisms import ( + ChurnHazardParams, + ExpansionPropensityParams, + PaymentFailureParams, + ) + +__all__ = [ + "MAX_PROBABILITY", + "churn_probability", + "expansion_probability", + "is_renewal_week", + "payment_failure_probability", +] + +# Probabilities are capped below 1.0 so extreme latent combinations never make +# an event *certain* — the simulation stays stochastic at the tails, and a +# mis-calibrated base rate degrades visibly instead of saturating silently. +# Public: the cap is part of the hazard contract (every function documents it) +# and the engine/tests may reference it. +MAX_PROBABILITY = 0.95 + +# Neutral latent value: traits absent from the latent dict contribute nothing. +# NOTE: latents are *not* range-validated in these hot-path functions — the +# population builder clamps all traits to [0, 1], and MAX_PROBABILITY bounds +# the damage of any out-of-range value. Validating per call would cost real +# time at ~customers x weeks call volume in the engine loop. +_NEUTRAL_LATENT = 0.5 + +# Floor of the feature-depth expansion multiplier: depth d maps to a factor of +# (_DEPTH_MULTIPLIER_FLOOR + d), i.e. x0.5 at zero depth, x1.0 at the neutral +# depth 0.5, x1.5 at full depth. Coincidentally equal to _NEUTRAL_LATENT but +# semantically unrelated — kept as its own constant so recentring latents can +# never silently change the health modulation. +_DEPTH_MULTIPLIER_FLOOR = 0.5 + +# Onboarding churn elevation: hazard starts at peak × base in week 0 and decays +# exponentially toward 1× with this time-constant. At week 12 the residual +# elevation is < 8% — effectively steady-state. Deliberately uniform across +# motif families (like the lead-scoring follow-up ramp): onboarding instability +# is a customer-success process constant; per-motif differentiation comes from +# the latent weights and base rates, not the tenure shape. +_ONBOARDING_PEAK_MULTIPLIER = 2.5 +_ONBOARDING_DECAY_WEEKS = 4.0 + +# Weeks per month for contract-anniversary arithmetic (52-week year). +_WEEKS_PER_MONTH = 52.0 / 12.0 + + +def _latent_multiplier(latents: Mapping[str, float], weights: Mapping[str, float]) -> float: + """Return the Cox-style multiplicative factor for *latents* under *weights*.""" + score = sum( + weight * (latents.get(trait, _NEUTRAL_LATENT) - _NEUTRAL_LATENT) + for trait, weight in weights.items() + ) + return math.exp(score) + + +def _onboarding_multiplier(week_of_tenure: int) -> float: + """Return the early-tenure churn elevation factor (≥ 1.0, → 1.0 with tenure).""" + return 1.0 + (_ONBOARDING_PEAK_MULTIPLIER - 1.0) * math.exp( + -week_of_tenure / _ONBOARDING_DECAY_WEEKS + ) + + +def is_renewal_week(week_of_tenure: int, contract_term_months: int) -> bool: + """Return ``True`` iff *week_of_tenure* contains a contract anniversary. + + Anniversaries fall at ``round(k · contract_term_months · 52/12)`` weeks for + ``k = 1, 2, …`` — e.g. a 12-month contract renews at weeks 52, 104, …; a + 24-month contract at weeks 104, 208, …. Week 0 (signing week) is never a + renewal week. + + Exposed publicly so the simulation engine can use the same boundary to emit + ``renewal`` events that it uses for the churn spike. + + Raises: + ValueError: if *week_of_tenure* is negative or *contract_term_months* + is not a positive integer. + """ + if week_of_tenure < 0: + raise ValueError(f"week_of_tenure must be >= 0, got {week_of_tenure}") + if contract_term_months < 1: + raise ValueError(f"contract_term_months must be >= 1, got {contract_term_months}") + if week_of_tenure == 0: + return False + term_weeks = contract_term_months * _WEEKS_PER_MONTH + k = round(week_of_tenure / term_weeks) + # Banker's rounding is provably safe here: term_weeks = 13m/3, so the + # fractional part of k*term_weeks is always in {0, 1/3, 2/3} — never .5. + return k >= 1 and round(k * term_weeks) == week_of_tenure + + +def churn_probability( + params: ChurnHazardParams, + latents: Mapping[str, float], + week_of_tenure: int, + contract_term_months: int, +) -> float: + """Return the weekly churn probability for one customer at one week. + + Composition: ``base_weekly_rate × latent multiplier × onboarding + elevation``, and on a renewal week additionally ``× + renewal_hazard_multiplier × renewal latent multiplier``. Capped at + ``MAX_PROBABILITY``. + + Args: + params: Motif-family churn parameters from + :func:`~leadforge.schemes.lifecycle.mechanisms.assign_lifecycle_mechanisms`. + latents: Merged customer + account latent traits in ``[0, 1]``. + Missing traits are treated as neutral (0.5). + week_of_tenure: Whole weeks since ``customer_start_at`` (0-based). + contract_term_months: The customer's contract term, for the + anniversary spike. + + Raises: + ValueError: via :func:`is_renewal_week` on negative tenure or + non-positive contract term. + """ + p = params.base_weekly_rate + p *= _latent_multiplier(latents, params.latent_weights) + p *= _onboarding_multiplier(week_of_tenure) + if is_renewal_week(week_of_tenure, contract_term_months): + p *= params.renewal_hazard_multiplier + p *= _latent_multiplier(latents, params.renewal_latent_weights) + return min(p, MAX_PROBABILITY) + + +def expansion_probability( + params: ExpansionPropensityParams, + latents: Mapping[str, float], + feature_depth_score: float | None = None, +) -> float: + """Return the weekly expansion (upsell / seat-add) probability. + + Composition: ``base_weekly_rate × latent multiplier``, optionally + ``× (0.5 + feature_depth_score)`` when the current health signal is + supplied — depth 0.5 is neutral (×1.0), full depth 1.0 raises the + propensity by half, zero depth halves it. Capped at ``MAX_PROBABILITY``. + + Args: + params: Motif-family expansion parameters. + latents: Merged latent traits; missing traits are neutral. + feature_depth_score: Optional current ``feature_depth_score`` health + signal in ``[0, 1]``; ``None`` skips the health modulation (the + engine passes it once health signals exist for the week). + """ + p = params.base_weekly_rate * _latent_multiplier(latents, params.latent_weights) + if feature_depth_score is not None: + if not 0.0 <= feature_depth_score <= 1.0: + raise ValueError(f"feature_depth_score must be in [0, 1], got {feature_depth_score}") + p *= _DEPTH_MULTIPLIER_FLOOR + feature_depth_score + return min(p, MAX_PROBABILITY) + + +def payment_failure_probability( + params: PaymentFailureParams, + latents: Mapping[str, float], +) -> float: + """Return the monthly invoice payment-failure probability. + + Composition: ``base_monthly_rate × latent multiplier`` (the dominant weight + is on ``latent_budget_stability``, negative — stable budgets fail less). + Capped at ``MAX_PROBABILITY``. + + Args: + params: Motif-family payment-failure parameters. + latents: Merged latent traits; missing traits are neutral. + """ + p = params.base_monthly_rate * _latent_multiplier(latents, params.latent_weights) + return min(p, MAX_PROBABILITY) diff --git a/leadforge/schemes/lifecycle/mechanisms.py b/leadforge/schemes/lifecycle/mechanisms.py index e8f057c..6b5fa64 100644 --- a/leadforge/schemes/lifecycle/mechanisms.py +++ b/leadforge/schemes/lifecycle/mechanisms.py @@ -133,11 +133,17 @@ class LifecycleMechanismAssignment: # 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 hazard base weekly rates. +# IMPORTANT — the per-motif "% annual" figures below are the BASE-RATE-ONLY +# equivalents (1 - (1-r)^52) at neutral latents. The hazard functions in +# hazards.py add material churn mass on top: the onboarding elevation +# contributes ~6.8 x base_rate of extra first-year mass and each renewal spike +# adds (multiplier - 1) x base_rate, so true first-year churn runs roughly +# 5-14 points above these figures (e.g. churner_dominated ~52%, not 37.5%). +# Final calibration against the difficulty-profile bands +# intro [0.10, 0.20] / intermediate [0.20, 0.35] / advanced [0.30, 0.50] +# happens in the engine tests (LTV-Pk), where these base rates are expected to +# be tuned DOWN to land inside the bands once the full tenure shape applies. _CHURN_BASE_WEEKLY: dict[str, float] = { # Exact annual equivalent: 1 - (1-r)^52. "product_led_retention": 0.0042, # 19.7% annual diff --git a/tests/schemes/lifecycle/test_hazards.py b/tests/schemes/lifecycle/test_hazards.py new file mode 100644 index 0000000..dd53f4e --- /dev/null +++ b/tests/schemes/lifecycle/test_hazards.py @@ -0,0 +1,273 @@ +"""Tests for the lifecycle hazard functions (LTV-Pj).""" + +import math + +import pytest + +from leadforge.schemes.lifecycle.hazards import ( + MAX_PROBABILITY, + churn_probability, + expansion_probability, + is_renewal_week, + payment_failure_probability, +) +from leadforge.schemes.lifecycle.mechanisms import ( + ChurnHazardParams, + ExpansionPropensityParams, + PaymentFailureParams, + assign_lifecycle_mechanisms, +) +from leadforge.schemes.lifecycle.population import LIFECYCLE_MOTIF_FAMILIES + +_NEUTRAL = { + "latent_product_fit": 0.5, + "latent_adoption_velocity": 0.5, + "latent_budget_stability": 0.5, + "latent_champion_strength": 0.5, + "latent_organizational_stability": 0.5, +} + +# A steady-state week well past onboarding and well before any renewal. +_STEADY_WEEK = 30 +_TERM_12MO = 12 + + +def _churn(motif: str = "product_led_retention") -> ChurnHazardParams: + return assign_lifecycle_mechanisms(motif).churn_hazard + + +def _expansion(motif: str = "expansion_led_growth") -> ExpansionPropensityParams: + return assign_lifecycle_mechanisms(motif).expansion_propensity + + +def _payment(motif: str = "payment_fragile") -> PaymentFailureParams: + return assign_lifecycle_mechanisms(motif).payment_failure + + +# --------------------------------------------------------------------------- +# is_renewal_week +# --------------------------------------------------------------------------- + + +def test_renewal_week_12mo_first_anniversary() -> None: + assert is_renewal_week(52, 12) + + +def test_renewal_week_12mo_second_anniversary() -> None: + assert is_renewal_week(104, 12) + + +def test_renewal_week_24mo_first_anniversary() -> None: + assert is_renewal_week(104, 24) + + +def test_week_52_is_not_renewal_for_24mo_term() -> None: + # A 24-month contract has no anniversary at week 52 — the spike (and the + # engine's renewal event) must not fire mid-contract. + assert not is_renewal_week(52, 24) + + +def test_adjacent_weeks_are_not_renewal() -> None: + assert not is_renewal_week(51, 12) + assert not is_renewal_week(53, 12) + + +def test_week_zero_is_never_renewal() -> None: + assert not is_renewal_week(0, 12) + assert not is_renewal_week(0, 1) + + +def test_renewal_week_non_integer_term_weeks() -> None: + # 13-month term → 56.33 weeks → anniversary at week 56. + assert is_renewal_week(56, 13) + assert not is_renewal_week(57, 13) + + +def test_renewal_week_rejects_negative_tenure() -> None: + with pytest.raises(ValueError, match="week_of_tenure"): + is_renewal_week(-1, 12) + + +def test_renewal_week_rejects_bad_term() -> None: + with pytest.raises(ValueError, match="contract_term_months"): + is_renewal_week(10, 0) + + +# --------------------------------------------------------------------------- +# churn_probability +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_churn_in_unit_interval_at_extremes(motif: str) -> None: + params = _churn(motif) + lo = dict.fromkeys(_NEUTRAL, 0.0) + hi = dict.fromkeys(_NEUTRAL, 1.0) + for latents in (lo, hi, _NEUTRAL): + for week in (0, 1, _STEADY_WEEK, 52): + p = churn_probability(params, latents, week, _TERM_12MO) + assert 0.0 < p <= MAX_PROBABILITY, f"{motif} week={week}: p={p}" + + +def test_churn_neutral_latents_steady_state_near_base_rate() -> None: + params = _churn() + p = churn_probability(params, _NEUTRAL, _STEADY_WEEK, _TERM_12MO) + # At week 30 the onboarding elevation residual is exp(-30/4) ≈ 5.5e-4. + assert math.isclose(p, params.base_weekly_rate, rel_tol=1e-2) + + +def test_churn_decreases_with_product_fit() -> None: + # product_led_retention weights latent_product_fit at -2.0. + params = _churn("product_led_retention") + poor_fit = {**_NEUTRAL, "latent_product_fit": 0.1} + good_fit = {**_NEUTRAL, "latent_product_fit": 0.9} + p_poor = churn_probability(params, poor_fit, _STEADY_WEEK, _TERM_12MO) + p_good = churn_probability(params, good_fit, _STEADY_WEEK, _TERM_12MO) + assert p_poor > p_good + + +def test_churn_onboarding_elevated_vs_steady_state() -> None: + params = _churn() + p_week0 = churn_probability(params, _NEUTRAL, 0, _TERM_12MO) + p_steady = churn_probability(params, _NEUTRAL, _STEADY_WEEK, _TERM_12MO) + assert p_week0 > 2.0 * p_steady # peak multiplier is 2.5 + + +def test_churn_onboarding_decays_monotonically() -> None: + params = _churn() + probs = [churn_probability(params, _NEUTRAL, w, _TERM_12MO) for w in range(13)] + assert probs == sorted(probs, reverse=True) + + +def test_churn_renewal_week_spikes() -> None: + params = _churn("relationship_led_retention") # multiplier 12.0 + p_renewal = churn_probability(params, _NEUTRAL, 52, _TERM_12MO) + p_before = churn_probability(params, _NEUTRAL, 51, _TERM_12MO) + p_after = churn_probability(params, _NEUTRAL, 53, _TERM_12MO) + assert p_renewal > 5.0 * p_before + assert p_renewal > 5.0 * p_after + + +def test_churn_strong_champion_dampens_renewal_spike() -> None: + params = _churn("relationship_led_retention") + weak = {**_NEUTRAL, "latent_champion_strength": 0.1} + strong = {**_NEUTRAL, "latent_champion_strength": 0.9} + p_weak = churn_probability(params, weak, 52, _TERM_12MO) + p_strong = churn_probability(params, strong, 52, _TERM_12MO) + assert p_weak > p_strong + + +def test_churn_missing_latents_treated_as_neutral() -> None: + params = _churn() + p_empty = churn_probability(params, {}, _STEADY_WEEK, _TERM_12MO) + p_neutral = churn_probability(params, _NEUTRAL, _STEADY_WEEK, _TERM_12MO) + assert p_empty == p_neutral + + +def test_churn_probability_is_capped() -> None: + params = ChurnHazardParams( + base_weekly_rate=0.5, + latent_weights=_churn().latent_weights, + renewal_hazard_multiplier=100.0, + renewal_latent_weights=_churn().renewal_latent_weights, + ) + p = churn_probability(params, _NEUTRAL, 52, _TERM_12MO) + assert p == MAX_PROBABILITY + + +def test_churn_is_deterministic() -> None: + params = _churn() + a = churn_probability(params, _NEUTRAL, 7, _TERM_12MO) + b = churn_probability(params, _NEUTRAL, 7, _TERM_12MO) + assert a == b + + +# --------------------------------------------------------------------------- +# expansion_probability +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_expansion_in_unit_interval_at_extremes(motif: str) -> None: + params = _expansion(motif) + for latents in (dict.fromkeys(_NEUTRAL, 0.0), dict.fromkeys(_NEUTRAL, 1.0), _NEUTRAL): + for depth in (None, 0.0, 0.5, 1.0): + p = expansion_probability(params, latents, depth) + assert 0.0 < p <= MAX_PROBABILITY + + +def test_expansion_increases_with_adoption_velocity() -> None: + params = _expansion("expansion_led_growth") # velocity weight +2.0 + slow = {**_NEUTRAL, "latent_adoption_velocity": 0.1} + fast = {**_NEUTRAL, "latent_adoption_velocity": 0.9} + assert expansion_probability(params, fast) > expansion_probability(params, slow) + + +def test_expansion_neutral_depth_is_no_op() -> None: + params = _expansion() + assert expansion_probability(params, _NEUTRAL, 0.5) == expansion_probability( + params, _NEUTRAL, None + ) + + +def test_expansion_depth_modulates_monotonically() -> None: + params = _expansion() + p_lo = expansion_probability(params, _NEUTRAL, 0.0) + p_mid = expansion_probability(params, _NEUTRAL, 0.5) + p_hi = expansion_probability(params, _NEUTRAL, 1.0) + assert p_lo < p_mid < p_hi + + +def test_expansion_rejects_out_of_range_depth() -> None: + params = _expansion() + with pytest.raises(ValueError, match="feature_depth_score"): + expansion_probability(params, _NEUTRAL, 1.5) + with pytest.raises(ValueError, match="feature_depth_score"): + expansion_probability(params, _NEUTRAL, -0.1) + + +# --------------------------------------------------------------------------- +# payment_failure_probability +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("motif", LIFECYCLE_MOTIF_FAMILIES) +def test_payment_failure_in_unit_interval_at_extremes(motif: str) -> None: + params = _payment(motif) + for latents in (dict.fromkeys(_NEUTRAL, 0.0), dict.fromkeys(_NEUTRAL, 1.0), _NEUTRAL): + p = payment_failure_probability(params, latents) + assert 0.0 < p <= MAX_PROBABILITY + + +def test_payment_failure_decreases_with_budget_stability() -> None: + params = _payment("payment_fragile") # stability weight -3.0 + fragile = {**_NEUTRAL, "latent_budget_stability": 0.1} + stable = {**_NEUTRAL, "latent_budget_stability": 0.9} + p_fragile = payment_failure_probability(params, fragile) + p_stable = payment_failure_probability(params, stable) + assert p_fragile > p_stable + # With weight -3.0 the spread between the tails is large (exp(2.4) ≈ 11×). + assert p_fragile > 5.0 * p_stable + + +def test_payment_failure_neutral_equals_base_rate() -> None: + params = _payment() + p = payment_failure_probability(params, _NEUTRAL) + assert math.isclose(p, params.base_monthly_rate, rel_tol=1e-9) + + +# --------------------------------------------------------------------------- +# Cross-mechanism sanity: motif identity flows through to probabilities +# --------------------------------------------------------------------------- + + +def test_fragile_world_fails_payments_more_than_product_led() -> None: + p_fragile = payment_failure_probability(_payment("payment_fragile"), _NEUTRAL) + p_plr = payment_failure_probability(_payment("product_led_retention"), _NEUTRAL) + assert p_fragile > 2.0 * p_plr + + +def test_churner_world_churns_more_than_growth_world() -> None: + p_churner = churn_probability(_churn("churner_dominated"), _NEUTRAL, _STEADY_WEEK, 12) + p_growth = churn_probability(_churn("expansion_led_growth"), _NEUTRAL, _STEADY_WEEK, 12) + assert p_churner > p_growth