MPS-20: lib/auction.py — bid placement, proxy, soft-close

Pure functions:
- validate_bid: state must be ACTIVE; first bid >= start_price; subsequent
  bids >= current_high + bid_increment; positive integer; max_proxy >= amount
- is_within_soft_close: now_ms inside (end - soft_close_seconds*1000, end]
- extended_end_timestamp: now_ms + soft_close_seconds*1000
- resolve_proxy: eBay-style — higher proxy wins; loser auto-bids defending
  bidder up to min(loser_proxy + increment, winner_proxy); ties go to
  the existing top (first-in wins)

Orchestrator place_bid:
- writes the new bid, marks the prior winning bid is_winning=False with
  outbid_timestamp, applies proxy resolution to choose visible amounts,
  applies soft-close to extend end_timestamp when bid lands in window,
  bumps auction.updated_timestamp

BidRejected exception carries reason in .args[0].

Tests:
- 16 unit tests: validate_bid matrix (state, start_price, increment,
  proxy >= amount), soft-close window math, 6 proxy resolution edge
  cases (no proxy, defending proxy auto-increments, breaking through
  top proxy, tie tie-break, capped at ceiling)
- 8 integration tests: first bid wins, increment floor, outbid marks
  prior bid is_winning=False, exactly-one-winner invariant, proxy
  defending bidder auto-increments, soft-close fires only inside window,
  inactive auction rejects

Self-bid (bidder == seller) blocking is the view layer's responsibility —
the pure validate_bid does not have visibility into seller identity.

Total: 859 tests pass (was 835 + 24).
This commit is contained in:
russell@unturf.com 2026-05-09 18:49:55 -04:00
parent 2f1a41d089
commit 5b8f6bcb58
No known key found for this signature in database
3 changed files with 609 additions and 0 deletions

View file

