#!/usr/bin/env python3 """Offline synthetic trade-in quote workflow. Standard library; no network calls.""" from dataclasses import dataclass, replace from datetime import datetime, timedelta, timezone from decimal import Decimal, ROUND_FLOOR, ROUND_HALF_UP from statistics import median D = Decimal CENT = D("0.01") @dataclass(frozen=True) class Costs: selling_fee_rate: Decimal | None outbound_shipping: Decimal | None refurbishment: Decimal | None handling: Decimal | None returns_reserve: Decimal | None @dataclass(frozen=True) class Plan: version: str policy_version: str currency: str planned_item_revenue: Decimal | None target_contribution_margin: Decimal | None costs: Costs @dataclass(frozen=True) class Evidence: cohort_id: str currency: str accepted_count: int retrieved_at: datetime median_displayed_item_price: Decimal | None reviewed: bool @dataclass(frozen=True) class Policy: minimum_comps: int = 5 maximum_evidence_age_hours: int = 48 offer_valid_hours: int = 24 @dataclass(frozen=True) class Decision: status: str reason: str ceiling: Decimal | None = None modeled_contribution_at_ceiling: Decimal | None = None @dataclass(frozen=True) class ConditionalOffer: amount: Decimal currency: str plan_version: str policy_version: str cohort_id: str approved_by: str approved_at: datetime expires_at: datetime status: str = "conditional_pending_inspection" def decimal_valid(value, *, positive=False, money=False): if not isinstance(value, D) or not value.is_finite(): return False if value <= 0 if positive else value < 0: return False # A cent is this example's chosen currency precision. return not money or value == value.quantize(CENT) def aware(value): return isinstance(value, datetime) and value.utcoffset() is not None def fixed_costs(plan): c = plan.costs return c.outbound_shipping + c.refurbishment + c.handling + c.returns_reserve def selling_fee(plan): return (plan.planned_item_revenue * plan.costs.selling_fee_rate).quantize( CENT, rounding=ROUND_HALF_UP ) def contribution(plan, acquisition_amount): return plan.planned_item_revenue - selling_fee(plan) - fixed_costs(plan) - acquisition_amount def evaluate(plan, evidence, policy, now): """Return a ceiling for staff review, or a hold with no quote amount.""" hold = lambda reason: Decision("hold", reason) if not aware(now) or not aware(evidence.retrieved_at): return hold("Timezone-aware evaluation and retrieval timestamps required") if (policy.minimum_comps < 1 or policy.maximum_evidence_age_hours <= 0 or policy.offer_valid_hours <= 0): return hold("Invalid local policy") if not plan.version or not plan.policy_version or not evidence.cohort_id: return hold("Versioned plan, policy and cohort required") if not plan.currency or evidence.currency != plan.currency: return hold("Evidence and plan currencies must match") if not evidence.reviewed: return hold("Comparable evidence needs review") if evidence.accepted_count < policy.minimum_comps: return hold("Sparse reviewed evidence") age = now - evidence.retrieved_at if age < timedelta(0) or age > timedelta(hours=policy.maximum_evidence_age_hours): return hold("Evidence is stale or future-dated") if not decimal_valid(evidence.median_displayed_item_price, positive=True, money=True): return hold("Usable displayed-price summary required") if not decimal_valid(plan.planned_item_revenue, positive=True, money=True): return hold("Merchant-selected planned revenue required") c = plan.costs fixed = (c.outbound_shipping, c.refurbishment, c.handling, c.returns_reserve) if any(not decimal_valid(value, money=True) for value in fixed): return hold("Every fixed cost and reserve must be explicit and nonnegative") if (not decimal_valid(c.selling_fee_rate) or c.selling_fee_rate >= 1 or not decimal_valid(plan.target_contribution_margin) or plan.target_contribution_margin >= 1): return hold("Fee and target margin rates must be between zero and one") target = plan.planned_item_revenue * plan.target_contribution_margin ceiling = (plan.planned_item_revenue - selling_fee(plan) - fixed_costs(plan) - target).quantize( CENT, rounding=ROUND_FLOOR ) if ceiling <= 0: return hold("Scenario cannot fund a positive acquisition offer") return Decision("review_required", "Staff approval required", ceiling, contribution(plan, ceiling)) def approve_quote(plan, evidence, policy, now, amount, reviewer): """Record an approved conditional offer; never bypass current eligibility.""" decision = evaluate(plan, evidence, policy, now) if decision.status != "review_required": raise ValueError(decision.reason) if not reviewer or not reviewer.strip(): raise ValueError("A named staff reviewer is required") if not decimal_valid(amount, positive=True, money=True) or amount > decision.ceiling: raise ValueError("Approved amount must be positive and within the current ceiling") return ConditionalOffer(amount, plan.currency, plan.version, plan.policy_version, evidence.cohort_id, reviewer.strip(), now, now + timedelta(hours=policy.offer_valid_hours)) def inspect_offer(offer, now, current_plan_version, current_policy_version, *, identity_matches, working, locks_clear, condition_matches): """Return an inspection outcome; failed assumptions require a fresh quote.""" if not aware(now) or now < offer.approved_at: return "invalid_inspection_time" if now >= offer.expires_at: return "expired" if (current_plan_version != offer.plan_version or current_policy_version != offer.policy_version): return "requote_required" if not all(value is True for value in (identity_matches, working, locks_clear, condition_matches)): return "requote_required" return "inspection_passed" def fixtures(): # These are invented reviewed listing observations, not current market data. amounts = [D(v) for v in ("280", "300", "310", "320", "320", "330", "340", "360")] now = datetime(2026, 10, 27, 12, tzinfo=timezone.utc) evidence = Evidence("example-phone-128gb-unlocked-working", "GBP", len(amounts), now - timedelta(hours=12), median(amounts), True) plan = Plan("resale-scenario-v1", "trade-in-policy-v1", "GBP", D("300.00"), D("0.20"), Costs(D("0.10"), D("12.00"), D("18.00"), D("8.00"), D("12.00"))) return plan, evidence, Policy(), now def run_checks(): plan, evidence, policy, now = fixtures() checks = 0 def check(condition): nonlocal checks if not condition: raise AssertionError("Offline self-check failed") checks += 1 def held(p=plan, e=evidence): result = evaluate(p, e, policy, now) return result.status == "hold" and result.ceiling is None def rejected(call): try: call() except ValueError: return True return False base = evaluate(plan, evidence, policy, now) check(evidence.median_displayed_item_price == D("320")) check(base.status == "review_required" and base.ceiling == D("160.00")) check(base.modeled_contribution_at_ceiling == D("60.00")) check(evaluate(replace(plan, planned_item_revenue=D("280.00")), evidence, policy, now).ceiling == D("146.00")) check(held(replace(plan, planned_item_revenue=None))) check(held(replace(plan, costs=replace(plan.costs, outbound_shipping=None)))) check(held(replace(plan, costs=replace(plan.costs, returns_reserve=None)))) check(held(replace(plan, costs=replace(plan.costs, refurbishment=D("NaN"))))) check(held(e=replace(evidence, accepted_count=4))) check(held(e=replace(evidence, retrieved_at=now - timedelta(hours=49)))) check(held(e=replace(evidence, retrieved_at=now + timedelta(hours=1)))) check(held(e=replace(evidence, reviewed=False))) check(held(e=replace(evidence, currency="USD"))) check(held(e=replace(evidence, median_displayed_item_price=None))) check(held(replace(plan, costs=replace(plan.costs, refurbishment=D("500.00"))))) check(held(replace(plan, costs=replace(plan.costs, selling_fee_rate=D("1.00"))))) check(rejected(lambda: approve_quote(plan, evidence, policy, now, D("160.01"), "Staff A"))) check(rejected(lambda: approve_quote(plan, evidence, policy, now, D("150.00"), ""))) offer = approve_quote(plan, evidence, policy, now, D("150.00"), "Staff A") check(offer.status == "conditional_pending_inspection" and contribution(plan, offer.amount) == D("70.00")) inspection = dict(identity_matches=True, working=True, locks_clear=True, condition_matches=True) check(inspect_offer(offer, now + timedelta(hours=1), plan.version, plan.policy_version, **inspection) == "inspection_passed") check(inspect_offer(offer, offer.expires_at, plan.version, plan.policy_version, **inspection) == "expired") check(inspect_offer(offer, now, "resale-scenario-v2", plan.policy_version, **inspection) == "requote_required") check(inspect_offer(offer, now, plan.version, plan.policy_version, **{**inspection, "locks_clear": False}) == "requote_required") return checks def main(): checks = run_checks() plan, evidence, policy, now = fixtures() base = evaluate(plan, evidence, policy, now) downside = evaluate(replace(plan, planned_item_revenue=D("280.00")), evidence, policy, now) offer = approve_quote(plan, evidence, policy, now, D("150.00"), "Staff A") amount = contribution(plan, offer.amount) margin = amount / plan.planned_item_revenue * D("100") print(f"Base: {base.status} | ceiling GBP {base.ceiling:.2f} | modeled contribution GBP {base.modeled_contribution_at_ceiling:.2f}") print(f"Downside: {downside.status} | ceiling GBP {downside.ceiling:.2f} | modeled contribution GBP {downside.modeled_contribution_at_ceiling:.2f}") print(f"Staff: conditional offer GBP {offer.amount:.2f} | modeled contribution GBP {amount:.2f} ({margin:.2f}%)") print(f"Self-checks passed: {checks}") if __name__ == "__main__": main()