#!/usr/bin/env python3 """Offline seller-price suggestion demo. Synthetic fixtures; no network calls. Run: PYTHONDONTWRITEBYTECODE=1 python3 resale-marketplace-price-suggestions.py Application identity/review fields are not SoldFetch API fields. This policy summarizes reviewed displayed prices; it does not verify transaction amounts. The approval helper needs server authentication, persistence and an atomic compare-and-update of the checked draft revision in a real application. """ from collections import Counter from copy import deepcopy from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation import hashlib import json from statistics import median, quantiles POLICY_VERSION = "seller-suggestion-demo-v1" MINIMUM_COMPS = 5 MAX_EVIDENCE_AGE = timedelta(hours=24) SNAPSHOT_LIFETIME = timedelta(minutes=30) DRAFT_FIELDS = ( "listingId", "revision", "identity", "conditionPolicy", "ebaySite", "currency", "priceBasis", ) def timestamp(value): """Require a timezone-aware ISO timestamp; naive times are ambiguous.""" parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError("Timestamp needs a timezone") return parsed.astimezone(timezone.utc) def iso(value): return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") def digest(value): payload = json.dumps(value, sort_keys=True, separators=(",", ":")) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def draft_fingerprint(draft): identity = draft.get("identity") if not isinstance(identity, dict) or not identity or not all(identity.values()): raise ValueError("Complete application-owned identity is required") if draft.get("priceBasis") != "displayed_item_amount": raise ValueError("Demo only supports displayed item amounts") return digest({field: draft[field] for field in DRAFT_FIELDS}) def positive_amount(value): try: amount = Decimal(str(value)) except (InvalidOperation, ValueError): raise ValueError("Amount must be a positive decimal") from None if not amount.is_finite() or amount <= 0: raise ValueError("Amount must be a positive decimal") return amount def build_snapshot(draft, reviewed_rows, now): """Build an immutable result using fresh, explicitly reviewed observations. Each input nests the untouched API item under listing. identity, conditionPolicy, reviewStatus, ebaySite and responseFetchedAt are owned by the app; responseFetchedAt preserves the API response's fetchedAt value. """ fingerprint = draft_fingerprint(draft) now = timestamp(iso(now)) reasons, seen, accepted = Counter(), set(), [] for row in reviewed_rows: listing = row.get("listing", {}) key = (row.get("ebaySite"), listing.get("itemId")) if (row.get("reviewStatus") != "accepted" or row.get("identity") != draft["identity"] or row.get("conditionPolicy") != draft["conditionPolicy"]): reasons["unapproved_or_mismatched_identity"] += 1 continue if (key[0] != draft["ebaySite"] or not key[1] or listing.get("listingType") != "sold" or listing.get("soldCurrency") != draft["currency"] or listing.get("quality", {}).get("priceBasis") != "displayed_listing_price"): reasons["incompatible_listing"] += 1 continue try: amount = positive_amount(listing.get("soldPrice")) observation_times = (timestamp(row["responseFetchedAt"]), timestamp(listing["scrapedAt"])) observed_at = min(observation_times) except (ValueError, KeyError, TypeError, AttributeError): reasons["invalid_amount_or_timestamp"] += 1 continue if any(at > now for at in observation_times) or now - observed_at >= MAX_EVIDENCE_AGE: reasons["stale_or_future_evidence"] += 1 continue if key in seen: reasons["duplicate_item"] += 1 continue seen.add(key) accepted.append((key[1], amount, observed_at)) expires_at = now + SNAPSHOT_LIFETIME if accepted: expires_at = min(expires_at, min(at for _, _, at in accepted) + MAX_EVIDENCE_AGE) snapshot = { "policyVersion": POLICY_VERSION, "draftFingerprint": fingerprint, "generatedAt": iso(now), "expiresAt": iso(expires_at), "currency": draft["currency"], "priceBasis": draft["priceBasis"], "sampleCount": len(accepted), "exclusions": dict(sorted(reasons.items())), "status": "insufficient_evidence", "range": None, "median": None, "oldestObservationAt": min((iso(x[2]) for x in accepted), default=None), "evidenceDigest": digest(sorted((item, str(value), iso(at)) for item, value, at in accepted)), } if not accepted and reasons["stale_or_future_evidence"]: snapshot["status"] = "stale_evidence" if len(accepted) >= MINIMUM_COMPS: values = sorted(value for _, value, _ in accepted) q1, _, q3 = quantiles(values, n=4, method="inclusive") snapshot.update(status="ready", range=[str(q1), str(q3)], median=str(median(values))) snapshot["snapshotId"] = digest(snapshot) return snapshot def approve_price(current_draft, saved_snapshot, selected_price, now): """Validate an explicit seller choice against a server-loaded snapshot.""" snapshot_body = {key: value for key, value in saved_snapshot.items() if key != "snapshotId"} if saved_snapshot.get("snapshotId") != digest(snapshot_body): raise ValueError("Snapshot changed; load the saved version") if saved_snapshot["draftFingerprint"] != draft_fingerprint(current_draft): raise ValueError("Draft changed; generate a new suggestion") if saved_snapshot["status"] != "ready": raise ValueError("No usable suggestion to approve") if now < timestamp(saved_snapshot["generatedAt"]) or now >= timestamp(saved_snapshot["expiresAt"]): raise ValueError("Suggestion expired or approval time is invalid") selected = positive_amount(selected_price) if selected != selected.quantize(Decimal("0.01")): raise ValueError("USD example requires whole cents") return { "listingId": current_draft["listingId"], "expectedDraftRevision": current_draft["revision"], "snapshotId": saved_snapshot["snapshotId"], "sellerAskingPrice": format(selected, ".2f"), "currency": current_draft["currency"], "choice": ("use_suggested_median" if selected == Decimal(saved_snapshot["median"]) else "seller_override"), "approvedAt": iso(now), } def fixtures(): """All names, IDs and prices below are fictional; do not fetch their URLs.""" draft = { "listingId": "draft-camera-7", "revision": 3, "identity": {"maker": "Northstar", "model": "C2", "variant": "standard", "bundle": "body only"}, "conditionPolicy": "working-used", "ebaySite": "ebay.com", "currency": "USD", "priceBasis": "displayed_item_amount", } rows = [] for index, price in enumerate(("80.00", "90.00", "100.00", "110.00", "120.00")): rows.append({ "identity": deepcopy(draft["identity"]), "conditionPolicy": "working-used", "reviewStatus": "accepted", "ebaySite": "ebay.com", "responseFetchedAt": "2026-10-27T15:00:00Z", "listing": { "itemId": str(100000000000 + index), "listingType": "sold", "title": "Synthetic Northstar C2 body only", "soldPrice": price, "soldCurrency": "USD", "scrapedAt": "2026-10-27T15:00:00Z", "quality": {"priceBasis": "displayed_listing_price", "saleDateVerified": False}, }, }) return draft, rows def must_reject(action): try: action() except ValueError: return raise AssertionError("Expected rejection") def self_checks(): now = timestamp("2026-10-27T16:00:00Z") draft, rows = fixtures() original_draft, original_rows = deepcopy(draft), deepcopy(rows) snapshot = build_snapshot(draft, rows, now) assert snapshot["status"] == "ready" and snapshot["sampleCount"] == 5 assert list(map(Decimal, snapshot["range"])) == [Decimal("90"), Decimal("110")] assert Decimal(snapshot["median"]) == Decimal("100") assert snapshot["oldestObservationAt"] == "2026-10-27T15:00:00Z" approved = approve_price(draft, snapshot, "125.00", now) assert approved["choice"] == "seller_override" assert approved["sellerAskingPrice"] == "125.00" assert approve_price(draft, snapshot, "100.00", now)["choice"] == "use_suggested_median" assert build_snapshot(draft, rows + [deepcopy(rows[0])], now)["sampleCount"] == 5 assert build_snapshot(draft, rows[:4], now)["status"] == "insufficient_evidence" assert build_snapshot(draft, [], now)["range"] is None mismatched = deepcopy(rows) mismatched[0]["identity"]["bundle"] = "body and lens" assert build_snapshot(draft, mismatched, now)["sampleCount"] == 4 stale = deepcopy(rows) for row in stale: row["listing"]["scrapedAt"] = "2026-10-25T15:00:00Z" assert build_snapshot(draft, stale, now)["status"] == "stale_evidence" nearly_stale = deepcopy(rows) for row in nearly_stale: row["listing"]["scrapedAt"] = iso(now - MAX_EVIDENCE_AGE + timedelta(minutes=10)) expiring = build_snapshot(draft, nearly_stale, now) assert timestamp(expiring["expiresAt"]) == now + timedelta(minutes=10) must_reject(lambda: approve_price(draft, expiring, "100.00", now + timedelta(minutes=11))) future = deepcopy(rows) future[0]["responseFetchedAt"] = iso(now + timedelta(minutes=1)) assert build_snapshot(draft, future, now)["sampleCount"] == 4 for field, value in (("soldCurrency", "EUR"), ("soldPrice", None), ("soldPrice", "NaN"), ("listingType", "active")): incompatible = deepcopy(rows) incompatible[0]["listing"][field] = value assert build_snapshot(draft, incompatible, now)["sampleCount"] == 4 lower_bound = deepcopy(rows) lower_bound[0]["listing"]["quality"]["priceBasis"] = "range_lower_bound" assert build_snapshot(draft, lower_bound, now)["sampleCount"] == 4 changed = deepcopy(draft) changed["identity"]["bundle"] = "body and lens" must_reject(lambda: approve_price(changed, snapshot, "100.00", now)) changed = deepcopy(draft) changed["revision"] += 1 must_reject(lambda: approve_price(changed, snapshot, "100.00", now)) must_reject(lambda: approve_price(draft, snapshot, "100.00", now + SNAPSHOT_LIFETIME)) must_reject(lambda: approve_price(draft, snapshot, "0", now)) altered = deepcopy(snapshot) altered["median"] = "1.00" must_reject(lambda: approve_price(draft, altered, "1.00", now)) assert draft == original_draft and rows == original_rows return snapshot, approved if __name__ == "__main__": snapshot, approval = self_checks() print(json.dumps({"snapshot": snapshot, "approval": approval}, indent=2)) print("All offline checks passed. No network requests were made.")