coulomb/adaptive-pricing
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement comparable LTV engine and close WP-0005

Browse Source
This commit is contained in:
codex
2026-07-02 22:50:16 +02:00
parent 656bbb81a5
commit 386c8a46fe
13 changed files with 1060 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
adaptive_pricing_core/__init__.py
View File

@@ -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",

496
adaptive_pricing_core/comparable_ltv.py Normal file
View File

@@ -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,
)