@ -0,0 +1,224 @@
"""Auction logic — MPS-20.
Pure functions for bid validation, soft-close math, and proxy resolution.
An orchestrator (place_bid) writes the bid + applies side effects.
Soft-close: a bid placed within `soft_close_seconds` of `end_timestamp`
extends the end by `soft_close_seconds`. eBay-style anti-snipe.
Proxy bidding (eBay-style):
- Each bidder may submit a `max_proxy_in_cents` the secret ceiling they're
willing to pay. The visible bid (`amount_in_cents`) is what shows on the
auction page; the proxy is hidden.
- When a new bid arrives, we compare the new bidder's max_proxy to the
current winner's max_proxy. The bidder with the higher proxy stays
winning at min(loser_proxy + increment, winner_proxy). The loser sees
their bid recorded at their actual amount, but they're outbid.
- A new bid where max_proxy is None defaults to the visible amount
no auto-incrementing.
Tie-breaking: identical proxies the existing winner stays (first-in wins).
"""
from ..models.auction import (
AUCTION_STATE_ACTIVE,
MpsAuction,
MpsBid,
now_timestamp,
)
class BidRejected(Exception):
"""Raised when a bid cannot be accepted. Reason in .args[0]."""
# ── Pure helpers ─────────────────────────────────────────────────────────────
def validate_bid(
auction_state,
current_high_in_cents,
bid_increment_in_cents,
start_price_in_cents,
has_bids,
amount_in_cents,
max_proxy_in_cents=None,
):
"""Raise BidRejected if the bid is invalid; return None otherwise.
Pure: no side effects, no DB. Caller passes raw scalars so this fn is
100% unit-testable without a fixture.
Rules:
- auction must be in ACTIVE state
- amount must be a positive integer
- first bid: amount >= start_price
- subsequent bids: amount >= current_high + increment
- max_proxy (if set) must be >= amount
"""
if auction_state != AUCTION_STATE_ACTIVE:
raise BidRejected("auction is not active")
if not isinstance(amount_in_cents, int) or amount_in_cents <= 0:
raise BidRejected("bid amount must be a positive integer (cents)")
if max_proxy_in_cents is not None:
if not isinstance(max_proxy_in_cents, int) or max_proxy_in_cents < amount_in_cents:
raise BidRejected("max proxy must be >= bid amount")
if not has_bids:
if amount_in_cents < start_price_in_cents:
raise BidRejected(
f"bid below start price ({start_price_in_cents} cents)"
)
else:
floor = current_high_in_cents + bid_increment_in_cents
if amount_in_cents < floor:
raise BidRejected(
f"bid below current high + increment ({floor} cents)"
)
def is_within_soft_close(end_timestamp, now_ms, soft_close_seconds):
"""Pure: is `now_ms` inside the soft-close window?"""
if not end_timestamp or not soft_close_seconds:
return False
window_ms = soft_close_seconds * 1000
return (end_timestamp - now_ms) < window_ms and now_ms <= end_timestamp
def extended_end_timestamp(now_ms, soft_close_seconds):
"""Pure: new end_timestamp after soft-close fires."""
return now_ms + (soft_close_seconds * 1000)
def resolve_proxy(
top_amount_in_cents,
top_max_proxy_in_cents,
new_amount_in_cents,
new_max_proxy_in_cents,
bid_increment_in_cents,
):
"""Pure: given the current top bid and a new incoming bid (with optional
proxies), return (winning_amount_in_cents, new_bidder_wins).
- top_max_proxy and new_max_proxy default to their respective visible
amounts when callers pass None (no proxy submitted).
- Tie on max_proxy: existing top stays winning (first-in wins).
Returns:
(winning_amount_in_cents, new_bidder_wins) where new_bidder_wins
is True iff the incoming bidder takes the lead.
"""
top_proxy = (
top_max_proxy_in_cents
if top_max_proxy_in_cents is not None
else top_amount_in_cents
)
new_proxy = (
new_max_proxy_in_cents
if new_max_proxy_in_cents is not None
else new_amount_in_cents
)
if new_proxy > top_proxy:
# New bidder wins. Their visible amount auto-increments past
# the prior top's proxy, capped at their own proxy ceiling.
winning = min(top_proxy + bid_increment_in_cents, new_proxy)
# Floor: at least max(new_amount_in_cents, top_amount_in_cents + increment)
winning = max(winning, new_amount_in_cents)
return (winning, True)
# Existing top stays. Their visible amount auto-increments to defend
# against the new bid, capped at their own proxy ceiling.
winning = min(new_proxy + bid_increment_in_cents, top_proxy)
# Floor: never less than current top's visible amount.
winning = max(winning, top_amount_in_cents)
return (winning, False)
# ── Orchestrator ─────────────────────────────────────────────────────────────
def place_bid(
auction,
bidder,
amount_in_cents,
max_proxy_in_cents=None,
now_ms=None,
):
"""Place a bid on `auction` by `bidder`. Writes the bid, updates the
auction (soft-close, winning bid pointer), marks the prior winning bid
as outbid. Caller is responsible for transaction.commit().
Returns the new MpsBid on success. Raises BidRejected on rejection.
Block self-bidding (bidder == seller): caller's responsibility — the
pure validate_bid does not have visibility into seller identity.
"""
if now_ms is None:
now_ms = now_timestamp()
has_bids = auction.bids.count() > 0
validate_bid(
auction_state=auction.state,
current_high_in_cents=auction.current_high_in_cents,
bid_increment_in_cents=auction.bid_increment_in_cents,
start_price_in_cents=auction.start_price_in_cents,
has_bids=has_bids,
amount_in_cents=amount_in_cents,
max_proxy_in_cents=max_proxy_in_cents,
)
dbsession = auction.dbsession
# Find the current winning bid (if any) — only one per auction.
prior_winner = (
dbsession.query(MpsBid)
.filter(MpsBid.auction_id == auction.id, MpsBid.is_winning.is_(True))
.one_or_none()
)
bid = MpsBid(
auction=auction,
bidder=bidder,
amount_in_cents=amount_in_cents,
max_proxy_in_cents=max_proxy_in_cents,
)
if prior_winner is None:
# First bid takes the lead at face value.
bid.is_winning = True
dbsession.add(bid)
else:
winning_amount, new_bidder_wins = resolve_proxy(
top_amount_in_cents=prior_winner.amount_in_cents,
top_max_proxy_in_cents=prior_winner.max_proxy_in_cents,
new_amount_in_cents=amount_in_cents,
new_max_proxy_in_cents=max_proxy_in_cents,
bid_increment_in_cents=auction.bid_increment_in_cents,
)
if new_bidder_wins:
prior_winner.is_winning = False
prior_winner.outbid_timestamp = now_ms
bid.amount_in_cents = winning_amount
bid.is_winning = True
dbsession.add(bid)
else:
# New bid is recorded but loses; defending top auto-bids up.
bid.is_winning = False
bid.outbid_timestamp = now_ms
dbsession.add(bid)
prior_winner.amount_in_cents = winning_amount
# Soft-close: extend if bid arrived in the closing window.
if is_within_soft_close(
auction.end_timestamp, now_ms, auction.soft_close_seconds,
):
auction.end_timestamp = extended_end_timestamp(
now_ms, auction.soft_close_seconds,
)
auction.updated_timestamp = now_ms
dbsession.flush()
return bid

