From 386c8a46fea7a292e6f6ce4baa72842a331a2213 Mon Sep 17 00:00:00 2001 From: codex Date: Thu, 2 Jul 2026 22:50:16 +0200 Subject: [PATCH] Implement comparable LTV engine and close WP-0005 --- adaptive_pricing_core/__init__.py | 24 + adaptive_pricing_core/comparable_ltv.py | 496 ++++++++++++++++++ docs/ComparableCustomerLTV.md | 99 ++++ .../coulomb-pricing/data/ltv_scenarios.json | 76 +++ projects/coulomb-pricing/observatory/api.py | 10 +- projects/coulomb-pricing/observatory/load.py | 4 + projects/coulomb-pricing/observatory/ltv.py | 210 ++++++++ .../observatory/recommendations.py | 4 +- .../coulomb-pricing/observatory/simulator.py | 106 ++-- projects/coulomb-pricing/tests/test_api.py | 4 + .../tests/test_comparable_ltv.py | 69 +++ .../coulomb-pricing/tests/test_mvp_sprints.py | 13 +- ...ble-customer-ltv-and-simulation-upgrade.md | 13 +- 13 files changed, 1060 insertions(+), 68 deletions(-) create mode 100644 adaptive_pricing_core/comparable_ltv.py create mode 100644 docs/ComparableCustomerLTV.md create mode 100644 projects/coulomb-pricing/data/ltv_scenarios.json create mode 100644 projects/coulomb-pricing/observatory/ltv.py create mode 100644 projects/coulomb-pricing/tests/test_comparable_ltv.py diff --git a/adaptive_pricing_core/__init__.py b/adaptive_pricing_core/__init__.py index c1f86eb..8119742 100644 --- a/adaptive_pricing_core/__init__.py +++ b/adaptive_pricing_core/__init__.py @@ -7,6 +7,19 @@ from .boundary_engine import ( default_commitment_terms, validate_pricing_configuration, ) +from .comparable_ltv import ( + ComparableCustomerProfile, + ComparableLTVEstimate, + LTVPolicy, + PricingComparison, + PricingModelComparison, + SensitivityCase, + SensitivityOutcome, + compare_pricing_configurations, + default_sensitivity_cases, + estimate_comparable_customer_ltv, + select_reference_estimate, +) from .pricing_models import ( ChargeComponent, Commitment, @@ -21,16 +34,27 @@ from .pricing_models import ( __all__ = [ "BoundaryPolicy", "ChargeComponent", + "ComparableCustomerProfile", + "ComparableLTVEstimate", "Commitment", "CommitmentTerms", "ConstraintResult", + "LTVPolicy", "PricingModel", + "PricingComparison", + "PricingModelComparison", "PricingModelStatus", "PricingConfiguration", + "SensitivityCase", + "SensitivityOutcome", "TunableParameter", "ValidationResult", + "compare_pricing_configurations", + "default_sensitivity_cases", "default_commitment_terms", + "estimate_comparable_customer_ltv", "load_pricing_models", + "select_reference_estimate", "validate_pricing_configuration", "validate_pricing_catalog", "validate_pricing_model", diff --git a/adaptive_pricing_core/comparable_ltv.py b/adaptive_pricing_core/comparable_ltv.py new file mode 100644 index 0000000..34a0605 --- /dev/null +++ b/adaptive_pricing_core/comparable_ltv.py @@ -0,0 +1,496 @@ +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from decimal import Decimal, ROUND_HALF_UP +from typing import Any + +from .boundary_engine import ( + BoundaryPolicy, + PricingConfiguration, + ValidationResult, + validate_pricing_configuration, +) + +TWOPLACES = Decimal("0.01") +PCTPLACES = Decimal("0.1") + + +def _money(value: Decimal) -> Decimal: + return value.quantize(TWOPLACES, rounding=ROUND_HALF_UP) + + +def _percent(value: Decimal) -> Decimal: + return value.quantize(PCTPLACES, rounding=ROUND_HALF_UP) + + +def _decimal(value: Decimal | str | int | float | None) -> Decimal: + if value in (None, ""): + return Decimal("0") + return Decimal(str(value)) + + +@dataclass(frozen=True) +class ComparableCustomerProfile: + id: str + name: str + segment: str + eligible_model_ids: tuple[str, ...] = () + members_per_customer: int = 1 + expected_monthly_usage_units: Decimal = Decimal("0") + usage_variance_pct: Decimal = Decimal("0") + monthly_churn_pct: Decimal = Decimal("0") + monthly_default_pct: Decimal = Decimal("0") + monthly_support_cost: Decimal = Decimal("0") + monthly_risk_cost: Decimal = Decimal("0") + acquisition_cost: Decimal = Decimal("0") + upfront_investment_cost: Decimal = Decimal("0") + allocated_fixed_cost: Decimal | None = None + notes: str = "" + + +@dataclass(frozen=True) +class LTVPolicy: + horizon_months: int = 24 + monthly_discount_rate_pct: Decimal = Decimal("1") + required_improvement_factor: Decimal = Decimal("1.05") + + +@dataclass(frozen=True) +class SensitivityCase: + id: str + name: str + usage_multiplier: Decimal = Decimal("1") + usage_variance_delta_pct: Decimal = Decimal("0") + monthly_churn_delta_pct: Decimal = Decimal("0") + monthly_default_delta_pct: Decimal = Decimal("0") + monthly_support_cost_delta: Decimal = Decimal("0") + monthly_risk_cost_delta: Decimal = Decimal("0") + + +@dataclass(frozen=True) +class ComparableLTVEstimate: + model_id: str + model_name: str + validation: ValidationResult + average_comparable_customer_lifetime_value: Decimal + discounted_margin_ltv: Decimal + expected_lifetime_months: Decimal + payback_months: int | None + adjusted_monthly_churn_pct: Decimal + adjusted_monthly_default_pct: Decimal + acquisition_cost: Decimal + upfront_investment_cost: Decimal + seller_cost_recovery_months: int | None + assumptions: dict[str, Any] + explanation: str + + +@dataclass(frozen=True) +class SensitivityOutcome: + case_id: str + case_name: str + average_comparable_customer_lifetime_value: Decimal + delta_vs_base: Decimal + delta_vs_base_pct: Decimal | None + decision: str + + +@dataclass(frozen=True) +class PricingModelComparison: + model_id: str + model_name: str + model_type: str + status: str + validation_decision: str + valid: bool + requires_approval: bool + average_comparable_customer_lifetime_value: Decimal + expected_lifetime_months: Decimal + base_monthly_margin: Decimal + base_margin_pct: Decimal + payment_fee_pct: Decimal + contract_duration_months: int + reference_model_id: str | None + passes_required_improvement: bool + required_improvement_threshold: Decimal | None + improvement_vs_reference_pct: Decimal | None + sensitivity: tuple[SensitivityOutcome, ...] + sensitivity_floor_ltv: Decimal + sensitivity_ceiling_ltv: Decimal + key_drivers: tuple[str, ...] + comparison_summary: str + + +@dataclass(frozen=True) +class PricingComparison: + profile: ComparableCustomerProfile + policy: LTVPolicy + boundary_policy: BoundaryPolicy + reference_model_id: str | None + best_ltv_model_id: str | None + best_valid_model_id: str | None + comparisons: tuple[PricingModelComparison, ...] + notes: tuple[str, ...] + + +def default_sensitivity_cases() -> tuple[SensitivityCase, ...]: + return ( + SensitivityCase( + id="usage-downside", + name="Usage downside", + usage_multiplier=Decimal("0.75"), + monthly_churn_delta_pct=Decimal("1.5"), + monthly_risk_cost_delta=Decimal("0.05"), + ), + SensitivityCase( + id="usage-upside", + name="Usage upside", + usage_multiplier=Decimal("1.35"), + usage_variance_delta_pct=Decimal("10"), + ), + SensitivityCase( + id="risk-spike", + name="Risk spike", + usage_multiplier=Decimal("1.00"), + monthly_churn_delta_pct=Decimal("3.0"), + monthly_default_delta_pct=Decimal("1.0"), + monthly_support_cost_delta=Decimal("0.25"), + monthly_risk_cost_delta=Decimal("0.20"), + ), + ) + + +def _risk_multiplier_for_default(validation: ValidationResult) -> Decimal: + metrics = validation.metrics + multiplier = Decimal("1") + if metrics.prepaid_amount >= metrics.effective_monthly_revenue and metrics.prepaid_amount > Decimal("0"): + multiplier *= Decimal("0.50") + if ( + metrics.guaranteed_platform_fee >= metrics.effective_monthly_revenue + and metrics.guaranteed_platform_fee > Decimal("0") + ): + multiplier *= Decimal("0.75") + return multiplier + + +def _risk_multiplier_for_churn(validation: ValidationResult) -> Decimal: + metrics = validation.metrics + multiplier = Decimal("1") + if metrics.reduced_cancellation_flexibility: + multiplier *= Decimal("0.85") + return multiplier + + +def _discount_rate(policy: LTVPolicy) -> Decimal: + return Decimal("1") + (policy.monthly_discount_rate_pct / Decimal("100")) + + +def _required_threshold(reference_ltv: Decimal, factor: Decimal) -> Decimal: + if reference_ltv >= Decimal("0"): + return _money(reference_ltv * factor) + improvement = abs(reference_ltv) * (factor - Decimal("1")) + return _money(reference_ltv + improvement) + + +def _pct_delta(candidate: Decimal, reference: Decimal) -> Decimal | None: + if reference == Decimal("0"): + return None + return _percent(((candidate - reference) / abs(reference)) * Decimal("100")) + + +def estimate_comparable_customer_ltv( + configuration: PricingConfiguration, + profile: ComparableCustomerProfile, + boundary_policy: BoundaryPolicy, + policy: LTVPolicy, +) -> ComparableLTVEstimate: + validation = validate_pricing_configuration(configuration, boundary_policy) + + base_churn_pct = max(profile.monthly_churn_pct, Decimal("0")) + base_default_pct = max(profile.monthly_default_pct, Decimal("0")) + adjusted_churn_pct = _percent(base_churn_pct * _risk_multiplier_for_churn(validation)) + adjusted_default_pct = _percent(base_default_pct * _risk_multiplier_for_default(validation)) + + monthly_margin = validation.metrics.monthly_margin + acquisition_cost = _money(profile.acquisition_cost) + upfront_investment_cost = _money(profile.upfront_investment_cost) + committed_months = max(validation.metrics.contract_duration_months, 1) + + survival = Decimal("1") + expected_lifetime_months = Decimal("0") + discounted_margin = Decimal("0") + cumulative_discounted_margin = Decimal("0") + payback_months: int | None = None + recovery_months: int | None = None + hurdle = acquisition_cost + upfront_investment_cost + + adjusted_churn_rate = adjusted_churn_pct / Decimal("100") + adjusted_default_rate = adjusted_default_pct / Decimal("100") + discount_rate = _discount_rate(policy) + + for month in range(1, policy.horizon_months + 1): + expected_lifetime_months += survival + discounted_month_margin = (monthly_margin * survival) / (discount_rate ** month) + discounted_margin += discounted_month_margin + cumulative_discounted_margin += discounted_month_margin + + if payback_months is None and cumulative_discounted_margin >= hurdle: + payback_months = month + if recovery_months is None and cumulative_discounted_margin >= Decimal("0"): + recovery_months = month + + if month < committed_months: + survival *= Decimal("1") - adjusted_default_rate + else: + survival *= (Decimal("1") - adjusted_default_rate) * (Decimal("1") - adjusted_churn_rate) + + average_ltv = _money(discounted_margin - hurdle) + explanation = ( + f"Monthly margin {validation.metrics.monthly_margin} {validation.metrics.currency}, " + f"expected lifetime {expected_lifetime_months.quantize(Decimal('0.1'))} months, " + f"discounted seller LTV {average_ltv} {validation.metrics.currency}." + ) + + return ComparableLTVEstimate( + model_id=configuration.model.id, + model_name=configuration.model.name, + validation=validation, + average_comparable_customer_lifetime_value=average_ltv, + discounted_margin_ltv=_money(discounted_margin), + expected_lifetime_months=expected_lifetime_months.quantize(Decimal("0.1")), + payback_months=payback_months, + adjusted_monthly_churn_pct=adjusted_churn_pct, + adjusted_monthly_default_pct=adjusted_default_pct, + acquisition_cost=acquisition_cost, + upfront_investment_cost=upfront_investment_cost, + seller_cost_recovery_months=recovery_months, + assumptions={ + "horizon_months": policy.horizon_months, + "monthly_discount_rate_pct": _percent(policy.monthly_discount_rate_pct), + "base_monthly_churn_pct": _percent(base_churn_pct), + "base_monthly_default_pct": _percent(base_default_pct), + "committed_months": committed_months, + }, + explanation=explanation, + ) + + +def _apply_sensitivity( + configuration: PricingConfiguration, + profile: ComparableCustomerProfile, + case: SensitivityCase, +) -> tuple[PricingConfiguration, ComparableCustomerProfile]: + return ( + replace( + configuration, + expected_usage_units=_money(configuration.expected_usage_units * case.usage_multiplier), + expected_usage_variance_pct=max( + configuration.expected_usage_variance_pct + case.usage_variance_delta_pct, + Decimal("0"), + ), + support_cost=max(configuration.support_cost + case.monthly_support_cost_delta, Decimal("0")), + risk_cost=max(configuration.risk_cost + case.monthly_risk_cost_delta, Decimal("0")), + ), + replace( + profile, + monthly_churn_pct=max(profile.monthly_churn_pct + case.monthly_churn_delta_pct, Decimal("0")), + monthly_default_pct=max(profile.monthly_default_pct + case.monthly_default_delta_pct, Decimal("0")), + ), + ) + + +def select_reference_estimate( + estimates: list[ComparableLTVEstimate], + eligible_model_ids: tuple[str, ...] = (), +) -> ComparableLTVEstimate | None: + candidates = estimates + if eligible_model_ids: + allowed = set(eligible_model_ids) + candidates = [estimate for estimate in estimates if estimate.model_id in allowed] + + valid = [estimate for estimate in candidates if estimate.validation.valid] + if valid: + return max(valid, key=lambda item: item.average_comparable_customer_lifetime_value) + if candidates: + return max(candidates, key=lambda item: item.average_comparable_customer_lifetime_value) + return None + + +def _key_drivers( + estimate: ComparableLTVEstimate, + reference: ComparableLTVEstimate | None, +) -> tuple[str, ...]: + drivers: list[str] = [] + metrics = estimate.validation.metrics + if reference is None: + return ("no_reference_model_available",) + + ref_metrics = reference.validation.metrics + if metrics.monthly_margin > ref_metrics.monthly_margin: + drivers.append("higher_monthly_margin") + if estimate.expected_lifetime_months > reference.expected_lifetime_months: + drivers.append("longer_expected_lifetime") + if metrics.contract_duration_months > ref_metrics.contract_duration_months: + drivers.append("longer_commitment_window") + if metrics.payment_fee_pct < ref_metrics.payment_fee_pct: + drivers.append("lower_payment_fee_burden") + if metrics.minimum_monthly_turnover > ref_metrics.minimum_monthly_turnover: + drivers.append("stronger_revenue_commitment") + if not drivers: + drivers.append("reference_like_economics") + return tuple(drivers) + + +def _comparison_summary( + estimate: ComparableLTVEstimate, + reference: ComparableLTVEstimate | None, + passes_required_improvement: bool, + threshold: Decimal | None, +) -> str: + if reference is None: + return f"{estimate.model_name}: no reference model available for improvement comparison." + + improvement = _pct_delta( + estimate.average_comparable_customer_lifetime_value, + reference.average_comparable_customer_lifetime_value, + ) + if estimate.model_id == reference.model_id: + return f"{estimate.model_name}: reference model for this comparable-customer segment." + if passes_required_improvement: + return ( + f"{estimate.model_name}: LTV {estimate.average_comparable_customer_lifetime_value} " + f"meets threshold {threshold} with improvement {improvement}%." + ) + return ( + f"{estimate.model_name}: LTV {estimate.average_comparable_customer_lifetime_value} " + f"misses threshold {threshold} with improvement {improvement}%." + ) + + +def compare_pricing_configurations( + configurations: list[PricingConfiguration], + profile: ComparableCustomerProfile, + boundary_policy: BoundaryPolicy, + policy: LTVPolicy, + sensitivity_cases: tuple[SensitivityCase, ...] | None = None, +) -> PricingComparison: + base_estimates = [ + estimate_comparable_customer_ltv(configuration, profile, boundary_policy, policy) + for configuration in configurations + ] + reference = select_reference_estimate(base_estimates, profile.eligible_model_ids) + + comparisons: list[PricingModelComparison] = [] + for configuration, estimate in zip(configurations, base_estimates, strict=True): + outcomes: list[SensitivityOutcome] = [] + for case in sensitivity_cases or default_sensitivity_cases(): + scenario_configuration, scenario_profile = _apply_sensitivity(configuration, profile, case) + scenario_estimate = estimate_comparable_customer_ltv( + scenario_configuration, + scenario_profile, + boundary_policy, + policy, + ) + delta_vs_base = _money( + scenario_estimate.average_comparable_customer_lifetime_value + - estimate.average_comparable_customer_lifetime_value + ) + outcomes.append( + SensitivityOutcome( + case_id=case.id, + case_name=case.name, + average_comparable_customer_lifetime_value=scenario_estimate.average_comparable_customer_lifetime_value, + delta_vs_base=delta_vs_base, + delta_vs_base_pct=_pct_delta( + scenario_estimate.average_comparable_customer_lifetime_value, + estimate.average_comparable_customer_lifetime_value, + ), + decision=scenario_estimate.validation.decision, + ) + ) + + threshold: Decimal | None = None + passes_required_improvement = True + if reference is not None and estimate.model_id != reference.model_id: + threshold = _required_threshold( + reference.average_comparable_customer_lifetime_value, + policy.required_improvement_factor, + ) + passes_required_improvement = ( + estimate.average_comparable_customer_lifetime_value >= threshold + ) + + improvement_vs_reference_pct = ( + Decimal("0.0") + if reference is not None and estimate.model_id == reference.model_id + else ( + _pct_delta( + estimate.average_comparable_customer_lifetime_value, + reference.average_comparable_customer_lifetime_value, + ) + if reference is not None + else None + ) + ) + + sensitivity_floor = min( + [estimate.average_comparable_customer_lifetime_value] + + [item.average_comparable_customer_lifetime_value for item in outcomes] + ) + sensitivity_ceiling = max( + [estimate.average_comparable_customer_lifetime_value] + + [item.average_comparable_customer_lifetime_value for item in outcomes] + ) + + comparisons.append( + PricingModelComparison( + model_id=estimate.model_id, + model_name=estimate.model_name, + model_type=configuration.model.model_type, + status=configuration.model.status, + validation_decision=estimate.validation.decision, + valid=estimate.validation.valid, + requires_approval=estimate.validation.requires_approval, + average_comparable_customer_lifetime_value=estimate.average_comparable_customer_lifetime_value, + expected_lifetime_months=estimate.expected_lifetime_months, + base_monthly_margin=estimate.validation.metrics.monthly_margin, + base_margin_pct=estimate.validation.metrics.margin_pct, + payment_fee_pct=estimate.validation.metrics.payment_fee_pct, + contract_duration_months=estimate.validation.metrics.contract_duration_months, + reference_model_id=reference.model_id if reference else None, + passes_required_improvement=passes_required_improvement, + required_improvement_threshold=threshold, + improvement_vs_reference_pct=improvement_vs_reference_pct, + sensitivity=tuple(outcomes), + sensitivity_floor_ltv=sensitivity_floor, + sensitivity_ceiling_ltv=sensitivity_ceiling, + key_drivers=_key_drivers(estimate, reference), + comparison_summary=_comparison_summary( + estimate, + reference, + passes_required_improvement, + threshold, + ), + ) + ) + + best_ltv = max(comparisons, key=lambda item: item.average_comparable_customer_lifetime_value, default=None) + valid_comparisons = [item for item in comparisons if item.valid] + best_valid = max(valid_comparisons, key=lambda item: item.average_comparable_customer_lifetime_value, default=None) + + notes = ( + "Comparable-customer LTV uses discounted expected seller margin over a finite horizon.", + "Reference selection prefers valid eligible predefined models with the highest seller LTV.", + "Negative reference LTV uses additive improvement semantics so tuned models must become less negative, not more negative.", + ) + + return PricingComparison( + profile=profile, + policy=policy, + boundary_policy=boundary_policy, + reference_model_id=reference.model_id if reference else None, + best_ltv_model_id=best_ltv.model_id if best_ltv else None, + best_valid_model_id=best_valid.model_id if best_valid else None, + comparisons=tuple(comparisons), + notes=notes, + ) diff --git a/docs/ComparableCustomerLTV.md b/docs/ComparableCustomerLTV.md new file mode 100644 index 0000000..7c80ee3 --- /dev/null +++ b/docs/ComparableCustomerLTV.md @@ -0,0 +1,99 @@ +# Comparable Customer LTV + +Status: implementation-facing MVP for `ADAPTIVE-WP-0005`. + +## Purpose + +This document defines the first operational form of +`average_comparable_customer_lifetime_value`. + +The goal is to compare pricing configurations using expected seller economics +over time instead of only the current-period observatory snapshot. + +## Core Definition + +For the current MVP: + +`average_comparable_customer_lifetime_value` + +means discounted expected seller margin for a comparable customer profile over a +finite horizon, minus acquisition and upfront seller investment costs. + +Inputs include: + +- validated monthly pricing economics from the boundary engine +- comparable-customer usage expectations +- churn and default risk assumptions +- contract duration and commitment protections +- seller acquisition and upfront investment costs +- a seller-configurable discount rate and required-improvement factor + +## Reference Model Selection + +The comparison engine selects: + +`most_favorable_predefined_model` + +as the highest-LTV valid predefined model available to the comparable-customer +profile. If no valid model exists, it falls back to the highest-LTV eligible +predefined model so the comparison still produces an inspectable anchor. + +Eligibility is currently supplied by the simulation profile rather than derived +solely from model metadata. + +## Required Improvement Semantics + +For non-reference configurations: + +```text +average_comparable_customer_lifetime_value(candidate) +>= average_comparable_customer_lifetime_value(reference) + × required_improvement_factor +``` + +When the reference LTV is positive, the threshold is multiplicative. + +When the reference LTV is negative, the engine switches to additive improvement +semantics so the candidate must become less negative by the requested +percentage. This avoids the invalid outcome where multiplying a negative value +would reward a worse configuration. + +## Risk Model + +Current risk handling is intentionally simple and explicit: + +- monthly churn risk applies after committed months expire +- monthly default risk applies throughout the horizon +- prepayment and guaranteed-fee commitments reduce default exposure +- reduced cancellation flexibility lowers modeled churn exposure + +This is a policy approximation, not a retention model trained from history. + +## Sensitivity Model + +Each comparison runs the base case plus named sensitivity cases. The current +Coulomb adapter includes: + +- usage downside +- usage upside +- risk spike + +Sensitivity output reports: + +- scenario LTV +- delta versus base LTV +- whether the configuration remains accepted, approval-only, or rejected + +## Coulomb Calibration + +The Coulomb observatory currently calibrates the generic engine with: + +- observed payment-fee rate from `payment_records.json` +- observed AI usage unit cost from `usage_records.json` +- segment profiles from `ltv_scenarios.json` +- profile-specific fixed-cost allocation overrides for comparable future + customers + +Those fixed-cost overrides are deliberate: the current single-member pilot cost +structure is too distorted to act as a reusable comparable-customer baseline on +its own. diff --git a/projects/coulomb-pricing/data/ltv_scenarios.json b/projects/coulomb-pricing/data/ltv_scenarios.json new file mode 100644 index 0000000..77f3590 --- /dev/null +++ b/projects/coulomb-pricing/data/ltv_scenarios.json @@ -0,0 +1,76 @@ +{ + "version": 1, + "currency": "EUR", + "horizon_months": 24, + "monthly_discount_rate_pct": "1.0", + "required_improvement_factor": "1.05", + "profiles": [ + { + "id": "solo-builder", + "name": "Solo builder", + "segment": "coulomb-social-members", + "eligible_model_ids": [ + "flat-899-eur-monthly", + "membership-plus-credits", + "membership-plus-overage" + ], + "members_per_customer": 1, + "expected_monthly_usage_units": "48200", + "usage_variance_pct": "20", + "monthly_churn_pct": "6.0", + "monthly_default_pct": "1.0", + "monthly_support_cost": "0.25", + "monthly_risk_cost": "0.10", + "acquisition_cost": "2.00", + "upfront_investment_cost": "0.00", + "allocated_fixed_cost": "5.00", + "notes": "Calibrated from the current founding-member usage record and payment ledger." + }, + { + "id": "small-team", + "name": "Small product team", + "segment": "coulomb-social-members", + "eligible_model_ids": [ + "flat-899-eur-monthly", + "membership-plus-credits", + "membership-plus-overage" + ], + "members_per_customer": 3, + "expected_monthly_usage_units": "180000", + "usage_variance_pct": "35", + "monthly_churn_pct": "3.5", + "monthly_default_pct": "1.0", + "monthly_support_cost": "1.50", + "monthly_risk_cost": "0.20", + "acquisition_cost": "8.00", + "upfront_investment_cost": "1.50", + "allocated_fixed_cost": "12.00", + "notes": "Hypothesis scenario for a higher-usage small team considering a multi-seat relationship." + } + ], + "sensitivity_cases": [ + { + "id": "usage-downside", + "name": "Usage downside", + "usage_multiplier": "0.75", + "monthly_churn_delta_pct": "1.5", + "monthly_risk_cost_delta": "0.05" + }, + { + "id": "usage-upside", + "name": "Usage upside", + "usage_multiplier": "1.35", + "usage_variance_delta_pct": "10.0" + }, + { + "id": "risk-spike", + "name": "Risk spike", + "usage_multiplier": "1.00", + "monthly_churn_delta_pct": "3.0", + "monthly_default_delta_pct": "1.0", + "monthly_support_cost_delta": "0.25", + "monthly_risk_cost_delta": "0.20" + } + ], + "notes": "First-pass comparable-customer LTV assumptions for the Coulomb observatory. These scenarios are meant for simulation and comparison, not billing execution." +} diff --git a/projects/coulomb-pricing/observatory/api.py b/projects/coulomb-pricing/observatory/api.py index 91a17d8..f46dde5 100644 --- a/projects/coulomb-pricing/observatory/api.py +++ b/projects/coulomb-pricing/observatory/api.py @@ -9,6 +9,7 @@ from .economics import build_liquidity_summary, build_snapshot from .load import ( default_data_dir, latest_period, + load_ltv_scenarios, load_budget, load_expense_records, load_market_signals, @@ -88,12 +89,19 @@ def build_dashboard_payload(data_dir: Path | None = None, period: str | None = N market_raw = load_market_signals(root) usage_records = load_usage_records(root) usage_summary = build_usage_summary(usage_records, target_period) + ltv_scenarios = load_ltv_scenarios(root) cost_floor = build_cost_floor(snapshot, models) value_range = build_value_range_view(value_range_raw, snapshot, product, models) market_price = build_market_price_view(market_raw) cost_allocation = build_cost_allocation(snapshot, usage_records) ai_cost_per_member = usage_summary["cost_per_active_user_eur"] - simulations = build_pricing_simulations(snapshot, models, ai_cost_per_member) + simulations = build_pricing_simulations( + snapshot, + models, + ai_cost_per_member, + usage_records=usage_records, + scenario_catalog=ltv_scenarios, + ) boundary_validation = build_boundary_validation(snapshot, models, usage_records) credit_wallets = load_credit_wallets(root) credit_summary = build_credit_summary( diff --git a/projects/coulomb-pricing/observatory/load.py b/projects/coulomb-pricing/observatory/load.py index bc97e66..8d4b168 100644 --- a/projects/coulomb-pricing/observatory/load.py +++ b/projects/coulomb-pricing/observatory/load.py @@ -115,6 +115,10 @@ def load_market_signals(data_dir: Path | None = None) -> dict: return _read_json((data_dir or default_data_dir()) / "market_signals.json") +def load_ltv_scenarios(data_dir: Path | None = None) -> dict: + return _read_json((data_dir or default_data_dir()) / "ltv_scenarios.json") + + def load_membership(data_dir: Path | None = None) -> list[MembershipRecord]: raw = _read_json((data_dir or default_data_dir()) / "membership.json") return [ diff --git a/projects/coulomb-pricing/observatory/ltv.py b/projects/coulomb-pricing/observatory/ltv.py new file mode 100644 index 0000000..32a71d3 --- /dev/null +++ b/projects/coulomb-pricing/observatory/ltv.py @@ -0,0 +1,210 @@ +from __future__ import annotations + +from decimal import Decimal +from typing import Any + +from ._repo_root import ensure_repo_root_on_syspath +from .boundary import build_boundary_policy +from .models import EconomicsSnapshot, PricingModel + +ensure_repo_root_on_syspath() + +from adaptive_pricing_core.boundary_engine import PricingConfiguration # noqa: E402 +from adaptive_pricing_core.comparable_ltv import ( # noqa: E402 + ComparableCustomerProfile, + LTVPolicy, + SensitivityCase, + compare_pricing_configurations, +) + + +def _serialize(value: Any) -> Any: + if isinstance(value, Decimal): + return str(value) + if hasattr(value, "__dataclass_fields__"): + return {key: _serialize(getattr(value, key)) for key in value.__dataclass_fields__} + if isinstance(value, tuple): + return [_serialize(item) for item in value] + if isinstance(value, list): + return [_serialize(item) for item in value] + if isinstance(value, dict): + return {key: _serialize(item) for key, item in value.items()} + return value + + +def _decimal(value: Decimal | str | int | float | None) -> Decimal: + if value in (None, ""): + return Decimal("0") + return Decimal(str(value)) + + +def _usage_component(model: PricingModel): + return next((component for component in model.charge_components if component.kind == "usage"), None) + + +def _included_units(model: PricingModel, members_per_customer: int) -> Decimal | None: + usage = _usage_component(model) + if not usage or usage.included_units is None: + return None + return usage.included_units * Decimal(members_per_customer) + + +def _usage_unit_price(model: PricingModel) -> Decimal | None: + usage = _usage_component(model) + if not usage or usage.unit_price is None: + return None + return usage.unit_price + + +def _usage_unit_cost(records: list[dict[str, Any]], period: str) -> Decimal: + period_rows = [row for row in records if row.get("period") == period] + total_units = sum(_decimal(row.get("tokens")) for row in period_rows) + total_cost = sum(_decimal(row.get("cost_eur")) for row in period_rows) + if total_units <= Decimal("0"): + return Decimal("0") + return total_cost / total_units + + +def _payment_fee_rate(snapshot: EconomicsSnapshot) -> Decimal: + if snapshot.monthly_revenue <= Decimal("0"): + return Decimal("0") + return (snapshot.monthly_payment_processing_cost / snapshot.monthly_revenue) * Decimal("100") + + +def _profile(raw: dict[str, Any]) -> ComparableCustomerProfile: + return ComparableCustomerProfile( + id=raw["id"], + name=raw["name"], + segment=raw["segment"], + eligible_model_ids=tuple(raw.get("eligible_model_ids", [])), + members_per_customer=int(raw.get("members_per_customer", 1)), + expected_monthly_usage_units=_decimal(raw.get("expected_monthly_usage_units")), + usage_variance_pct=_decimal(raw.get("usage_variance_pct")), + monthly_churn_pct=_decimal(raw.get("monthly_churn_pct")), + monthly_default_pct=_decimal(raw.get("monthly_default_pct")), + monthly_support_cost=_decimal(raw.get("monthly_support_cost")), + monthly_risk_cost=_decimal(raw.get("monthly_risk_cost")), + acquisition_cost=_decimal(raw.get("acquisition_cost")), + upfront_investment_cost=_decimal(raw.get("upfront_investment_cost")), + allocated_fixed_cost=_decimal(raw["allocated_fixed_cost"]) if raw.get("allocated_fixed_cost") else None, + notes=raw.get("notes", ""), + ) + + +def _sensitivity_case(raw: dict[str, Any]) -> SensitivityCase: + return SensitivityCase( + id=raw["id"], + name=raw["name"], + usage_multiplier=_decimal(raw.get("usage_multiplier", "1")), + usage_variance_delta_pct=_decimal(raw.get("usage_variance_delta_pct")), + monthly_churn_delta_pct=_decimal(raw.get("monthly_churn_delta_pct")), + monthly_default_delta_pct=_decimal(raw.get("monthly_default_delta_pct")), + monthly_support_cost_delta=_decimal(raw.get("monthly_support_cost_delta")), + monthly_risk_cost_delta=_decimal(raw.get("monthly_risk_cost_delta")), + ) + + +def _ltv_policy(raw: dict[str, Any]) -> LTVPolicy: + return LTVPolicy( + horizon_months=int(raw.get("horizon_months", 24)), + monthly_discount_rate_pct=_decimal(raw.get("monthly_discount_rate_pct", "1.0")), + required_improvement_factor=_decimal(raw.get("required_improvement_factor", "1.05")), + ) + + +def _configuration( + model: PricingModel, + profile: ComparableCustomerProfile, + snapshot: EconomicsSnapshot, + usage_unit_cost: Decimal, +) -> PricingConfiguration: + members_per_customer = max(profile.members_per_customer, 1) + per_member_fixed_cost = ( + snapshot.monthly_infrastructure_cost / snapshot.active_members + if snapshot.active_members + else snapshot.monthly_infrastructure_cost + ) + allocated_fixed_cost = ( + profile.allocated_fixed_cost + if profile.allocated_fixed_cost is not None + else per_member_fixed_cost * Decimal(members_per_customer) + ) + return PricingConfiguration( + model=model, + segment=profile.segment, + expected_usage_units=profile.expected_monthly_usage_units, + expected_usage_variance_pct=profile.usage_variance_pct, + allocated_fixed_cost=allocated_fixed_cost, + unit_cost=usage_unit_cost, + support_cost=profile.monthly_support_cost, + risk_cost=profile.monthly_risk_cost, + payment_fee_rate_pct=_payment_fee_rate(snapshot), + access_fee_amount=model.access_fee_amount * Decimal(members_per_customer), + included_units=_included_units(model, members_per_customer), + usage_unit_price=_usage_unit_price(model), + ) + + +def build_ltv_simulations( + snapshot: EconomicsSnapshot, + models: list[PricingModel], + usage_records: list[dict[str, Any]], + scenario_catalog: dict[str, Any], +) -> dict[str, Any]: + policy = _ltv_policy(scenario_catalog) + boundary_policy = build_boundary_policy(snapshot) + sensitivity_cases = tuple(_sensitivity_case(item) for item in scenario_catalog.get("sensitivity_cases", [])) + observed_usage_unit_cost = _usage_unit_cost(usage_records, snapshot.period) + + profile_results = [] + for raw_profile in scenario_catalog.get("profiles", []): + profile = _profile(raw_profile) + configurations = [ + _configuration(model, profile, snapshot, observed_usage_unit_cost) + for model in models + if model.status in ("active", "candidate") + ] + profile_results.append( + compare_pricing_configurations( + configurations, + profile, + boundary_policy, + policy, + sensitivity_cases=sensitivity_cases, + ) + ) + + primary = profile_results[0] if profile_results else None + primary_scenarios = list(primary.comparisons) if primary else [] + active_model = next((model for model in models if model.status == "active"), None) + best_margin = max(primary_scenarios, key=lambda item: item.base_monthly_margin, default=None) + best_ltv = max( + primary_scenarios, + key=lambda item: item.average_comparable_customer_lifetime_value, + default=None, + ) + + return _serialize({ + "period": snapshot.period, + "currency": snapshot.currency, + "required_improvement_factor": policy.required_improvement_factor, + "horizon_months": policy.horizon_months, + "monthly_discount_rate_pct": policy.monthly_discount_rate_pct, + "active_scenario_id": active_model.id if active_model else None, + "best_margin_scenario_id": best_margin.model_id if best_margin else None, + "best_ltv_scenario_id": best_ltv.model_id if best_ltv else None, + "reference_model_id": primary.reference_model_id if primary else None, + "primary_profile_id": primary.profile.id if primary else None, + "scenarios": primary_scenarios, + "profile_comparisons": profile_results, + "calibration": { + "observed_usage_unit_cost": observed_usage_unit_cost, + "observed_payment_fee_rate_pct": _payment_fee_rate(snapshot), + "profile_count": len(profile_results), + }, + "notes": [ + scenario_catalog.get("notes", ""), + "Primary scenarios expose the first configured comparable-customer profile for backward-compatible UI consumers.", + "Profile comparisons compare candidate models using discounted seller LTV rather than only current-period gross margin.", + ], + }) diff --git a/projects/coulomb-pricing/observatory/recommendations.py b/projects/coulomb-pricing/observatory/recommendations.py index 214b3a6..2c83668 100644 --- a/projects/coulomb-pricing/observatory/recommendations.py +++ b/projects/coulomb-pricing/observatory/recommendations.py @@ -30,7 +30,7 @@ def build_pricing_recommendations( if ai_spend > Decimal("0") and cost_per_member > Decimal("0"): ai_ratio = (ai_spend / cost_per_member) * Decimal("100") if ai_ratio > Decimal("15"): - best = simulations.get("best_margin_scenario_id") + best = simulations.get("best_ltv_scenario_id") or simulations.get("best_margin_scenario_id") recommendations.append( { "id": "usage-pricing-signal", @@ -65,4 +65,4 @@ def build_pricing_recommendations( } ) - return recommendations \ No newline at end of file + return recommendations diff --git a/projects/coulomb-pricing/observatory/simulator.py b/projects/coulomb-pricing/observatory/simulator.py index d15be2b..1d7d253 100644 --- a/projects/coulomb-pricing/observatory/simulator.py +++ b/projects/coulomb-pricing/observatory/simulator.py @@ -1,70 +1,64 @@ from __future__ import annotations -from decimal import Decimal, ROUND_HALF_UP +from decimal import Decimal +from typing import Any +from .ltv import build_ltv_simulations from .models import EconomicsSnapshot, PricingModel -TWOPLACES = Decimal("0.01") -OVERAGE_RATE = Decimal("0.002") # EUR per token above allowance (observatory estimate) - - -def _money(value: Decimal) -> Decimal: - return value.quantize(TWOPLACES, rounding=ROUND_HALF_UP) - - -def _simulate_model( - model: PricingModel, - snapshot: EconomicsSnapshot, - ai_cost_per_member: Decimal, - included_tokens: int = 100_000, - actual_tokens: int = 120_000, -) -> dict: - members = snapshot.active_members or 1 - subscription_revenue = model.access_fee_amount * members - overage_revenue = Decimal("0") - if model.model_type == "hybrid_subscription_usage" and actual_tokens > included_tokens: - overage_tokens = actual_tokens - included_tokens - overage_revenue = OVERAGE_RATE * overage_tokens * members - - gross_revenue = subscription_revenue + overage_revenue - platform_cost = snapshot.monthly_total_platform_cost + (ai_cost_per_member * members) - margin = gross_revenue - platform_cost - margin_pct = (margin / gross_revenue * Decimal("100")) if gross_revenue else Decimal("0") +def _fallback_catalog(models: list[PricingModel]) -> dict[str, Any]: return { - "model_id": model.id, - "model_name": model.name, - "model_type": model.model_type, - "status": model.status, - "access_fee_eur": model.access_fee_amount, - "projected_revenue_eur": _money(gross_revenue), - "projected_overage_eur": _money(overage_revenue), - "projected_platform_cost_eur": _money(platform_cost), - "projected_margin_eur": _money(margin), - "projected_margin_pct": _money(margin_pct), - "assumed_tokens_per_member": actual_tokens, - "included_tokens": included_tokens if model.model_type != "flat_subscription" else None, + "version": 1, + "currency": "EUR", + "horizon_months": 24, + "monthly_discount_rate_pct": "1.0", + "required_improvement_factor": "1.05", + "profiles": [ + { + "id": "observatory-default", + "name": "Observatory default", + "segment": "coulomb-social-members", + "eligible_model_ids": [model.id for model in models if model.status in ("active", "candidate")], + "members_per_customer": 1, + "expected_monthly_usage_units": "120000", + "usage_variance_pct": "25", + "monthly_churn_pct": "5.0", + "monthly_default_pct": "1.0", + "monthly_support_cost": "0.00", + "monthly_risk_cost": "0.00", + "acquisition_cost": "0.00", + "upfront_investment_cost": "0.00", + "notes": "Fallback scenario when no explicit LTV scenario catalog is provided." + } + ], + "notes": "Fallback scenario catalog generated inside observatory.simulator.", } +def _fallback_usage_records(snapshot: EconomicsSnapshot, ai_cost_per_member: Decimal) -> list[dict[str, Any]]: + return [ + { + "id": "fallback-usage", + "period": snapshot.period, + "member_id": "fallback", + "tokens": 120000, + "cost_eur": ai_cost_per_member, + "source": "fallback", + } + ] + + def build_pricing_simulations( snapshot: EconomicsSnapshot, models: list[PricingModel], ai_cost_per_member: Decimal, -) -> dict: - scenarios = [ - _simulate_model(model, snapshot, ai_cost_per_member) - for model in models - if model.status in ("active", "candidate") - ] - active = next((item for item in scenarios if item["status"] == "active"), scenarios[0]) - best_margin = max(scenarios, key=lambda item: item["projected_margin_eur"]) - - return { - "period": snapshot.period, - "currency": snapshot.currency, - "active_scenario_id": active["model_id"], - "best_margin_scenario_id": best_margin["model_id"], - "scenarios": scenarios, - "notes": "Projections hold member count and infrastructure cost constant; overage uses observatory token estimate.", - } \ No newline at end of file + usage_records: list[dict[str, Any]] | None = None, + scenario_catalog: dict[str, Any] | None = None, +) -> dict[str, Any]: + return build_ltv_simulations( + snapshot, + models, + usage_records or _fallback_usage_records(snapshot, ai_cost_per_member), + scenario_catalog or _fallback_catalog(models), + ) diff --git a/projects/coulomb-pricing/tests/test_api.py b/projects/coulomb-pricing/tests/test_api.py index 026dd14..398a452 100644 --- a/projects/coulomb-pricing/tests/test_api.py +++ b/projects/coulomb-pricing/tests/test_api.py @@ -24,6 +24,10 @@ def test_dashboard_payload_contains_live_ledger_totals() -> None: assert payload["membership_analytics"]["active_members"] == 1 assert payload["usage"]["record_count"] == 1 assert len(payload["pricing_simulations"]["scenarios"]) == 3 + assert len(payload["pricing_simulations"]["profile_comparisons"]) == 2 + assert payload["pricing_simulations"]["primary_profile_id"] == "solo-builder" + assert payload["pricing_simulations"]["required_improvement_factor"] == "1.05" + assert payload["pricing_simulations"]["reference_model_id"] is not None assert len(payload["boundary_validation"]["model_results"]) == 3 assert payload["boundary_validation"]["policy"]["target_margin_pct"] == "15" assert any( diff --git a/projects/coulomb-pricing/tests/test_comparable_ltv.py b/projects/coulomb-pricing/tests/test_comparable_ltv.py new file mode 100644 index 0000000..34bc3cc --- /dev/null +++ b/projects/coulomb-pricing/tests/test_comparable_ltv.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +from pathlib import Path + +from observatory.economics import build_snapshot +from observatory.load import ( + load_ltv_scenarios, + load_membership, + load_monthly_ledger, + load_payment_records, + load_pricing_models, + load_product, +) +from observatory.simulator import build_pricing_simulations +from observatory.usage import load_usage_records + +DATA_DIR = Path(__file__).resolve().parent.parent / "data" + + +def _snapshot(period: str = "2026-06"): + product = load_product(DATA_DIR) + models = load_pricing_models(DATA_DIR) + members = load_membership(DATA_DIR) + payments = load_payment_records(DATA_DIR) + ledger = load_monthly_ledger(DATA_DIR) + return build_snapshot(period, product, models, members, payments, ledger) + + +def test_simulations_include_reference_model_and_profile_comparisons() -> None: + snapshot = _snapshot() + models = load_pricing_models(DATA_DIR) + simulations = build_pricing_simulations( + snapshot, + models, + snapshot.cost_per_member, + usage_records=load_usage_records(DATA_DIR), + scenario_catalog=load_ltv_scenarios(DATA_DIR), + ) + + assert simulations["primary_profile_id"] == "solo-builder" + assert simulations["reference_model_id"] is not None + assert simulations["best_ltv_scenario_id"] is not None + assert len(simulations["profile_comparisons"]) == 2 + assert simulations["scenarios"][0]["average_comparable_customer_lifetime_value"] is not None + assert simulations["scenarios"][0]["sensitivity"] + + +def test_small_team_profile_has_reference_and_non_passing_candidates() -> None: + snapshot = _snapshot() + models = load_pricing_models(DATA_DIR) + simulations = build_pricing_simulations( + snapshot, + models, + snapshot.cost_per_member, + usage_records=load_usage_records(DATA_DIR), + scenario_catalog=load_ltv_scenarios(DATA_DIR), + ) + + small_team = next( + item for item in simulations["profile_comparisons"] if item["profile"]["id"] == "small-team" + ) + + assert small_team["reference_model_id"] is not None + assert small_team["best_valid_model_id"] is not None + assert any( + not comparison["passes_required_improvement"] + for comparison in small_team["comparisons"] + if comparison["model_id"] != small_team["reference_model_id"] + ) diff --git a/projects/coulomb-pricing/tests/test_mvp_sprints.py b/projects/coulomb-pricing/tests/test_mvp_sprints.py index 159d93f..73c6653 100644 --- a/projects/coulomb-pricing/tests/test_mvp_sprints.py +++ b/projects/coulomb-pricing/tests/test_mvp_sprints.py @@ -8,6 +8,7 @@ from observatory.api import build_dashboard_payload from observatory.credits import build_credit_summary, load_credit_wallets from observatory.economics import build_snapshot from observatory.load import ( + load_ltv_scenarios, load_membership, load_monthly_ledger, load_payment_records, @@ -58,10 +59,18 @@ def test_cost_allocation_includes_ai_variable_cost() -> None: def test_pricing_simulator_compares_candidate_models() -> None: snapshot = _snapshot() models = load_pricing_models(DATA_DIR) - simulations = build_pricing_simulations(snapshot, models, Decimal("0.06")) + simulations = build_pricing_simulations( + snapshot, + models, + Decimal("0.06"), + usage_records=load_usage_records(DATA_DIR), + scenario_catalog=load_ltv_scenarios(DATA_DIR), + ) assert len(simulations["scenarios"]) == 3 assert simulations["active_scenario_id"] == "flat-899-eur-monthly" + assert simulations["best_ltv_scenario_id"] is not None + assert simulations["reference_model_id"] is not None def test_credit_summary_tracks_remaining_allowance() -> None: @@ -93,4 +102,4 @@ def test_dashboard_payload_includes_mvp_sections() -> None: assert "cost_allocation" in payload assert "pricing_simulations" in payload assert "credit_wallets" in payload - assert "recommendations" in payload \ No newline at end of file + assert "recommendations" in payload diff --git a/workplans/ADAPTIVE-WP-0005-comparable-customer-ltv-and-simulation-upgrade.md b/workplans/ADAPTIVE-WP-0005-comparable-customer-ltv-and-simulation-upgrade.md index 112d357..320e1f6 100644 --- a/workplans/ADAPTIVE-WP-0005-comparable-customer-ltv-and-simulation-upgrade.md +++ b/workplans/ADAPTIVE-WP-0005-comparable-customer-ltv-and-simulation-upgrade.md @@ -4,7 +4,7 @@ type: workplan title: "Comparable customer LTV and simulation upgrade" domain: financials repo: adaptive-pricing -status: backlog +status: finished owner: codex topic_slug: helix-forge created: "2026-07-02" @@ -22,7 +22,7 @@ current-period Coulomb snapshot. ```task id: ADAPTIVE-WP-0005-T01 -status: todo +status: done priority: high state_hub_task_id: "2c6ff5b9-de47-46c1-aaa8-5e6e226f96d0" ``` @@ -35,7 +35,7 @@ required for reliable estimates. ```task id: ADAPTIVE-WP-0005-T02 -status: todo +status: done priority: high state_hub_task_id: "fbf4e274-70ec-468f-988c-cbebeef304d2" ``` @@ -48,7 +48,7 @@ commitment offsets. ```task id: ADAPTIVE-WP-0005-T03 -status: todo +status: done priority: medium state_hub_task_id: "084bf0bb-5892-429e-bd8a-1f9edc273eb3" ``` @@ -61,7 +61,7 @@ to forecast changes. ```task id: ADAPTIVE-WP-0005-T04 -status: todo +status: done priority: medium state_hub_task_id: "9c60c8d2-7cd0-4035-92dc-cae8c4cd85d7" ``` @@ -74,7 +74,7 @@ other products. ```task id: ADAPTIVE-WP-0005-T05 -status: todo +status: done priority: medium state_hub_task_id: "fabd1f1b-d3b7-4bfb-a2fc-212ec31c5f31" ``` @@ -83,4 +83,3 @@ Exit when simulations can compare pricing models using an explicit LTV-oriented economic model with documented assumptions and test coverage. Dependencies: `ADAPTIVE-WP-0003`, `ADAPTIVE-WP-0004`. -