Files
adaptive-pricing/adaptive_pricing_core/boundary_engine.py
2026-07-02 22:07:56 +02:00

938 lines
38 KiB
Python

from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal, ROUND_HALF_UP
from typing import Any, Callable, Literal
from .pricing_models import PricingModel
ConstraintSeverity = Literal["hard", "soft"]
ConstraintStatus = Literal["pass", "fail", "review"]
ValidationDecision = Literal["accepted", "requires_approval", "rejected"]
ConstraintEvaluator = Callable[
["PricingConfiguration", "BoundaryPolicy", "PricingMetrics", "PricingMetrics"],
"ConstraintResult",
]
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))
def _recurring_non_usage_component_revenue(model: PricingModel) -> Decimal:
total = Decimal("0")
for component in model.charge_components:
if component.kind in {"access", "usage", "setup"}:
continue
if component.amount is None:
continue
if component.cadence == "one_time":
continue
total += component.amount
return total
def _access_fee_amount(model: PricingModel) -> Decimal:
for component in model.charge_components:
if component.kind == "access" and component.amount is not None:
return component.amount
return model.access_fee_amount
def _usage_component(model: PricingModel):
return next((component for component in model.charge_components if component.kind == "usage"), None)
def _tunable_default(model: PricingModel, key: str) -> Decimal | None:
parameter = next((item for item in model.tunable_parameters if item.key == key), None)
if not parameter or parameter.default_value in (None, ""):
return None
return Decimal(str(parameter.default_value))
def _default_included_units(model: PricingModel) -> Decimal:
usage = _usage_component(model)
if usage and usage.included_units is not None:
return usage.included_units
value = _tunable_default(model, "included_tokens")
return value if value is not None else Decimal("0")
def _default_usage_unit_price(model: PricingModel) -> Decimal:
usage = _usage_component(model)
if usage and usage.unit_price is not None:
return usage.unit_price
value = _tunable_default(model, "overage_unit_price")
return value if value is not None else Decimal("0")
def _months_from_commitment(value: Decimal, unit: str | None) -> int:
normalized = (unit or "month").lower()
if normalized in {"month", "months"}:
return int(value)
if normalized in {"year", "years"}:
return int(value * Decimal("12"))
return int(value)
@dataclass(frozen=True)
class CommitmentTerms:
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 = False
def default_commitment_terms(model: PricingModel) -> CommitmentTerms:
contract_duration = None
minimum_turnover = Decimal("0")
prepaid_amount = Decimal("0")
guaranteed_platform_fee = Decimal("0")
for commitment in model.commitments:
raw_value = _decimal(commitment.value)
if commitment.kind == "contract_duration":
contract_duration = _months_from_commitment(raw_value, commitment.unit)
elif commitment.kind in {"minimum_turnover", "minimum_monthly_turnover"}:
minimum_turnover = raw_value
elif commitment.kind in {"guaranteed_platform_fee", "minimum_platform_fee"}:
guaranteed_platform_fee = raw_value
elif commitment.kind == "prepayment" and (commitment.unit or "").lower() in {"eur", "usd"}:
prepaid_amount = raw_value
tunable_contract_duration = _tunable_default(model, "contract_duration_months")
if contract_duration is None and tunable_contract_duration is not None:
contract_duration = int(tunable_contract_duration)
return CommitmentTerms(
contract_duration_months=contract_duration,
minimum_monthly_turnover=minimum_turnover,
prepaid_amount=prepaid_amount,
guaranteed_platform_fee=guaranteed_platform_fee,
)
@dataclass(frozen=True)
class PricingConfiguration:
model: PricingModel
segment: str | None = None
expected_usage_units: Decimal = Decimal("0")
expected_usage_variance_pct: Decimal = Decimal("0")
allocated_fixed_cost: Decimal = Decimal("0")
direct_cost_amount: Decimal = Decimal("0")
unit_cost: Decimal = Decimal("0")
support_cost: Decimal = Decimal("0")
onboarding_cost: Decimal = Decimal("0")
risk_cost: Decimal = Decimal("0")
payment_fee_fixed: Decimal = Decimal("0")
payment_fee_rate_pct: Decimal = Decimal("0")
access_fee_amount: Decimal | None = None
included_units: Decimal | None = None
usage_unit_price: Decimal | None = None
discount_amount: Decimal = Decimal("0")
commitment_terms: CommitmentTerms = field(default_factory=CommitmentTerms)
@dataclass(frozen=True)
class BoundaryPolicy:
minimum_margin_pct: Decimal = Decimal("0")
target_margin_pct: Decimal = Decimal("15")
max_payment_fee_pct: Decimal = Decimal("10")
max_expected_usage_variance_pct: Decimal = Decimal("50")
approval_discount_pct: Decimal = Decimal("10")
max_discount_pct: Decimal = Decimal("25")
minimum_contract_duration_for_discount_months: int = 3
minimum_turnover_multiple_for_discount: Decimal = Decimal("1")
minimum_prepayment_months_for_discount: Decimal = Decimal("1")
@dataclass(frozen=True)
class PricingMetrics:
currency: str
monthly_revenue: Decimal
effective_monthly_revenue: Decimal
payment_fees: Decimal
payment_fee_pct: Decimal
allocated_fixed_cost: Decimal
direct_cost_amount: Decimal
variable_usage_cost: Decimal
support_cost: Decimal
onboarding_cost: Decimal
customer_funded_onboarding: Decimal
risk_cost: Decimal
total_monthly_cost: Decimal
monthly_margin: Decimal
margin_pct: Decimal
cost_floor_revenue: Decimal
minimum_margin_revenue: Decimal
target_margin_revenue: Decimal
expected_usage_units: Decimal
included_units: Decimal
billable_usage_units: Decimal
unit_cost: Decimal
usage_unit_price: Decimal
access_fee_amount: Decimal
contract_duration_months: int
minimum_monthly_turnover: Decimal
prepaid_amount: Decimal
guaranteed_platform_fee: Decimal
concession_value: Decimal
concession_pct: Decimal
baseline_monthly_revenue: Decimal
baseline_margin: Decimal
baseline_margin_pct: Decimal
meaningful_commitment_signals: tuple[str, ...]
reduced_cancellation_flexibility: bool
@dataclass(frozen=True)
class ConstraintResult:
id: str
title: str
severity: ConstraintSeverity
status: ConstraintStatus
summary: str
reason: str
actual_value: Decimal | str | int | None = None
threshold_value: Decimal | str | int | None = None
unit: str | None = None
details: dict[str, Any] = field(default_factory=dict)
suggested_action: str | None = None
@dataclass(frozen=True)
class BoundaryConstraint:
id: str
title: str
severity: ConstraintSeverity
evaluator: ConstraintEvaluator
@dataclass(frozen=True)
class ValidationResult:
model_id: str
model_name: str
decision: ValidationDecision
valid: bool
requires_approval: bool
summary: str
configuration: dict[str, Any]
metrics: PricingMetrics
policy: BoundaryPolicy
constraints: tuple[ConstraintResult, ...]
def _required_revenue(cost: Decimal, margin_pct: Decimal) -> Decimal:
if margin_pct >= Decimal("100"):
return Decimal("Infinity")
ratio = Decimal("1") - (margin_pct / Decimal("100"))
if ratio <= Decimal("0"):
return Decimal("Infinity")
return _money(cost / ratio)
def _meaningful_commitment_signals(
metrics: PricingMetrics,
baseline_metrics: PricingMetrics,
policy: BoundaryPolicy,
) -> tuple[str, ...]:
signals: list[str] = []
if metrics.minimum_monthly_turnover >= (
metrics.monthly_revenue * policy.minimum_turnover_multiple_for_discount
) and metrics.minimum_monthly_turnover > Decimal("0"):
signals.append("minimum_monthly_turnover")
if metrics.prepaid_amount >= (
metrics.monthly_revenue * policy.minimum_prepayment_months_for_discount
) and metrics.prepaid_amount > Decimal("0"):
signals.append("prepayment")
if metrics.guaranteed_platform_fee >= metrics.monthly_revenue and metrics.guaranteed_platform_fee > Decimal("0"):
signals.append("guaranteed_platform_fee")
if metrics.customer_funded_onboarding >= metrics.onboarding_cost and metrics.onboarding_cost > Decimal("0"):
signals.append("customer_funded_onboarding")
if (
metrics.contract_duration_months >= policy.minimum_contract_duration_for_discount_months
and metrics.contract_duration_months > baseline_metrics.contract_duration_months
):
signals.append("longer_contract_duration")
if metrics.reduced_cancellation_flexibility:
signals.append("reduced_cancellation_flexibility")
return tuple(signals)
def _build_metrics(
configuration: PricingConfiguration,
policy: BoundaryPolicy,
*,
baseline_metrics: PricingMetrics | None = None,
) -> PricingMetrics:
model = configuration.model
defaults = default_commitment_terms(model)
access_fee_amount = configuration.access_fee_amount
if access_fee_amount is None:
access_fee_amount = _access_fee_amount(model)
included_units = configuration.included_units
if included_units is None:
included_units = _default_included_units(model)
usage_unit_price = configuration.usage_unit_price
if usage_unit_price is None:
usage_unit_price = _default_usage_unit_price(model)
contract_duration_months = configuration.commitment_terms.contract_duration_months
if contract_duration_months is None:
contract_duration_months = defaults.contract_duration_months or 1
minimum_monthly_turnover = (
configuration.commitment_terms.minimum_monthly_turnover
if configuration.commitment_terms.minimum_monthly_turnover > Decimal("0")
else defaults.minimum_monthly_turnover
)
prepaid_amount = (
configuration.commitment_terms.prepaid_amount
if configuration.commitment_terms.prepaid_amount > Decimal("0")
else defaults.prepaid_amount
)
guaranteed_platform_fee = (
configuration.commitment_terms.guaranteed_platform_fee
if configuration.commitment_terms.guaranteed_platform_fee > Decimal("0")
else defaults.guaranteed_platform_fee
)
customer_funded_onboarding = configuration.commitment_terms.customer_funded_onboarding
reduced_cancellation_flexibility = configuration.commitment_terms.reduced_cancellation_flexibility
expected_usage_units = _decimal(configuration.expected_usage_units)
billable_usage_units = max(expected_usage_units - included_units, Decimal("0"))
recurring_revenue = (
access_fee_amount
+ _recurring_non_usage_component_revenue(model)
+ (usage_unit_price * billable_usage_units)
- configuration.discount_amount
)
monthly_revenue = _money(max(recurring_revenue, Decimal("0")))
effective_monthly_revenue = _money(
max(monthly_revenue, minimum_monthly_turnover, guaranteed_platform_fee)
)
payment_fees = _money(
configuration.payment_fee_fixed
+ (effective_monthly_revenue * configuration.payment_fee_rate_pct / Decimal("100"))
)
if effective_monthly_revenue > Decimal("0"):
payment_fee_pct = _percent(
(payment_fees / effective_monthly_revenue) * Decimal("100")
)
else:
payment_fee_pct = Decimal("100.0") if payment_fees > Decimal("0") else Decimal("0.0")
variable_usage_cost = _money(expected_usage_units * configuration.unit_cost)
residual_onboarding_cost = max(
configuration.onboarding_cost - customer_funded_onboarding,
Decimal("0"),
)
total_monthly_cost = _money(
configuration.allocated_fixed_cost
+ configuration.direct_cost_amount
+ variable_usage_cost
+ configuration.support_cost
+ residual_onboarding_cost
+ configuration.risk_cost
+ payment_fees
)
monthly_margin = _money(effective_monthly_revenue - total_monthly_cost)
if effective_monthly_revenue > Decimal("0"):
margin_pct = _percent((monthly_margin / effective_monthly_revenue) * Decimal("100"))
else:
margin_pct = Decimal("-100.0") if total_monthly_cost > Decimal("0") else Decimal("0.0")
cost_floor_revenue = _money(total_monthly_cost)
minimum_margin_revenue = _required_revenue(total_monthly_cost, policy.minimum_margin_pct)
target_margin_revenue = _required_revenue(total_monthly_cost, policy.target_margin_pct)
baseline_revenue = Decimal("0")
baseline_margin = Decimal("0")
baseline_margin_pct = Decimal("0.0")
concession_value = Decimal("0")
concession_pct = Decimal("0.0")
meaningful_commitment_signals: tuple[str, ...] = ()
baseline_contract_duration = contract_duration_months
if baseline_metrics is not None:
baseline_revenue = baseline_metrics.effective_monthly_revenue
baseline_margin = baseline_metrics.monthly_margin
baseline_margin_pct = baseline_metrics.margin_pct
concession_value = _money(max(baseline_metrics.monthly_margin - monthly_margin, Decimal("0")))
if baseline_metrics.effective_monthly_revenue > Decimal("0"):
concession_pct = _percent(
(concession_value / baseline_metrics.effective_monthly_revenue) * Decimal("100")
)
else:
concession_pct = Decimal("0.0")
baseline_contract_duration = baseline_metrics.contract_duration_months
probe_baseline = baseline_metrics
if probe_baseline is None:
probe_baseline = PricingMetrics(
currency=model.currency,
monthly_revenue=monthly_revenue,
effective_monthly_revenue=effective_monthly_revenue,
payment_fees=payment_fees,
payment_fee_pct=payment_fee_pct,
allocated_fixed_cost=_money(configuration.allocated_fixed_cost),
direct_cost_amount=_money(configuration.direct_cost_amount),
variable_usage_cost=variable_usage_cost,
support_cost=_money(configuration.support_cost),
onboarding_cost=_money(configuration.onboarding_cost),
customer_funded_onboarding=_money(customer_funded_onboarding),
risk_cost=_money(configuration.risk_cost),
total_monthly_cost=total_monthly_cost,
monthly_margin=monthly_margin,
margin_pct=margin_pct,
cost_floor_revenue=cost_floor_revenue,
minimum_margin_revenue=minimum_margin_revenue,
target_margin_revenue=target_margin_revenue,
expected_usage_units=expected_usage_units,
included_units=included_units,
billable_usage_units=billable_usage_units,
unit_cost=configuration.unit_cost,
usage_unit_price=usage_unit_price,
access_fee_amount=access_fee_amount,
contract_duration_months=baseline_contract_duration,
minimum_monthly_turnover=minimum_monthly_turnover,
prepaid_amount=prepaid_amount,
guaranteed_platform_fee=guaranteed_platform_fee,
concession_value=Decimal("0"),
concession_pct=Decimal("0.0"),
baseline_monthly_revenue=effective_monthly_revenue,
baseline_margin=monthly_margin,
baseline_margin_pct=margin_pct,
meaningful_commitment_signals=(),
reduced_cancellation_flexibility=reduced_cancellation_flexibility,
)
meaningful_commitment_signals = _meaningful_commitment_signals(
PricingMetrics(
currency=model.currency,
monthly_revenue=monthly_revenue,
effective_monthly_revenue=effective_monthly_revenue,
payment_fees=payment_fees,
payment_fee_pct=payment_fee_pct,
allocated_fixed_cost=_money(configuration.allocated_fixed_cost),
direct_cost_amount=_money(configuration.direct_cost_amount),
variable_usage_cost=variable_usage_cost,
support_cost=_money(configuration.support_cost),
onboarding_cost=_money(configuration.onboarding_cost),
customer_funded_onboarding=_money(customer_funded_onboarding),
risk_cost=_money(configuration.risk_cost),
total_monthly_cost=total_monthly_cost,
monthly_margin=monthly_margin,
margin_pct=margin_pct,
cost_floor_revenue=cost_floor_revenue,
minimum_margin_revenue=minimum_margin_revenue,
target_margin_revenue=target_margin_revenue,
expected_usage_units=expected_usage_units,
included_units=included_units,
billable_usage_units=billable_usage_units,
unit_cost=configuration.unit_cost,
usage_unit_price=usage_unit_price,
access_fee_amount=access_fee_amount,
contract_duration_months=contract_duration_months,
minimum_monthly_turnover=minimum_monthly_turnover,
prepaid_amount=prepaid_amount,
guaranteed_platform_fee=guaranteed_platform_fee,
concession_value=concession_value,
concession_pct=concession_pct,
baseline_monthly_revenue=baseline_revenue,
baseline_margin=baseline_margin,
baseline_margin_pct=baseline_margin_pct,
meaningful_commitment_signals=(),
reduced_cancellation_flexibility=reduced_cancellation_flexibility,
),
probe_baseline,
policy,
)
return PricingMetrics(
currency=model.currency,
monthly_revenue=monthly_revenue,
effective_monthly_revenue=effective_monthly_revenue,
payment_fees=payment_fees,
payment_fee_pct=payment_fee_pct,
allocated_fixed_cost=_money(configuration.allocated_fixed_cost),
direct_cost_amount=_money(configuration.direct_cost_amount),
variable_usage_cost=variable_usage_cost,
support_cost=_money(configuration.support_cost),
onboarding_cost=_money(configuration.onboarding_cost),
customer_funded_onboarding=_money(customer_funded_onboarding),
risk_cost=_money(configuration.risk_cost),
total_monthly_cost=total_monthly_cost,
monthly_margin=monthly_margin,
margin_pct=margin_pct,
cost_floor_revenue=cost_floor_revenue,
minimum_margin_revenue=minimum_margin_revenue,
target_margin_revenue=target_margin_revenue,
expected_usage_units=expected_usage_units,
included_units=included_units,
billable_usage_units=billable_usage_units,
unit_cost=configuration.unit_cost,
usage_unit_price=usage_unit_price,
access_fee_amount=access_fee_amount,
contract_duration_months=contract_duration_months,
minimum_monthly_turnover=minimum_monthly_turnover,
prepaid_amount=prepaid_amount,
guaranteed_platform_fee=guaranteed_platform_fee,
concession_value=concession_value,
concession_pct=concession_pct,
baseline_monthly_revenue=_money(baseline_revenue),
baseline_margin=_money(baseline_margin),
baseline_margin_pct=baseline_margin_pct,
meaningful_commitment_signals=meaningful_commitment_signals,
reduced_cancellation_flexibility=reduced_cancellation_flexibility,
)
def _baseline_configuration(configuration: PricingConfiguration) -> PricingConfiguration:
return PricingConfiguration(
model=configuration.model,
segment=configuration.segment,
expected_usage_units=configuration.expected_usage_units,
expected_usage_variance_pct=configuration.expected_usage_variance_pct,
allocated_fixed_cost=configuration.allocated_fixed_cost,
direct_cost_amount=configuration.direct_cost_amount,
unit_cost=configuration.unit_cost,
support_cost=configuration.support_cost,
onboarding_cost=configuration.onboarding_cost,
risk_cost=configuration.risk_cost,
payment_fee_fixed=configuration.payment_fee_fixed,
payment_fee_rate_pct=configuration.payment_fee_rate_pct,
commitment_terms=default_commitment_terms(configuration.model),
)
def _segment_eligibility(
configuration: PricingConfiguration,
_policy: BoundaryPolicy,
_metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if not configuration.segment or not configuration.model.eligibility:
return ConstraintResult(
id="segment-eligibility",
title="Segment eligibility",
severity="hard",
status="pass",
summary="Segment eligibility not restrictive for this evaluation.",
reason="No customer segment was supplied or the model declares no eligibility list.",
)
if configuration.segment in configuration.model.eligibility:
return ConstraintResult(
id="segment-eligibility",
title="Segment eligibility",
severity="hard",
status="pass",
summary=f"Segment '{configuration.segment}' is eligible.",
reason="The supplied customer segment is listed in the model eligibility rules.",
actual_value=configuration.segment,
)
return ConstraintResult(
id="segment-eligibility",
title="Segment eligibility",
severity="hard",
status="fail",
summary=f"Segment '{configuration.segment}' is not eligible for this model.",
reason="The model declares an explicit eligibility list and the supplied segment is outside it.",
actual_value=configuration.segment,
details={"eligible_segments": list(configuration.model.eligibility)},
suggested_action="Choose an eligible segment or use a different pricing model.",
)
def _usage_variance_limit(
configuration: PricingConfiguration,
policy: BoundaryPolicy,
_metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
actual = _percent(configuration.expected_usage_variance_pct)
if actual <= policy.max_expected_usage_variance_pct:
return ConstraintResult(
id="usage-variance-limit",
title="Usage variance limit",
severity="hard",
status="pass",
summary=f"Expected usage variance {actual}% is within the allowed range.",
reason="The scenario stays inside the configured usage-variance guardrail.",
actual_value=actual,
threshold_value=policy.max_expected_usage_variance_pct,
unit="percent",
)
return ConstraintResult(
id="usage-variance-limit",
title="Usage variance limit",
severity="hard",
status="fail",
summary=f"Expected usage variance {actual}% exceeds the allowed {policy.max_expected_usage_variance_pct}%.",
reason="High-variance usage invalidates the pricing configuration until the seller widens the guardrail or changes the package.",
actual_value=actual,
threshold_value=policy.max_expected_usage_variance_pct,
unit="percent",
suggested_action="Reduce exposure to volatile usage or tighten the included-usage assumptions.",
)
def _payment_fee_limit(
_configuration: PricingConfiguration,
policy: BoundaryPolicy,
metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if metrics.payment_fee_pct <= policy.max_payment_fee_pct:
return ConstraintResult(
id="payment-fee-limit",
title="Payment fee limit",
severity="hard",
status="pass",
summary=f"Payment fees consume {metrics.payment_fee_pct}% of revenue, within policy.",
reason="Payment-provider fees remain inside the configured coverage ceiling.",
actual_value=metrics.payment_fee_pct,
threshold_value=policy.max_payment_fee_pct,
unit="percent",
)
return ConstraintResult(
id="payment-fee-limit",
title="Payment fee limit",
severity="hard",
status="fail",
summary=f"Payment fees consume {metrics.payment_fee_pct}% of revenue, above the {policy.max_payment_fee_pct}% ceiling.",
reason="The pricing configuration leaves too little contribution margin after provider fees.",
actual_value=metrics.payment_fee_pct,
threshold_value=policy.max_payment_fee_pct,
unit="percent",
suggested_action="Increase revenue, reduce fee burden, or change the collection method.",
)
def _cost_floor_coverage(
_configuration: PricingConfiguration,
_policy: BoundaryPolicy,
metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if metrics.monthly_margin >= Decimal("0"):
return ConstraintResult(
id="cost-floor-coverage",
title="Cost floor coverage",
severity="hard",
status="pass",
summary=f"Monthly revenue clears the cost floor by {metrics.monthly_margin} {metrics.currency}.",
reason="Expected revenue covers allocated fixed cost, variable cost, and payment fees.",
actual_value=metrics.monthly_margin,
threshold_value=Decimal("0.00"),
unit=metrics.currency,
)
return ConstraintResult(
id="cost-floor-coverage",
title="Cost floor coverage",
severity="hard",
status="fail",
summary=f"Monthly revenue misses the cost floor by {abs(metrics.monthly_margin)} {metrics.currency}.",
reason=f"At least {metrics.cost_floor_revenue} {metrics.currency} monthly revenue is required to break even under these assumptions.",
actual_value=metrics.monthly_margin,
threshold_value=Decimal("0.00"),
unit=metrics.currency,
suggested_action="Raise price, lower cost, or add stronger commitments before offering this configuration.",
)
def _minimum_margin(
_configuration: PricingConfiguration,
policy: BoundaryPolicy,
metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if metrics.margin_pct >= policy.minimum_margin_pct:
return ConstraintResult(
id="minimum-margin",
title="Minimum margin",
severity="hard",
status="pass",
summary=f"Margin {metrics.margin_pct}% satisfies the hard minimum.",
reason="The configuration meets the seller's minimum margin boundary.",
actual_value=metrics.margin_pct,
threshold_value=policy.minimum_margin_pct,
unit="percent",
)
return ConstraintResult(
id="minimum-margin",
title="Minimum margin",
severity="hard",
status="fail",
summary=f"Margin {metrics.margin_pct}% is below the hard minimum of {policy.minimum_margin_pct}%.",
reason=f"At least {metrics.minimum_margin_revenue} {metrics.currency} monthly revenue is required to satisfy the minimum margin boundary.",
actual_value=metrics.margin_pct,
threshold_value=policy.minimum_margin_pct,
unit="percent",
suggested_action="Increase price, reduce costs, or add commitment-backed protection.",
)
def _target_margin_approval(
_configuration: PricingConfiguration,
policy: BoundaryPolicy,
metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if metrics.margin_pct >= policy.target_margin_pct:
return ConstraintResult(
id="target-margin-approval",
title="Target margin approval threshold",
severity="soft",
status="pass",
summary=f"Margin {metrics.margin_pct}% satisfies the target margin threshold.",
reason="No sales or pricing approval is required on margin grounds.",
actual_value=metrics.margin_pct,
threshold_value=policy.target_margin_pct,
unit="percent",
)
return ConstraintResult(
id="target-margin-approval",
title="Target margin approval threshold",
severity="soft",
status="review",
summary=f"Margin {metrics.margin_pct}% is below the target threshold of {policy.target_margin_pct}%.",
reason=f"The configuration is economically viable but falls short of the seller's preferred target margin of {policy.target_margin_pct}%.",
actual_value=metrics.margin_pct,
threshold_value=policy.target_margin_pct,
unit="percent",
suggested_action="Route through approval or improve economics before release.",
)
def _discount_exposure_limit(
_configuration: PricingConfiguration,
policy: BoundaryPolicy,
metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if metrics.concession_pct <= policy.max_discount_pct:
return ConstraintResult(
id="discount-exposure-limit",
title="Discount exposure limit",
severity="hard",
status="pass",
summary=f"Economic concession {metrics.concession_pct}% stays inside the hard discount ceiling.",
reason="The configuration does not exceed the seller's maximum discount exposure.",
actual_value=metrics.concession_pct,
threshold_value=policy.max_discount_pct,
unit="percent",
)
return ConstraintResult(
id="discount-exposure-limit",
title="Discount exposure limit",
severity="hard",
status="fail",
summary=f"Economic concession {metrics.concession_pct}% exceeds the hard discount ceiling of {policy.max_discount_pct}%.",
reason="The proposed economics reduce seller value too far relative to the model baseline.",
actual_value=metrics.concession_pct,
threshold_value=policy.max_discount_pct,
unit="percent",
suggested_action="Reduce the concession or offset it with a materially stronger commitment structure.",
)
def _discount_approval_threshold(
_configuration: PricingConfiguration,
policy: BoundaryPolicy,
metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if metrics.concession_pct <= policy.approval_discount_pct:
return ConstraintResult(
id="discount-approval-threshold",
title="Discount approval threshold",
severity="soft",
status="pass",
summary=f"Economic concession {metrics.concession_pct}% is inside the self-serve threshold.",
reason="No extra approval is needed for discount exposure.",
actual_value=metrics.concession_pct,
threshold_value=policy.approval_discount_pct,
unit="percent",
)
return ConstraintResult(
id="discount-approval-threshold",
title="Discount approval threshold",
severity="soft",
status="review",
summary=f"Economic concession {metrics.concession_pct}% exceeds the self-serve threshold of {policy.approval_discount_pct}%.",
reason="The configuration may still be acceptable, but it now requires seller approval instead of self-serve acceptance.",
actual_value=metrics.concession_pct,
threshold_value=policy.approval_discount_pct,
unit="percent",
suggested_action="Escalate for approval or reduce the concession magnitude.",
)
def _commitment_backed_concession(
_configuration: PricingConfiguration,
_policy: BoundaryPolicy,
metrics: PricingMetrics,
_baseline: PricingMetrics,
) -> ConstraintResult:
if metrics.concession_value <= Decimal("0"):
return ConstraintResult(
id="commitment-backed-concession",
title="Commitment-backed concession",
severity="hard",
status="pass",
summary="No economic concession was introduced relative to the model baseline.",
reason="The configuration does not weaken seller economics versus the baseline assumptions.",
actual_value=metrics.concession_value,
threshold_value=Decimal("0.00"),
unit=metrics.currency,
)
if metrics.meaningful_commitment_signals:
return ConstraintResult(
id="commitment-backed-concession",
title="Commitment-backed concession",
severity="hard",
status="pass",
summary="The economic concession is backed by explicit commitments.",
reason="The configuration introduces weaker unit economics, but it also adds meaningful seller protections.",
actual_value=metrics.concession_value,
threshold_value=Decimal("0.00"),
unit=metrics.currency,
details={"signals": list(metrics.meaningful_commitment_signals)},
)
return ConstraintResult(
id="commitment-backed-concession",
title="Commitment-backed concession",
severity="hard",
status="fail",
summary="The configuration introduces weaker seller economics without an offsetting commitment.",
reason="Discounts and improved customer unit economics must be tied to enforceable or economically meaningful commitments.",
actual_value=metrics.concession_value,
threshold_value=Decimal("0.00"),
unit=metrics.currency,
suggested_action="Add minimum turnover, prepayment, guaranteed fees, or a materially longer contract before offering this concession.",
)
def default_constraints() -> tuple[BoundaryConstraint, ...]:
return (
BoundaryConstraint("segment-eligibility", "Segment eligibility", "hard", _segment_eligibility),
BoundaryConstraint("usage-variance-limit", "Usage variance limit", "hard", _usage_variance_limit),
BoundaryConstraint("payment-fee-limit", "Payment fee limit", "hard", _payment_fee_limit),
BoundaryConstraint("cost-floor-coverage", "Cost floor coverage", "hard", _cost_floor_coverage),
BoundaryConstraint("minimum-margin", "Minimum margin", "hard", _minimum_margin),
BoundaryConstraint("target-margin-approval", "Target margin approval threshold", "soft", _target_margin_approval),
BoundaryConstraint("discount-exposure-limit", "Discount exposure limit", "hard", _discount_exposure_limit),
BoundaryConstraint("discount-approval-threshold", "Discount approval threshold", "soft", _discount_approval_threshold),
BoundaryConstraint("commitment-backed-concession", "Commitment-backed concession", "hard", _commitment_backed_concession),
)
def _configuration_view(configuration: PricingConfiguration, metrics: PricingMetrics) -> dict[str, Any]:
return {
"segment": configuration.segment,
"currency": metrics.currency,
"access_fee_amount": metrics.access_fee_amount,
"included_units": metrics.included_units,
"usage_unit_price": metrics.usage_unit_price,
"expected_usage_units": metrics.expected_usage_units,
"expected_usage_variance_pct": _percent(configuration.expected_usage_variance_pct),
"allocated_fixed_cost": metrics.allocated_fixed_cost,
"direct_cost_amount": metrics.direct_cost_amount,
"unit_cost": metrics.unit_cost,
"support_cost": metrics.support_cost,
"onboarding_cost": metrics.onboarding_cost,
"risk_cost": metrics.risk_cost,
"payment_fee_fixed": _money(configuration.payment_fee_fixed),
"payment_fee_rate_pct": _percent(configuration.payment_fee_rate_pct),
"discount_amount": _money(configuration.discount_amount),
"commitment_terms": {
"contract_duration_months": metrics.contract_duration_months,
"minimum_monthly_turnover": metrics.minimum_monthly_turnover,
"prepaid_amount": metrics.prepaid_amount,
"guaranteed_platform_fee": metrics.guaranteed_platform_fee,
"customer_funded_onboarding": metrics.customer_funded_onboarding,
"reduced_cancellation_flexibility": metrics.reduced_cancellation_flexibility,
},
}
def validate_pricing_configuration(
configuration: PricingConfiguration,
policy: BoundaryPolicy,
constraints: tuple[BoundaryConstraint, ...] | None = None,
) -> ValidationResult:
baseline_metrics = _build_metrics(_baseline_configuration(configuration), policy)
metrics = _build_metrics(configuration, policy, baseline_metrics=baseline_metrics)
results = tuple(
constraint.evaluator(configuration, policy, metrics, baseline_metrics)
for constraint in (constraints or default_constraints())
)
hard_failures = [result for result in results if result.status == "fail" and result.severity == "hard"]
soft_reviews = [result for result in results if result.status == "review"]
valid = not hard_failures
requires_approval = valid and bool(soft_reviews)
if hard_failures:
decision: ValidationDecision = "rejected"
summary = "Rejected: " + ", ".join(result.title for result in hard_failures) + "."
elif soft_reviews:
decision = "requires_approval"
summary = "Approval required: " + ", ".join(result.title for result in soft_reviews) + "."
else:
decision = "accepted"
summary = "Accepted: all boundary constraints passed."
return ValidationResult(
model_id=configuration.model.id,
model_name=configuration.model.name,
decision=decision,
valid=valid,
requires_approval=requires_approval,
summary=summary,
configuration=_configuration_view(configuration, metrics),
metrics=metrics,
policy=policy,
constraints=results,
)