View file

@ -4363,3 +4363,160 @@ class TestPricingModePersistence(DatabaseIntegrationTests):
# Booleans/nullable: offer_enabled defaults None (no server_default set)
# offer_min_in_cents nullable.
self.assertIsNone(fetched.offer_min_in_cents)
class TestAuctionPlaceBid(DatabaseIntegrationTests):
"""MPS-20: lib/auction.place_bid orchestrator — writes bid, applies
soft-close, marks prior winning bid outbid. Real DB."""
def _make_active_auction(self, start_price=1000, increment=100,
soft_close=60, end_in_ms=3_600_000):
from ..models.auction import (
MpsAuction, AUCTION_STATE_ACTIVE, now_timestamp,
)
shop = Shop(
name="Active Shop", phone_number="x", billing_address="x",
description="x",
)
shop.stripe_public_api_key = "pk_test"
shop.stripe_secret_api_key = "sk_test"
shop.domain_name = "active.test"
self.dbsession.add(shop)
product = Product(title="Hot item", description="...")
product.shop = shop
product.price_in_cents = start_price
product.is_physical = False
product.is_sellable = True
product.pricing_mode = 1
self.dbsession.add(product)
self.dbsession.flush()
auction = MpsAuction(
product=product, shop=shop,
start_price_in_cents=start_price,
bid_increment_in_cents=increment,
soft_close_seconds=soft_close,
)
auction.state = AUCTION_STATE_ACTIVE
auction.start_timestamp = now_timestamp() - 60_000
auction.end_timestamp = now_timestamp() + end_in_ms
auction.original_end_timestamp = auction.end_timestamp
self.dbsession.add(auction)
self.dbsession.flush()
return shop, product, auction
def _make_user(self, email):
u = get_or_create_user_by_email(self.dbsession, email)
self.dbsession.add(u)
self.dbsession.flush()
return u
def test_place_first_bid(self):
from ..lib.auction import place_bid
shop, product, auction = self._make_active_auction()
u = self._make_user("first@example.com")
bid = place_bid(auction, u, amount_in_cents=1000)
self.assertEqual(bid.amount_in_cents, 1000)
self.assertTrue(bid.is_winning)
self.assertEqual(auction.bids.count(), 1)
self.assertEqual(auction.current_high_in_cents, 1000)
def test_place_bid_below_start_rejected(self):
from ..lib.auction import place_bid, BidRejected
shop, product, auction = self._make_active_auction(start_price=1000)
u = self._make_user("low@example.com")
with self.assertRaises(BidRejected):
place_bid(auction, u, amount_in_cents=999)
def test_place_bid_below_increment_rejected(self):
from ..lib.auction import place_bid, BidRejected
shop, product, auction = self._make_active_auction(
start_price=1000, increment=100,
)
u1 = self._make_user("u1@example.com")
u2 = self._make_user("u2@example.com")
place_bid(auction, u1, amount_in_cents=1000)
with self.assertRaises(BidRejected):
place_bid(auction, u2, amount_in_cents=1099)
def test_outbid_marks_prior_winning_false(self):
from ..lib.auction import place_bid
from ..models.auction import MpsBid
shop, product, auction = self._make_active_auction()
u1 = self._make_user("u1@example.com")
u2 = self._make_user("u2@example.com")
bid1 = place_bid(auction, u1, amount_in_cents=1000)
bid2 = place_bid(auction, u2, amount_in_cents=1100)
self.dbsession.refresh(bid1)
self.assertFalse(bid1.is_winning)
self.assertIsNotNone(bid1.outbid_timestamp)
self.assertTrue(bid2.is_winning)
self.assertEqual(auction.current_high_in_cents, 1100)
# Exactly one winning bid at any time.
winning_count = (
self.dbsession.query(MpsBid)
.filter(MpsBid.auction_id == auction.id, MpsBid.is_winning.is_(True))
.count()
)
self.assertEqual(winning_count, 1)
def test_proxy_defending_bidder_auto_increments(self):
from ..lib.auction import place_bid
shop, product, auction = self._make_active_auction(
start_price=1000, increment=100,
)
u1 = self._make_user("proxy_top@example.com")
u2 = self._make_user("proxy_chal@example.com")
# u1 bids 1000 with proxy ceiling 5000.
place_bid(auction, u1, amount_in_cents=1000, max_proxy_in_cents=5000)
# u2 bids 1500 with proxy ceiling 2000.
place_bid(auction, u2, amount_in_cents=1500, max_proxy_in_cents=2000)
# u1 still winning at min(2000 + 100, 5000) = 2100.
self.assertEqual(auction.current_high_in_cents, 2100)
def test_soft_close_extends_end_timestamp(self):
from ..lib.auction import place_bid
from ..models.auction import now_timestamp
# Set up an auction ending in 30 seconds with a 60s soft-close window.
shop, product, auction = self._make_active_auction(
soft_close=60, end_in_ms=30_000,
)
u = self._make_user("late@example.com")
original_end = auction.end_timestamp
place_bid(auction, u, amount_in_cents=1000)
# End should now be roughly now + 60s.
self.assertGreater(auction.end_timestamp, original_end)
approx_target = now_timestamp() + 60_000
self.assertLess(abs(auction.end_timestamp - approx_target), 5_000)
# original_end_timestamp preserved.
self.assertEqual(auction.original_end_timestamp, original_end)
def test_no_soft_close_when_not_in_window(self):
from ..lib.auction import place_bid
# Auction with 1 hour remaining, 60s soft-close → not in window.
shop, product, auction = self._make_active_auction(
soft_close=60, end_in_ms=3_600_000,
)
u = self._make_user("early@example.com")
original_end = auction.end_timestamp
place_bid(auction, u, amount_in_cents=1000)
self.assertEqual(auction.end_timestamp, original_end)
def test_inactive_auction_rejects_bid(self):
from ..lib.auction import place_bid, BidRejected
from ..models.auction import AUCTION_STATE_DRAFT
shop, product, auction = self._make_active_auction()
auction.state = AUCTION_STATE_DRAFT
u = self._make_user("draft@example.com")
with self.assertRaises(BidRejected):
place_bid(auction, u, amount_in_cents=1000)

