BanditDB

from banditdb import Client, BanditDBError
# Connect to the BanditDB server.# Pass api_key if BANDITDB_API_KEY is set on the server.db = Client(    url="http://localhost:8080",    timeout=2.0,    api_key="your-secret-key",   # omit if server runs without auth)
try:    # 1. Create a campaign (run once at startup)    # algorithm defaults to "linucb"; use "thompson_sampling" for Bayesian exploration    db.create_campaign(        campaign_id="checkout_upsell",        arms=["offer_discount", "offer_free_shipping"],        features=["is_mobile", "cart_value", "is_returning_user"],    )    # or: db.create_campaign(..., algorithm="thompson_sampling")
    # 2. A user arrives — ask the database what to show them.    # Scale each value to 0–1; the SDK puts them in order, adds a constant 1    # and normalises the vector.    arm_id, interaction_id = db.predict("checkout_upsell", {        "is_mobile": 1.0,        "cart_value": min(cart_total / 200, 1.0),        "is_returning_user": 0.0,    })    print(f"Showing: {arm_id}")  # e.g., "offer_free_shipping"
    # 3. The user clicked — send the reward    db.reward(interaction_id, reward=1.0)
except BanditDBError as e:    print(f"Database error: {e}")

from banditdb import Client, PacingConfig, ResourceConstraint
db.create_campaign(    "promos", ["none", "discount_10", "discount_20"], feature_dim=4,    pacing=PacingConfig(resources=[        ResourceConstraint(            name="promo_budget",            budget=5_000.0,      # units available over the horizon            horizon=100_000,     # decisions the budget should last            arm_costs={"discount_10": 1.0, "discount_20": 2.0},  # "none" costs 0        ),    ]),)
# Arms added later declare their own costsdb.add_arm("promos", "discount_30", costs={"promo_budget": 3.0})
for r in db.pacing("promos")["resources"]:    print(f"{r['name']}: {r['utilization']:.0%} spent, lambda={r['lambda']:.3f}")

# Checkpoint: snapshot models, write Parquet, rotate the WAL.# Call this on a schedule or after significant traffic.summary = db.checkpoint()print(summary)# "Checkpoint written and WAL rotated: 2 campaigns, offset 4821 bytes,#  150 interactions exported, 3 in-flight re-emitted"
# List which Parquet files are availableprint(db.export())# 'Parquet files in /data/exports: ["llm_routing.parquet"]'
# Load directly from the mounted volume into Polars.# Flat schema: interaction_id | arm_id | reward | predicted_at | rewarded_at | propensity | feature_0 | ...import polars as pldf = pl.read_parquet("/data/exports/llm_routing.parquet")print(df.head())print(df.columns)

import polars as plfrom banditdb.eval import replay, ips, doubly_robust
df = pl.read_parquet("/data/exports/llm_routing.parquet")
# How much reward would a uniform random policy have earned?print(replay(df))# OPEResult(method='replay', estimate=0.4821, std_error=0.0312, coverage=22.1% [33/149])
print(ips(df))# OPEResult(method='ips', estimate=0.5103, std_error=0.0187, coverage=100.0% [149/149])
print(doubly_robust(df))# OPEResult(method='doubly_robust', estimate=0.5219, std_error=0.0141, coverage=100.0% [149/149])
# Compare against the observed reward of the logging policy:print("Observed (logging policy):", df["reward"].mean())# If observed >> estimate, the campaign has learned something real — it outperforms random.

from banditdb import Client, NeuralLinUCBConfig, NeuralTSConfig, ProgressiveConfig
db = Client("http://localhost:8080")
# LinUCB (default)db.create_campaign("routing", ["fast", "cheap"], feature_dim=4, alpha=1.5)
# Thompson Sampling — natural Bayesian exploration, alpha=1.0 is idealdb.create_campaign("routing_ts", ["fast", "cheap"], feature_dim=4,                   algorithm="thompson_sampling")
# NeuralLinUCB — learns a deep embedding of the context, then applies LinUCBcfg = NeuralLinUCBConfig(    context_dim=4,     # must match feature_dim    embed_dim=32,      # arm matrix dimension (default 32)    hidden_dim=128,    # MLP hidden layer width (default 128)    retrain_every=200, # retrain the MLP every N cumulative rewards)db.create_campaign("routing_neural", ["fast", "cheap"], feature_dim=4, algorithm=cfg)
# NeuralTS — same parameters as NeuralLinUCB, posterior-sampling explorationcfg = NeuralTSConfig(context_dim=4, embed_dim=32)db.create_campaign("routing_nts", ["fast", "cheap"], feature_dim=4, algorithm=cfg)
# Progressive — runs LinUCB vs NeuralLinUCB, shifts traffic to whoever wins SNIPS checkpointscfg = ProgressiveConfig(    base="linucb",    challenger=NeuralLinUCBConfig(context_dim=4, embed_dim=32),    min_obs=100,       # minimum buffer entries per arm before any traffic shift    required_wins=3,   # consecutive checkpoint wins to earn one traffic step    step_bps=1000,     # traffic delta per win run, in basis points (1000 = 10%))db.create_campaign("routing_prog", ["fast", "cheap"], feature_dim=4, algorithm=cfg)