94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from hashlib import sha256
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RuntimeVariantCandidate:
|
|
id: str
|
|
variant_key: str
|
|
traffic_weight: int
|
|
content_config: dict[str, Any] | list[Any] | None = None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AssignmentDecision:
|
|
experiment_id: str
|
|
experiment_key: str
|
|
variant_id: str
|
|
variant_key: str
|
|
bucket: int
|
|
reason: str = "weighted_assignment"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RuntimeExperimentContext:
|
|
experiment_id: str
|
|
experiment_key: str
|
|
status: str
|
|
site_key: str | None = None
|
|
assignment_salt: str | None = None
|
|
release_id: str | None = None
|
|
release_version: int | None = None
|
|
payload: dict[str, Any] | list[Any] | None = None
|
|
variants: list[RuntimeVariantCandidate] = field(default_factory=list)
|
|
|
|
|
|
def build_bucket(seed: str) -> int:
|
|
"""Create a stable 0-9999 bucket from a visitor+experiment seed."""
|
|
|
|
digest = sha256(seed.encode("utf-8")).hexdigest()
|
|
return int(digest[:8], 16) % 10000
|
|
|
|
|
|
def choose_variant(
|
|
*,
|
|
experiment_id: str,
|
|
experiment_key: str,
|
|
visitor_id: str,
|
|
site_key: str | None = None,
|
|
assignment_salt: str | None = None,
|
|
variants: list[RuntimeVariantCandidate],
|
|
) -> AssignmentDecision | None:
|
|
"""Deterministically assign a visitor based on weighted variants.
|
|
|
|
This first version intentionally lives in pure Python with no storage
|
|
dependency so we can review the bucketing rule before wiring persistence.
|
|
"""
|
|
|
|
active_variants = [variant for variant in variants if variant.traffic_weight > 0]
|
|
if not active_variants:
|
|
return None
|
|
|
|
total_weight = sum(variant.traffic_weight for variant in active_variants)
|
|
if total_weight <= 0:
|
|
return None
|
|
|
|
seed = f"{site_key or 'global'}:{assignment_salt or experiment_key}:{visitor_id}:{experiment_key}"
|
|
bucket = build_bucket(seed)
|
|
scaled_bucket = bucket % total_weight
|
|
|
|
running_total = 0
|
|
for variant in active_variants:
|
|
running_total += variant.traffic_weight
|
|
if scaled_bucket < running_total:
|
|
return AssignmentDecision(
|
|
experiment_id=experiment_id,
|
|
experiment_key=experiment_key,
|
|
variant_id=variant.id,
|
|
variant_key=variant.variant_key,
|
|
bucket=bucket,
|
|
)
|
|
|
|
chosen = active_variants[-1]
|
|
return AssignmentDecision(
|
|
experiment_id=experiment_id,
|
|
experiment_key=experiment_key,
|
|
variant_id=chosen.id,
|
|
variant_key=chosen.variant_key,
|
|
bucket=bucket,
|
|
reason="weighted_assignment_fallback",
|
|
)
|