1"""Chapter 35 (case study: leaderboard / top-K, fantasy cricket) deep-dive code.
21. A sorted set (what Redis ZSETs give you): update, rank, top-K.
32. Real-time updates: one real-world event changes millions of scores; batch them.
43. Ties: a composite score so equal points rank deterministically.
54. Sharding: merge per-shard top-K; rank = sum of counts above you; approximate percentile.
6Run: python3 ch35_leaderboard.py"""
7import bisect
8import random
9from collections import defaultdict
10
11
12# ---- 1. a sorted set ---------------------------------------------------------------
13
14class SortedSet:
15 """Members ordered by (score desc). Kept as a sorted list of (-score, member)."""
16
17 def __init__(self):
18 self.items, self.score = [], {}
19
20 def add(self, member, score): # ZADD
21 if member in self.score:
22 self.items.remove((-self.score[member], member))
23 self.score[member] = score
24 bisect.insort(self.items, (-score, member))
25
26 def rank(self, member): # ZREVRANK (0-based)
27 return bisect.bisect_left(self.items, (-self.score[member], member))
28
29 def top(self, k): # ZREVRANGE 0 k-1 WITHSCORES
30 return [(m, -s) for s, m in self.items[:k]]
31
32 def count_above(self, score): # ZCOUNT (score, +inf)
33 return bisect.bisect_left(self.items, (-score, ""))
34
35
36# ---- 2. scoring fantasy teams ------------------------------------------------------
37
38def team_points(team, player_points):
39 """Captain counts double, vice-captain one and a half times."""
40 total = 0.0
41 for p in team["players"]:
42 mult = 2.0 if p == team["captain"] else 1.5 if p == team["vice"] else 1.0
43 total += player_points.get(p, 0) * mult
44 return total
45
46
47def affected(teams, player):
48 return [t for t in teams if player in t["players"]]
49
50
51# ---- 3. ties -----------------------------------------------------------------------
52
53def composite(points, joined_at, max_ts=10**10):
54 """Higher points first; for equal points, whoever joined earlier ranks higher.
55 Points are in half-points (multipliers 1.5x), so scale by 2 to keep integers."""
56 return int(points * 2) * max_ts + (max_ts - joined_at)
57
58
59def competition_ranks(scores):
60 """Standard '1224' ranking for display: equal scores share a rank."""
61 ordered = sorted(scores, reverse=True)
62 out, prev, rank = [], None, 0
63 for i, s in enumerate(ordered):
64 if s != prev:
65 rank, prev = i + 1, s
66 out.append((rank, s))
67 return out
68
69
70# ---- 4. sharding -------------------------------------------------------------------
71
72def shard_of(user, n):
73 return hash(user) % n
74
75
76def global_top(shards, k):
77 merged = []
78 for s in shards:
79 merged.extend(s.top(k)) # each shard's top k is enough
80 return sorted(merged, key=lambda x: (-x[1], x[0]))[:k]
81
82
83def global_rank(shards, score):
84 return sum(s.count_above(score) for s in shards) # one ZCOUNT per shard
85
86
87def approx_percentile(bucket_counts, score, bucket_size):
88 """Histogram of scores in buckets: 'you're in the top x%' without exact rank."""
89 b = int(score // bucket_size)
90 above = sum(c for k, c in bucket_counts.items() if k > b)
91 total = sum(bucket_counts.values())
92 return 100 * above / total
93
94
95# ---- checks ------------------------------------------------------------------------
96
97def run_checks():
98 rnd = random.Random(35)
99 squad = [f"p{i}" for i in range(22)] # 22 players in the match
100
101 teams = []
102 for u in range(5_000):
103 picks = rnd.sample(squad[:6], 4) + rnd.sample(squad[6:], 7) # p0-p5 are popular picks
104 teams.append({"user": f"u{u}", "players": picks, "captain": picks[0],
105 "vice": picks[1], "joined_at": rnd.randint(0, 10_000)})
106
107 # 1. sorted set basics
108 points = {p: rnd.randint(0, 60) for p in squad}
109 lb = SortedSet()
110 for t in teams:
111 lb.add(t["user"], team_points(t, points))
112 best = lb.top(3)
113 assert best[0][1] >= best[1][1] >= best[2][1]
114 u = teams[123]["user"]
115 assert lb.rank(u) == sum(1 for t in teams if lb.score[t["user"]] > lb.score[u]) + \
116 sum(1 for t in teams if lb.score[t["user"]] == lb.score[u] and t["user"] < u)
117
118 # 2. one six by a popular player changes many scores
119 hit = affected(teams, "p0")
120 share = len(hit) / len(teams)
121 assert 0.6 < share < 0.75 # ~2/3 of teams picked p0
122 assert round(20_000_000 * 0.66 / 1_000_000) == 13 # ~13M updates for one ball in a 20M contest
123 assert round(13_200_000 / 100_000) == 132 # one shard at 100k updates/s: over 2 minutes
124 assert round(13_200_000 / (100_000 * 20)) == 7 # 20 shards: ~7 s
125
126 # 3. ties
127 a = composite(120.5, joined_at=500)
128 b = composite(120.5, joined_at=900)
129 c = composite(121.0, joined_at=9_999)
130 assert c > a > b # more points wins; then earlier join
131 assert competition_ranks([90, 85, 85, 70]) == [(1, 90), (2, 85), (2, 85), (4, 70)]
132
133 # 4. sharding by user: global top-K and rank from shards equal the single set
134 n = 8
135 shards = [SortedSet() for _ in range(n)]
136 for t in teams:
137 shards[shard_of(t["user"], n)].add(t["user"], lb.score[t["user"]])
138 assert [s for _, s in global_top(shards, 10)] == [s for _, s in lb.top(10)]
139 my = lb.score[u]
140 assert global_rank(shards, my) == lb.count_above(my)
141
142 # approximate percentile from a 10-point histogram
143 buckets = defaultdict(int)
144 for t in teams:
145 buckets[int(lb.score[t["user"]] // 10)] += 1
146 pct = approx_percentile(buckets, my, 10)
147 exact = 100 * lb.count_above(my) / len(teams)
148 assert abs(pct - exact) < 10 # close enough for "top x%"
149
150 # memory: 20M members at ~100 B each in a sorted set (assumed)
151 assert 20_000_000 * 100 / 1e9 == 2.0
152
153 return {"top3": best, "p0_share": round(share, 2), "rank_u123": lb.rank(u),
154 "pct_approx_vs_exact": (round(pct, 1), round(exact, 1))}
155
156
157if __name__ == "__main__":
158 print("All checks pass.", run_checks())