离线、影子与在线 A/B

mkdir l7-online
cd l7-online

"""回放退款请求，比较离线、影子与按用户固定分组的本地执行。"""

import hashlib

EXPERIMENT = "refund-policy-v1"
REQUESTS = [
    {"request": "r1", "user": "u1", "days": 10},
    {"request": "r2", "user": "u2", "days": 40},
    {"request": "r3", "user": "u3", "days": 5},
    {"request": "r4", "user": "u1", "days": 10},
]


def assignment(user, experiment=EXPERIMENT):
    # Python 的 hash() 跨进程可能变化；稳定摘要加实验编号可重现分组。
    raw = f"{experiment}:{user}".encode("utf-8")
    bucket = int(hashlib.sha256(raw).hexdigest(), 16) % 100
    return "B" if bucket < 50 else "A"


def propose(request, variant):
    limit = {"A": 7, "B": 30}[variant]
    return {"request": request["request"],
            "action": "refund" if request["days"] <= limit else "review"}


class LocalSink:
    def __init__(self):
        self.records = {}

    def execute(self, intent):
        # 副作用集中在这里：只写本地内存，不包含网络或真实退款接口。
        key = intent["request"]
        if key in self.records and self.records[key] != intent:
            raise ValueError("conflicting intent")
        self.records[key] = dict(intent)


def evaluate(mode, requests, sink):
    rows = []
    for request in requests:
        a = propose(request, "A")
        b = propose(request, "B")
        if mode == "offline":
            rows.append({"request": request["request"], "candidate": b["action"]})
        elif mode == "shadow":
            # 影子候选仅生成意图，执行器根本不会被调用。
            rows.append({"request": request["request"], "disagree": a != b})
        elif mode == "online":
            group = assignment(request["user"])
            chosen = a if group == "A" else b
            sink.execute(chosen)
            rows.append({"request": request["request"], "user": request["user"],
                         "group": group, "action": chosen["action"]})
        else:
            raise ValueError("unknown mode")
    return rows


def main():
    for mode in ("offline", "shadow", "online"):
        sink = LocalSink()
        rows = evaluate(mode, REQUESTS, sink)
        if mode == "offline":
            print(f"offline candidates={len(rows)} local_writes={len(sink.records)}")
        elif mode == "shadow":
            print(f"shadow disagreements={sum(r['disagree'] for r in rows)} "
                  f"local_writes={len(sink.records)}")
        else:
            for row in rows:
                print(f"online {row['request']} user={row['user']} group={row['group']} "
                      f"action={row['action']}")
            print(f"online local_writes={len(sink.records)} "
                  f"A_requests={sum(r['group'] == 'A' for r in rows)} "
                  f"B_requests={sum(r['group'] == 'B' for r in rows)}")
            unique = {r["user"]: r["group"] for r in rows}
            print(f"online A_users={sum(g == 'A' for g in unique.values())} "
                  f"B_users={sum(g == 'B' for g in unique.values())}")


if __name__ == "__main__":
    main()

ls
python -X utf8 online_lab.py

offline candidates=4 local_writes=0
shadow disagreements=2 local_writes=0
online r1 user=u1 group=B action=refund
online r2 user=u2 group=B action=review
online r3 user=u3 group=A action=refund
online r4 user=u1 group=B action=refund
online local_writes=4 A_requests=1 B_requests=3
online A_users=1 B_users=2

ls
python -X utf8 -m unittest -v test_online_lab.py