View file

@ -4419,3 +4419,231 @@ class TestMpsOfferStateHelpers(unittest.TestCase):
state=OFFER_STATE_PAID, expires_timestamp=past,
)
self.assertFalse(terminal_past.is_expired)
class TestAuctionLibPureFunctions(unittest.TestCase):
"""MPS-20: lib/auction.py pure functions — no DB."""
def test_validate_bid_rejects_inactive(self):
from ..lib.auction import validate_bid, BidRejected
from ..models.auction import (
AUCTION_STATE_DRAFT, AUCTION_STATE_SCHEDULED,
AUCTION_STATE_ENDED, AUCTION_STATE_CANCELLED,
)
for state in (AUCTION_STATE_DRAFT, AUCTION_STATE_SCHEDULED,
AUCTION_STATE_ENDED, AUCTION_STATE_CANCELLED):
with self.assertRaises(BidRejected):
validate_bid(
auction_state=state,
current_high_in_cents=0,
bid_increment_in_cents=100,
start_price_in_cents=1000,
has_bids=False,
amount_in_cents=1000,
)
def test_validate_bid_rejects_below_start_price_first_bid(self):
from ..lib.auction import validate_bid, BidRejected
from ..models.auction import AUCTION_STATE_ACTIVE
with self.assertRaises(BidRejected):
validate_bid(
auction_state=AUCTION_STATE_ACTIVE,
current_high_in_cents=1000,
bid_increment_in_cents=100,
start_price_in_cents=1000,
has_bids=False,
amount_in_cents=999,
)
def test_validate_bid_accepts_first_bid_at_start_price(self):
from ..lib.auction import validate_bid
from ..models.auction import AUCTION_STATE_ACTIVE
validate_bid(
auction_state=AUCTION_STATE_ACTIVE,
current_high_in_cents=1000,
bid_increment_in_cents=100,
start_price_in_cents=1000,
has_bids=False,
amount_in_cents=1000,
) # no raise
def test_validate_bid_rejects_below_increment_floor(self):
from ..lib.auction import validate_bid, BidRejected
from ..models.auction import AUCTION_STATE_ACTIVE
# current high = 1000, increment = 100 → floor = 1100
with self.assertRaises(BidRejected):
validate_bid(
auction_state=AUCTION_STATE_ACTIVE,
current_high_in_cents=1000,
bid_increment_in_cents=100,
start_price_in_cents=500,
has_bids=True,
amount_in_cents=1099,
)
def test_validate_bid_accepts_at_increment_floor(self):
from ..lib.auction import validate_bid
from ..models.auction import AUCTION_STATE_ACTIVE
validate_bid(
auction_state=AUCTION_STATE_ACTIVE,
current_high_in_cents=1000,
bid_increment_in_cents=100,
start_price_in_cents=500,
has_bids=True,
amount_in_cents=1100,
)
def test_validate_bid_rejects_negative_or_zero_amount(self):
from ..lib.auction import validate_bid, BidRejected
from ..models.auction import AUCTION_STATE_ACTIVE
for bad in (0, -1, -1000):
with self.assertRaises(BidRejected):
validate_bid(
auction_state=AUCTION_STATE_ACTIVE,
current_high_in_cents=0,
bid_increment_in_cents=100,
start_price_in_cents=0,
has_bids=False,
amount_in_cents=bad,
)
def test_validate_bid_rejects_proxy_below_amount(self):
from ..lib.auction import validate_bid, BidRejected
from ..models.auction import AUCTION_STATE_ACTIVE
with self.assertRaises(BidRejected):
validate_bid(
auction_state=AUCTION_STATE_ACTIVE,
current_high_in_cents=0,
bid_increment_in_cents=100,
start_price_in_cents=0,
has_bids=False,
amount_in_cents=1000,
max_proxy_in_cents=999,
)
def test_validate_bid_accepts_proxy_equal_to_amount(self):
from ..lib.auction import validate_bid
from ..models.auction import AUCTION_STATE_ACTIVE
validate_bid(
auction_state=AUCTION_STATE_ACTIVE,
current_high_in_cents=0,
bid_increment_in_cents=100,
start_price_in_cents=0,
has_bids=False,
amount_in_cents=1000,
max_proxy_in_cents=1000,
)
def test_is_within_soft_close(self):
from ..lib.auction import is_within_soft_close
# 60-second window, 30 seconds remaining → inside.
end = 1_000_000
now_inside = end - 30_000
self.assertTrue(is_within_soft_close(end, now_inside, 60))
# 60-second window, 90 seconds remaining → outside.
now_outside = end - 90_000
self.assertFalse(is_within_soft_close(end, now_outside, 60))
# No end timestamp → never inside.
self.assertFalse(is_within_soft_close(None, now_inside, 60))
# No window → never inside.
self.assertFalse(is_within_soft_close(end, now_inside, 0))
# Past end → not inside (already over).
self.assertFalse(is_within_soft_close(end, end + 1_000, 60))
def test_extended_end_timestamp(self):
from ..lib.auction import extended_end_timestamp
self.assertEqual(extended_end_timestamp(1_000_000, 60), 1_060_000)
self.assertEqual(extended_end_timestamp(2_000, 0), 2_000)
class TestAuctionLibProxyResolution(unittest.TestCase):
"""MPS-20: resolve_proxy edge cases — pure math, no DB."""
def test_no_existing_proxy_new_bid_wins_at_face_value(self):
# Top: 1000 (no proxy). New: 1100 (no proxy).
# New > top, but no proxy auto-bid is needed since neither has a proxy.
# Floor: max(top + increment, new_amount) = max(1100, 1100) = 1100.
from ..lib.auction import resolve_proxy
winning, new_wins = resolve_proxy(
top_amount_in_cents=1000,
top_max_proxy_in_cents=None,
new_amount_in_cents=1100,
new_max_proxy_in_cents=None,
bid_increment_in_cents=100,
)
self.assertTrue(new_wins)
self.assertEqual(winning, 1100)
def test_top_proxy_defends_against_lower_proxy(self):
# Top: visible 1000, proxy 2000. New: visible 1100, proxy 1500.
# Top stays. Their visible amount = min(1500 + 100, 2000) = 1600.
from ..lib.auction import resolve_proxy
winning, new_wins = resolve_proxy(
top_amount_in_cents=1000,
top_max_proxy_in_cents=2000,
new_amount_in_cents=1100,
new_max_proxy_in_cents=1500,
bid_increment_in_cents=100,
)
self.assertFalse(new_wins)
self.assertEqual(winning, 1600)
def test_new_proxy_breaks_through_top_proxy(self):
# Top: visible 1000, proxy 1500. New: visible 1100, proxy 2500.
# New wins. Their visible amount = min(1500 + 100, 2500) = 1600.
from ..lib.auction import resolve_proxy
winning, new_wins = resolve_proxy(
top_amount_in_cents=1000,
top_max_proxy_in_cents=1500,
new_amount_in_cents=1100,
new_max_proxy_in_cents=2500,
bid_increment_in_cents=100,
)
self.assertTrue(new_wins)
self.assertEqual(winning, 1600)
def test_tie_on_proxy_existing_top_stays(self):
# Top: visible 1000, proxy 2000. New: visible 1500, proxy 2000.
# Tie → top stays. Top auto-bids to min(2000 + 100, 2000) = 2000
# (capped at own ceiling).
from ..lib.auction import resolve_proxy
winning, new_wins = resolve_proxy(
top_amount_in_cents=1000,
top_max_proxy_in_cents=2000,
new_amount_in_cents=1500,
new_max_proxy_in_cents=2000,
bid_increment_in_cents=100,
)
self.assertFalse(new_wins)
self.assertEqual(winning, 2000)
def test_new_amount_above_top_proxy_new_wins(self):
# Top: visible 1000, proxy 1500. New: visible 2000, no proxy.
# New visible (2000) > top proxy (1500) → new wins.
# Winning = min(1500 + 100, 2000) = 1600 (capped at new's effective proxy=2000)
# Floor: max(1600, 2000) = 2000.
from ..lib.auction import resolve_proxy
winning, new_wins = resolve_proxy(
top_amount_in_cents=1000,
top_max_proxy_in_cents=1500,
new_amount_in_cents=2000,
new_max_proxy_in_cents=None,
bid_increment_in_cents=100,
)
self.assertTrue(new_wins)
self.assertEqual(winning, 2000)
def test_proxy_capped_at_ceiling(self):
# Top: visible 1000, proxy 5000. New: visible 1100, proxy 4000.
# Top stays. Their visible amount = min(4000 + 100, 5000) = 4100.
# Top doesn't max out at 5000 because 4100 is enough to outbid new.
from ..lib.auction import resolve_proxy
winning, new_wins = resolve_proxy(
top_amount_in_cents=1000,
top_max_proxy_in_cents=5000,
new_amount_in_cents=1100,
new_max_proxy_in_cents=4000,
bid_increment_in_cents=100,
)
self.assertFalse(new_wins)
self.assertEqual(winning, 4100)