The ranking feedback loop

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""The ranking feedback loop.

52 weeks of catalog search with two tables. A starts with 60 reads, B with 40.
Each week 100 users arrive: 70% take the top-ranked result; the other 30%
compare both and 60% of them pick B, the better table. Ranking by total reads,
then with a certification bonus for B (swept to find the smallest that lets B
take over), then with recent reads weighted more (an exponential decay).
"""
import argparse, json, random

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--weeks", type=int, default=52, help="Weeks to simulate.")
p.add_argument("--users", type=int, default=100, help="New users per week.")
p.add_argument("--top-share", type=float, default=0.7, help="Share of users who take the top result.")
p.add_argument("--b-share", type=float, default=0.6, help="Share of comparing users who pick B.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

def simulate(bonus=0.0, decay=1.0, seed=args.seed, trace=None):
    rng = random.Random(seed); reads = {"A": 60.0, "B": 40.0}; score = dict(reads)
    for week in range(1, args.weeks + 1):
        top = "B" if score["B"] + bonus > score["A"] else "A"
        got = {"A": 0, "B": 0}
        for _ in range(args.users):
            if rng.random() < args.top_share: got[top] += 1
            else: got["B" if rng.random() < args.b_share else "A"] += 1
        for t in reads: reads[t] += got[t]; score[t] = score[t] * decay + got[t]
        if trace is not None: trace.append({"week": week, "A": reads["A"], "B": reads["B"], "top": top})
    return reads, top

timeline = []; reads, top = simulate(trace=timeline)
sweep = []
for bonus in range(0, 41):
    wins = sum(simulate(bonus=bonus, seed=s)[1] == "B" for s in range(20))
    sweep.append({"bonus": bonus, "b_on_top_pct": wins / 20 * 100})
smallest = next((r["bonus"] for r in sweep if r["b_on_top_pct"] == 100), None)
decays = []
for decay in (1.0, 0.9, 0.7, 0.5):
    wins = sum(simulate(decay=decay, seed=s)[1] == "B" for s in range(20))
    decays.append({"decay": decay, "b_on_top_pct": wins / 20 * 100})
summary = {"a_reads": reads["A"], "b_reads": reads["B"], "top": top, "smallest_bonus": smallest, "initial_gap": 20,
           "a_share_pct": reads["A"] / (reads["A"] + reads["B"]) * 100}
print(f"ranked by total reads, after {args.weeks} weeks: A {reads['A']:.0f} reads, B {reads['B']:.0f}; on top: {top}")
print(f"smallest certification bonus for B that puts B on top in all 20 seeds: {smallest}")
for r in decays: print(f"recency weighting (weekly decay {r['decay']}): B on top in {r['b_on_top_pct']:.0f}% of seeds")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"summary": summary, "tables": {"timeline": timeline, "bonus": sweep, "decay": decays}}, f)