diff --git a/adaptive_pricing_core/comparable_ltv.py b/adaptive_pricing_core/comparable_ltv.py index 34a0605..3b6a573 100644 --- a/adaptive_pricing_core/comparable_ltv.py +++ b/adaptive_pricing_core/comparable_ltv.py @@ -185,7 +185,7 @@ 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: +def required_improvement_threshold(reference_ltv: Decimal, factor: Decimal) -> Decimal: if reference_ltv >= Decimal("0"): return _money(reference_ltv * factor) improvement = abs(reference_ltv) * (factor - Decimal("1")) @@ -412,7 +412,7 @@ def compare_pricing_configurations( threshold: Decimal | None = None passes_required_improvement = True if reference is not None and estimate.model_id != reference.model_id: - threshold = _required_threshold( + threshold = required_improvement_threshold( reference.average_comparable_customer_lifetime_value, policy.required_improvement_factor, ) diff --git a/adaptive_pricing_core/customer_tuning.py b/adaptive_pricing_core/customer_tuning.py new file mode 100644 index 0000000..76dd1b3 --- /dev/null +++ b/adaptive_pricing_core/customer_tuning.py @@ -0,0 +1,511 @@ +from __future__ import annotations + +from dataclasses import dataclass, replace +from decimal import Decimal, ROUND_HALF_UP +from typing import Literal + +from .boundary_engine import ( + BoundaryPolicy, + CommitmentTerms, + ConstraintResult, + PricingConfiguration, + ValidationResult, +) +from .comparable_ltv import ( + ComparableCustomerProfile, + ComparableLTVEstimate, + LTVPolicy, + estimate_comparable_customer_ltv, + required_improvement_threshold, + select_reference_estimate, +) + +SolverPreference = Literal["lower_usage_price", "seller_ltv"] +ApprovalMode = Literal["self_serve_only", "allow_approval"] +TuningDecision = Literal["accepted", "requires_approval", "rejected"] + +TWOPLACES = Decimal("0.01") + + +def _money(value: Decimal) -> Decimal: + return value.quantize(TWOPLACES, rounding=ROUND_HALF_UP) + + +def _usage_component(configuration: PricingConfiguration): + return next( + (component for component in configuration.model.charge_components if component.kind == "usage"), + None, + ) + + +def _default_usage_unit_price(configuration: PricingConfiguration) -> Decimal: + usage_component = _usage_component(configuration) + if configuration.usage_unit_price is not None: + return configuration.usage_unit_price + if usage_component and usage_component.unit_price is not None: + return usage_component.unit_price + for parameter in configuration.model.tunable_parameters: + if parameter.key == "overage_unit_price" and parameter.default_value not in (None, ""): + return Decimal(str(parameter.default_value)) + return Decimal("0") + + +def _percent_delta(candidate: Decimal, reference: Decimal) -> Decimal | None: + if reference == Decimal("0"): + return None + return _money(((candidate - reference) / abs(reference)) * Decimal("100")) + + +@dataclass(frozen=True) +class UsagePriceSearchPolicy: + min_usage_unit_price: Decimal | None = None + max_usage_unit_price: Decimal | None = None + usage_unit_price_step: Decimal = Decimal("0.0001") + max_usage_price_multiplier: Decimal = Decimal("4") + + +@dataclass(frozen=True) +class CustomerTuningRequest: + included_units: Decimal | None = None + contract_duration_months: int | None = None + minimum_monthly_turnover: Decimal = Decimal("0") + prepaid_amount: Decimal = Decimal("0") + guaranteed_platform_fee: Decimal = Decimal("0") + customer_funded_onboarding: Decimal = Decimal("0") + reduced_cancellation_flexibility: bool | None = None + preference: SolverPreference = "lower_usage_price" + approval_mode: ApprovalMode = "self_serve_only" + + +@dataclass(frozen=True) +class CustomerTuningOutcome: + model_id: str + model_name: str + decision: TuningDecision + valid: bool + requires_approval: bool + preference: SolverPreference + approval_mode: ApprovalMode + request: CustomerTuningRequest + solved_configuration: dict[str, object] + solved_usage_unit_price: Decimal + reference_model_id: str | None + reference_model_name: str | None + reference_ltv: Decimal | None + required_improvement_threshold: Decimal | None + average_comparable_customer_lifetime_value: Decimal + improvement_vs_reference_pct: Decimal | None + passes_required_improvement: bool + evaluated_candidates: int + tradeoffs: tuple[str, ...] + binding_constraints: tuple[ConstraintResult, ...] + validation: ValidationResult + explanation: str + + +@dataclass(frozen=True) +class _CandidateAssessment: + configuration: PricingConfiguration + estimate: ComparableLTVEstimate + decision: TuningDecision + passes_required_improvement: bool + improvement_vs_reference_pct: Decimal | None + + +def _price_range( + configuration: PricingConfiguration, + search_policy: UsagePriceSearchPolicy, +) -> tuple[Decimal, ...]: + step = search_policy.usage_unit_price_step + if step <= Decimal("0"): + raise ValueError("usage_unit_price_step must be positive") + + default_usage_price = _default_usage_unit_price(configuration) + min_usage_price = search_policy.min_usage_unit_price + if min_usage_price is None: + min_usage_price = max(configuration.unit_cost, default_usage_price / Decimal("10"), step) + max_usage_price = search_policy.max_usage_unit_price + if max_usage_price is None: + base = default_usage_price if default_usage_price > Decimal("0") else step + max_usage_price = max(min_usage_price, base * search_policy.max_usage_price_multiplier) + + if max_usage_price < min_usage_price: + max_usage_price = min_usage_price + + values: list[Decimal] = [] + current = min_usage_price + while current <= max_usage_price: + values.append(current) + current += step + if not values or values[-1] != max_usage_price: + values.append(max_usage_price) + return tuple(dict.fromkeys(values)) + + +def _resolved_search_policy( + configuration: PricingConfiguration, + request: CustomerTuningRequest, + search_policy: UsagePriceSearchPolicy | None, +) -> UsagePriceSearchPolicy: + policy = search_policy or UsagePriceSearchPolicy() + if request.preference != "lower_usage_price" or policy.max_usage_unit_price is not None: + return policy + + return replace( + policy, + max_usage_unit_price=_default_usage_unit_price(configuration), + ) + + +def _commitment_terms( + base_terms: CommitmentTerms, + request: CustomerTuningRequest, +) -> CommitmentTerms: + return replace( + base_terms, + contract_duration_months=( + request.contract_duration_months + if request.contract_duration_months is not None + else base_terms.contract_duration_months + ), + minimum_monthly_turnover=request.minimum_monthly_turnover, + prepaid_amount=request.prepaid_amount, + guaranteed_platform_fee=request.guaranteed_platform_fee, + customer_funded_onboarding=request.customer_funded_onboarding, + reduced_cancellation_flexibility=( + request.reduced_cancellation_flexibility + if request.reduced_cancellation_flexibility is not None + else base_terms.reduced_cancellation_flexibility + ), + ) + + +def _candidate_configuration( + base_configuration: PricingConfiguration, + request: CustomerTuningRequest, + usage_unit_price: Decimal, +) -> PricingConfiguration: + return replace( + base_configuration, + included_units=( + request.included_units + if request.included_units is not None + else base_configuration.included_units + ), + usage_unit_price=usage_unit_price, + commitment_terms=_commitment_terms(base_configuration.commitment_terms, request), + ) + + +def _candidate_decision( + validation: ValidationResult, + passes_required_improvement: bool, + approval_mode: ApprovalMode, +) -> TuningDecision: + if not validation.valid or not passes_required_improvement: + return "rejected" + if validation.requires_approval: + return "requires_approval" if approval_mode == "allow_approval" else "rejected" + return "accepted" + + +def _headroom_by_constraint( + configuration: PricingConfiguration, + validation: ValidationResult, +) -> dict[str, Decimal]: + metrics = validation.metrics + policy = validation.policy + return { + "usage-variance-limit": policy.max_expected_usage_variance_pct - configuration.expected_usage_variance_pct, + "payment-fee-limit": policy.max_payment_fee_pct - metrics.payment_fee_pct, + "cost-floor-coverage": metrics.monthly_margin, + "minimum-margin": metrics.margin_pct - policy.minimum_margin_pct, + "target-margin-approval": metrics.margin_pct - policy.target_margin_pct, + "discount-exposure-limit": policy.max_discount_pct - metrics.concession_pct, + "discount-approval-threshold": policy.approval_discount_pct - metrics.concession_pct, + } + + +def _binding_constraints( + configuration: PricingConfiguration, + validation: ValidationResult, +) -> tuple[ConstraintResult, ...]: + flagged = tuple(result for result in validation.constraints if result.status != "pass") + if flagged: + return flagged + + headroom = _headroom_by_constraint(configuration, validation) + ordered_ids = [ + constraint_id + for constraint_id, _headroom in sorted(headroom.items(), key=lambda item: item[1]) + if constraint_id in {result.id for result in validation.constraints} + ] + selected_ids = ordered_ids[:2] + if not selected_ids: + return () + + return tuple( + result for result in validation.constraints if result.id in selected_ids + ) + + +def _tradeoffs( + base_configuration: PricingConfiguration, + candidate_configuration: PricingConfiguration, + validation: ValidationResult, +) -> tuple[str, ...]: + tradeoffs: list[str] = [] + + if ( + base_configuration.included_units is not None + and candidate_configuration.included_units is not None + and candidate_configuration.included_units < base_configuration.included_units + ): + tradeoffs.append("lower_included_usage") + if ( + base_configuration.included_units is not None + and candidate_configuration.included_units is not None + and candidate_configuration.included_units > base_configuration.included_units + ): + tradeoffs.append("higher_included_usage") + if ( + base_configuration.usage_unit_price is not None + and candidate_configuration.usage_unit_price is not None + and candidate_configuration.usage_unit_price < base_configuration.usage_unit_price + ): + tradeoffs.append("lower_usage_price") + if ( + base_configuration.usage_unit_price is not None + and candidate_configuration.usage_unit_price is not None + and candidate_configuration.usage_unit_price > base_configuration.usage_unit_price + ): + tradeoffs.append("higher_usage_price") + + baseline_duration = base_configuration.commitment_terms.contract_duration_months or 0 + candidate_duration = validation.metrics.contract_duration_months + if candidate_duration > baseline_duration: + tradeoffs.append("longer_contract_duration") + if validation.metrics.minimum_monthly_turnover > Decimal("0"): + tradeoffs.append("minimum_monthly_turnover") + if validation.metrics.prepaid_amount > Decimal("0"): + tradeoffs.append("prepayment") + if validation.metrics.guaranteed_platform_fee > Decimal("0"): + tradeoffs.append("guaranteed_platform_fee") + if validation.metrics.customer_funded_onboarding > Decimal("0"): + tradeoffs.append("customer_funded_onboarding") + if validation.metrics.reduced_cancellation_flexibility: + tradeoffs.append("reduced_cancellation_flexibility") + + for signal in validation.metrics.meaningful_commitment_signals: + if signal not in tradeoffs: + tradeoffs.append(signal) + + return tuple(tradeoffs) + + +def _explanation( + assessment: _CandidateAssessment, + request: CustomerTuningRequest, + reference_estimate: ComparableLTVEstimate | None, + threshold: Decimal | None, + tradeoffs: tuple[str, ...], + binding_constraints: tuple[ConstraintResult, ...], +) -> str: + validation = assessment.estimate.validation + metrics = validation.metrics + if assessment.decision in {"accepted", "requires_approval"}: + outcome = ( + "Accepted self-serve tuning" + if assessment.decision == "accepted" + else "Requires seller approval" + ) + parts = [ + f"{outcome} at {metrics.usage_unit_price} {metrics.currency} usage price.", + ( + f"Comparable-customer LTV {assessment.estimate.average_comparable_customer_lifetime_value} " + f"{metrics.currency}" + ), + ] + if reference_estimate is not None and threshold is not None: + parts.append( + f"clears threshold {threshold} {metrics.currency} versus {reference_estimate.model_name}." + ) + if tradeoffs: + parts.append("Trade-offs: " + ", ".join(tradeoffs) + ".") + return " ".join(parts) + + failed_constraints = [result.title for result in binding_constraints if result.status == "fail"] + review_constraints = [result.title for result in binding_constraints if result.status == "review"] + parts = ["Rejected tuning request."] + if not assessment.passes_required_improvement and reference_estimate is not None and threshold is not None: + parts.append( + ( + f"LTV {assessment.estimate.average_comparable_customer_lifetime_value} {metrics.currency} " + f"misses threshold {threshold} {metrics.currency} versus {reference_estimate.model_name}." + ) + ) + if failed_constraints: + parts.append("Hard blockers: " + ", ".join(failed_constraints) + ".") + if review_constraints and request.approval_mode == "self_serve_only": + parts.append("Self-serve blockers: " + ", ".join(review_constraints) + ".") + if tradeoffs: + parts.append("Attempted trade-offs: " + ", ".join(tradeoffs) + ".") + return " ".join(parts) + + +def _acceptable_candidates( + candidates: tuple[_CandidateAssessment, ...], +) -> tuple[_CandidateAssessment, ...]: + return tuple(candidate for candidate in candidates if candidate.decision in {"accepted", "requires_approval"}) + + +def _candidate_sort_key( + candidate: _CandidateAssessment, + preference: SolverPreference, +) -> tuple[Decimal, Decimal]: + usage_price = candidate.estimate.validation.metrics.usage_unit_price + ltv = candidate.estimate.average_comparable_customer_lifetime_value + if preference == "lower_usage_price": + return (usage_price, -ltv) + return (-ltv, usage_price) + + +def _fallback_sort_key( + candidate: _CandidateAssessment, + preference: SolverPreference, +) -> tuple[int, int, int, Decimal, Decimal]: + usage_price = candidate.estimate.validation.metrics.usage_unit_price + ltv = candidate.estimate.average_comparable_customer_lifetime_value + return ( + 0 if candidate.passes_required_improvement else 1, + 0 if candidate.estimate.validation.valid else 1, + 0 if not candidate.estimate.validation.requires_approval else 1, + usage_price if preference == "lower_usage_price" else -ltv, + -ltv if preference == "lower_usage_price" else usage_price, + ) + + +def _select_candidate( + candidates: tuple[_CandidateAssessment, ...], + preference: SolverPreference, +) -> _CandidateAssessment: + acceptable = _acceptable_candidates(candidates) + if acceptable: + return min(acceptable, key=lambda candidate: _candidate_sort_key(candidate, preference)) + return min(candidates, key=lambda candidate: _fallback_sort_key(candidate, preference)) + + +def solve_customer_tuning( + base_configuration: PricingConfiguration, + reference_configurations: list[PricingConfiguration], + profile: ComparableCustomerProfile, + boundary_policy: BoundaryPolicy, + ltv_policy: LTVPolicy, + request: CustomerTuningRequest, + search_policy: UsagePriceSearchPolicy | None = None, +) -> CustomerTuningOutcome: + if _usage_component(base_configuration) is None: + raise ValueError("customer tuning prototype currently requires a usage-priced model") + + reference_estimates = [ + estimate_comparable_customer_ltv(configuration, profile, boundary_policy, ltv_policy) + for configuration in reference_configurations + ] + reference_estimate = select_reference_estimate(reference_estimates, profile.eligible_model_ids) + threshold = ( + required_improvement_threshold( + reference_estimate.average_comparable_customer_lifetime_value, + ltv_policy.required_improvement_factor, + ) + if reference_estimate is not None + else None + ) + + candidates: list[_CandidateAssessment] = [] + for usage_unit_price in _price_range( + base_configuration, + _resolved_search_policy(base_configuration, request, search_policy), + ): + configuration = _candidate_configuration(base_configuration, request, usage_unit_price) + estimate = estimate_comparable_customer_ltv( + configuration, + profile, + boundary_policy, + ltv_policy, + ) + passes_required_improvement = ( + True + if threshold is None + else estimate.average_comparable_customer_lifetime_value >= threshold + ) + decision = _candidate_decision( + estimate.validation, + passes_required_improvement, + request.approval_mode, + ) + candidates.append( + _CandidateAssessment( + configuration=configuration, + estimate=estimate, + decision=decision, + passes_required_improvement=passes_required_improvement, + improvement_vs_reference_pct=( + _percent_delta( + estimate.average_comparable_customer_lifetime_value, + reference_estimate.average_comparable_customer_lifetime_value, + ) + if reference_estimate is not None + else None + ), + ) + ) + + if not candidates: + raise ValueError("customer tuning search produced no candidates") + + selected = _select_candidate(tuple(candidates), request.preference) + binding_constraints = _binding_constraints(selected.configuration, selected.estimate.validation) + tradeoffs = _tradeoffs( + base_configuration, + selected.configuration, + selected.estimate.validation, + ) + explanation = _explanation( + selected, + request, + reference_estimate, + threshold, + tradeoffs, + binding_constraints, + ) + + return CustomerTuningOutcome( + model_id=base_configuration.model.id, + model_name=base_configuration.model.name, + decision=selected.decision, + valid=selected.estimate.validation.valid, + requires_approval=selected.estimate.validation.requires_approval, + preference=request.preference, + approval_mode=request.approval_mode, + request=request, + solved_configuration=selected.estimate.validation.configuration, + solved_usage_unit_price=selected.estimate.validation.metrics.usage_unit_price, + reference_model_id=reference_estimate.model_id if reference_estimate else None, + reference_model_name=reference_estimate.model_name if reference_estimate else None, + reference_ltv=( + reference_estimate.average_comparable_customer_lifetime_value + if reference_estimate is not None + else None + ), + required_improvement_threshold=threshold, + average_comparable_customer_lifetime_value=( + selected.estimate.average_comparable_customer_lifetime_value + ), + improvement_vs_reference_pct=selected.improvement_vs_reference_pct, + passes_required_improvement=selected.passes_required_improvement, + evaluated_candidates=len(candidates), + tradeoffs=tradeoffs, + binding_constraints=binding_constraints, + validation=selected.estimate.validation, + explanation=explanation, + ) diff --git a/docs/CustomerTuningSolver.md b/docs/CustomerTuningSolver.md new file mode 100644 index 0000000..1ca6abb --- /dev/null +++ b/docs/CustomerTuningSolver.md @@ -0,0 +1,93 @@ +# Customer-Tuning Solver Prototype + +Status: MVP for `ADAPTIVE-WP-0006`. + +## Purpose + +This milestone adds the first executable customer-tuning flow described in +`INTENT.md`. + +The solver now accepts selected customer-tunable inputs, solves the remaining +usage-price parameter, validates the tuned configuration against boundary +constraints, and checks seller-side comparable-customer LTV against the best +available predefined reference model. + +## Generic Solver Contract + +Core module: `adaptive_pricing_core/customer_tuning.py` + +Inputs: + +- a baseline `PricingConfiguration` +- the comparable-customer profile +- boundary policy +- LTV policy +- a `CustomerTuningRequest` +- the set of predefined reference configurations available to that profile + +Current request fields: + +- `included_units` +- `contract_duration_months` +- `minimum_monthly_turnover` +- `prepaid_amount` +- `guaranteed_platform_fee` +- `customer_funded_onboarding` +- `reduced_cancellation_flexibility` +- preference: `lower_usage_price` or `seller_ltv` +- approval mode: `self_serve_only` or `allow_approval` + +Current solved field: + +- `usage_unit_price` + +## Decision Logic + +For each candidate usage price in the search range, the solver: + +1. builds a tuned `PricingConfiguration` +2. runs boundary validation +3. estimates `average_comparable_customer_lifetime_value` +4. compares the tuned result with the best predefined reference model for the + profile + +A tuned configuration is only accepted when: + +- boundary validation is valid +- no seller approval is required when the request is `self_serve_only` +- tuned comparable-customer LTV meets the configured improvement threshold + +The solver returns structured output including: + +- accepted / rejected / requires approval decision +- solved configuration +- reference model and required LTV threshold +- binding constraints +- chosen trade-offs +- explanation text + +## Coulomb Pilot + +Pilot module: `projects/coulomb-pricing/observatory/tuning.py` + +Pilot request catalog: + +- `projects/coulomb-pricing/data/tuning_requests.json` + +The Coulomb pilot currently targets `membership-plus-overage` against the +`small-team` comparable-customer profile. + +Two pilot requests are shipped: + +- a seller-safe lower-usage-price request that succeeds +- a high-included-usage request that is rejected for self-serve + +## Current Modeling Note + +The observatory simulation path still scales default hybrid included usage by +`members_per_customer`. + +The tuning pilot interprets request-level `included_tokens` values as total +package allowances, then maps them into canonical configuration fields before +running the solver. This keeps the prototype aligned with the catalog’s tunable +bounds while avoiding a broader simulation recalibration inside this milestone. diff --git a/projects/coulomb-pricing/data/tuning_requests.json b/projects/coulomb-pricing/data/tuning_requests.json new file mode 100644 index 0000000..2e76cee --- /dev/null +++ b/projects/coulomb-pricing/data/tuning_requests.json @@ -0,0 +1,38 @@ +{ + "version": 1, + "requests": [ + { + "id": "small-team-lower-usage-price", + "name": "Small team lower usage price", + "profile_id": "small-team", + "model_id": "membership-plus-overage", + "preference": "lower_usage_price", + "approval_mode": "self_serve_only", + "selected_tunables": { + "included_tokens": "50000", + "contract_duration_months": 3 + }, + "search_policy": { + "min_usage_unit_price": "0.0005", + "usage_unit_price_step": "0.0001" + } + }, + { + "id": "small-team-high-included-bundle", + "name": "Small team high included bundle", + "profile_id": "small-team", + "model_id": "membership-plus-overage", + "preference": "lower_usage_price", + "approval_mode": "self_serve_only", + "selected_tunables": { + "included_tokens": "150000", + "contract_duration_months": 3 + }, + "search_policy": { + "min_usage_unit_price": "0.0005", + "usage_unit_price_step": "0.0001" + } + } + ], + "notes": "Customer-tuning pilot requests for the Coulomb hybrid overage prototype." +} diff --git a/projects/coulomb-pricing/observatory/api.py b/projects/coulomb-pricing/observatory/api.py index f46dde5..2f2dbe1 100644 --- a/projects/coulomb-pricing/observatory/api.py +++ b/projects/coulomb-pricing/observatory/api.py @@ -18,6 +18,7 @@ from .load import ( load_payment_records, load_pricing_models, load_product, + load_tuning_requests, load_value_range, ) from .allocation import build_cost_allocation @@ -27,6 +28,7 @@ from .membership_analytics import build_membership_analytics from .pricing_context import build_cost_floor, build_market_price_view, build_value_range_view from .recommendations import build_pricing_recommendations from .simulator import build_pricing_simulations +from .tuning import build_customer_tuning_pilot from .usage import build_usage_summary, load_usage_records @@ -90,6 +92,7 @@ def build_dashboard_payload(data_dir: Path | None = None, period: str | None = N usage_records = load_usage_records(root) usage_summary = build_usage_summary(usage_records, target_period) ltv_scenarios = load_ltv_scenarios(root) + tuning_requests = load_tuning_requests(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) @@ -102,6 +105,13 @@ def build_dashboard_payload(data_dir: Path | None = None, period: str | None = N usage_records=usage_records, scenario_catalog=ltv_scenarios, ) + customer_tuning = build_customer_tuning_pilot( + snapshot, + models, + usage_records, + ltv_scenarios, + tuning_requests, + ) boundary_validation = build_boundary_validation(snapshot, models, usage_records) credit_wallets = load_credit_wallets(root) credit_summary = build_credit_summary( @@ -135,6 +145,7 @@ def build_dashboard_payload(data_dir: Path | None = None, period: str | None = N "usage": usage_summary, "cost_allocation": cost_allocation, "pricing_simulations": simulations, + "customer_tuning": customer_tuning, "boundary_validation": boundary_validation, "credit_wallets": credit_summary, "recommendations": recommendations, diff --git a/projects/coulomb-pricing/observatory/load.py b/projects/coulomb-pricing/observatory/load.py index 8d4b168..2b391f1 100644 --- a/projects/coulomb-pricing/observatory/load.py +++ b/projects/coulomb-pricing/observatory/load.py @@ -119,6 +119,13 @@ def load_ltv_scenarios(data_dir: Path | None = None) -> dict: return _read_json((data_dir or default_data_dir()) / "ltv_scenarios.json") +def load_tuning_requests(data_dir: Path | None = None) -> dict: + path = (data_dir or default_data_dir()) / "tuning_requests.json" + if not path.exists(): + return {} + return _read_json(path) + + 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/tuning.py b/projects/coulomb-pricing/observatory/tuning.py new file mode 100644 index 0000000..a7044d7 --- /dev/null +++ b/projects/coulomb-pricing/observatory/tuning.py @@ -0,0 +1,178 @@ +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 .ltv import _configuration, _ltv_policy, _profile, _usage_unit_cost +from .models import EconomicsSnapshot, PricingModel + +ensure_repo_root_on_syspath() + +from adaptive_pricing_core.customer_tuning import ( # noqa: E402 + CustomerTuningRequest, + UsagePriceSearchPolicy, + solve_customer_tuning, +) + + +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 | None: + if value in (None, ""): + return None + return Decimal(str(value)) + + +def _customer_tunable_keys(model: PricingModel) -> set[str]: + return { + parameter.key + for parameter in model.tunable_parameters + if parameter.parameter_class == "customer_tunable" + } + + +def _validate_request_surface(model: PricingModel, selected_tunables: dict[str, Any]) -> tuple[str, ...]: + tunable_keys = _customer_tunable_keys(model) + issues: list[str] = [] + for key in selected_tunables: + if key not in tunable_keys: + issues.append(f"{key} is not customer-tunable on {model.id}") + return tuple(issues) + + +def _request(raw: dict[str, Any]) -> CustomerTuningRequest: + selected = raw.get("selected_tunables", {}) + return CustomerTuningRequest( + included_units=_decimal(selected.get("included_tokens")), + contract_duration_months=( + int(selected["contract_duration_months"]) + if selected.get("contract_duration_months") is not None + else None + ), + minimum_monthly_turnover=_decimal(selected.get("minimum_monthly_turnover")) or Decimal("0"), + prepaid_amount=_decimal(selected.get("prepaid_amount")) or Decimal("0"), + guaranteed_platform_fee=_decimal(selected.get("guaranteed_platform_fee")) or Decimal("0"), + customer_funded_onboarding=_decimal(selected.get("customer_funded_onboarding")) or Decimal("0"), + reduced_cancellation_flexibility=raw.get("reduced_cancellation_flexibility"), + preference=raw.get("preference", "lower_usage_price"), + approval_mode=raw.get("approval_mode", "self_serve_only"), + ) + + +def _search_policy(raw: dict[str, Any]) -> UsagePriceSearchPolicy | None: + if not raw: + return None + return UsagePriceSearchPolicy( + min_usage_unit_price=_decimal(raw.get("min_usage_unit_price")), + max_usage_unit_price=_decimal(raw.get("max_usage_unit_price")), + usage_unit_price_step=_decimal(raw.get("usage_unit_price_step")) or Decimal("0.0001"), + max_usage_price_multiplier=_decimal(raw.get("max_usage_price_multiplier")) or Decimal("4"), + ) + + +def build_customer_tuning_pilot( + snapshot: EconomicsSnapshot, + models: list[PricingModel], + usage_records: list[dict[str, Any]], + scenario_catalog: dict[str, Any], + request_catalog: dict[str, Any] | None = None, +) -> dict[str, Any]: + request_catalog = request_catalog or {} + if not request_catalog.get("requests"): + return { + "period": snapshot.period, + "currency": snapshot.currency, + "requests": [], + "notes": [ + "No customer-tuning pilot requests are configured for this observatory deployment.", + ], + } + + profile_index = {item["id"]: _profile(item) for item in scenario_catalog.get("profiles", [])} + model_index = {model.id: model for model in models if model.status in ("active", "candidate")} + policy = _ltv_policy(scenario_catalog) + boundary_policy = build_boundary_policy(snapshot) + observed_usage_unit_cost = _usage_unit_cost(usage_records, snapshot.period) + + results = [] + for raw_request in request_catalog.get("requests", []): + model = model_index[raw_request["model_id"]] + profile = profile_index[raw_request["profile_id"]] + selected_tunables = raw_request.get("selected_tunables", {}) + issues = _validate_request_surface(model, selected_tunables) + + if issues: + results.append( + { + "id": raw_request["id"], + "name": raw_request["name"], + "decision": "rejected", + "issues": list(issues), + "profile_id": profile.id, + "model_id": model.id, + } + ) + continue + + base_configuration = _configuration(model, profile, snapshot, observed_usage_unit_cost) + reference_configurations = [ + _configuration(candidate, profile, snapshot, observed_usage_unit_cost) + for candidate in models + if candidate.status in ("active", "candidate") + ] + outcome = solve_customer_tuning( + base_configuration, + reference_configurations, + profile, + boundary_policy, + policy, + _request(raw_request), + search_policy=_search_policy(raw_request.get("search_policy", {})), + ) + results.append( + { + "id": raw_request["id"], + "name": raw_request["name"], + "profile_id": profile.id, + "profile_name": profile.name, + "model_id": model.id, + "model_name": model.name, + "selected_tunables": selected_tunables, + "result": outcome, + } + ) + + accepted = [ + item["id"] + for item in results + if item.get("result") is not None and getattr(item["result"], "decision", None) == "accepted" + ] + + return _serialize( + { + "period": snapshot.period, + "currency": snapshot.currency, + "request_count": len(results), + "accepted_request_ids": accepted, + "requests": results, + "notes": [ + request_catalog.get("notes", ""), + "Pilot requests map product-level tunables into canonical pricing configuration fields before running the generic solver.", + "For Coulomb's hybrid prototype, selected included token values are treated as total package allowances rather than per-seat multipliers.", + ], + } + ) diff --git a/projects/coulomb-pricing/tests/test_api.py b/projects/coulomb-pricing/tests/test_api.py index 398a452..0f9dc94 100644 --- a/projects/coulomb-pricing/tests/test_api.py +++ b/projects/coulomb-pricing/tests/test_api.py @@ -28,6 +28,8 @@ def test_dashboard_payload_contains_live_ledger_totals() -> None: 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 payload["customer_tuning"]["request_count"] == 2 + assert payload["customer_tuning"]["accepted_request_ids"] == ["small-team-lower-usage-price"] 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_customer_tuning.py b/projects/coulomb-pricing/tests/test_customer_tuning.py new file mode 100644 index 0000000..29ed66f --- /dev/null +++ b/projects/coulomb-pricing/tests/test_customer_tuning.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +from decimal import Decimal +from pathlib import Path + +from adaptive_pricing_core.customer_tuning import CustomerTuningRequest, solve_customer_tuning +from observatory.boundary import build_boundary_policy +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.ltv import _configuration, _ltv_policy, _profile, _usage_unit_cost +from observatory.tuning import build_customer_tuning_pilot +from observatory.usage import load_usage_records + +DATA_DIR = Path(__file__).resolve().parent.parent / "data" + + +def _scenario_inputs(profile_id: str = "small-team"): + 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) + snapshot = build_snapshot("2026-06", product, models, members, payments, ledger) + usage_records = load_usage_records(DATA_DIR) + scenario_catalog = load_ltv_scenarios(DATA_DIR) + profile = _profile(next(item for item in scenario_catalog["profiles"] if item["id"] == profile_id)) + observed_usage_unit_cost = _usage_unit_cost(usage_records, snapshot.period) + return snapshot, models, usage_records, scenario_catalog, profile, observed_usage_unit_cost + + +def test_lower_usage_price_request_can_stay_seller_safe() -> None: + ( + snapshot, + models, + usage_records, + scenario_catalog, + profile, + observed_usage_unit_cost, + ) = _scenario_inputs() + model = next(item for item in models if item.id == "membership-plus-overage") + outcome = solve_customer_tuning( + _configuration(model, profile, snapshot, observed_usage_unit_cost), + [ + _configuration(candidate, profile, snapshot, observed_usage_unit_cost) + for candidate in models + if candidate.status in ("active", "candidate") + ], + profile, + build_boundary_policy(snapshot), + _ltv_policy(scenario_catalog), + CustomerTuningRequest( + included_units=Decimal("50000"), + contract_duration_months=3, + preference="lower_usage_price", + approval_mode="self_serve_only", + ), + ) + + assert outcome.decision == "accepted" + assert outcome.reference_model_id == "flat-899-eur-monthly" + assert outcome.passes_required_improvement is True + assert outcome.solved_usage_unit_price < Decimal("0.002") + assert "lower_included_usage" in outcome.tradeoffs + assert "longer_contract_duration" in outcome.tradeoffs + + +def test_high_included_request_is_rejected_for_self_serve() -> None: + ( + snapshot, + models, + _usage_records, + scenario_catalog, + profile, + observed_usage_unit_cost, + ) = _scenario_inputs() + model = next(item for item in models if item.id == "membership-plus-overage") + outcome = solve_customer_tuning( + _configuration(model, profile, snapshot, observed_usage_unit_cost), + [ + _configuration(candidate, profile, snapshot, observed_usage_unit_cost) + for candidate in models + if candidate.status in ("active", "candidate") + ], + profile, + build_boundary_policy(snapshot), + _ltv_policy(scenario_catalog), + CustomerTuningRequest( + included_units=Decimal("150000"), + contract_duration_months=3, + preference="lower_usage_price", + approval_mode="self_serve_only", + ), + ) + + assert outcome.decision == "rejected" + assert outcome.passes_required_improvement is True + assert any( + constraint.id == "discount-exposure-limit" + for constraint in outcome.binding_constraints + ) + + +def test_customer_tuning_pilot_surfaces_accepted_and_rejected_requests() -> None: + snapshot, models, usage_records, scenario_catalog, _profile_data, _usage_unit_cost_value = _scenario_inputs() + pilot = build_customer_tuning_pilot( + snapshot, + models, + usage_records, + scenario_catalog, + { + "requests": [ + { + "id": "accepted", + "name": "Accepted", + "profile_id": "small-team", + "model_id": "membership-plus-overage", + "preference": "lower_usage_price", + "approval_mode": "self_serve_only", + "selected_tunables": { + "included_tokens": "50000", + "contract_duration_months": 3, + }, + }, + { + "id": "rejected", + "name": "Rejected", + "profile_id": "small-team", + "model_id": "membership-plus-overage", + "preference": "lower_usage_price", + "approval_mode": "self_serve_only", + "selected_tunables": { + "included_tokens": "150000", + "contract_duration_months": 3, + }, + }, + ] + }, + ) + + assert pilot["request_count"] == 2 + assert pilot["accepted_request_ids"] == ["accepted"] + assert {item["result"]["decision"] for item in pilot["requests"]} == {"accepted", "rejected"} diff --git a/workplans/ADAPTIVE-WP-0006-customer-tuning-solver-prototype.md b/workplans/ADAPTIVE-WP-0006-customer-tuning-solver-prototype.md index 42b402e..791ba99 100644 --- a/workplans/ADAPTIVE-WP-0006-customer-tuning-solver-prototype.md +++ b/workplans/ADAPTIVE-WP-0006-customer-tuning-solver-prototype.md @@ -4,7 +4,7 @@ type: workplan title: "Customer-tuning solver prototype" domain: financials repo: adaptive-pricing -status: backlog +status: finished owner: codex topic_slug: helix-forge created: "2026-07-02" @@ -20,7 +20,7 @@ Implement the first seller-safe customer-tuning flow described in `INTENT.md`. ```task id: ADAPTIVE-WP-0006-T01 -status: todo +status: done priority: high state_hub_task_id: "ad2df753-ae93-4813-bd35-fb8ee78149cc" ``` @@ -32,7 +32,7 @@ expressed, and which parts remain seller-controlled or calculated. ```task id: ADAPTIVE-WP-0006-T02 -status: todo +status: done priority: high state_hub_task_id: "9146f1dc-2e86-4617-8e3e-0b5e4799daa3" ``` @@ -45,7 +45,7 @@ requirements. ```task id: ADAPTIVE-WP-0006-T03 -status: todo +status: done priority: high state_hub_task_id: "57d6066d-088b-413c-817b-9e374f60f83a" ``` @@ -57,7 +57,7 @@ including which constraints bound the result and which trade-offs were chosen. ```task id: ADAPTIVE-WP-0006-T04 -status: todo +status: done priority: medium state_hub_task_id: "86423454-71e4-4f8e-afe0-61ceb70314c2" ``` @@ -70,7 +70,7 @@ trusted. ```task id: ADAPTIVE-WP-0006-T05 -status: todo +status: done priority: medium state_hub_task_id: "72f7d7b4-6ad2-4b63-b9de-acb7a7b81832" ``` @@ -79,4 +79,3 @@ Exit when the repo can produce and explain seller-safe tuned configurations for at least one hybrid pricing model family. Dependencies: `ADAPTIVE-WP-0004`, `ADAPTIVE-WP-0005`. -