When I first implemented production A/B testing for our LLM-powered features, I spent three weeks debugging inconsistent results caused by model version drift, temperature variations, and hidden latency spikes. The framework I'm about to share would have saved me that time entirely. This tutorial walks through building a production-ready AI model A/B testing infrastructure using HolySheep AI as your unified gateway, cutting costs by 85% while maintaining enterprise-grade reliability.
Why Your A/B Testing Approach Determines AI ROI
Most teams test AI models by manually sending the same prompt to different endpoints and eyeballing responses. This approach fails for three critical reasons: statistical insignificance from small sample sizes, confounding variables like network latency affecting perceived quality, and exponential cost growth as you test multiple model configurations.
Before diving into the technical implementation, let's address the foundational question: which API provider should you use for systematic A/B testing at scale?
HolySheep vs Official API vs Relay Services Comparison
| Feature | HolySheep AI | Official OpenAI/Anthropic | Other Relay Services |
|---|---|---|---|
| Pricing (GPT-4.1) | $8.00/MTok output | $15.00/MTok output | $9.50-12.00/MTok |
| Pricing (Claude Sonnet 4.5) | $15.00/MTok output | $18.00/MTok output | $16.00-17.50/MTok |
| Pricing (DeepSeek V3.2) | $0.42/MTok output | $0.55/MTok output | $0.50-0.52/MTok |
| Rate Advantage | ¥1=$1 (85%+ savings vs ¥7.3) | Standard USD rates | Variable markup |
| Latency (P50) | <50ms | 80-150ms | 60-120ms |
| Payment Methods | WeChat, Alipay, USDT | Credit card only | Limited options |
| Free Credits | Yes, on signup | No | Usually no |
| A/B Testing Support | Built-in traffic splitting | None (DIY) | Basic routing only |
| Model Variety | GPT-4.1, Claude 4.5, Gemini 2.5 Flash, DeepSeek V3.2 | Single provider | Mixed, inconsistent |
Sign up here for HolySheep AI and receive free credits to start your A/B testing journey immediately.
Framework Architecture Overview
A production-grade AI A/B testing framework requires four core components working in concert. First, a traffic router that distributes requests according to your configured split ratios. Second, a metrics collector capturing latency, cost, response quality, and custom business metrics. Third, a statistical analyzer that determines significance without requiring massive sample sizes. Fourth, a configuration manager enabling hot-swapping of weights without deployment.
Implementation: The Complete A/B Testing Framework
The following implementation uses Python with async capabilities for high-throughput testing. All requests route through HolySheep's unified gateway at https://api.holysheep.ai/v1, eliminating the need to manage separate provider credentials.
Step 1: Core A/B Testing Client
# ab_testing_framework.py
import asyncio
import hashlib
import time
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from datetime import datetime
import statistics
class ModelConfig:
"""Configuration for a single model variant in A/B test."""
def __init__(
self,
model_id: str,
weight: float = 1.0,
temperature: float = 0.7,
max_tokens: int = 2048,
base_url: str = "https://api.holysheep.ai/v1"
):
self.model_id = model_id
self.weight = weight
self.temperature = temperature
self.max_tokens = max_tokens
self.base_url = base_url
self.api_endpoint = f"{base_url}/chat/completions"
@dataclass
class RequestResult:
"""Container for individual request metrics."""
model_id: str
latency_ms: float
input_tokens: int
output_tokens: int
cost_usd: float
response_text: str
quality_score: Optional[float] = None
custom_metrics: Dict = field(default_factory=dict)
timestamp: datetime = field(default_factory=datetime.now)
error: Optional[str] = None
class ABMetricsCollector:
"""Collects and analyzes metrics across A/B test variants."""
# 2026 pricing from HolySheep (per million output tokens)
MODEL_PRICING = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
def __init__(self):
self.results: Dict[str, List[RequestResult]] = {}
def record(self, variant_name: str, result: RequestResult):
if variant_name not in self.results:
self.results[variant_name] = []
self.results[variant_name].append(result)
def calculate_cost_per_1k(self, variant_name: str) -> float:
"""Calculate cost per 1,000 requests for a variant."""
if variant_name not in self.results:
return float('inf')
total_cost = sum(r.cost_usd for r in self.results[variant_name])
count = len(self.results[variant_name])
return (total_cost / count) * 1000 if count > 0 else float('inf')
def get_latency_stats(self, variant_name: str) -> Dict[str, float]:
"""Get P50, P95, P99 latency for a variant."""
if variant_name not in self.results or not self.results[variant_name]:
return {"p50": 0, "p95": 0, "p99": 0}
latencies = sorted([r.latency_ms for r in self.results[variant_name]])
n = len(latencies)
return {
"p50": latencies[int(n * 0.50)],
"p95": latencies[int(n * 0.95)],
"p99": latencies[int(n * 0.99)],
"mean": statistics.mean(latencies),
"stddev": statistics.stdev(latencies) if n > 1 else 0
}
def calculate_statistical_significance(self, variant_a: str, variant_b: str) -> Dict:
"""
Two-proportion Z-test for comparing conversion/success rates.
Returns z-score and p-value.
"""
results_a = self.results.get(variant_a, [])
results_b = self.results.get(variant_b, [])
if not results_a or not results_b:
return {"error": "Insufficient data"}
# Example: comparing quality scores if available
scores_a = [r.quality_score for r in results_a if r.quality_score is not None]
scores_b = [r.quality_score for r in results_b if r.quality_score is not None]
if len(scores_a) < 10 or len(scores_b) < 10:
return {"error": "Need at least 10 samples per variant"}
mean_a, mean_b = statistics.mean(scores_a), statistics.mean(scores_b)
std_a = statistics.stdev(scores_a) if len(scores_a) > 1 else 0
std_b = statistics.stdev(scores_b) if len(scores_b) > 1 else 0
# Pooled standard error
se = ((std_a ** 2 / len(scores_a)) + (std_b ** 2 / len(scores_b))) ** 0.5
if se == 0:
return {"error": "No variance difference detected"}
z_score = (mean_a - mean_b) / se
# Approximate p-value (two-tailed)
import math
p_value = 2 * (1 - 0.5 * (1 + math.erf(abs(z_score) / math.sqrt(2))))
return {
"z_score": z_score,
"p_value": p_value,
"significant": p_value < 0.05,
"variant_a_mean": mean_a,
"variant_b_mean": mean_b,
"winner": variant_a if mean_a > mean_b else variant_b,
"improvement_pct": abs(mean_a - mean_b) / min(mean_a, mean_b) * 100
}
def generate_report(self) -> str:
"""Generate a comprehensive A/B test report."""
report_lines = [
"=" * 60,
"AI MODEL A/B TESTING REPORT",
f"Generated: {datetime.now().isoformat()}",
"=" * 60,
""
]
for variant, results in self.results.items():
report_lines.append(f"\n### Variant: {variant} ###")
report_lines.append(f"Total Requests: {len(results)}")
errors = sum(1 for r in results if r.error)
report_lines.append(f"Success Rate: {(len(results) - errors) / len(results) * 100:.2f}%")
latency_stats = self.get_latency_stats(variant)
report_lines.append(f"Latency P50: {latency_stats['p50']:.2f}ms")
report_lines.append(f"Latency P95: {latency_stats['p95']:.2f}ms")
report_lines.append(f"Latency P99: {latency_stats['p99']:.2f}ms")
cost_per_1k = self.calculate_cost_per_1k(variant)
report_lines.append(f"Cost per 1K requests: ${cost_per_1k:.4f}")
total_tokens = sum(r.output_tokens for r in results)
report_lines.append(f"Total output tokens: {total_tokens:,}")
return "\n".join(report_lines)
print("A/B Testing Framework imported successfully!")Step 2: Traffic Router with Weighted Distribution
# traffic_router.py
import hashlib
from typing import Tuple, List, Dict
from ab_testing_framework import ModelConfig, ABMetricsCollector
class TrafficRouter:
"""
Routes requests to different model variants based on configurable weights.
Uses consistent hashing to ensure same user gets same variant (sticky sessions).
"""
def __init__(self, variants: List[ModelConfig], collector: ABMetricsCollector):
self.variants = variants
self.collector = collector
self.total_weight = sum(v.weight for v in variants)
self._validate_weights()
def _validate_weights(self):
if self.total_weight <= 0:
raise ValueError("Total weight must be positive")
if not self.variants:
raise ValueError("At least one variant required")
def select_variant(self, user_id: str = None, request_id: str = None) -> ModelConfig:
"""
Select variant using weighted random selection.
If user_id provided, uses consistent hashing for sticky sessions.
"""
# Generate deterministic seed for consistent hashing
if user_id:
seed_input = f"{user_id}:{self.__class__.__name__}"
elif request_id:
seed_input = f"{request_id}:{self.__class__.__name__}"
else:
import time
seed_input = f"{time.time()}:{id(self)}"
hash_value = int(hashlib.md5(seed_input.encode()).hexdigest(), 16)
normalized_value = (hash_value % 10000) / 10000.0 # 0.0 to 1.0
cumulative = 0.0
for variant in self.variants:
cumulative += variant.weight / self.total_weight
if normalized_value < cumulative:
return variant
return self.variants[-1] # Fallback to last variant
def split_traffic(self, percentages: Dict[str, float]) -> None:
"""
Programmatically adjust traffic split percentages.
Example: {'gpt-4.1': 50, 'deepseek-v3.2': 50} for 50/50 split
"""
total_pct = sum(percentages.values())
if abs(total_pct - 100) > 0.01:
raise ValueError(f"Percentages must sum to 100, got {total_pct}")
for variant in self.variants:
if variant.model_id in percentages:
variant.weight = percentages[variant.model_id]
else:
variant.weight = 0.0
self.total_weight = sum(v.weight for v in self.variants)
self._validate_weights()
def get_current_distribution(self) -> Dict[str, float]:
"""Get current traffic distribution percentages."""
return {
variant.model_id: (variant.weight / self.total_weight) * 100
for variant in self.variants
}
Example usage for testing
if __name__ == "__main__":
collector = ABMetricsCollector()
variants = [
ModelConfig(model_id="gpt-4.1", weight=50, temperature=0.7),
ModelConfig(model_id="deepseek-v3.2", weight=50, temperature=0.7),
]
router = TrafficRouter(variants, collector)
# Simulate 1000 routing decisions
distribution = {"gpt-4.1": 0, "deepseek-v3.2": 0}
for i in range(1000):
selected = router.select_variant(user_id=f"user_{i % 100}")
distribution[selected.model_id] += 1
print("Simulated 1000 requests:")
for model, count in distribution.items():
print(f" {model}: {count} ({count/10:.1f}%)")
print(f"\nExpected: ~50% each")
print(f"Actual distribution: {router.get_current_distribution()}")Step 3: Production API Client with A/B Testing Integration
# holy_sheep_ab_client.py
import aiohttp
import asyncio
import json
from typing import List, Dict, Optional, Any
from ab_testing_framework import ABMetricsCollector, ModelConfig, RequestResult
class HolySheepABClient:
"""
Production-ready client for A/B testing AI models via HolySheep gateway.
Base URL: https://api.holysheep.ai/v1
Supports: GPT-4.1 ($8/MTok), Claude Sonnet 4.5 ($15/MTok),
Gemini 2.5 Flash ($2.50/MTok), DeepSeek V3.2 ($0.42/MTok)
"""
BASE_URL = "https://api.holysheep.ai/v1"
# Token pricing per million output tokens (2026 rates)
OUTPUT_PRICING = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
def __init__(self, api_key: str, collector: ABMetricsCollector):
self.api_key = api_key
self.collector = collector
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
def _calculate_cost(self, model_id: str, output_tokens: int) -> float:
"""Calculate cost in USD for output tokens."""
price_per_million = self.OUTPUT_PRICING.get(model_id, 8.00)
return (output_tokens / 1_000_000) * price_per_million
async def chat_completions(
self,
model_config: ModelConfig,
messages: List[Dict[str, str]],
quality_scorer: Optional[callable] = None,
custom_metrics: Optional[Dict] = None
) -> RequestResult:
"""
Send a single request to the selected model variant.
Records full metrics for A/B analysis.
"""
start_time = asyncio.get_event_loop().time()
payload = {
"model": model_config.model_id,
"messages": messages,
"temperature": model_config.temperature,
"max_tokens": model_config.max_tokens,
}
try:
async with self.session.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
end_time = asyncio.get_event_loop().time()
latency_ms = (end_time - start_time) * 1000
if response.status != 200:
error_text = await response.text()
return RequestResult(
model_id=model_config.model_id,
latency_ms=latency_ms,
input_tokens=0,
output_tokens=0,
cost_usd=0,
response_text="",
error=f"HTTP {response.status}: {error_text}"
)
data = await response.json()
# Extract usage metrics
usage = data.get("usage", {})
output_tokens = usage.get("completion_tokens", 0)
input_tokens = usage.get("prompt_tokens", 0)
cost_usd = self._calculate_cost(model_config.model_id, output_tokens)
response_text = data.get("choices", [{}])[0].get("message", {}).get("content", "")
# Calculate quality score if scorer provided
quality_score = None
if quality_scorer:
quality_score = quality_scorer(response_text, messages)
result = RequestResult(
model_id=model_config.model_id,
latency_ms=latency_ms,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost_usd,
response_text=response_text,
quality_score=quality_score,
custom_metrics=custom_metrics or {}
)
self.collector.record(model_config.model_id, result)
return result
except asyncio.TimeoutError:
return RequestResult(
model_id=model_config.model_id,
latency_ms=30000,
input_tokens=0,
output_tokens=0,
cost_usd=0,
response_text="",
error="Request timeout after 30 seconds"
)
except Exception as e:
return RequestResult(
model_id=model_config.model_id,
latency_ms=0,
input_tokens=0,
output_tokens=0,
cost_usd=0,
response_text="",
error=str(e)
)
async def run_ab_test(
self,
variants: List[ModelConfig],
messages: List[Dict[str, str]],
iterations: int = 100,
quality_scorer: Optional[callable] = None,
custom_metrics_fn: Optional[callable] = None
) -> Dict[str, Any]:
"""
Run a complete A/B test across multiple variants.
Returns comprehensive metrics and statistical analysis.
"""
from traffic_router import TrafficRouter
router = TrafficRouter(variants, self.collector)
tasks = []
for i in range(iterations):
selected_variant = router.select_variant(request_id=f"test_{i}")
custom_metrics = custom_metrics_fn(i) if custom_metrics_fn else None
task = self.chat_completions(
model_config=selected_variant,
messages=messages,
quality_scorer=quality_scorer,
custom_metrics=custom_metrics
)
tasks.append(task)
# Execute all requests concurrently (respecting rate limits)
results = await asyncio.gather(*tasks, return_exceptions=True)
return {
"total_requests": iterations,
"successful": sum(1 for r in results if not isinstance(r, Exception) and not r.error),
"distribution": router.get_current_distribution(),
"report": self.collector.generate_report()
}
Example quality scorer function
def simple_quality_scorer(response: str, messages: List[Dict]) -> float:
"""
Example quality scoring function.
Returns 0-100 score based on response characteristics.
"""
score = 50.0 # Base score
# Length bonus (prefer substantial responses)
if len(response) > 200:
score += 10
if len(response) > 500:
score += 10
# Coherence bonus (basic check)
if response.count('.') >= 3:
score += 10
# Penalize empty or very short responses
if len(response) < 50:
score -= 20
return max(0, min(100, score)) # Clamp to 0-100
Demo execution
async def main():
# Initialize with your HolySheep API key
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Replace with actual key
collector = ABMetricsCollector()
test_messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain quantum computing in simple terms."}
]
variants = [
ModelConfig(model_id="gpt-4.1", weight=33, temperature=0.7),
ModelConfig(model_id="deepseek-v3.2", weight=33, temperature=0.7),
ModelConfig(model_id="gemini-2.5-flash", weight=34, temperature=0.7),
]
async with HolySheepABClient(API_KEY, collector) as client:
results = await client.run_ab_test(
variants=variants,
messages=test_messages,
iterations=50,
quality_scorer=simple_quality_scorer
)
print(json.dumps(results, indent=2, default=str))
# Statistical analysis
if len(collector.results) >= 2:
variants_list = list(collector.results.keys())
analysis = collector.calculate_statistical_significance(
variants_list[0], variants_list[1]
)
print("\nStatistical Analysis:")
print(json.dumps(analysis, indent=2))
if __name__ == "__main__":
# Note: Run with actual API key for production testing
print("HolySheep A/B Testing Client ready!")
print("Replace YOUR_HOLYSHEEP_API_KEY with your actual key to test.")Implementing Real-World Quality Assessment
Raw latency and cost metrics only tell part of the story. For meaningful A/B testing, you need quality assessment that aligns with your business objectives. Here's a production-grade quality scoring system that evaluates responses across multiple dimensions.
# quality_evaluator.py
import re
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class QualityMetrics:
"""Container for multi-dimensional quality assessment."""
relevance: float # 0-100: How well response addresses the query
coherence: float # 0-100: Logical flow and consistency
completeness: float # 0-100: Coverage of the topic
safety: float # 0-100: Absence of harmful content
efficiency: float # 0-100: Conciseness vs verbosity balance
overall: float # Weighted average
class QualityEvaluator:
"""
Multi-dimensional quality evaluator for AI responses.
Designed for A/B testing comparison across model variants.
"""
def __init__(self, weights: Dict[str, float] = None):
self.weights = weights or {
"relevance": 0.30,
"coherence": 0.20,
"completeness": 0.20,
"safety": 0.20,
"efficiency": 0.10
}
def evaluate(self, response: str, context: Dict = None) -> QualityMetrics:
"""Evaluate a response across multiple quality dimensions."""
# Relevance: Check for key topic coverage
relevance = self._evaluate_relevance(response, context)
# Coherence: Analyze sentence structure and flow
coherence = self._evaluate_coherence(response)
# Completeness: Check for thorough coverage
completeness = self._evaluate_completeness(response, context)
# Safety: Basic content safety checks
safety = self._evaluate_safety(response)
# Efficiency: Balance between thoroughness and conciseness
efficiency = self._evaluate_efficiency(response)
# Calculate weighted overall score
overall = (
relevance * self.weights["relevance"] +
coherence * self.weights["coherence"] +
completeness * self.weights["completeness"] +
safety * self.weights["safety"] +
efficiency * self.weights["efficiency"]
)
return QualityMetrics(
relevance=relevance,
coherence=coherence,
completeness=completeness,
safety=safety,
efficiency=efficiency,
overall=overall
)
def _evaluate_relevance(self, response: str, context: Dict = None) -> float:
"""Score relevance based on keyword presence and topic alignment."""
if not response:
return 0.0
score = 60.0 # Base score
# Penalize completely generic responses
generic_phrases = ["i'm not sure", "as an ai", "i cannot", "undefined"]
for phrase in generic_phrases:
if phrase.lower() in response.lower():
score -= 15
# Length indicates effort (correlation with relevance)
if len(response) > 300:
score += 20
elif len(response) > 150:
score += 10
# Check for structured response (often indicates thoughtful answer)
if response.count('\n') >= 2 or response.count(';') >= 3:
score += 10
return max(0, min(100, score))
def _evaluate_coherence(self, response: str) -> float:
"""Score logical coherence and flow."""
if not response:
return 0.0
score = 70.0 # Base score
# Check for sentence completion
sentences = re.split(r'[.!?]+', response)
complete_sentences = sum(1 for s in sentences if len(s.strip()) > 10)
if complete_sentences >= 3:
score += 15
elif complete_sentences == 0:
score -= 30
# Penalize excessive repetition
words = response.lower().split()
if len(words) > 20:
unique_ratio = len(set(words)) / len(words)
if unique_ratio < 0.5:
score -= 25
# Bonus for list/structured format
if re.search(r'^\d+[.)]\s|\n-|\n\*', response, re.MULTILINE):
score += 10
return max(0, min(100, score))
def _evaluate_completeness(self, response: str, context: Dict = None) -> float:
"""Score completeness of information coverage."""
if not response:
return 0.0
score = 50.0 # Base score
# Longer responses often more complete (but not always better)
if len(response) > 400:
score += 25
elif len(response) > 200:
score += 15
elif len(response) < 50:
score -= 30
# Check for multi-paragraph structure (depth indicator)
paragraphs = [p for p in response.split('\n\n') if p.strip()]
if len(paragraphs) >= 3:
score += 15
elif len(paragraphs) == 2:
score += 5
# Presence of explanatory phrases
explanatory_markers = ["because", "therefore", "however", "additionally", "for example"]
marker_count = sum(response.lower().count(m) for m in explanatory_markers)
if marker_count >= 2:
score += 10
return max(0, min(100, score))
def _evaluate_safety(self, response: str) -> float:
"""Basic safety assessment (replace with production ML classifier)."""
if not response:
return 100.0
score = 100.0 # Start with perfect score
# Basic harmful content patterns (extend for production)
harmful_patterns = [
(r'\b(buy|manufacture)\s+(weapons|bombs|explosives)\b', -50),
(r'\bhow\s+to\s+hack\b', -30),
(r'\bstep-by-step\s+instructions\s+for\s+(harm|danger)', -40),
]
response_lower = response.lower()
for pattern, penalty in harmful_patterns:
if re.search(pattern, response_lower):
score += penalty
return max(0, min(100, score))
def _evaluate_efficiency(self, response: str) -> float:
"""Score efficiency: right amount of information, not too verbose."""
if not response:
return 0.0
word_count = len(response.split())
# Sweet spot between 100-300 words
if 100 <= word_count <= 300:
score = 100.0
elif word_count > 300:
# Penalize verbose responses
score = 100 - ((word_count - 300) / 10)
else:
# Penalize too brief
score = word_count
# Bonus for good information density
sentences = re.split(r'[.!?]+', response)
if sentences and len(sentences) > 1:
avg_words_per_sentence = word_count / len(sentences)
if 10 <= avg_words_per_sentence <= 25:
score += 10
return max(0, min(100, score))
def compare_models(self, results_a: List[QualityMetrics], results_b: List[QualityMetrics]) -> Dict:
"""Compare quality metrics between two model variants."""
def avg_scores(results):
return {
"overall": sum(r.overall for r in results) / len(results),
"relevance": sum(r.relevance for r in results) / len(results),
"coherence": sum(r.coherence for r in results) / len(results),
"completeness": sum(r.completeness for r in results) / len(results),
}
scores_a = avg_scores(results_a)
scores_b = avg_scores(results_b)
return {
"variant_a_scores": scores_a,
"variant_b_scores": scores_b,
"improvement": {
metric: (scores_a[metric] - scores_b[metric])
for metric in scores_a
}
}
Integration with A/B framework
def create_quality_scorer(evaluator: QualityEvaluator, context: Dict = None):
"""Factory function to create quality scorer for A/B testing."""
def scorer(response: str, messages: List[Dict]) -> float:
quality = evaluator.evaluate(response, context)
return quality.overall
return scorer
if __name__ == "__main__":
evaluator = QualityEvaluator()
# Test with example responses
test_responses = [
"Quantum computing uses quantum mechanics principles to process information. Unlike classical computers that use bits (0 or 1), quantum computers use qubits that can exist in multiple states simultaneously through superposition. This allows quantum computers to solve certain problems exponentially faster than classical computers.",
"It's complicated.",
]
for i, response in enumerate(test_responses, 1):
metrics = evaluator.evaluate(response)
print(f"Response {i}: Overall quality = {metrics.overall:.1f}")
print(f" Relevance: {metrics.relevance:.1f}")
print(f" Coherence: {metrics.coherence:.1f}")
print(f" Completeness: {metrics.completeness:.1f}")
print(f" Safety: {metrics.safety:.1f}")
print(f" Efficiency: {metrics.efficiency:.1f}")
print()Monitoring Dashboard Implementation
For ongoing production A/B tests, you need real-time visibility into performance metrics. Here's a simple monitoring dashboard that tracks your experiment health and automatically detects statistical anomalies.
# ab_monitoring.py
import time
from collections import deque
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import json
class RealtimeMonitor:
"""
Real-time monitoring for A/B test experiments.
Tracks key metrics and alerts on statistical anomalies.
"""
def __init__(self, window_size: int = 100):
self.window_size = window_size