#!/usr/bin/env python3 """Synthetic inventory pricing review queue. Standard library only; no network. Snapshots contain already-reviewed same-cohort, USD, item-only observations. This example does not match raw eBay rows or perform marketplace price writes. Production callers load proposals by ID from immutable trusted server storage; the browser submits only the proposal ID and approval intent. Current inventory and evidence must also be loaded server-side. This file does not implement auth. """ from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation, ROUND_CEILING from statistics import median D = Decimal CENT = D("0.01") POLICY = "inventory-policy-v1" NOW = datetime(2026, 10, 27, 12, tzinfo=timezone.utc) def amount(value, *, rate=False): if isinstance(value, bool) or value is None: raise ValueError("Amount is missing or invalid") try: result = D(str(value)) except (InvalidOperation, ValueError): raise ValueError("Amount is invalid") from None if not result.is_finite() or result < 0 or (rate and result >= 1): raise ValueError("Amount is outside the policy domain") return result def timestamp(value): try: result = datetime.fromisoformat(value.replace("Z", "+00:00")) except (AttributeError, TypeError, ValueError): raise ValueError("Timestamp is invalid") from None if result.tzinfo is None or result.utcoffset() is None: raise ValueError("Timestamp needs a timezone") return result def review(sku, snapshots, now=NOW): """Return an explanation and proposal; mutate neither inventory nor prices.""" out = {"sku": sku["sku"], "state": "hold", "reason": None, "median": None, "floor": None, "proposed": None} try: current = amount(sku["currentAskingPrice"]) acquired = timestamp(sku["acquiredAt"]) if acquired > now or sku["currency"] != "USD": raise ValueError("Invalid acquisition date or unsupported currency") cost = amount(sku["landedCost"]) fixed = amount(sku["fixedSellingCosts"]) profit = amount(sku["targetProfit"]) fee = amount(sku["sellingFeeRate"], rate=True) floor = ((cost + fixed + profit) / (1 - fee)).quantize( CENT, rounding=ROUND_CEILING) out.update(current=current, floor=floor, ageDays=(now-acquired).days) except (KeyError, ValueError): return dict(out, reason="invalid_inventory") evidence = snapshots.get(sku["cohortId"]) if evidence is None: return dict(out, reason="no_snapshot") try: observed = timestamp(evidence["observedAt"]) if observed > now or evidence["currency"] != sku["currency"]: raise ValueError("Invalid evidence date or currency") if evidence["cohortId"] != sku["cohortId"] or not evidence["id"]: raise ValueError("Evidence identity mismatch") prices = [amount(x) for x in evidence["reviewedItemPrices"]] if any(x <= 0 for x in prices): raise ValueError("Reviewed prices must be positive") except (KeyError, TypeError, ValueError): return dict(out, reason="invalid_evidence") if now - observed > timedelta(days=7): return dict(out, reason="stale_evidence") if len(prices) < 4: return dict(out, reason="insufficient_evidence") mid = median(prices).quantize(CENT) out["median"] = mid if mid < floor: return dict(out, reason="below_floor") if abs(mid-current) >= D("10"): return dict(out, state="review", reason="price_difference", proposed=mid, expectedVersion=sku["version"], evidenceId=evidence["id"], policyVersion=POLICY, expiresAt=min( now+timedelta(hours=24), observed+timedelta(days=7)).isoformat()) return dict(out, state="inspect" if out["ageDays"] >= 90 else "keep", reason="aging_stock" if out["ageDays"] >= 90 else "within_threshold") def approve(proposal, current_sku, current_evidence, policy=POLICY, now=NOW): """Re-evaluate an approval; return only recomputed amounts. Load proposal by ID from immutable trusted server storage. Do not accept a proposal object or amount from the browser. Load current SKU and the complete current evidence snapshot server-side as well. This function is not an auth layer; a real DB must check/update atomically under authorized access. """ try: if proposal["state"] != "review": raise ValueError("No actionable proposal") if (policy != POLICY or proposal["policyVersion"] != policy or proposal["sku"] != current_sku["sku"] or proposal["expectedVersion"] != current_sku["version"] or proposal["evidenceId"] != current_evidence["id"]): raise ValueError("Proposal inputs changed; regenerate") if now >= timestamp(proposal["expiresAt"]): raise ValueError("Proposal expired") fresh = review(current_sku, {current_sku["cohortId"]: current_evidence}, now) if fresh["state"] != "review": raise ValueError("Current evidence or merchant policy blocks approval") if (amount(proposal["proposed"]) != fresh["proposed"] or amount(proposal["current"]) != fresh["current"]): raise ValueError("Proposal amount changed; regenerate") return {"sku": fresh["sku"], "approvedPrice": fresh["proposed"], "expectedVersion": fresh["expectedVersion"], "evidenceId": fresh["evidenceId"]} except (KeyError, TypeError): raise ValueError("Approval records are incomplete") from None def sample_sku(key, cohort, current, cost="90", fixed="15", profit="25", age=100): return {"sku": key, "version": 4, "cohortId": cohort, "currency": "USD", "currentAskingPrice": current, "landedCost": cost, "fixedSellingCosts": fixed, "targetProfit": profit, "sellingFeeRate": "0.10", "acquiredAt": (NOW-timedelta(days=age)).isoformat()} def sample_snapshot(cohort, prices, age=1): return {"id": "snapshot-"+cohort, "cohortId": cohort, "currency": "USD", "observedAt": (NOW-timedelta(days=age)).isoformat(), "reviewedItemPrices": prices} def fixtures(): skus = [sample_sku("AUDIO-001", "audio", "185"), sample_sku("CAM-002", "camera", "175"), sample_sku("LENS-003", "lens", "120", "60", "10", "20"), sample_sku("GAME-004", "game", "60", "20", "5", "15"), sample_sku("CTRL-005", "controller", "80", "30", "5", "15")] snapshots = {"audio": sample_snapshot("audio", ["145", "155", "165", "175"]), "camera": sample_snapshot("camera", ["115", "125", "135", "145"]), "lens": sample_snapshot("lens", ["110", "120"]), "controller": sample_snapshot("controller", ["70", "75", "85", "90"], 20)} return skus, snapshots def must_reject(call): try: call() except ValueError: return raise AssertionError("Expected rejection") def self_check(): skus, snapshots = fixtures() first = review(skus[0], snapshots) assert first["median"] == D("160.00") and first["floor"] == D("144.45") assert first["proposed"] == D("160.00") and first["ageDays"] == 100 assert [review(s, snapshots)["reason"] for s in skus[1:]] == [ "below_floor", "insufficient_evidence", "no_snapshot", "stale_evidence"] for bad in (None, "-1", "NaN", "Infinity"): assert review(dict(skus[0], landedCost=bad), snapshots)["reason"] == "invalid_inventory" for bad in ("1", "-0.1", "NaN"): assert review(dict(skus[0], sellingFeeRate=bad), snapshots)["reason"] == "invalid_inventory" for bad in ("bad date", "2026-10-27", "2026-10-28T12:00:00+00:00"): assert review(dict(skus[0], acquiredAt=bad), snapshots)["reason"] == "invalid_inventory" invalid = dict(snapshots, audio=dict(snapshots["audio"], observedAt="bad date")) assert review(skus[0], invalid)["reason"] == "invalid_evidence" current = snapshots["audio"] assert approve(first, skus[0], current)["approvedPrice"] == D("160") must_reject(lambda: approve(first, dict(skus[0], version=5), current)) must_reject(lambda: approve(first, skus[0], dict(current, id="new-snapshot"))) must_reject(lambda: approve(first, skus[0], current, policy="v2")) must_reject(lambda: approve(first, skus[0], current, now=NOW+timedelta(hours=24))) for changed_price in (D("1"), D("159"), D("NaN"), D("-1")): must_reject(lambda: approve(dict(first, proposed=changed_price), skus[0], current)) must_reject(lambda: approve(first, dict(skus[0], landedCost="200"), current)) must_reject(lambda: approve(first, dict(skus[0], currentAskingPrice="200"), current)) changed = dict(current, reviewedItemPrices=["120", "130", "140", "150"]) must_reject(lambda: approve(first, skus[0], changed)) must_reject(lambda: approve(first, skus[0], dict(current, reviewedItemPrices=["150"]))) stale = dict(current, observedAt=(NOW-timedelta(days=8)).isoformat()) must_reject(lambda: approve(first, skus[0], stale)) near_expiry = dict(snapshots, audio=dict(snapshots["audio"], observedAt=(NOW-timedelta(days=6, hours=23)).isoformat())) expiring = review(skus[0], near_expiry) assert timestamp(expiring["expiresAt"]) == NOW+timedelta(hours=1) must_reject(lambda: approve(expiring, skus[0], near_expiry["audio"], now=NOW+timedelta(hours=1))) must_reject(lambda: approve(review(skus[1], snapshots), skus[1], snapshots["camera"])) if __name__ == "__main__": self_check() records, evidence = fixtures() print("Offline checks passed. Synthetic USD amounts; no marketplace writes.") for record in records: result = review(record, evidence) print(result["sku"], "current="+str(result.get("current")), "median="+str(result["median"]), "floor="+str(result["floor"]), "state="+result["state"], "reason="+str(result["reason"]), "proposed="+str(result["proposed"]))