elo-leaderboard

第6章 · Agent 的评估 · 配套项目 chapter6/elo-leaderboard

项目说明

Elo Rating Leaderboard from Pairwise Comparisons

Experiment 6-6: Building Model Leaderboard from Pairwise Comparison Data

This project implements an Elo rating system from scratch to analyze model performance using Chatbot Arena's public voting data. The implementation demonstrates how the Bradley-Terry model extracts relative model capabilities from millions of pairwise comparison votes.

Overview

The Elo rating system is a method for calculating the relative skill levels of players (or in this case, AI models) in zero-sum games. Originally developed for chess, it has been adapted to rank AI language models based on head-to-head comparisons from user votes.

Key Features

• High-performance implementation: NumPy + Numba JIT + parallel processing for optimal speed
• Real voting data analysis: Uses actual Chatbot Arena voting data with millions of pairwise comparisons
• Win rate prediction: Calculates expected win probabilities between any two models
• Historical tracking: Builds time-series snapshots showing ranking evolution
• Interactive visualizations: Multiple visualization types including animated bar chart races
• Scalable: Efficiently handles 2GB datasets with hundreds of thousands of matches

Mathematical Foundation

The Elo system is based on the Bradley-Terry model, which models the probability that model A beats model B as:

P(A beats B) = 1 / (1 + 10^((R_B - R_A) / 400))

After each match, ratings are updated using:

R_A_new = R_A + K * (S_A - E_A)

Where: - R_A is the current rating of model A - K is the learning rate (K-factor) - S_A is the actual score (1 for win, 0 for loss, 0.5 for tie) - E_A is the expected score (predicted win probability)

Requirements

• Disk Space: At least 3GB free (2GB for data file, 1GB for processing)
• RAM: 4GB+ recommended for full dataset analysis
• Internet: Stable connection for ~2GB download
• Python: 3.8+

Installation

cd projects/week6/elo-leaderboard
pip install -r requirements.txt

命令行工具 / Command-Line Interface (cli.py)

cli.py 是本实验统一的 argparse 命令行入口（中文 --help），把整条流水线拆成子命令： 对战 (battle) -> 计算评分 (elo) -> 展示排行榜 (leaderboard)，并提供 pipeline 一步到位。

python cli.py --help              # 查看全部子命令
python cli.py battle --help       # 查看某个子命令的参数

# 默认离线端到端演示：模拟对战 -> 在线 Elo -> 最终排行榜表格（无需任何数据/API）
python cli.py                     # 等价于 python cli.py pipeline

子命令

子命令	作用	关键参数
battle	生成两两对战结果	--source {simulate,arena,llm}、--num-battles、--tie-prob、--seed、--sample、--output
elo	从对战结果计算评分	--method {online-elo,bradley-terry}、--k、--bootstrap、--input、--output
leaderboard	渲染最终排行榜表格	--input（对战或评分文件）、--method、--bootstrap、--top-n
pipeline	一步跑完 对战 -> Elo -> 排行榜	上述参数的并集

三种对战来源（--source）

• simulate（默认，纯离线）：从已知的潜在实力分模拟对战。因为真值已知，可用来校验恢复出的排行榜排序是否正确；--tie-prob 控制平局比例，用于演练平局处理。
• arena（离线）：加载真实 Chatbot Arena 投票数据（默认 arena_data.json，约 2GB），可用 --sample N 抽样。
• llm（需 API）：用 LLM 做配对评判，并内置位置偏差消除——每对交换顺序各评一次，两次判决一致才计胜负、否则记为平局（对应书中 6.4 位置偏差讨论）。仅此来源需要 LLM API Key。

两种评判后端（--judge-backend {anthropic,openrouter,auto}，默认 auto）： - anthropic：官方 anthropic SDK，用 ANTHROPIC_API_KEY。 - openrouter：OpenAI 兼容 SDK 指向 https://openrouter.ai/api/v1，用 OPENROUTER_API_KEY。内部 Claude 名字会自动映射为 OpenRouter id（claude-opus-4-8 → anthropic/claude-opus-4.8，claude-haiku-4-5 → anthropic/claude-haiku-4.5）；已含 / 的 id（如 openai/gpt-5.6-luna）原样透传。当直连 Anthropic key 缺失或失效时用它兜底。 - auto（默认）：有 ANTHROPIC_API_KEY 走 anthropic，否则回退 openrouter。注意 auto 只看 key 是否存在、不校验有效性；若 ANTHROPIC_API_KEY 存在但已失效，请显式 --judge-backend openrouter。

位置偏差消除与 A/B/tie 解析逻辑与后端无关，两条路径完全一致。

分步示例

# 1) 模拟 5000 场对战（含 10% 平局）
python cli.py battle --source simulate --num-battles 5000 --output battles.json

# 2) 用官方 Bradley-Terry MLE + 100 轮 bootstrap 置信区间计算评分
python cli.py elo --input battles.json --method bradley-terry --bootstrap 100

# 3) 展示前 20 名排行榜（也可直接读评分文件）
python cli.py leaderboard --input battles.json --top-n 20

# 用真实 Arena 数据抽样跑（离线）
python cli.py pipeline --source arena --arena-file arena_data.json --sample 50000 --method bradley-terry --bootstrap 100

# LLM 评判对战（需要 API Key）——官方 Anthropic
export ANTHROPIC_API_KEY=sk-...
python cli.py battle --source llm --candidate-models claude-opus-4-8 claude-haiku-4-5

# LLM 评判对战——通过 OpenRouter 兜底（直连 Anthropic key 缺失/失效时）
export OPENROUTER_API_KEY=sk-or-...
python cli.py battle --source llm --judge-backend openrouter \
  --judge-model claude-opus-4-8 \
  --candidate-models anthropic/claude-haiku-4.5 openai/gpt-5.6-luna

模拟来源会同时打印真值潜在实力，方便和恢复出的排行榜对照；在线 Elo 与 Bradley-Terry 两种方法都应恢复出与真值一致的排名（分值不必精确对齐，见下文说明）。

Quick Start

The project implements two ranking methods following official Chatbot Arena:

1. Bradley-Terry Model (Default - Recommended)

python main.py
# or explicitly:
python main.py bradley-terry

Use this for: Official leaderboard, stable rankings, production use

Key features: - ✅ Official Chatbot Arena method - ✅ Uses sklearn LogisticRegression for Maximum Likelihood Estimation - ✅ Order-independent (processes all matches simultaneously) - ✅ Includes 95% confidence intervals via bootstrap (100 samples) - ✅ More stable and reliable rankings

Processing time: ~2-3 minutes (including bootstrap)

2. Online Elo (K=4)

python main.py online-elo

Use this for: Understanding Elo mechanics, educational purposes, faster computation

Key features: - ✅ K-factor = 4 (official value used by Chatbot Arena) - ✅ Simple sequential rating updates - ✅ Order-dependent (processes matches chronologically) - ✅ Faster computation (~30 seconds) - ⚠️ Less stable, can vary based on match order

Method Comparison

Feature	Bradley-Terry	Online Elo
Stability	High (MLE fit)	Medium (sequential)
Order dependence	None	High
Confidence intervals	Yes (bootstrap)	No
Speed	Slower (~3 min)	Faster (~30 sec)
Official method	✅ Yes	For comparison only
Recommended	✅ Production	Educational

What Both Methods Do

1. Download Chatbot Arena voting data (~2GB, 5-15 minutes depending on connection)
2. Apply official filters:
3. Anonymous votes only (blind evaluation)
4. Deduplication (removes top 0.1% redundant prompts)
5. Compute model ratings using selected method
6. Calculate predicted win rates between all model pairs
7. Generate visualizations:
8. leaderboard.png - Top 20 models ranked by rating
9. rating_distribution.png - Rating histogram and statistics
10. win_rate_matrix.png - Predicted win rates (top 30 models)

Note: The initial data download is ~2GB and may take several minutes. A progress bar shows download status.

Quick Demo (Synthetic Data)

To quickly understand Elo mechanics without downloading 2GB:

python quickstart.py

This runs a small demo with synthetic matchups between GPT-4, Claude, Llama, and Gemini.

Benchmark

To compare both methods:

python benchmark.py

This shows performance and accuracy differences between online Elo and Bradley-Terry approaches.

Project Structure

elo-leaderboard/
├── cli.py                      # Unified argparse CLI (battle / elo / leaderboard / pipeline)
├── battle_simulator.py         # Offline synthetic pairwise-battle generator
├── llm_judge.py                # LLM-as-judge battles with position-bias mitigation (needs API)
├── main.py                     # Main analysis script
├── optimized_elo.py            # NumPy + Numba Elo rating system
├── parallel_processing.py      # Multi-core parallel processing utilities
├── data_loader.py              # Data download and preprocessing
├── leaderboard.py              # Leaderboard calculation and analysis
├── visualization.py            # Static and interactive visualizations
├── animation.py                # Animated bar chart race generator
├── benchmark.py                # Performance benchmark tool
├── quickstart.py               # Quick demo with synthetic data
├── elo_rating.py               # Reference implementation (for comparison)
├── test_elo.py                 # Unit tests
├── requirements.txt            # Python dependencies
└── README.md                   # This file

Usage Examples

Building Elo Leaderboard

from optimized_elo import build_leaderboard_optimized

# Build Elo leaderboard from DataFrame
elo = build_leaderboard_optimized(
    df,  # DataFrame with columns: model_a, model_b, winner
    initial_rating=1000.0,
    k_factor=32.0,
    show_progress=True
)

# Get leaderboard
leaderboard = elo.get_leaderboard()
for rank, (model, rating, matches, wins) in enumerate(leaderboard[:10], 1):
    win_rate = wins / matches * 100 if matches > 0 else 0
    print(f"{rank}. {model}: {rating:.1f} ({matches} matches, {win_rate:.1f}% win rate)")

Loading and Filtering Data

from data_loader import load_arena_data, filter_data

# Load data
df = load_arena_data("arena_data.json")

# Filter for blind votes only (reduces bias)
df_filtered = filter_data(
    df,
    anony_only=True,      # Only anonymous votes
    language="English",    # Specific language
    min_turn=1            # Minimum conversation turn
)

Building Historical Leaderboards

from data_loader import get_time_slices
from leaderboard import build_historical_leaderboards, get_rating_history

# Create weekly time slices
time_slices = get_time_slices(df, interval='W')

# Build leaderboard for each time point
historical_leaderboards = build_historical_leaderboards(
    df, time_slices, initial_rating=1000.0, k_factor=32.0
)

# Get rating history DataFrame
history_df = get_rating_history(historical_leaderboards)

Creating Visualizations

from visualization import (
    plot_leaderboard,
    plot_win_rate_matrix,
    plot_rating_history,
    create_interactive_leaderboard
)

# Static leaderboard chart
plot_leaderboard(leaderboard, top_n=20, save_path="leaderboard.png")

# Win rate heatmap
win_rate_df = calculate_win_rate_matrix_from_data(df)
plot_win_rate_matrix(win_rate_df, top_n=15, save_path="matrix.png")

# Rating evolution
plot_rating_history(history_df, models=["gpt-4", "claude-v1"], 
                   save_path="history.png")

# Interactive chart
fig = create_interactive_leaderboard(history_df, top_n=15)
fig.write_html("interactive.html")

Creating Animated Bar Chart Race

from animation import create_simple_animation

# Generate animated HTML
animation_file = create_simple_animation(
    history_df,
    output_path="animation.html",
    top_n=15
)

# Open animation.html in browser to view

Output Files

After running main.py, the following files are generated:

Static Images (PNG)

• leaderboard.png - Current top 20 models ranked by Elo rating
• rating_distribution.png - Histogram and box plot of rating distribution
• win_rate_matrix.png - Heatmap showing pairwise win rates
• rating_history.png - Line chart showing rating evolution over time

Interactive Visualizations (HTML)

• interactive_rating_evolution.html - Interactive chart with zoom/pan
• interactive_rank_evolution.html - Interactive rank tracking
• leaderboard_animation.html - Animated bar chart race showing ranking evolution

Key Parameters

Elo System Parameters (Online Elo Method)

• initial_rating (default: 1000.0): Starting rating for all models
• k_factor (default: 4.0): Learning rate controlling update magnitude
• Official Chatbot Arena uses K=4 for stability
• Higher K-factor (e.g., 32): More volatile, faster adaptation to new data
• Lower K-factor (e.g., 4): More stable, less influenced by recent matches

Bradley-Terry Parameters

• SCALE (400): Elo scale parameter - determines rating point interpretation
• BASE (10): Base for logistic function - standard for Elo calculations
• INIT_RATING (1000): Initial rating for all models
• bootstrap_rounds (100): Number of bootstrap samples for confidence intervals

Time Slice Intervals

For historical analysis, you can adjust the time granularity: - 'D' - Daily snapshots - 'W' - Weekly snapshots (recommended) - 'M' - Monthly snapshots

Visualization Parameters

• top_n: Number of top models to display (10-20 recommended)
• Animation speed: Adjustable in the HTML interface (1x to 10x)

Data Format

The Chatbot Arena data includes the following fields:

• model_a: Identifier for first model
• model_b: Identifier for second model
• winner: Match outcome ('model_a', 'model_b', or 'tie')
• tstamp: Unix timestamp of the vote
• judge: User who made the vote
• turn: Conversation turn number
• anony: Whether vote was anonymous/blind
• language: Language of the conversation

Validation

The implementation validates the Elo predictions against empirical win rates:

from leaderboard import compare_win_rates

# Compare predicted vs actual win rates
comparison = compare_win_rates(elo_system, empirical_win_rates)
mean_error = comparison['error'].mean()
print(f"Mean Absolute Error: {mean_error:.4f}")

A low MAE (< 0.05) indicates the Elo model fits the data well.

Analysis Insights

The project helps identify:

1. Current Rankings: Which models are currently strongest
2. Rating Trends: How model performance evolves over time
3. Breakthrough Moments: When new models enter or shake up rankings
4. Competitive Dynamics: Which models are closely matched
5. Long-term Trajectories: Models in ascent vs. decline
6. Rating Stability: Volatility in model performance

Performance Architecture

The implementation is designed for high performance on large datasets (2GB+).

Core Optimizations

1. NumPy + Numba JIT Compilation

Uses NumPy arrays and Numba's just-in-time compilation: - NumPy arrays for O(1) integer indexing (vs O(n) dictionary lookups) - Numba JIT compiles hot loops to machine code (50-100x speedup) - Pre-allocated arrays eliminate dynamic memory allocation overhead - Integer indices instead of string model names for cache-friendly access

2. Multi-Core Parallel Processing

Parallelizes independent operations across all CPU cores: - Historical analysis: Each time slice processed independently - Win rate matrices: Model pairs computed in parallel chunks - Data filtering: DataFrame operations distributed across cores

from parallel_processing import build_historical_leaderboards_parallel

# Automatically uses all available CPU cores
historical_lb = build_historical_leaderboards_parallel(
    df, time_slices, n_jobs=-1
)

3. Memory Optimization

Reduces memory footprint through intelligent data types: - Downcasts numeric types (int64 → int32, float64 → float32) - Converts repetitive strings to categorical types - Achieves 30-50% memory reduction

from parallel_processing import optimize_dataframe

df = optimize_dataframe(df)  # Automatic memory optimization

Performance Characteristics

On typical hardware (4-8 core CPU) with the full 2GB dataset:

Component	Technique	Impact
Elo Computation	NumPy + Numba JIT	50-100x faster
Historical Analysis	Multi-core parallel	4-8x faster
Win Rate Matrix	Parallel processing	4-8x faster
Memory Usage	Type optimization	30-50% reduction
Overall	Combined	~10-15x speedup

Processing time: 1-2 minutes for full dataset (hundreds of thousands of matches)

Advanced Usage

Custom Analysis

The modular design allows for flexible customization:

Focused Analysis

from optimized_elo import build_leaderboard_optimized
from data_loader import load_arena_data, filter_data

df = load_arena_data("arena_data.json")

# Analyze only recent data
df_recent = filter_data(df, min_date="2024-01-01")
elo_recent = build_leaderboard_optimized(df_recent)

# Analyze specific model family
gpt_models = [m for m in df['model_a'].unique() if 'gpt' in m.lower()]
df_gpt = df[df['model_a'].isin(gpt_models) & df['model_b'].isin(gpt_models)]
elo_gpt = build_leaderboard_optimized(df_gpt)

Export Results

import pandas as pd

# Export leaderboard to CSV
lb_df = pd.DataFrame(leaderboard, columns=['model', 'rating', 'matches', 'wins'])
lb_df.to_csv('leaderboard.csv', index=False)

# Export rating history
history_df.to_csv('rating_history.csv', index=False)

Troubleshooting

Data Download Issues

If automatic download fails: 1. Manually download from: https://storage.googleapis.com/arena_external_data/public/clean_battle_20240814_public.json 2. Save as arena_data.json in the project directory 3. Run python main.py again

Memory Issues

The dataset is large (~2GB, hundreds of thousands of battles). If you encounter memory issues on systems with limited RAM:

from data_loader import load_arena_data, filter_data, get_time_slices
from optimized_elo import build_leaderboard_optimized

# Load and immediately filter to reduce memory usage
df = load_arena_data("arena_data.json")

# Filter to recent data only
df_filtered = filter_data(df, min_date="2024-01-01", anony_only=True)

# Use monthly instead of weekly intervals for historical analysis
time_slices = get_time_slices(df_filtered, interval='M')  # vs 'W' for weekly

# Analyze with smaller top_n for visualizations
elo = build_leaderboard_optimized(df_filtered)

The built-in memory optimization reduces footprint by 30-50%, but very large analyses may still require 4GB+ RAM.

Visualization Issues

• Ensure matplotlib, seaborn, and plotly are installed: pip install -r requirements.txt
• For HTML animations, use a modern web browser (Chrome, Firefox, Safari, Edge)
• If plots don't display in Jupyter, use %matplotlib inline or save to file

References

• Chatbot Arena: https://chat.lmsys.org/
• Elo Rating System: https://en.wikipedia.org/wiki/Elo_rating_system
• Bradley-Terry Model: https://en.wikipedia.org/wiki/Bradley–Terry_model
• LMSYS Paper: "Judging LLM-as-a-Judge with MT-Bench and Chatbot Arena"

Learning Objectives

This experiment demonstrates:

1. Statistical Modeling: How pairwise comparisons reveal relative abilities
2. Online Learning: Incremental rating updates as new data arrives
3. Probabilistic Prediction: Converting rating differences to win probabilities
4. Data Visualization: Effective techniques for showing temporal dynamics
5. Model Evaluation: Alternative to traditional benchmark approaches

Technical Details

Why Elo Computation is Hard to Parallelize

Elo rating computation is inherently sequential because each match's rating update depends on the current ratings, which were modified by all previous matches. This is why we can't simply split matches into chunks and process them independently.

However, we can still achieve significant speedups through:

1. Algorithmic optimization: NumPy arrays + Numba JIT
2. Parallelizing independent operations: Historical analysis, win rate matrices
3. Memory efficiency: Better cache utilization
4. Data structure optimization: Integer indexing, pre-allocation

Numba JIT Compilation

The core Elo update loop is compiled to machine code using Numba:

@jit(nopython=True)
def process_elo_updates_vectorized(ratings, model_a_indices, model_b_indices, 
                                   outcomes, k_factor, match_counts, win_counts):
    for i in range(len(model_a_indices)):
        # This loop runs at C speed, not Python speed
        # Typical speedup: 50-100x over pure Python
        ...

Memory Layout

Using NumPy arrays with proper data types: - ratings: float64 array (8 bytes per model) - match_counts: int32 array (4 bytes per model) - model_indices: int32 array (4 bytes per match)

For 500 models and 500K matches: ~10 MB vs ~500 MB for dictionaries.

Extensions

Potential enhancements:

• Implement Glicko or Glicko-2 rating systems (account for rating uncertainty)
• Add confidence intervals for rating estimates
• Analyze rating by language or task type
• Compare with other ranking methods (e.g., TrueSkill, PageRank)
• Implement time-decay for older matches
• Add statistical significance testing
• Build prediction model for future rankings
• GPU acceleration using CuPy for even larger datasets
• Distributed processing using Dask for multi-machine scaling

License

This project is part of the AI Agent practical training course materials.

Contact

For questions or issues, please refer to the course materials or discussion forums.

源代码

animation.py

"""
Create animated bar chart race showing leaderboard evolution over time
"""
import pandas as pd
import numpy as np
from typing import List, Tuple
import json
import os


def prepare_animation_data(history_df: pd.DataFrame, top_n: int = 15) -> dict:
    """
    Prepare data for D3.js bar chart race animation.

    Args:
        history_df: DataFrame with columns: date, model, rating, rank
        top_n: Number of top models to show at each time point

    Returns:
        Dictionary with animation data
    """
    # Get all unique dates
    dates = sorted(history_df['date'].unique())

    # For each date, get top N models
    frames = []
    for date in dates:
        date_data = history_df[history_df['date'] == date].nlargest(top_n, 'rating')

        frame = {
            'date': date.strftime('%Y-%m-%d'),
            'timestamp': int(date.timestamp()),
            'models': []
        }

        for rank, row in enumerate(date_data.itertuples(), 1):
            frame['models'].append({
                'rank': rank,
                'name': row.model,
                'rating': float(row.rating),
                'matches': int(row.matches),
                'wins': int(row.wins)
            })

        frames.append(frame)

    animation_data = {
        'frames': frames,
        'total_frames': len(frames),
        'top_n': top_n,
        'start_date': dates[0].strftime('%Y-%m-%d'),
        'end_date': dates[-1].strftime('%Y-%m-%d')
    }

    return animation_data


def generate_html_animation(animation_data: dict, output_path: str = "leaderboard_animation.html"):
    """
    Generate standalone HTML file with D3.js bar chart race animation.

    Args:
        animation_data: Dictionary from prepare_animation_data
        output_path: Path to save HTML file
    """
    html_template = """<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Model Leaderboard Evolution</title>
    <script src="https://d3js.org/d3.v7.min.js"></script>
    <style>
        body {
            font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
            margin: 0;
            padding: 20px;
            background: #f5f5f5;
        }

        #container {
            max-width: 1200px;
            margin: 0 auto;
            background: white;
            padding: 30px;
            border-radius: 10px;
            box-shadow: 0 2px 10px rgba(0,0,0,0.1);
        }

        h1 {
            text-align: center;
            color: #333;
            margin-bottom: 10px;
        }

        #date-display {
            text-align: center;
            font-size: 24px;
            font-weight: bold;
            color: #666;
            margin-bottom: 20px;
        }

        #chart {
            margin: 20px 0;
        }

        .bar {
            fill: steelblue;
            cursor: pointer;
            transition: fill 0.3s;
        }

        .bar:hover {
            fill: #4682b4;
        }

        .bar-label {
            font-size: 14px;
            fill: white;
            font-weight: bold;
        }

        .bar-value {
            font-size: 12px;
            fill: #333;
        }

        .rank-label {
            font-size: 18px;
            fill: #666;
            font-weight: bold;
        }

        #controls {
            text-align: center;
            margin-top: 30px;
        }

        button {
            padding: 10px 20px;
            margin: 0 5px;
            font-size: 16px;
            cursor: pointer;
            border: none;
            border-radius: 5px;
            background: #4CAF50;
            color: white;
            transition: background 0.3s;
        }

        button:hover {
            background: #45a049;
        }

        button:disabled {
            background: #ccc;
            cursor: not-allowed;
        }

        #progress-bar {
            width: 100%;
            height: 5px;
            background: #e0e0e0;
            margin-top: 20px;
            border-radius: 3px;
            overflow: hidden;
        }

        #progress {
            height: 100%;
            background: #4CAF50;
            width: 0%;
            transition: width 0.5s;
        }

        #speed-control {
            margin-top: 20px;
            text-align: center;
        }

        #speed-slider {
            width: 300px;
            margin: 0 10px;
        }

        .info-box {
            background: #f9f9f9;
            padding: 15px;
            border-radius: 5px;
            margin-top: 20px;
            font-size: 14px;
            color: #666;
        }
    </style>
</head>
<body>
    <div id="container">
        <h1>🏆 Model Leaderboard Evolution</h1>
        <div id="date-display">Loading...</div>
        <div id="chart"></div>
        <div id="controls">
            <button id="play-btn">▶ Play</button>
            <button id="pause-btn" disabled>⏸ Pause</button>
            <button id="reset-btn">↺ Reset</button>
        </div>
        <div id="progress-bar">
            <div id="progress"></div>
        </div>
        <div id="speed-control">
            <label>Speed: </label>
            <input type="range" id="speed-slider" min="1" max="10" value="5">
            <span id="speed-value">5x</span>
        </div>
        <div class="info-box">
            <strong>About:</strong> This animation shows the evolution of model rankings based on Elo ratings 
            calculated from Chatbot Arena voting data. Each frame represents a snapshot in time, 
            with models ranked by their current Elo rating. Bars show the rating value, 
            and the animation reveals how models compete and evolve over time.
        </div>
    </div>

    <script>
        const data = """ + json.dumps(animation_data, indent=2) + """;

        // Configuration
        const margin = {top: 20, right: 100, bottom: 40, left: 50};
        const width = 1100 - margin.left - margin.right;
        const height = 600 - margin.top - margin.bottom;
        const barHeight = height / data.top_n - 5;

        // Create SVG
        const svg = d3.select("#chart")
            .append("svg")
            .attr("width", width + margin.left + margin.right)
            .attr("height", height + margin.top + margin.bottom)
            .append("g")
            .attr("transform", `translate(${margin.left},${margin.top})`);

        // Scales
        const xScale = d3.scaleLinear()
            .domain([0, d3.max(data.frames.flatMap(f => f.models.map(m => m.rating)))])
            .range([0, width - 200]);

        // Color scale
        const colorScale = d3.scaleOrdinal(d3.schemeCategory10);

        // Animation state
        let currentFrame = 0;
        let isPlaying = false;
        let animationInterval = null;
        let animationSpeed = 500; // milliseconds per frame

        // Update speed based on slider
        d3.select("#speed-slider").on("input", function() {
            const speed = +this.value;
            animationSpeed = 1000 / speed;
            d3.select("#speed-value").text(`${speed}x`);
            if (isPlaying) {
                stopAnimation();
                startAnimation();
            }
        });

        function updateChart(frameIndex) {
            const frame = data.frames[frameIndex];

            // Update date display
            d3.select("#date-display").text(frame.date);

            // Update progress bar
            const progress = ((frameIndex + 1) / data.total_frames) * 100;
            d3.select("#progress").style("width", `${progress}%`);

            // Update max value for scale
            const maxRating = d3.max(frame.models, d => d.rating);
            xScale.domain([0, maxRating * 1.1]);

            // Bind data
            const bars = svg.selectAll(".bar-group")
                .data(frame.models, d => d.name);

            // Remove old bars
            bars.exit()
                .transition()
                .duration(animationSpeed * 0.8)
                .style("opacity", 0)
                .remove();

            // Add new bars
            const enter = bars.enter()
                .append("g")
                .attr("class", "bar-group")
                .style("opacity", 0);

            enter.append("rect")
                .attr("class", "bar")
                .attr("height", barHeight);

            enter.append("text")
                .attr("class", "bar-label")
                .attr("x", 10)
                .attr("y", barHeight / 2)
                .attr("dy", "0.35em");

            enter.append("text")
                .attr("class", "bar-value")
                .attr("y", barHeight / 2)
                .attr("dy", "0.35em");

            enter.append("text")
                .attr("class", "rank-label")
                .attr("x", -40)
                .attr("y", barHeight / 2)
                .attr("dy", "0.35em")
                .attr("text-anchor", "middle");

            // Update all bars
            const merged = enter.merge(bars);

            merged.transition()
                .duration(animationSpeed * 0.8)
                .style("opacity", 1)
                .attr("transform", (d, i) => `translate(0,${i * (barHeight + 5)})`);

            merged.select(".bar")
                .transition()
                .duration(animationSpeed * 0.8)
                .attr("width", d => xScale(d.rating))
                .attr("fill", d => colorScale(d.name));

            merged.select(".bar-label")
                .text(d => d.name);

            merged.select(".bar-value")
                .transition()
                .duration(animationSpeed * 0.8)
                .attr("x", d => xScale(d.rating) + 10)
                .text(d => `${Math.round(d.rating)} (${d.matches} matches)`);

            merged.select(".rank-label")
                .text(d => `#${d.rank}`);
        }

        function startAnimation() {
            if (currentFrame >= data.total_frames - 1) {
                currentFrame = 0;
            }

            isPlaying = true;
            d3.select("#play-btn").property("disabled", true);
            d3.select("#pause-btn").property("disabled", false);

            animationInterval = setInterval(() => {
                updateChart(currentFrame);
                currentFrame++;

                if (currentFrame >= data.total_frames) {
                    stopAnimation();
                    currentFrame = data.total_frames - 1;
                }
            }, animationSpeed);
        }

        function stopAnimation() {
            isPlaying = false;
            d3.select("#play-btn").property("disabled", false);
            d3.select("#pause-btn").property("disabled", true);

            if (animationInterval) {
                clearInterval(animationInterval);
                animationInterval = null;
            }
        }

        function resetAnimation() {
            stopAnimation();
            currentFrame = 0;
            updateChart(currentFrame);
            d3.select("#progress").style("width", "0%");
        }

        // Button handlers
        d3.select("#play-btn").on("click", startAnimation);
        d3.select("#pause-btn").on("click", stopAnimation);
        d3.select("#reset-btn").on("click", resetAnimation);

        // Initialize with first frame
        updateChart(0);

        // Auto-play on load
        setTimeout(startAnimation, 1000);
    </script>
</body>
</html>"""

    # Write to file
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(html_template)

    print(f"Generated animation HTML at: {output_path}")
    print(f"Open the file in a web browser to view the animation.")


def create_simple_animation(history_df: pd.DataFrame, output_path: str = "leaderboard_animation.html", top_n: int = 15):
    """
    Convenience function to create animation in one step.

    Args:
        history_df: DataFrame with rating history
        output_path: Path to save HTML file
        top_n: Number of top models to show
    """
    print("Preparing animation data...")
    animation_data = prepare_animation_data(history_df, top_n)

    print(f"Generating HTML animation with {animation_data['total_frames']} frames...")
    generate_html_animation(animation_data, output_path)

    return output_path

battle_simulator.py

"""
Synthetic pairwise battle generator (offline).

Generates head-to-head "battle" outcomes from a set of known latent skill
scores, so the whole battles -> Elo -> leaderboard pipeline can be demonstrated
end-to-end without downloading the 2GB Chatbot Arena dataset or calling any API.

Because the ground-truth skills are known, the recovered Elo leaderboard can be
checked against them: the ranking should match, which validates the
implementation. Ties are produced with a configurable probability to exercise
the tie-handling paths in both the online Elo and Bradley-Terry code.
"""
import random
from typing import Dict, List, Optional


# Default roster with plausible latent skills (in Elo points). The exact numbers
# are only used to *generate* battles; the experiment then tries to recover them.
DEFAULT_TRUE_SKILLS: Dict[str, float] = {
    "gpt-4": 1250.0,
    "claude-3-opus": 1225.0,
    "gemini-1.5-pro": 1180.0,
    "llama-3-70b": 1120.0,
    "mixtral-8x7b": 1075.0,
    "gpt-3.5-turbo": 1035.0,
    "llama-2-13b": 980.0,
    "vicuna-13b": 935.0,
}


def expected_score(rating_a: float, rating_b: float,
                   base: float = 10.0, scale: float = 400.0) -> float:
    """Bradley-Terry / Elo win probability of A against B."""
    return 1.0 / (1.0 + base ** ((rating_b - rating_a) / scale))


def simulate_battles(true_skills: Dict[str, float],
                     num_battles: int,
                     tie_prob: float = 0.1,
                     seed: Optional[int] = None) -> List[dict]:
    """
    Simulate `num_battles` random pairwise battles.

    For each battle two distinct models are drawn uniformly at random. With
    probability `tie_prob` the outcome is a tie; otherwise the winner is sampled
    according to the Bradley-Terry win probability implied by the latent skills
    (so upsets happen, but stronger models win more often).

    Args:
        true_skills: Mapping of model name -> latent skill (Elo points).
        num_battles: Number of battles to generate.
        tie_prob: Probability that a battle ends in a tie.
        seed: Optional RNG seed for reproducibility.

    Returns:
        List of dicts with keys 'model_a', 'model_b', 'winner'
        (winner in {'model_a', 'model_b', 'tie'}), matching the Chatbot Arena
        schema consumed by the Elo / Bradley-Terry code.
    """
    if len(true_skills) < 2:
        raise ValueError("Need at least 2 models to simulate battles")

    rng = random.Random(seed)
    models = list(true_skills.keys())
    battles: List[dict] = []

    for _ in range(num_battles):
        model_a, model_b = rng.sample(models, 2)
        if rng.random() < tie_prob:
            winner = "tie"
        elif rng.random() < expected_score(true_skills[model_a], true_skills[model_b]):
            winner = "model_a"
        else:
            winner = "model_b"
        battles.append({"model_a": model_a, "model_b": model_b, "winner": winner})

    return battles

benchmark.py

"""
Benchmark script to compare performance of different Elo implementations
"""
import time
import pandas as pd
import numpy as np
from elo_rating import EloRatingSystem
from optimized_elo import build_leaderboard_optimized
from data_loader import load_arena_data, filter_data


def benchmark_basic_elo(df: pd.DataFrame) -> float:
    """Benchmark the basic Elo implementation."""
    print("\n" + "="*80)
    print("Benchmarking Basic Elo Implementation (Python dict)")
    print("="*80)

    start_time = time.time()

    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    for _, row in df.iterrows():
        elo.update_ratings(row['model_a'], row['model_b'], row['winner'])

    end_time = time.time()
    elapsed = end_time - start_time

    leaderboard = elo.get_leaderboard()

    print(f"✓ Processed {len(df)} matches in {elapsed:.2f} seconds")
    print(f"  Speed: {len(df)/elapsed:.0f} matches/second")
    print(f"  Top 3 models: {[m[0] for m in leaderboard[:3]]}")

    return elapsed


def benchmark_optimized_elo(df: pd.DataFrame) -> float:
    """Benchmark the NumPy + Numba optimized implementation."""
    print("\n" + "="*80)
    print("Benchmarking Optimized Elo Implementation (NumPy + Numba JIT)")
    print("="*80)

    start_time = time.time()

    elo = build_leaderboard_optimized(
        df,
        initial_rating=1000.0,
        k_factor=32.0,
        show_progress=False
    )

    end_time = time.time()
    elapsed = end_time - start_time

    leaderboard = elo.get_leaderboard()

    print(f"✓ Processed {len(df)} matches in {elapsed:.2f} seconds")
    print(f"  Speed: {len(df)/elapsed:.0f} matches/second")
    print(f"  Top 3 models: {[m[0] for m in leaderboard[:3]]}")

    return elapsed


def main():
    """Run benchmark comparison."""
    print("="*80)
    print("ELO RATING COMPUTATION BENCHMARK")
    print("="*80)
    print("\nThis benchmark compares the performance of different Elo implementations")
    print("on Chatbot Arena voting data.\n")

    # Load data
    print("Loading data...")
    try:
        df = load_arena_data("arena_data.json")
    except FileNotFoundError:
        print("Error: arena_data.json not found. Please run main.py first to download the data.")
        return

    # Filter for blind votes
    print("Filtering data...")
    df_filtered = filter_data(df, anony_only=True, min_turn=1)

    # Use a subset for quick benchmarking (can change to full dataset)
    sample_size = 50000
    if len(df_filtered) > sample_size:
        print(f"\nUsing a sample of {sample_size} matches for benchmarking")
        print("(To benchmark on full dataset, set sample_size = len(df_filtered))")
        df_sample = df_filtered.head(sample_size).copy()
    else:
        df_sample = df_filtered.copy()

    print(f"\nBenchmark dataset: {len(df_sample)} matches")
    print(f"Unique models: {len(set(df_sample['model_a'].unique()) | set(df_sample['model_b'].unique()))}")

    # Warm up Numba JIT (first run compiles the functions)
    print("\n" + "-"*80)
    print("Warming up Numba JIT compiler (first run)...")
    print("-"*80)
    df_tiny = df_sample.head(1000)
    build_leaderboard_optimized(df_tiny, show_progress=False)
    print("✓ JIT compilation complete")

    # Run benchmarks
    time_basic = benchmark_basic_elo(df_sample)
    time_optimized = benchmark_optimized_elo(df_sample)

    # Results summary
    print("\n" + "="*80)
    print("BENCHMARK RESULTS")
    print("="*80)

    speedup = time_basic / time_optimized if time_optimized > 0 else 0

    print(f"\nBasic Implementation:     {time_basic:8.2f} seconds")
    print(f"Optimized Implementation: {time_optimized:8.2f} seconds")
    print(f"\nSpeedup: {speedup:.1f}x faster")
    print(f"Time saved: {time_basic - time_optimized:.2f} seconds ({(1-time_optimized/time_basic)*100:.1f}% reduction)")

    # Extrapolate to full dataset
    if len(df_sample) < len(df_filtered):
        full_time_basic = time_basic * (len(df_filtered) / len(df_sample))
        full_time_optimized = time_optimized * (len(df_filtered) / len(df_sample))

        print(f"\nExtrapolated times for full dataset ({len(df_filtered)} matches):")
        print(f"  Basic:     ~{full_time_basic/60:.1f} minutes")
        print(f"  Optimized: ~{full_time_optimized/60:.1f} minutes")
        print(f"  Time saved: ~{(full_time_basic - full_time_optimized)/60:.1f} minutes")

    print("\n" + "="*80)
    print("\nOptimization Techniques Applied:")
    print("  • NumPy arrays instead of Python dicts (O(1) integer indexing)")
    print("  • Numba JIT compilation (compiles hot loops to machine code)")
    print("  • Pre-allocated arrays (no dynamic memory allocation)")
    print("  • Integer model indices (no string lookups)")
    print("  • Vectorized operations where possible")
    print("\nFor the full optimized pipeline with parallel processing, run main_optimized.py")
    print("="*80 + "\n")


if __name__ == "__main__":
    main()

bradley_terry.py

"""
Bradley-Terry Model Implementation
Official Chatbot Arena leaderboard calculation method
"""
import numpy as np
import pandas as pd
import math
from typing import Dict
from sklearn.linear_model import LogisticRegression


def compute_mle_elo(df: pd.DataFrame, 
                    SCALE: int = 400, 
                    BASE: int = 10, 
                    INIT_RATING: int = 1000,
                    calibration_model: str = None,
                    calibration_rating: int = None) -> pd.Series:
    """
    Compute Elo ratings using Bradley-Terry model with Maximum Likelihood Estimation.

    This is the official method used by Chatbot Arena for their leaderboard.
    It uses sklearn's LogisticRegression to fit a Bradley-Terry model.

    Args:
        df: DataFrame with columns 'model_a', 'model_b', 'winner'
        SCALE: Elo scale parameter (default 400)
        BASE: Base for logistic function (default 10)
        INIT_RATING: Initial rating (default 1000)
        calibration_model: Model name to calibrate ratings to
        calibration_rating: Target rating for calibration model

    Returns:
        Series of Elo ratings indexed by model name
    """
    # Create pivot tables for wins
    ptbl_a_win = pd.pivot_table(
        df[df["winner"] == "model_a"],
        index="model_a",
        columns="model_b",
        aggfunc="size",
        fill_value=0,
        observed=False
    )

    # Handle ties (including "tie (bothbad)")
    if sum(df["winner"].isin(["tie", "tie (bothbad)"])) == 0:
        ptbl_tie = pd.DataFrame(0, index=ptbl_a_win.index, columns=ptbl_a_win.columns)
    else:
        ptbl_tie = pd.pivot_table(
            df[df["winner"].isin(["tie", "tie (bothbad)"])],
            index="model_a",
            columns="model_b",
            aggfunc="size",
            fill_value=0,
            observed=False
        )
        ptbl_tie = ptbl_tie + ptbl_tie.T

    ptbl_b_win = pd.pivot_table(
        df[df["winner"] == "model_b"],
        index="model_a",
        columns="model_b",
        aggfunc="size",
        fill_value=0,
        observed=False
    )

    # Compute win matrix (A wins * 2 + B wins * 2 + ties)
    ptbl_win = ptbl_a_win * 2 + ptbl_b_win.T * 2 + ptbl_tie

    # Map models to indices
    models = pd.Series(np.arange(len(ptbl_win.index)), index=ptbl_win.index)

    p = len(models)
    X = np.zeros([p * (p - 1) * 2, p])
    Y = np.zeros(p * (p - 1) * 2)

    cur_row = 0
    sample_weights = []

    for m_a in ptbl_win.index:
        for m_b in ptbl_win.columns:
            if m_a == m_b:
                continue
            # Skip if nan
            if math.isnan(ptbl_win.loc[m_a, m_b]) or math.isnan(ptbl_win.loc[m_b, m_a]):
                continue

            X[cur_row, models[m_a]] = +math.log(BASE)
            X[cur_row, models[m_b]] = -math.log(BASE)
            Y[cur_row] = 1.0
            sample_weights.append(ptbl_win.loc[m_a, m_b])

            X[cur_row + 1, models[m_a]] = math.log(BASE)
            X[cur_row + 1, models[m_b]] = -math.log(BASE)
            Y[cur_row + 1] = 0.0
            sample_weights.append(ptbl_win.loc[m_b, m_a])
            cur_row += 2

    X = X[:cur_row]
    Y = Y[:cur_row]

    # Fit logistic regression
    lr = LogisticRegression(fit_intercept=False, penalty=None, tol=1e-6)
    lr.fit(X, Y, sample_weight=sample_weights)

    # Convert to Elo scores
    elo_scores = SCALE * lr.coef_[0] + INIT_RATING

    # Calibrate to reference model if provided
    if calibration_model and calibration_model in models.index:
        elo_scores += calibration_rating - elo_scores[models[calibration_model]]

    return pd.Series(elo_scores, index=models.index).sort_values(ascending=False)


def predict_win_rate(elo_ratings: Dict[str, float], 
                     SCALE: int = 400, 
                     BASE: int = 10) -> pd.DataFrame:
    """
    Predict win rates between all model pairs using Elo ratings.

    Args:
        elo_ratings: Dictionary of model names to Elo ratings
        SCALE: Elo scale parameter
        BASE: Base for logistic function

    Returns:
        DataFrame with predicted win rates (row vs column)
    """
    from collections import defaultdict

    names = sorted(list(elo_ratings.keys()))
    wins = defaultdict(lambda: defaultdict(lambda: 0))

    for a in names:
        for b in names:
            ea = 1 / (1 + BASE ** ((elo_ratings[b] - elo_ratings[a]) / SCALE))
            wins[a][b] = ea
            wins[b][a] = 1 - ea

    data = {
        a: [wins[a][b] if a != b else np.NAN for b in names]
        for a in names
    }

    df = pd.DataFrame(data, index=names)
    df.index.name = "model_a"
    df.columns.name = "model_b"
    return df.T


def get_bootstrap_result(battles: pd.DataFrame, 
                        func_compute_elo, 
                        num_round: int = 100) -> pd.DataFrame:
    """
    Compute bootstrap confidence intervals for Elo ratings.

    Args:
        battles: DataFrame with battle data
        func_compute_elo: Function to compute Elo ratings
        num_round: Number of bootstrap rounds

    Returns:
        DataFrame with ratings from each bootstrap round
    """
    from tqdm import tqdm

    rows = []
    for i in tqdm(range(num_round), desc="Bootstrap sampling"):
        rows.append(func_compute_elo(battles.sample(frac=1.0, replace=True)))
    df = pd.DataFrame(rows)
    return df[df.median().sort_values(ascending=False).index]


def compute_bradley_terry_leaderboard(df: pd.DataFrame,
                                      bootstrap_rounds: int = 0) -> pd.DataFrame:
    """
    Compute leaderboard using Bradley-Terry model (official Chatbot Arena method).

    Args:
        df: DataFrame with columns 'model_a', 'model_b', 'winner'
        bootstrap_rounds: Number of bootstrap rounds for confidence intervals (0 = no bootstrap)

    Returns:
        DataFrame with model ratings (and confidence intervals if bootstrap > 0)
    """
    print(f"Computing Bradley-Terry model ratings...")

    # Compute MLE Elo ratings
    elo_ratings = compute_mle_elo(df)

    if bootstrap_rounds > 0:
        print(f"Computing {bootstrap_rounds} bootstrap samples for confidence intervals...")
        bootstrap_df = get_bootstrap_result(df, compute_mle_elo, bootstrap_rounds)

        # Compute confidence intervals
        result = pd.DataFrame({
            'rating': bootstrap_df.quantile(0.5),
            'lower_ci': bootstrap_df.quantile(0.025),
            'upper_ci': bootstrap_df.quantile(0.975)
        }).sort_values('rating', ascending=False)
    else:
        result = pd.DataFrame({
            'rating': elo_ratings
        }).sort_values('rating', ascending=False)

    result.index.name = 'model'
    return result.reset_index()

cli.py

#!/usr/bin/env python3
"""
实验 6-6：从配对比较数据构建模型排行榜 —— 命令行入口

统一的 argparse 命令行工具，把整个流程拆成三个子命令：

    battle        运行两两对战，生成对战结果（模拟 / Chatbot Arena 真实数据 / LLM 评判）
    elo           从对战结果计算 Elo 或 Bradley-Terry 评分
    leaderboard   把对战结果或评分渲染成最终排行榜表格
    pipeline      一步跑完 对战 -> Elo -> 排行榜（默认离线可复现）

其中 battle 的 simulate/arena 来源与 elo、leaderboard、pipeline 均为纯离线计算，
无需任何 API；只有 --source llm（LLM 评判对战）需要 LLM API Key：优先用官方
Anthropic（ANTHROPIC_API_KEY），若无则自动回退到 OpenRouter（OPENROUTER_API_KEY），
也可用 --judge-backend openrouter 强制走 OpenRouter（direct key 失效时）。

示例：
    # 离线一条龙：模拟对战 -> Elo -> 排行榜
    python cli.py pipeline

    # 分步运行
    python cli.py battle --source simulate --num-battles 5000 --output battles.json
    python cli.py elo --input battles.json --method bradley-terry --bootstrap 100
    python cli.py leaderboard --input battles.json --top-n 20
"""
import argparse
import json
import os
import sys
import warnings
from typing import List, Optional

import pandas as pd

# Bradley-Terry 的 LogisticRegression 在新版 sklearn 会对 penalty=None 抛
# FutureWarning；bootstrap 会重复上百次，这里静音以保持排行榜输出整洁。
warnings.filterwarnings("ignore", category=FutureWarning, module="sklearn")

from battle_simulator import DEFAULT_TRUE_SKILLS, simulate_battles
from elo_rating import EloRatingSystem


# --------------------------------------------------------------------------- #
# 通用辅助函数
# --------------------------------------------------------------------------- #
def _load_battles(path: str) -> pd.DataFrame:
    """从 JSON 文件加载对战结果，返回带 model_a/model_b/winner 列的 DataFrame。"""
    with open(path, "r", encoding="utf-8") as f:
        data = json.load(f)
    df = pd.DataFrame(data)
    required = {"model_a", "model_b", "winner"}
    if not required.issubset(df.columns):
        raise ValueError(
            f"对战文件 {path} 缺少必要字段 {required}，实际字段：{list(df.columns)}"
        )
    return df


def _save_json(obj, path: str) -> None:
    with open(path, "w", encoding="utf-8") as f:
        json.dump(obj, f, ensure_ascii=False, indent=2)


def _battle_stats(df: pd.DataFrame) -> dict:
    """统计每个模型的对战场数与胜场（平局按 0.5 计）。"""
    matches: dict = {}
    wins: dict = {}
    for model_a, model_b, winner in zip(df["model_a"], df["model_b"], df["winner"]):
        matches[model_a] = matches.get(model_a, 0) + 1
        matches[model_b] = matches.get(model_b, 0) + 1
        if winner == "model_a":
            wins[model_a] = wins.get(model_a, 0) + 1.0
        elif winner == "model_b":
            wins[model_b] = wins.get(model_b, 0) + 1.0
        else:  # tie / tie (bothbad)
            wins[model_a] = wins.get(model_a, 0) + 0.5
            wins[model_b] = wins.get(model_b, 0) + 0.5
    return {"matches": matches, "wins": wins}


def _compute_online_elo(df: pd.DataFrame, k: float, init_rating: float) -> pd.DataFrame:
    """在线增量 Elo（按记录顺序处理），返回带 model/rating 列的 DataFrame。"""
    elo = EloRatingSystem(initial_rating=init_rating, k_factor=k)
    for model_a, model_b, winner in zip(df["model_a"], df["model_b"], df["winner"]):
        elo.update_ratings(model_a, model_b, winner)
    rows = [(m, r) for m, r, *_ in elo.get_leaderboard()]
    return pd.DataFrame(rows, columns=["model", "rating"])


def _compute_bradley_terry(df: pd.DataFrame, bootstrap: int) -> pd.DataFrame:
    """Bradley-Terry MLE 评分（可选 bootstrap 置信区间）。"""
    # 延迟导入：Bradley-Terry 依赖 scikit-learn，仅在需要时加载。
    from bradley_terry import compute_bradley_terry_leaderboard
    return compute_bradley_terry_leaderboard(df, bootstrap_rounds=bootstrap)


def _compute_ratings(df: pd.DataFrame, method: str, k: float,
                     init_rating: float, bootstrap: int) -> pd.DataFrame:
    if method == "bradley-terry":
        return _compute_bradley_terry(df, bootstrap)
    return _compute_online_elo(df, k, init_rating)


def _print_leaderboard(ratings: pd.DataFrame, df: Optional[pd.DataFrame],
                       top_n: int, title: str) -> None:
    """打印最终排行榜表格。若评分含置信区间则展示 95% CI 列。"""
    has_ci = {"lower_ci", "upper_ci"}.issubset(ratings.columns)
    stats = _battle_stats(df) if df is not None else {"matches": {}, "wins": {}}

    ratings = ratings.sort_values("rating", ascending=False).reset_index(drop=True)

    print("=" * 78)
    print(title)
    print("=" * 78)
    if has_ci:
        header = f"{'排名':<6}{'模型':<24}{'Elo':>8}   {'95% 置信区间':<20}{'场数':>7}{'胜率':>9}"
    else:
        header = f"{'排名':<6}{'模型':<24}{'Elo':>8}   {'场数':>7}{'胜率':>9}"
    print(header)
    print("-" * 78)

    for idx, row in ratings.head(top_n).iterrows():
        model = str(row["model"])
        n = stats["matches"].get(model, 0)
        w = stats["wins"].get(model, 0.0)
        win_rate = (w / n * 100.0) if n else 0.0
        if has_ci:
            ci = f"[{row['lower_ci']:.0f}, {row['upper_ci']:.0f}]"
            print(f"{idx + 1:<6}{model:<24}{row['rating']:>8.1f}   "
                  f"{ci:<20}{n:>7}{win_rate:>8.1f}%")
        else:
            print(f"{idx + 1:<6}{model:<24}{row['rating']:>8.1f}   "
                  f"{n:>7}{win_rate:>8.1f}%")
    print("-" * 78)
    print(f"共 {len(ratings)} 个模型，"
          f"评分范围 {ratings['rating'].min():.1f} ~ {ratings['rating'].max():.1f}")
    if has_ci:
        avg_ci = (ratings["upper_ci"] - ratings["lower_ci"]).mean()
        print(f"平均 95% 置信区间宽度：{avg_ci:.1f} 分")
    print()


# --------------------------------------------------------------------------- #
# 子命令实现
# --------------------------------------------------------------------------- #
def _make_battles(args) -> List[dict]:
    if args.source == "simulate":
        skills = DEFAULT_TRUE_SKILLS
        if args.models:
            # 用户指定模型名时，围绕 1000 分等距分配潜在实力。
            n = len(args.models)
            skills = {m: 1000.0 + (n - 1 - 2 * i) * 40.0 for i, m in enumerate(args.models)}
        print(f"模拟 {args.num_battles} 场对战（{len(skills)} 个模型，"
              f"平局概率 {args.tie_prob}，随机种子 {args.seed}）...")
        battles = simulate_battles(skills, args.num_battles,
                                   tie_prob=args.tie_prob, seed=args.seed)
        print("真实潜在实力（用于事后对照）：")
        for m, s in sorted(skills.items(), key=lambda kv: -kv[1]):
            print(f"  {m:<24}{s:>8.1f}")
        return battles

    if args.source == "arena":
        from data_loader import load_arena_data, filter_data
        from parallel_processing import optimize_dataframe
        if not os.path.exists(args.arena_file):
            print(f"错误：找不到 Chatbot Arena 数据文件 {args.arena_file}。", file=sys.stderr)
            print("可从以下地址下载并保存为该文件名：", file=sys.stderr)
            print("https://storage.googleapis.com/arena_external_data/public/"
                  "clean_battle_20240814_public.json", file=sys.stderr)
            sys.exit(1)
        df = load_arena_data(args.arena_file)
        df = optimize_dataframe(df)
        df = filter_data(df, anony_only=True, use_dedup=True, min_turn=1)
        if args.sample and args.sample < len(df):
            df = df.sample(n=args.sample, random_state=args.seed).reset_index(drop=True)
            print(f"采样 {args.sample} 场对战。")
        return df[["model_a", "model_b", "winner"]].to_dict("records")

    # source == "llm"
    from llm_judge import run_llm_battles
    print("运行 LLM 评判对战（顺序交换以消除位置偏差）...")
    return run_llm_battles(
        candidate_models=args.candidate_models,
        judge_model=args.judge_model,
        backend=args.judge_backend,
    )


def cmd_battle(args) -> None:
    battles = _make_battles(args)
    _save_json(battles, args.output)
    print(f"\n已生成 {len(battles)} 场对战，写入 {args.output}")


def cmd_elo(args) -> None:
    df = _load_battles(args.input)
    print(f"从 {args.input} 加载 {len(df)} 场对战，方法：{args.method}")
    ratings = _compute_ratings(df, args.method, args.k, args.init_rating, args.bootstrap)
    _print_leaderboard(ratings, df, top_n=args.top_n,
                       title=f"Elo 评分（{args.method}）")
    if args.output:
        _save_json(ratings.to_dict("records"), args.output)
        print(f"评分已写入 {args.output}")


def cmd_leaderboard(args) -> None:
    with open(args.input, "r", encoding="utf-8") as f:
        data = json.load(f)
    sample = data[0] if isinstance(data, list) and data else {}
    if "rating" in sample:  # 输入已是评分文件，直接展示。
        ratings = pd.DataFrame(data)
        _print_leaderboard(ratings, None, top_n=args.top_n, title="模型排行榜")
        return
    # 否则视为对战文件：先计算评分再展示。
    df = _load_battles(args.input)
    print(f"从 {args.input} 加载 {len(df)} 场对战，方法：{args.method}")
    ratings = _compute_ratings(df, args.method, args.k, args.init_rating, args.bootstrap)
    _print_leaderboard(ratings, df, top_n=args.top_n, title="模型排行榜")


def cmd_pipeline(args) -> None:
    print("=" * 78)
    print("实验 6-6：对战 -> Elo -> 排行榜（端到端）")
    print("=" * 78)
    battles = _make_battles(args)
    if args.output:
        _save_json(battles, args.output)
        print(f"对战结果写入 {args.output}")
    df = pd.DataFrame(battles)
    print(f"\n用 {args.method} 方法从 {len(df)} 场对战计算评分...")
    ratings = _compute_ratings(df, args.method, args.k, args.init_rating, args.bootstrap)
    _print_leaderboard(ratings, df, top_n=args.top_n, title="最终排行榜")


# --------------------------------------------------------------------------- #
# 参数解析
# --------------------------------------------------------------------------- #
def _add_source_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--source", choices=["simulate", "arena", "llm"],
                        default="simulate",
                        help="对战来源：simulate=离线模拟(默认)，arena=真实 Chatbot Arena 数据，"
                             "llm=LLM 评判(需 API)")
    parser.add_argument("--models", nargs="+", default=None,
                        help="simulate：自定义模型名列表（默认使用内置 8 个模型）")
    parser.add_argument("--num-battles", type=int, default=3000,
                        help="simulate：模拟对战场数（默认 3000）")
    parser.add_argument("--tie-prob", type=float, default=0.1,
                        help="simulate：平局概率（默认 0.1）")
    parser.add_argument("--seed", type=int, default=42,
                        help="随机种子（默认 42）")
    parser.add_argument("--arena-file", default="arena_data.json",
                        help="arena：Chatbot Arena 数据文件路径（默认 arena_data.json）")
    parser.add_argument("--sample", type=int, default=0,
                        help="arena：随机采样 N 场对战，0 表示全部（默认 0）")
    parser.add_argument("--candidate-models", nargs="+", default=None,
                        help="llm：参与对战的候选模型（默认 Claude 系列）")
    parser.add_argument("--judge-model", default="claude-opus-4-8",
                        help="llm：评判模型（默认 claude-opus-4-8）")
    parser.add_argument("--judge-backend", choices=["anthropic", "openrouter", "auto"],
                        default="auto",
                        help="llm：评判后端。auto=有 ANTHROPIC_API_KEY 用官方 Anthropic，"
                             "否则回退到 OpenRouter（OPENROUTER_API_KEY）；"
                             "openrouter=强制走 OpenRouter（direct key 失效时用）")


def _add_rating_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--method", choices=["online-elo", "bradley-terry"],
                        default="online-elo",
                        help="评分方法：online-elo=在线增量 Elo(默认)，"
                             "bradley-terry=官方 MLE 拟合")
    parser.add_argument("--k", type=float, default=4.0,
                        help="online-elo：K 因子/学习率（默认 4.0，官方取值）")
    parser.add_argument("--init-rating", type=float, default=1000.0,
                        help="初始评分（默认 1000）")
    parser.add_argument("--bootstrap", type=int, default=0,
                        help="bradley-terry：bootstrap 轮数以估计 95%% 置信区间（默认 0=不估计）")
    parser.add_argument("--top-n", type=int, default=20,
                        help="排行榜展示的模型数量（默认 20）")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="cli.py",
        description="实验 6-6：从配对比较数据构建模型排行榜（对战 -> Elo -> 排行榜）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    sub = parser.add_subparsers(dest="command", metavar="{battle,elo,leaderboard,pipeline}")

    # battle
    p_battle = sub.add_parser("battle", help="运行两两对战，生成对战结果")
    _add_source_args(p_battle)
    p_battle.add_argument("--output", default="battles.json",
                          help="对战结果输出文件（默认 battles.json）")
    p_battle.set_defaults(func=cmd_battle)

    # elo
    p_elo = sub.add_parser("elo", help="从对战结果计算 Elo / Bradley-Terry 评分")
    p_elo.add_argument("--input", default="battles.json",
                       help="对战结果输入文件（默认 battles.json）")
    _add_rating_args(p_elo)
    p_elo.add_argument("--output", default=None,
                       help="把评分写入 JSON 文件（可选）")
    p_elo.set_defaults(func=cmd_elo)

    # leaderboard
    p_lb = sub.add_parser("leaderboard", help="显示最终排行榜表格")
    p_lb.add_argument("--input", default="battles.json",
                      help="对战结果或评分输入文件（默认 battles.json）")
    _add_rating_args(p_lb)
    p_lb.set_defaults(func=cmd_leaderboard)

    # pipeline
    p_pipe = sub.add_parser("pipeline", help="一步跑完 对战 -> Elo -> 排行榜（默认离线）")
    _add_source_args(p_pipe)
    _add_rating_args(p_pipe)
    p_pipe.add_argument("--output", default=None,
                        help="把对战结果写入 JSON 文件（可选）")
    p_pipe.set_defaults(func=cmd_pipeline)

    return parser


def main(argv: Optional[List[str]] = None) -> None:
    parser = build_parser()
    # 无子命令时默认运行离线端到端演示，保留开箱即用体验。
    args = parser.parse_args(argv if argv is not None else (sys.argv[1:] or ["pipeline"]))
    try:
        args.func(args)
    except (RuntimeError, FileNotFoundError, ValueError) as exc:
        print(f"错误：{exc}", file=sys.stderr)
        sys.exit(1)
    except Exception as exc:  # 例如无效 ANTHROPIC_API_KEY 触发的 anthropic.AuthenticationError
        print(f"错误：{type(exc).__name__}: {exc}", file=sys.stderr)
        print("（若为 LLM 评审路径，请检查对应 provider 的 API key 是否有效）", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

data_loader.py

"""
Data loading and preprocessing for Chatbot Arena voting data
"""
import pandas as pd
import requests
import os
from typing import Optional
from tqdm import tqdm


def download_arena_data(output_path: str = "arena_data.json", force_download: bool = False) -> str:
    """
    Download Chatbot Arena voting data via HTTPS.

    Args:
        output_path: Path to save downloaded file
        force_download: If True, re-download even if file exists

    Returns:
        Path to downloaded file
    """
    if os.path.exists(output_path) and not force_download:
        print(f"Data file already exists at {output_path}")
        file_size = os.path.getsize(output_path) / (1024 * 1024)
        print(f"File size: {file_size:.2f} MB")
        return output_path

    print("Downloading Chatbot Arena voting data...")
    url = "https://storage.googleapis.com/arena_external_data/public/clean_battle_20240814_public.json"

    try:
        # Stream download with progress bar
        response = requests.get(url, stream=True)
        response.raise_for_status()

        # Get total file size
        total_size = int(response.headers.get('content-length', 0))

        # Download with progress bar
        with open(output_path, 'wb') as f, tqdm(
            desc=output_path,
            total=total_size,
            unit='B',
            unit_scale=True,
            unit_divisor=1024,
        ) as pbar:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)
                    pbar.update(len(chunk))

        file_size = os.path.getsize(output_path) / (1024 * 1024)
        print(f"\nDownloaded data to {output_path} ({file_size:.2f} MB)")
        return output_path

    except Exception as e:
        print(f"Error downloading data: {e}")
        print("Please ensure you have internet connection and the URL is accessible.")
        raise


def load_arena_data(filepath: str) -> pd.DataFrame:
    """
    Load and preprocess Chatbot Arena voting data.

    Expected columns:
    - model_a: Identifier for first model
    - model_b: Identifier for second model
    - winner: Which model won ('model_a', 'model_b', or 'tie')
    - tstamp: Unix timestamp of the vote
    - judge: User who made the vote
    - turn: Conversation turn
    - anony: Whether vote was anonymous (blind)
    - language: Language of the conversation

    Args:
        filepath: Path to data file

    Returns:
        Preprocessed DataFrame sorted by timestamp
    """
    print(f"Loading data from {filepath}...")
    print("Note: This is a large file (~2GB), loading may take 1-2 minutes...")

    # Try different file formats
    if filepath.endswith('.json'):
        df = pd.read_json(filepath)
    elif filepath.endswith('.jsonl'):
        df = pd.read_json(filepath, lines=True)
    elif filepath.endswith('.csv'):
        df = pd.read_csv(filepath)
    else:
        # Try JSON by default
        try:
            df = pd.read_json(filepath)
        except:
            df = pd.read_json(filepath, lines=True)

    print(f"Loaded {len(df)} records")
    print(f"Columns: {df.columns.tolist()}")

    # Sort by timestamp
    if 'tstamp' in df.columns:
        df = df.sort_values('tstamp', ascending=True).reset_index(drop=True)
        print(f"Data spans from {pd.to_datetime(df['tstamp'].min(), unit='s')} to {pd.to_datetime(df['tstamp'].max(), unit='s')}")

    # Basic statistics
    if 'winner' in df.columns:
        print(f"\nOutcome distribution:")
        print(df['winner'].value_counts())

    if 'model_a' in df.columns and 'model_b' in df.columns:
        all_models = set(df['model_a'].unique()) | set(df['model_b'].unique())
        print(f"\nTotal unique models: {len(all_models)}")
        print(f"Top 10 models by appearance:")
        model_counts = pd.concat([df['model_a'], df['model_b']]).value_counts().head(10)
        print(model_counts)

    return df


def filter_data(df: pd.DataFrame, 
                min_date: Optional[str] = None,
                max_date: Optional[str] = None,
                anony_only: bool = True,
                language: Optional[str] = None,
                min_turn: int = 1,
                use_dedup: bool = True) -> pd.DataFrame:
    """
    Filter voting data based on various criteria (following official Chatbot Arena method).

    Args:
        df: Input DataFrame
        min_date: Minimum date (YYYY-MM-DD format)
        max_date: Maximum date (YYYY-MM-DD format)
        anony_only: If True, only include anonymous (blind) votes
        language: If specified, filter by language
        min_turn: Minimum conversation turn
        use_dedup: If True, apply deduplication filter (official Arena method)

    Returns:
        Filtered DataFrame
    """
    filtered = df.copy()

    print(f"Before filtering: {len(filtered)} records")

    # Filter by anonymous votes only (official method)
    if anony_only and 'anony' in filtered.columns:
        filtered = filtered[filtered['anony'] == True]
        print(f"  After anony filter: {len(filtered)} records")

    # Apply deduplication (official method removes top 0.1% redundant prompts)
    if use_dedup and 'dedup_tag' in filtered.columns:
        try:
            filtered = filtered[filtered["dedup_tag"].apply(lambda x: x.get("sampled", False) if isinstance(x, dict) else False)]
            print(f"  After dedup filter: {len(filtered)} records")
        except Exception as e:
            print(f"  Warning: Could not apply dedup filter: {e}")

    # Filter by date
    if 'tstamp' in filtered.columns:
        if min_date:
            min_timestamp = pd.to_datetime(min_date).timestamp()
            filtered = filtered[filtered['tstamp'] >= min_timestamp]
        if max_date:
            max_timestamp = pd.to_datetime(max_date).timestamp()
            filtered = filtered[filtered['tstamp'] <= max_timestamp]

    # Filter by language
    if language and 'language' in filtered.columns:
        filtered = filtered[filtered['language'] == language]

    # Filter by turn
    if 'turn' in filtered.columns:
        filtered = filtered[filtered['turn'] >= min_turn]

    pct = (len(filtered) / len(df) * 100) if len(df) else 0.0
    print(f"After filtering: {len(filtered)} records ({pct:.1f}% of original)")
    return filtered.reset_index(drop=True)


def get_time_slices(df: pd.DataFrame, interval: str = 'W') -> list:
    """
    Split data into time slices for historical analysis.

    Args:
        df: Input DataFrame with 'tstamp' column
        interval: Pandas frequency string ('D' for daily, 'W' for weekly, 'M' for monthly)

    Returns:
        List of (end_date, dataframe_slice) tuples
    """
    if 'tstamp' not in df.columns:
        raise ValueError("DataFrame must have 'tstamp' column")

    df['datetime'] = pd.to_datetime(df['tstamp'], unit='s')
    min_date = df['datetime'].min()
    max_date = df['datetime'].max()

    # Generate date ranges
    date_ranges = pd.date_range(start=min_date, end=max_date, freq=interval)

    slices = []
    for end_date in date_ranges:
        slice_df = df[df['datetime'] <= end_date].copy()
        if len(slice_df) > 0:
            slices.append((end_date, slice_df))

    # Add final slice with all data
    if date_ranges[-1] < max_date:
        slices.append((max_date, df.copy()))

    print(f"Created {len(slices)} time slices with interval '{interval}'")
    return slices

elo_rating.py

"""
Elo Rating System Implementation
Based on Bradley-Terry model for pairwise comparison
"""
import numpy as np
from typing import Dict, Tuple, Optional


class EloRatingSystem:
    """
    Implementation of Elo rating system for model comparison.

    The Elo system updates ratings based on pairwise comparison outcomes,
    where the rating difference between two models determines expected win probability.
    """

    def __init__(self, initial_rating: float = 1000.0, k_factor: float = 4.0):
        """
        Initialize Elo rating system.

        Args:
            initial_rating: Starting rating for all models
            k_factor: Learning rate controlling magnitude of rating updates
        """
        self.initial_rating = initial_rating
        self.k_factor = k_factor
        self.ratings: Dict[str, float] = {}
        self.match_counts: Dict[str, int] = {}
        self.win_counts: Dict[str, float] = {}  # ties add 0.5, so this is float

    def get_rating(self, model: str) -> float:
        """Get current rating for a model, initializing if necessary."""
        if model not in self.ratings:
            self.ratings[model] = self.initial_rating
            self.match_counts[model] = 0
            self.win_counts[model] = 0
        return self.ratings[model]

    def expected_score(self, rating_a: float, rating_b: float) -> float:
        """
        Calculate expected win probability for model A against model B.

        Uses logistic function: P(A wins) = 1 / (1 + 10^((R_B - R_A)/400))

        Args:
            rating_a: Rating of model A
            rating_b: Rating of model B

        Returns:
            Expected probability that A wins (between 0 and 1)
        """
        return 1.0 / (1.0 + 10.0 ** ((rating_b - rating_a) / 400.0))

    def update_ratings(self, model_a: str, model_b: str, outcome: str) -> Tuple[float, float]:
        """
        Update ratings after a match between two models.

        Args:
            model_a: Identifier for first model
            model_b: Identifier for second model
            outcome: Match result ('model_a', 'model_b', or 'tie')

        Returns:
            Tuple of (new_rating_a, new_rating_b)
        """
        # Get current ratings
        rating_a = self.get_rating(model_a)
        rating_b = self.get_rating(model_b)

        # Calculate expected scores
        expected_a = self.expected_score(rating_a, rating_b)
        expected_b = 1.0 - expected_a

        # Determine actual scores
        if outcome == 'model_a':
            score_a, score_b = 1.0, 0.0
            self.win_counts[model_a] = self.win_counts.get(model_a, 0) + 1
        elif outcome == 'model_b':
            score_a, score_b = 0.0, 1.0
            self.win_counts[model_b] = self.win_counts.get(model_b, 0) + 1
        else:  # tie
            score_a, score_b = 0.5, 0.5
            # A tie counts as half a win for each side, keeping win_counts (and
            # the win-rate derived from it) consistent with the 0.5-per-tie
            # convention used elsewhere (leaderboard win-rate matrix, CLI stats).
            self.win_counts[model_a] = self.win_counts.get(model_a, 0) + 0.5
            self.win_counts[model_b] = self.win_counts.get(model_b, 0) + 0.5

        # Update ratings using Elo formula
        new_rating_a = rating_a + self.k_factor * (score_a - expected_a)
        new_rating_b = rating_b + self.k_factor * (score_b - expected_b)

        # Store updated ratings
        self.ratings[model_a] = new_rating_a
        self.ratings[model_b] = new_rating_b

        # Update match counts
        self.match_counts[model_a] = self.match_counts.get(model_a, 0) + 1
        self.match_counts[model_b] = self.match_counts.get(model_b, 0) + 1

        return new_rating_a, new_rating_b

    def get_leaderboard(self) -> list:
        """
        Get current leaderboard sorted by rating.

        Returns:
            List of tuples (model, rating, matches, wins) sorted by rating descending
        """
        leaderboard = []
        for model in self.ratings:
            leaderboard.append((
                model,
                self.ratings[model],
                self.match_counts.get(model, 0),
                self.win_counts.get(model, 0)
            ))

        # Sort by rating descending
        leaderboard.sort(key=lambda x: x[1], reverse=True)
        return leaderboard

    def calculate_win_probability(self, model_a: str, model_b: str) -> float:
        """
        Calculate win probability of model_a against model_b based on current ratings.

        Args:
            model_a: First model identifier
            model_b: Second model identifier

        Returns:
            Probability that model_a wins (between 0 and 1)
        """
        rating_a = self.get_rating(model_a)
        rating_b = self.get_rating(model_b)
        return self.expected_score(rating_a, rating_b)

    def get_win_rate_matrix(self) -> Dict[Tuple[str, str], float]:
        """
        Calculate pairwise win probability matrix for all models.

        Returns:
            Dictionary mapping (model_a, model_b) to win probability of model_a
        """
        models = sorted(self.ratings.keys())
        matrix = {}

        for model_a in models:
            for model_b in models:
                if model_a != model_b:
                    prob = self.calculate_win_probability(model_a, model_b)
                    matrix[(model_a, model_b)] = prob

        return matrix

    def reset(self):
        """Reset all ratings to initial values."""
        self.ratings.clear()
        self.match_counts.clear()
        self.win_counts.clear()

    def copy(self) -> 'EloRatingSystem':
        """Create a deep copy of the current rating system."""
        new_system = EloRatingSystem(self.initial_rating, self.k_factor)
        new_system.ratings = self.ratings.copy()
        new_system.match_counts = self.match_counts.copy()
        new_system.win_counts = self.win_counts.copy()
        return new_system

leaderboard.py

"""
Leaderboard calculation and analysis
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from tqdm import tqdm
from elo_rating import EloRatingSystem


def build_leaderboard(df: pd.DataFrame, 
                      initial_rating: float = 1000.0,
                      k_factor: float = 32.0,
                      show_progress: bool = True) -> EloRatingSystem:
    """
    Build Elo leaderboard from voting data.

    Args:
        df: DataFrame with columns 'model_a', 'model_b', 'winner'
        initial_rating: Starting rating for all models
        k_factor: Elo learning rate
        show_progress: Whether to show progress bar

    Returns:
        EloRatingSystem with final ratings
    """
    elo = EloRatingSystem(initial_rating=initial_rating, k_factor=k_factor)

    iterator = tqdm(df.iterrows(), total=len(df), desc="Processing matches") if show_progress else df.iterrows()

    for idx, row in iterator:
        model_a = row['model_a']
        model_b = row['model_b']
        winner = row['winner']

        elo.update_ratings(model_a, model_b, winner)

    return elo


def calculate_win_rate_matrix_from_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Calculate empirical win rate matrix directly from vote data.

    Args:
        df: DataFrame with columns 'model_a', 'model_b', 'winner'

    Returns:
        DataFrame with win rates (rows beat columns)
    """
    # Get all unique models
    all_models = sorted(set(df['model_a'].unique()) | set(df['model_b'].unique()))

    # Initialize counts
    wins = {model: {opponent: 0 for opponent in all_models} for model in all_models}
    total = {model: {opponent: 0 for opponent in all_models} for model in all_models}

    # Count wins and totals
    for _, row in df.iterrows():
        model_a = row['model_a']
        model_b = row['model_b']
        winner = row['winner']

        total[model_a][model_b] += 1
        total[model_b][model_a] += 1

        if winner == 'model_a':
            wins[model_a][model_b] += 1
        elif winner == 'model_b':
            wins[model_b][model_a] += 1
        else:  # tie
            wins[model_a][model_b] += 0.5
            wins[model_b][model_a] += 0.5

    # Calculate win rates
    win_rates = {}
    for model in all_models:
        win_rates[model] = {}
        for opponent in all_models:
            if model == opponent:
                win_rates[model][opponent] = 0.5
            elif total[model][opponent] > 0:
                win_rates[model][opponent] = wins[model][opponent] / total[model][opponent]
            else:
                win_rates[model][opponent] = np.nan

    # Convert to DataFrame
    win_rate_df = pd.DataFrame(win_rates).T
    win_rate_df = win_rate_df[all_models]  # Ensure consistent ordering

    return win_rate_df


def compare_win_rates(elo_system: EloRatingSystem, empirical_df: pd.DataFrame) -> pd.DataFrame:
    """
    Compare predicted win rates from Elo with empirical win rates.

    Args:
        elo_system: Trained Elo rating system
        empirical_df: DataFrame with empirical win rates

    Returns:
        DataFrame with comparison statistics
    """
    models = empirical_df.index.tolist()

    comparisons = []
    for model_a in models:
        for model_b in models:
            if model_a != model_b:
                empirical = empirical_df.loc[model_a, model_b]
                if not np.isnan(empirical):
                    predicted = elo_system.calculate_win_probability(model_a, model_b)
                    error = abs(predicted - empirical)
                    comparisons.append({
                        'model_a': model_a,
                        'model_b': model_b,
                        'empirical': empirical,
                        'predicted': predicted,
                        'error': error
                    })

    comparison_df = pd.DataFrame(comparisons)
    return comparison_df


def build_historical_leaderboards(df: pd.DataFrame,
                                   time_slices: List[Tuple],
                                   initial_rating: float = 1000.0,
                                   k_factor: float = 32.0) -> List[Tuple]:
    """
    Build leaderboard snapshots at different time points.

    Args:
        df: Full voting DataFrame
        time_slices: List of (end_date, slice_df) tuples from get_time_slices
        initial_rating: Starting rating
        k_factor: Elo learning rate

    Returns:
        List of (date, leaderboard_data) tuples
    """
    historical_leaderboards = []

    for end_date, slice_df in tqdm(time_slices, desc="Building historical leaderboards"):
        elo = build_leaderboard(slice_df, initial_rating, k_factor, show_progress=False)
        leaderboard = elo.get_leaderboard()

        # Convert to DataFrame for easier handling
        lb_df = pd.DataFrame(leaderboard, columns=['model', 'rating', 'matches', 'wins'])
        lb_df['date'] = end_date
        lb_df['rank'] = range(1, len(lb_df) + 1)

        historical_leaderboards.append((end_date, lb_df))

    return historical_leaderboards


def get_rating_history(historical_leaderboards: List[Tuple]) -> pd.DataFrame:
    """
    Extract rating history for all models over time.

    Args:
        historical_leaderboards: List of (date, leaderboard_df) tuples

    Returns:
        DataFrame with columns: date, model, rating, rank
    """
    all_data = []

    for date, lb_df in historical_leaderboards:
        for _, row in lb_df.iterrows():
            all_data.append({
                'date': date,
                'model': row['model'],
                'rating': row['rating'],
                'rank': row['rank'],
                'matches': row['matches'],
                'wins': row['wins']
            })

    history_df = pd.DataFrame(all_data)
    return history_df


def analyze_rating_changes(history_df: pd.DataFrame, top_n: int = 20) -> pd.DataFrame:
    """
    Analyze rating changes over time for top models.

    Args:
        history_df: DataFrame from get_rating_history
        top_n: Number of top models to analyze

    Returns:
        DataFrame with change statistics
    """
    # Get final ratings
    final_date = history_df['date'].max()
    final_ratings = history_df[history_df['date'] == final_date].nlargest(top_n, 'rating')
    top_models = final_ratings['model'].tolist()

    # Calculate statistics for each model
    stats = []
    for model in top_models:
        model_data = history_df[history_df['model'] == model].sort_values('date')

        if len(model_data) > 0:
            initial_rating = model_data.iloc[0]['rating']
            final_rating = model_data.iloc[-1]['rating']
            max_rating = model_data['rating'].max()
            min_rating = model_data['rating'].min()
            rating_change = final_rating - initial_rating
            volatility = model_data['rating'].std()

            stats.append({
                'model': model,
                'final_rating': final_rating,
                'initial_rating': initial_rating,
                'rating_change': rating_change,
                'max_rating': max_rating,
                'min_rating': min_rating,
                'volatility': volatility,
                'total_matches': model_data.iloc[-1]['matches']
            })

    stats_df = pd.DataFrame(stats).sort_values('final_rating', ascending=False)
    return stats_df

llm_judge.py

"""
LLM-as-judge pairwise battles with position-bias mitigation.

This is the only battle source that needs network access; `simulate` and
`arena` run fully offline.

Two backends are supported, selected automatically or via ``backend=``:

  * ``anthropic``  – the official ``anthropic`` SDK, using ``ANTHROPIC_API_KEY``
    (the default when that key is present).
  * ``openrouter`` – the OpenAI-compatible ``openai`` SDK pointed at
    ``https://openrouter.ai/api/v1`` with ``OPENROUTER_API_KEY``. Internal
    Claude ids (e.g. ``claude-opus-4-8``) are mapped to their OpenRouter ids
    (``anthropic/claude-opus-4.8``); ids that already contain a ``/`` such as
    ``openai/gpt-5.6-luna`` are passed through untouched. This lets the judge run
    when a direct Anthropic key is missing or invalid.

The two backends are interchangeable: the position-bias swap-and-agree logic and
the A/B/tie response parsing are identical regardless of which one is used.

The book (实验 6-6, 位置偏差 discussion) notes that an LLM judge systematically
favours whichever answer appears in a fixed slot (usually the first). The
standard mitigation, implemented here, is to judge each pair twice with the
answers swapped and only record a winner when both judgements agree; a
disagreement is counted as a tie. This cancels the position bias instead of
letting it leak into the ratings.

The resulting battle list uses the same {'model_a', 'model_b', 'winner'} schema
as the simulated and Chatbot Arena data, so it feeds straight into the Elo /
Bradley-Terry pipeline.
"""
import os
from typing import Dict, List, Optional

# Default candidate roster and judge (Claude models). Kept small because every
# battle costs several API calls (two responses + two swapped judgements).
DEFAULT_CANDIDATE_MODELS = ["claude-opus-4-8", "claude-haiku-4-5"]
DEFAULT_JUDGE_MODEL = "claude-opus-4-8"

DEFAULT_PROMPTS = [
    "用一句话解释什么是 Transformer 的自注意力机制。",
    "Write a haiku about distributed systems.",
    "给出快速排序的时间复杂度，并简要说明最坏情况。",
]

OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"

# Map internal Claude ids -> OpenRouter model ids. Any id already containing a
# '/' (e.g. 'openai/gpt-5.6-luna') is treated as a native OpenRouter id and used
# verbatim; unknown ids are also passed through unchanged.
_OPENROUTER_MODEL_MAP = {
    "claude-opus-4-8": "anthropic/claude-opus-4.8",
    "claude-opus-4-1": "anthropic/claude-opus-4.1",
    "claude-sonnet-4-6": "anthropic/claude-sonnet-4.6",
    "claude-sonnet-4-5": "anthropic/claude-sonnet-4.5",
    "claude-haiku-4-5": "anthropic/claude-haiku-4.5",
}

_JUDGE_SYSTEM = (
    "你是一个严格的评委。用户会给你一个问题和两个候选回答（回答 A 和回答 B）。"
    "请只根据回答质量判断哪个更好，忽略它们出现的顺序。"
    "只输出一个词：A、B 或 tie。"
)


def _to_openrouter_model(model: str) -> str:
    """Translate an internal model id into an OpenRouter model id."""
    if "/" in model:  # already a native OpenRouter id
        return model
    return _OPENROUTER_MODEL_MAP.get(model, model)


class JudgeClient:
    """
    Thin adapter over either the Anthropic SDK or the OpenAI-compatible
    OpenRouter endpoint, exposing a single ``chat()`` method so the rest of the
    module is backend-agnostic.
    """

    def __init__(self, backend: str, impl):
        self.backend = backend
        self.impl = impl

    def chat(self, model: str, user: str, max_tokens: int,
             system: Optional[str] = None) -> str:
        """Send a single-turn chat and return the assistant's text reply."""
        if self.backend == "anthropic":
            kwargs = {
                "model": model,
                "max_tokens": max_tokens,
                "messages": [{"role": "user", "content": user}],
            }
            if system is not None:
                kwargs["system"] = system
            response = self.impl.messages.create(**kwargs)
            return "".join(
                block.text for block in response.content if block.type == "text"
            ).strip()

        # openrouter (OpenAI-compatible chat.completions)
        messages = []
        if system is not None:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": user})
        response = self.impl.chat.completions.create(
            model=_to_openrouter_model(model),
            max_tokens=max_tokens,
            messages=messages,
        )
        return (response.choices[0].message.content or "").strip()


def _resolve_backend(backend: str = "auto") -> str:
    """
    Resolve the effective backend.

    ``auto`` -> ``anthropic`` if ANTHROPIC_API_KEY is set, else ``openrouter``
    if OPENROUTER_API_KEY is set. Raises if neither key is available.
    """
    if backend not in ("anthropic", "openrouter", "auto"):
        raise ValueError(
            f"Unknown judge backend {backend!r}; expected 'anthropic', "
            "'openrouter' or 'auto'."
        )
    if backend != "auto":
        return backend
    if os.environ.get("ANTHROPIC_API_KEY"):
        return "anthropic"
    if os.environ.get("OPENROUTER_API_KEY"):
        return "openrouter"
    raise RuntimeError(
        "No LLM-judge credentials found. Set ANTHROPIC_API_KEY (direct Anthropic) "
        "or OPENROUTER_API_KEY (OpenRouter fallback); or use --source simulate / "
        "--source arena to run the experiment fully offline."
    )


def _get_client(backend: str = "auto") -> JudgeClient:
    """Create a JudgeClient for the resolved backend, with clear errors."""
    backend = _resolve_backend(backend)

    if backend == "anthropic":
        try:
            import anthropic
        except ImportError as exc:  # pragma: no cover - depends on environment
            raise RuntimeError(
                "The 'anthropic' package is required for the anthropic judge "
                "backend. Install it with: pip install anthropic"
            ) from exc
        if not os.environ.get("ANTHROPIC_API_KEY"):
            raise RuntimeError(
                "ANTHROPIC_API_KEY is not set. Set it, or use "
                "--judge-backend openrouter with OPENROUTER_API_KEY, or run "
                "--source simulate / --source arena fully offline."
            )
        return JudgeClient("anthropic", anthropic.Anthropic())

    # backend == "openrouter"
    try:
        import openai
    except ImportError as exc:  # pragma: no cover - depends on environment
        raise RuntimeError(
            "The 'openai' package is required for the openrouter judge backend. "
            "Install it with: pip install openai"
        ) from exc
    if not os.environ.get("OPENROUTER_API_KEY"):
        raise RuntimeError(
            "OPENROUTER_API_KEY is not set. Set it, or use --judge-backend "
            "anthropic with ANTHROPIC_API_KEY, or run --source simulate / "
            "--source arena fully offline."
        )
    return JudgeClient(
        "openrouter",
        openai.OpenAI(
            base_url=OPENROUTER_BASE_URL,
            api_key=os.environ["OPENROUTER_API_KEY"],
        ),
    )


def generate_response(client: JudgeClient, model: str, prompt: str,
                      max_tokens: int = 1024) -> str:
    """Generate a single model answer for a prompt."""
    return client.chat(model, prompt, max_tokens=max_tokens)


def _judge_once(client: JudgeClient, judge_model: str, prompt: str,
                answer_first: str, answer_second: str) -> str:
    """Ask the judge which slot is better; returns 'first', 'second' or 'tie'."""
    user = (
        f"问题：\n{prompt}\n\n"
        f"回答 A：\n{answer_first}\n\n"
        f"回答 B：\n{answer_second}\n\n"
        "哪个回答更好？只输出 A、B 或 tie。"
    )
    verdict = client.chat(judge_model, user, max_tokens=8, system=_JUDGE_SYSTEM).lower()
    if verdict.startswith("a"):
        return "first"
    if verdict.startswith("b"):
        return "second"
    return "tie"


def judge_pair(client: JudgeClient, judge_model: str, prompt: str,
               answer_a: str, answer_b: str) -> str:
    """
    Judge a pair with position-bias mitigation (swap order, tie on disagreement).

    Returns 'model_a', 'model_b', or 'tie'.
    """
    # First pass: A in slot 1, B in slot 2.
    first_pass = _judge_once(client, judge_model, prompt, answer_a, answer_b)
    # Second pass: swap the slots so B is now in slot 1.
    second_pass = _judge_once(client, judge_model, prompt, answer_b, answer_a)

    # Translate both judgements into "which real model won", then require
    # agreement. Slot 1 in the first pass is A; slot 1 in the second pass is B.
    winner_first = {"first": "model_a", "second": "model_b", "tie": "tie"}[first_pass]
    winner_second = {"first": "model_b", "second": "model_a", "tie": "tie"}[second_pass]

    if winner_first == winner_second:
        return winner_first
    return "tie"  # inconsistent under swap -> position bias, count as tie


def run_llm_battles(candidate_models: Optional[List[str]] = None,
                    prompts: Optional[List[str]] = None,
                    judge_model: str = DEFAULT_JUDGE_MODEL,
                    backend: str = "auto") -> List[dict]:
    """
    Run LLM-judged battles between every model pair over every prompt.

    Args:
        candidate_models: Models to compare (default: DEFAULT_CANDIDATE_MODELS).
        prompts: Prompts to battle on (default: DEFAULT_PROMPTS).
        judge_model: Model used as the judge.
        backend: 'anthropic', 'openrouter', or 'auto' (anthropic if
            ANTHROPIC_API_KEY else openrouter).

    Returns:
        List of battle dicts ({'model_a', 'model_b', 'winner'}).
    """
    candidate_models = candidate_models or DEFAULT_CANDIDATE_MODELS
    prompts = prompts or DEFAULT_PROMPTS
    if len(candidate_models) < 2:
        raise ValueError("Need at least 2 candidate models for LLM-judge battles")

    client = _get_client(backend)
    battles: List[dict] = []

    for prompt in prompts:
        # Cache each model's answer per prompt so it is generated only once.
        answers: Dict[str, str] = {
            model: generate_response(client, model, prompt) for model in candidate_models
        }
        for i, model_a in enumerate(candidate_models):
            for model_b in candidate_models[i + 1:]:
                winner = judge_pair(
                    client, judge_model, prompt, answers[model_a], answers[model_b]
                )
                battles.append(
                    {"model_a": model_a, "model_b": model_b, "winner": winner}
                )

    return battles

main.py

"""
Main script for Model Leaderboard Calculation
Experiment 6.8: Building Model Leaderboard from Pairwise Comparison Data

Supports two methods (following official Chatbot Arena):
1. Online Elo (K=4) - Simple but order-dependent
2. Bradley-Terry MLE - Official leaderboard method (more stable)
"""
import os
import sys
import pandas as pd
import numpy as np
from data_loader import download_arena_data, load_arena_data, filter_data
from bradley_terry import (
    compute_bradley_terry_leaderboard,
    predict_win_rate,
    get_bootstrap_result,
    compute_mle_elo
)
from elo_rating import EloRatingSystem
from parallel_processing import optimize_dataframe
from visualization import (
    plot_leaderboard,
    plot_win_rate_matrix,
    plot_rating_distribution
)


def compute_online_elo_leaderboard(df: pd.DataFrame) -> pd.DataFrame:
    """
    Compute leaderboard using online Elo updates (K=4, official value).

    This method updates ratings sequentially as matches are processed.
    It's simpler but can be unstable and order-dependent.

    Args:
        df: DataFrame with columns 'model_a', 'model_b', 'winner'

    Returns:
        DataFrame with model ratings
    """
    from tqdm import tqdm

    print("Computing online Elo ratings (K=4)...")

    elo = EloRatingSystem(initial_rating=1000.0, k_factor=4.0)

    # Process matches sequentially
    for _, row in tqdm(df.iterrows(), total=len(df), desc="Processing matches"):
        elo.update_ratings(row['model_a'], row['model_b'], row['winner'])

    # Get leaderboard
    leaderboard = elo.get_leaderboard()

    # Convert to DataFrame
    result = pd.DataFrame(leaderboard, columns=['model', 'rating', 'matches', 'wins'])
    return result


def main(method: str = 'bradley-terry'):
    """
    Run model leaderboard calculation.

    Args:
        method: 'bradley-terry' (default, official) or 'online-elo' (simple)
    """

    print("="*80)
    print("Experiment 6.8: Building Model Leaderboard from Pairwise Comparisons")
    if method == 'bradley-terry':
        print("Method: Bradley-Terry Model with MLE (Official Chatbot Arena)")
    else:
        print("Method: Online Elo Updates (K=4)")
    print("="*80)
    print()

    # Step 1: Download and load data
    print("Step 1: Loading Chatbot Arena voting data...")
    print("-" * 80)

    data_file = "arena_data.json"

    try:
        # Download data if not exists
        if not os.path.exists(data_file):
            data_file = download_arena_data(data_file)

        # Load data
        df = load_arena_data(data_file)

        # Optimize memory usage
        df = optimize_dataframe(df)

    except Exception as e:
        print(f"Error loading data: {e}")
        print("\nNote: If the data download fails, you can manually download the file from:")
        print("https://storage.googleapis.com/arena_external_data/public/clean_battle_20240814_public.json")
        print("and save it as 'arena_data.json' in the current directory.")
        return

    print()

    # Step 2: Filter data (official Chatbot Arena method)
    print("Step 2: Filtering data (following official Arena method)...")
    print("-" * 80)

    df_filtered = filter_data(
        df,
        anony_only=True,  # Only anonymous/blind votes
        use_dedup=True,   # Apply deduplication (removes top 0.1% redundant prompts)
        min_turn=1
    )

    print()

    # Step 3: Compute ratings using selected method
    print(f"Step 3: Computing ratings using {method} method...")
    print("-" * 80)

    if method == 'bradley-terry':
        print("Note: Bradley-Terry model uses sklearn LogisticRegression for MLE.")
        print("This is the official Chatbot Arena method - more stable than online Elo.")
        print()

        # Compute ratings with bootstrap for confidence intervals
        leaderboard_df = compute_bradley_terry_leaderboard(df_filtered, bootstrap_rounds=100)

        print(f"\nTop 20 models by Bradley-Terry rating:")
        print("-" * 80)
        print(f"{'Rank':<6}{'Model':<35}{'Rating':<10}{'95% CI':<20}")
        print("-" * 80)
        for idx, row in leaderboard_df.head(20).iterrows():
            if 'lower_ci' in row and 'upper_ci' in row:
                ci_str = f"[{row['lower_ci']:.1f}, {row['upper_ci']:.1f}]"
            else:
                ci_str = "N/A"
            print(f"{idx+1:<6}{row['model']:<35}{row['rating']:7.1f}   {ci_str:<20}")

    else:  # online-elo
        print("Note: Online Elo uses K=4 (official value) for stable ratings.")
        print("Processes matches sequentially - simpler but can be order-dependent.")
        print()

        # Compute online Elo ratings
        leaderboard_df = compute_online_elo_leaderboard(df_filtered)

        print(f"\nTop 20 models by Online Elo rating:")
        print("-" * 80)
        print(f"{'Rank':<6}{'Model':<35}{'Rating':<10}{'Matches':<10}{'Win Rate':<10}")
        print("-" * 80)
        for idx, row in leaderboard_df.head(20).iterrows():
            win_rate = row['wins'] / row['matches'] * 100 if row['matches'] > 0 else 0
            print(f"{idx+1:<6}{row['model']:<35}{row['rating']:7.1f}   {row['matches']:<10}{win_rate:6.1f}%")

    print()

    # Step 4: Predict win rates using Bradley-Terry model
    print("Step 4: Calculating predicted win rates...")
    print("-" * 80)

    # Get ratings as dictionary
    ratings_dict = dict(zip(leaderboard_df['model'], leaderboard_df['rating']))

    # Predict win rates
    predicted_win_rates = predict_win_rate(ratings_dict)

    print(f"Calculated predicted win rates for {len(ratings_dict)} models")
    print()

    # Step 5: Create visualizations
    print("Step 5: Creating visualizations...")
    print("-" * 80)

    # Convert leaderboard_df to format expected by visualization functions
    leaderboard_tuples = [(row['model'], row['rating'], 0, 0) for _, row in leaderboard_df.iterrows()]

    plot_leaderboard(leaderboard_tuples, top_n=20, save_path="leaderboard.png")
    plot_rating_distribution(leaderboard_tuples, save_path="rating_distribution.png")

    # Plot win rate matrix
    top_30_models = leaderboard_df.head(30)['model'].tolist()
    plot_win_rate_matrix(predicted_win_rates.loc[top_30_models, top_30_models], 
                         top_n=30, save_path="win_rate_matrix.png")

    print()

    # Summary
    print("="*80)
    print("Analysis complete!")
    print("="*80)
    print("\nGenerated files:")
    print("  - leaderboard.png                    : Top 20 models by Bradley-Terry rating")
    print("  - rating_distribution.png            : Distribution of ratings")
    print("  - win_rate_matrix.png                : Predicted win rate matrix (top 30 models)")
    print()

    # Method summary
    print("Method:")
    print("-" * 80)
    if method == 'bradley-terry':
        print("  ✓ Bradley-Terry model with Maximum Likelihood Estimation")
        print("  ✓ sklearn LogisticRegression for stable rating computation")
        print("  ✓ Bootstrap confidence intervals (100 samples)")
    else:
        print("  ✓ Online Elo with K=4 (official value)")
        print("  ✓ Sequential match processing")
        print("  ✓ Simple but order-dependent")
    print("  ✓ Deduplication filter (removes top 0.1% redundant prompts)")
    print("  ✓ Anonymous votes only (blind evaluation)")
    print()

    # Key insights
    print("Key Insights:")
    print("-" * 80)
    print(f"  • Total models analyzed: {len(leaderboard_df)}")
    print(f"  • Total battles: {len(df_filtered):,}")
    print(f"  • Rating range: {leaderboard_df['rating'].min():.1f} - {leaderboard_df['rating'].max():.1f}")
    print(f"  • Top model: {leaderboard_df.iloc[0]['model']} ({leaderboard_df.iloc[0]['rating']:.1f})")

    if 'lower_ci' in leaderboard_df.columns:
        avg_ci_width = (leaderboard_df['upper_ci'] - leaderboard_df['lower_ci']).mean()
        print(f"  • Average confidence interval width: {avg_ci_width:.1f} rating points")

    print()
    print("This implementation matches the official Chatbot Arena leaderboard calculation!")
    print("Source: https://colab.research.google.com/drive/1KdwokPjirkTmpO_P1WByFNFiqxWQquwH")
    print()


if __name__ == "__main__":
    # Allow method selection via command line argument
    import sys

    method = 'bradley-terry'  # Default to official method

    if len(sys.argv) > 1:
        if sys.argv[1] in ['bradley-terry', 'bt', 'mle']:
            method = 'bradley-terry'
        elif sys.argv[1] in ['online-elo', 'elo', 'online']:
            method = 'online-elo'
        else:
            print(f"Unknown method: {sys.argv[1]}")
            print("Usage: python main.py [bradley-terry|online-elo]")
            print("  bradley-terry (default): Official Arena method, more stable")
            print("  online-elo: Simple Elo updates with K=4")
            sys.exit(1)

    main(method)

optimized_elo.py

"""
Optimized Elo rating system using NumPy vectorization and Numba JIT
"""
import numpy as np
import pandas as pd
from typing import Dict, Tuple, List
from numba import jit
from tqdm import tqdm


@jit(nopython=True)
def expected_score_fast(rating_a: float, rating_b: float) -> float:
    """
    Fast expected score calculation using Numba JIT.

    Args:
        rating_a: Rating of model A
        rating_b: Rating of model B

    Returns:
        Expected probability that A wins
    """
    return 1.0 / (1.0 + 10.0 ** ((rating_b - rating_a) / 400.0))


@jit(nopython=True)
def process_elo_updates_vectorized(ratings: np.ndarray,
                                   model_a_indices: np.ndarray,
                                   model_b_indices: np.ndarray,
                                   outcomes: np.ndarray,
                                   k_factor: float,
                                   match_counts: np.ndarray,
                                   win_counts: np.ndarray) -> np.ndarray:
    """
    Process Elo updates using vectorized NumPy operations with Numba JIT.

    This is the core hot loop optimized with Numba for maximum performance.

    Args:
        ratings: Array of current ratings for all models
        model_a_indices: Indices of model A for each match
        model_b_indices: Indices of model B for each match
        outcomes: Match outcomes (1.0 = A wins, 0.0 = B wins, 0.5 = tie)
        k_factor: Elo K-factor
        match_counts: Array to track match counts per model
        win_counts: Array to track win counts per model

    Returns:
        Updated ratings array
    """
    n_matches = len(model_a_indices)

    for i in range(n_matches):
        idx_a = model_a_indices[i]
        idx_b = model_b_indices[i]
        outcome = outcomes[i]

        # Get current ratings
        rating_a = ratings[idx_a]
        rating_b = ratings[idx_b]

        # Calculate expected scores
        expected_a = 1.0 / (1.0 + 10.0 ** ((rating_b - rating_a) / 400.0))
        expected_b = 1.0 - expected_a

        # Update ratings
        ratings[idx_a] += k_factor * (outcome - expected_a)
        ratings[idx_b] += k_factor * ((1.0 - outcome) - expected_b)

        # Update counts
        match_counts[idx_a] += 1
        match_counts[idx_b] += 1
        win_counts[idx_a] += outcome
        win_counts[idx_b] += (1.0 - outcome)

    return ratings


@jit(nopython=True)
def calculate_expected_scores_vectorized(ratings_a: np.ndarray, 
                                        ratings_b: np.ndarray) -> np.ndarray:
    """
    Vectorized calculation of expected scores for multiple matches.

    Args:
        ratings_a: Array of ratings for model A
        ratings_b: Array of ratings for model B

    Returns:
        Array of expected scores for model A
    """
    return 1.0 / (1.0 + np.power(10.0, (ratings_b - ratings_a) / 400.0))


class NumpyEloRatingSystem:
    """
    Highly optimized Elo rating system using NumPy arrays and Numba JIT.

    Optimizations:
    - NumPy arrays for O(1) indexing instead of dictionary lookups
    - Numba JIT compilation of hot loops
    - Pre-allocated arrays to avoid memory reallocation
    - Integer indexing for models instead of string lookups
    """

    def __init__(self, initial_rating: float = 1000.0, k_factor: float = 4.0):
        """Initialize NumPy-based Elo system."""
        self.initial_rating = initial_rating
        self.k_factor = k_factor

        # Model name to index mapping
        self.model_to_idx: Dict[str, int] = {}
        self.idx_to_model: Dict[int, str] = {}

        # NumPy arrays for fast access
        self.ratings: np.ndarray = None
        self.match_counts: np.ndarray = None
        self.win_counts: np.ndarray = None

        self.n_models = 0

    def _prepare_data(self, df: pd.DataFrame):
        """
        Prepare NumPy arrays from DataFrame for fast processing.

        Args:
            df: DataFrame with columns 'model_a', 'model_b', 'winner'
        """
        print("Preparing data structures...")

        # Get all unique models
        all_models = sorted(set(df['model_a'].unique()) | set(df['model_b'].unique()))
        self.n_models = len(all_models)

        print(f"Found {self.n_models} unique models")

        # Create model mappings
        for idx, model in enumerate(all_models):
            self.model_to_idx[model] = idx
            self.idx_to_model[idx] = model

        # Initialize arrays
        self.ratings = np.full(self.n_models, self.initial_rating, dtype=np.float64)
        self.match_counts = np.zeros(self.n_models, dtype=np.int32)
        self.win_counts = np.zeros(self.n_models, dtype=np.float64)

        # Convert DataFrame columns to NumPy arrays with integer indices
        print("Converting model names to indices...")
        model_a_indices = df['model_a'].map(self.model_to_idx).values.astype(np.int32)
        model_b_indices = df['model_b'].map(self.model_to_idx).values.astype(np.int32)

        # Convert outcomes to numeric (1.0 for A wins, 0.0 for B wins, 0.5 for tie)
        print("Converting outcomes to numeric...")
        outcome_map = {'model_a': 1.0, 'model_b': 0.0, 'tie': 0.5}
        outcomes = df['winner'].map(outcome_map).values.astype(np.float64)

        return model_a_indices, model_b_indices, outcomes

    def process_matches_vectorized(self, df: pd.DataFrame, show_progress: bool = True):
        """
        Process all matches using vectorized NumPy operations and Numba JIT.

        This is the fastest way to compute Elo ratings for large datasets.

        Args:
            df: DataFrame with columns 'model_a', 'model_b', 'winner'
            show_progress: Whether to show progress bar
        """
        # Prepare data
        model_a_indices, model_b_indices, outcomes = self._prepare_data(df)

        print(f"\nProcessing {len(df)} matches with NumPy + Numba JIT...")

        # Process all matches using JIT-compiled function
        # This is where the magic happens - Numba compiles this to machine code
        if show_progress:
            # Process in chunks to show progress
            chunk_size = 50000
            n_chunks = (len(model_a_indices) + chunk_size - 1) // chunk_size

            for i in tqdm(range(n_chunks), desc="Processing matches"):
                start_idx = i * chunk_size
                end_idx = min((i + 1) * chunk_size, len(model_a_indices))

                self.ratings = process_elo_updates_vectorized(
                    self.ratings,
                    model_a_indices[start_idx:end_idx],
                    model_b_indices[start_idx:end_idx],
                    outcomes[start_idx:end_idx],
                    self.k_factor,
                    self.match_counts,
                    self.win_counts
                )
        else:
            self.ratings = process_elo_updates_vectorized(
                self.ratings,
                model_a_indices,
                model_b_indices,
                outcomes,
                self.k_factor,
                self.match_counts,
                self.win_counts
            )

        print("✓ Processing complete!")

    def get_leaderboard(self) -> List[Tuple]:
        """
        Get sorted leaderboard using NumPy's fast sorting.

        Returns:
            List of tuples (model, rating, matches, wins)
        """
        # Use NumPy's argsort for fast sorting
        sorted_indices = np.argsort(-self.ratings)  # Negative for descending order

        leaderboard = []
        for idx in sorted_indices:
            model = self.idx_to_model[idx]
            rating = float(self.ratings[idx])
            matches = int(self.match_counts[idx])
            wins = float(self.win_counts[idx])
            leaderboard.append((model, rating, matches, wins))

        return leaderboard

    def calculate_win_probability(self, model_a: str, model_b: str) -> float:
        """
        Calculate win probability using fast NumPy operations.

        Args:
            model_a: First model identifier
            model_b: Second model identifier

        Returns:
            Probability that model_a wins
        """
        if model_a not in self.model_to_idx or model_b not in self.model_to_idx:
            return 0.5

        idx_a = self.model_to_idx[model_a]
        idx_b = self.model_to_idx[model_b]

        rating_a = self.ratings[idx_a]
        rating_b = self.ratings[idx_b]

        return expected_score_fast(rating_a, rating_b)

    def get_win_rate_matrix(self) -> Dict[Tuple[str, str], float]:
        """
        Calculate pairwise win probability matrix using vectorized operations.

        Returns:
            Dictionary mapping (model_a, model_b) to win probability
        """
        matrix = {}

        # Vectorized calculation for all pairs
        for i in range(self.n_models):
            model_a = self.idx_to_model[i]

            # Calculate win probabilities against all other models at once
            ratings_a = np.full(self.n_models, self.ratings[i])
            win_probs = calculate_expected_scores_vectorized(ratings_a, self.ratings)

            for j in range(self.n_models):
                if i != j:
                    model_b = self.idx_to_model[j]
                    matrix[(model_a, model_b)] = float(win_probs[j])

        return matrix


def build_leaderboard_optimized(df: pd.DataFrame,
                                initial_rating: float = 1000.0,
                                k_factor: float = 4.0,
                                show_progress: bool = True) -> NumpyEloRatingSystem:
    """
    Build Elo leaderboard using highly optimized NumPy + Numba algorithm.

    This implementation is significantly faster than the basic version:
    - Uses NumPy arrays for O(1) indexing
    - Numba JIT compilation for hot loops
    - Pre-allocated arrays to avoid memory overhead
    - Integer-based model indexing instead of string lookups

    Args:
        df: DataFrame with match data (columns: model_a, model_b, winner)
        initial_rating: Starting rating for all models
        k_factor: Elo learning rate (K-factor)
        show_progress: Whether to display progress bar

    Returns:
        NumpyEloRatingSystem with final ratings
    """
    elo = NumpyEloRatingSystem(initial_rating=initial_rating, k_factor=k_factor)
    elo.process_matches_vectorized(df, show_progress=show_progress)
    return elo

parallel_processing.py

"""
Parallel processing utilities for Elo rating computation
"""
import pandas as pd
import numpy as np
from multiprocessing import Pool, cpu_count
from functools import partial
from typing import List, Tuple
from tqdm import tqdm
from elo_rating import EloRatingSystem


def process_time_slice(args: Tuple) -> Tuple:
    """
    Process a single time slice to build leaderboard.

    Args:
        args: Tuple of (end_date, slice_df, initial_rating, k_factor)

    Returns:
        Tuple of (end_date, leaderboard_data)
    """
    end_date, slice_df, initial_rating, k_factor = args

    # Build Elo system for this time slice
    elo = EloRatingSystem(initial_rating=initial_rating, k_factor=k_factor)

    # Process all matches in this slice
    for _, row in slice_df.iterrows():
        elo.update_ratings(row['model_a'], row['model_b'], row['winner'])

    # Get leaderboard
    leaderboard = elo.get_leaderboard()

    # Convert to list of dicts for easier handling
    lb_data = []
    for rank, (model, rating, matches, wins) in enumerate(leaderboard, 1):
        lb_data.append({
            'model': model,
            'rating': rating,
            'matches': matches,
            'wins': wins,
            'rank': rank,
            'date': end_date
        })

    return (end_date, lb_data)


def build_historical_leaderboards_parallel(df: pd.DataFrame,
                                           time_slices: List[Tuple],
                                           initial_rating: float = 1000.0,
                                           k_factor: float = 32.0,
                                           n_jobs: int = -1) -> List[Tuple]:
    """
    Build historical leaderboards using parallel processing.

    Args:
        df: Full voting DataFrame
        time_slices: List of (end_date, slice_df) tuples
        initial_rating: Starting rating
        k_factor: Elo learning rate
        n_jobs: Number of parallel jobs (-1 for all cores)

    Returns:
        List of (date, leaderboard_data) tuples
    """
    if n_jobs == -1:
        n_jobs = cpu_count()

    print(f"Building historical leaderboards using {n_jobs} cores...")

    # Prepare arguments for parallel processing
    args_list = [
        (end_date, slice_df, initial_rating, k_factor)
        for end_date, slice_df in time_slices
    ]

    # Process in parallel
    with Pool(processes=n_jobs) as pool:
        results = list(tqdm(
            pool.imap(process_time_slice, args_list),
            total=len(args_list),
            desc="Processing time slices"
        ))

    # Convert results to expected format
    historical_leaderboards = []
    for end_date, lb_data in results:
        lb_df = pd.DataFrame(lb_data)
        historical_leaderboards.append((end_date, lb_df))

    # Sort by date
    historical_leaderboards.sort(key=lambda x: x[0])

    return historical_leaderboards


def calculate_pairwise_win_rates_chunk(args: Tuple) -> List[dict]:
    """
    Calculate win rates for a chunk of model pairs.

    Args:
        args: Tuple of (model_pairs, df)

    Returns:
        List of win rate dictionaries
    """
    model_pairs, df = args

    results = []
    for model_a, model_b in model_pairs:
        # Filter matches between these two models
        matches = df[
            ((df['model_a'] == model_a) & (df['model_b'] == model_b)) |
            ((df['model_a'] == model_b) & (df['model_b'] == model_a))
        ]

        if len(matches) == 0:
            continue

        wins_a = 0
        total = len(matches)

        for _, row in matches.iterrows():
            if row['model_a'] == model_a:
                if row['winner'] == 'model_a':
                    wins_a += 1
                elif row['winner'] == 'tie':
                    wins_a += 0.5
            else:  # model_a is model_b in the row
                if row['winner'] == 'model_b':
                    wins_a += 1
                elif row['winner'] == 'tie':
                    wins_a += 0.5

        win_rate = wins_a / total if total > 0 else 0.5

        results.append({
            'model_a': model_a,
            'model_b': model_b,
            'win_rate': win_rate,
            'total_matches': total
        })

    return results


def calculate_win_rate_matrix_parallel(df: pd.DataFrame, 
                                       models: List[str] = None,
                                       n_jobs: int = -1) -> pd.DataFrame:
    """
    Calculate win rate matrix using parallel processing.

    Args:
        df: DataFrame with match data
        models: List of models to include (if None, use all)
        n_jobs: Number of parallel jobs

    Returns:
        DataFrame with win rates
    """
    if n_jobs == -1:
        n_jobs = cpu_count()

    if models is None:
        models = sorted(set(df['model_a'].unique()) | set(df['model_b'].unique()))

    print(f"Calculating win rate matrix for {len(models)} models using {n_jobs} cores...")

    # Generate all model pairs
    model_pairs = [(m1, m2) for i, m1 in enumerate(models) for m2 in models[i+1:]]

    # Split pairs into chunks for parallel processing
    chunk_size = max(1, len(model_pairs) // (n_jobs * 4))
    chunks = [model_pairs[i:i+chunk_size] for i in range(0, len(model_pairs), chunk_size)]

    # Prepare arguments
    args_list = [(chunk, df) for chunk in chunks]

    # Process in parallel
    with Pool(processes=n_jobs) as pool:
        results_chunks = list(tqdm(
            pool.imap(calculate_pairwise_win_rates_chunk, args_list),
            total=len(args_list),
            desc="Calculating win rates"
        ))

    # Flatten results
    all_results = [item for chunk in results_chunks for item in chunk]

    # Build matrix
    win_rates = {model: {opponent: 0.5 for opponent in models} for model in models}

    for result in all_results:
        model_a = result['model_a']
        model_b = result['model_b']
        win_rate = result['win_rate']

        win_rates[model_a][model_b] = win_rate
        win_rates[model_b][model_a] = 1.0 - win_rate

    # Convert to DataFrame
    win_rate_df = pd.DataFrame(win_rates).T
    win_rate_df = win_rate_df[models]

    return win_rate_df


def filter_data_parallel(df: pd.DataFrame, 
                        filters: dict,
                        n_jobs: int = -1) -> pd.DataFrame:
    """
    Filter large DataFrame using parallel processing.

    Args:
        df: Input DataFrame
        filters: Dictionary of filter conditions
        n_jobs: Number of parallel jobs

    Returns:
        Filtered DataFrame
    """
    if n_jobs == -1:
        n_jobs = min(cpu_count(), 4)  # Cap at 4 for filtering

    # Split DataFrame into chunks
    chunk_size = len(df) // n_jobs
    chunks = [df.iloc[i:i+chunk_size] for i in range(0, len(df), chunk_size)]

    def apply_filters(chunk):
        filtered = chunk.copy()

        # Apply each filter
        if 'anony_only' in filters and filters['anony_only'] and 'anony' in filtered.columns:
            filtered = filtered[filtered['anony'] == True]

        if 'language' in filters and filters['language'] and 'language' in filtered.columns:
            filtered = filtered[filtered['language'] == filters['language']]

        if 'min_turn' in filters and 'turn' in filtered.columns:
            filtered = filtered[filtered['turn'] >= filters['min_turn']]

        if 'min_date' in filters and 'tstamp' in filtered.columns:
            min_timestamp = pd.to_datetime(filters['min_date']).timestamp()
            filtered = filtered[filtered['tstamp'] >= min_timestamp]

        if 'max_date' in filters and 'tstamp' in filtered.columns:
            max_timestamp = pd.to_datetime(filters['max_date']).timestamp()
            filtered = filtered[filtered['tstamp'] <= max_timestamp]

        return filtered

    # Process chunks in parallel
    with Pool(processes=n_jobs) as pool:
        filtered_chunks = pool.map(apply_filters, chunks)

    # Combine results
    result = pd.concat(filtered_chunks, ignore_index=True)

    return result


def optimize_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    """
    Optimize DataFrame memory usage by downcasting numeric types.

    Args:
        df: Input DataFrame

    Returns:
        Optimized DataFrame
    """
    print("Optimizing DataFrame memory usage...")

    initial_memory = df.memory_usage(deep=True).sum() / 1024**2

    # Optimize numeric columns
    for col in df.columns:
        col_type = df[col].dtype

        if col_type == 'int64':
            df[col] = pd.to_numeric(df[col], downcast='integer')
        elif col_type == 'float64':
            df[col] = pd.to_numeric(df[col], downcast='float')

    # Convert string columns to category if they have few unique values
    for col in df.select_dtypes(include=['object']).columns:
        try:
            # Check if column contains hashable types (not dict, list, etc.)
            # Try to get unique values - will fail if unhashable
            num_unique = df[col].nunique()
            num_total = len(df[col])

            # If less than 50% unique values, convert to category
            if num_unique / num_total < 0.5:
                df[col] = df[col].astype('category')
        except (TypeError, AttributeError):
            # Column contains unhashable types (dicts, lists), skip optimization
            print(f"  Skipping column '{col}' (contains complex data types)")
            continue

    final_memory = df.memory_usage(deep=True).sum() / 1024**2
    reduction = (1 - final_memory / initial_memory) * 100

    print(f"Memory usage reduced from {initial_memory:.2f} MB to {final_memory:.2f} MB ({reduction:.1f}% reduction)")

    return df

quickstart.py

"""
Quick start demo - minimal example to get started quickly
"""
from elo_rating import EloRatingSystem


def demo_basic_elo():
    """Demonstrate basic Elo rating calculation with synthetic data."""

    print("="*60)
    print("Quick Start: Elo Rating System Demo")
    print("="*60)
    print()

    # Initialize Elo system
    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    # Simulate some matches
    matches = [
        ("GPT-4", "Claude-v1", "GPT-4"),
        ("GPT-4", "Llama-2", "GPT-4"),
        ("Claude-v1", "Llama-2", "Claude-v1"),
        ("GPT-4", "Claude-v1", "tie"),
        ("Llama-2", "Gemini", "Gemini"),
        ("GPT-4", "Gemini", "GPT-4"),
        ("Claude-v1", "Gemini", "Claude-v1"),
        ("GPT-4", "Llama-2", "GPT-4"),
        ("Claude-v1", "Llama-2", "Claude-v1"),
        ("Gemini", "Llama-2", "Gemini"),
    ]

    print("Processing matches:")
    print("-" * 60)
    for i, (model_a, model_b, winner) in enumerate(matches, 1):
        old_rating_a = elo.get_rating(model_a)
        old_rating_b = elo.get_rating(model_b)

        new_rating_a, new_rating_b = elo.update_ratings(model_a, model_b, winner)

        print(f"Match {i}: {model_a} vs {model_b} -> {winner} wins")
        print(f"  {model_a}: {old_rating_a:.1f} → {new_rating_a:.1f} ({new_rating_a-old_rating_a:+.1f})")
        print(f"  {model_b}: {old_rating_b:.1f} → {new_rating_b:.1f} ({new_rating_b-old_rating_b:+.1f})")
        print()

    # Show final leaderboard
    print("=" * 60)
    print("Final Leaderboard:")
    print("=" * 60)
    leaderboard = elo.get_leaderboard()
    for rank, (model, rating, matches, wins) in enumerate(leaderboard, 1):
        win_rate = (wins / matches * 100) if matches > 0 else 0
        print(f"{rank}. {model:15s} - Rating: {rating:7.1f} | "
              f"Matches: {matches:2d} | Wins: {wins:4.1f} | Win Rate: {win_rate:5.1f}%")

    print()

    # Show win probability predictions
    print("=" * 60)
    print("Win Probability Predictions:")
    print("=" * 60)

    models = [m[0] for m in leaderboard]
    for i, model_a in enumerate(models):
        for model_b in models[i+1:]:
            prob = elo.calculate_win_probability(model_a, model_b)
            print(f"{model_a} vs {model_b}: {prob*100:.1f}% - {(1-prob)*100:.1f}%")

    print()
    print("=" * 60)
    print("Demo complete! Check main.py for full analysis with real data.")
    print("=" * 60)


if __name__ == "__main__":
    demo_basic_elo()

test_elo.py

"""
Unit tests for Elo rating system
"""
import pytest
from elo_rating import EloRatingSystem
import math


def test_initial_rating():
    """Test that models start with initial rating."""
    elo = EloRatingSystem(initial_rating=1000.0)
    assert elo.get_rating("model_a") == 1000.0
    assert elo.get_rating("model_b") == 1000.0


def test_expected_score():
    """Test expected score calculation."""
    elo = EloRatingSystem()

    # Equal ratings should give 50% probability
    assert elo.expected_score(1000, 1000) == 0.5

    # Higher rated player should have > 50% probability
    assert elo.expected_score(1200, 1000) > 0.5
    assert elo.expected_score(1000, 1200) < 0.5

    # 400 point difference should give ~91% probability
    prob = elo.expected_score(1400, 1000)
    assert 0.90 < prob < 0.92


def test_rating_update_win():
    """Test rating update when model_a wins."""
    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    new_a, new_b = elo.update_ratings("model_a", "model_b", "model_a")

    # Winner should gain rating, loser should lose rating
    assert new_a > 1000.0
    assert new_b < 1000.0

    # Total rating should be conserved (zero-sum)
    assert abs((new_a + new_b) - 2000.0) < 0.01


def test_rating_update_tie():
    """Test rating update for a tie."""
    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    new_a, new_b = elo.update_ratings("model_a", "model_b", "tie")

    # With equal ratings, tie should not change ratings much
    assert abs(new_a - 1000.0) < 0.01
    assert abs(new_b - 1000.0) < 0.01


def test_upset_gives_larger_change():
    """Test that unexpected results cause larger rating changes."""
    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    # Give model_a higher rating
    elo.ratings["model_a"] = 1200.0
    elo.ratings["model_b"] = 1000.0

    # If weaker model wins (upset), changes should be larger
    new_a_upset, new_b_upset = elo.update_ratings("model_a", "model_b", "model_b")

    # Reset
    elo.ratings["model_a"] = 1200.0
    elo.ratings["model_b"] = 1000.0

    # If stronger model wins (expected), changes should be smaller
    new_a_expected, new_b_expected = elo.update_ratings("model_a", "model_b", "model_a")

    # Upset should cause larger change
    change_upset = abs(new_a_upset - 1200.0)
    change_expected = abs(new_a_expected - 1200.0)

    assert change_upset > change_expected


def test_leaderboard_sorting():
    """Test that leaderboard is sorted by rating."""
    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    # Create some matches to differentiate ratings
    elo.update_ratings("model_a", "model_b", "model_a")
    elo.update_ratings("model_a", "model_c", "model_a")
    elo.update_ratings("model_b", "model_c", "model_b")

    leaderboard = elo.get_leaderboard()

    # Check descending order
    for i in range(len(leaderboard) - 1):
        assert leaderboard[i][1] >= leaderboard[i+1][1]

    # model_a should be first (won all matches)
    assert leaderboard[0][0] == "model_a"


def test_win_probability_symmetry():
    """Test that win probabilities sum to 1."""
    elo = EloRatingSystem()
    elo.ratings["model_a"] = 1200.0
    elo.ratings["model_b"] = 1000.0

    prob_a = elo.calculate_win_probability("model_a", "model_b")
    prob_b = elo.calculate_win_probability("model_b", "model_a")

    # Should sum to 1
    assert abs(prob_a + prob_b - 1.0) < 0.001


def test_match_counting():
    """Test that match and win counts are tracked correctly."""
    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    elo.update_ratings("model_a", "model_b", "model_a")  # model_a wins
    elo.update_ratings("model_a", "model_c", "model_b")  # model_a loses (2nd slot wins)
    elo.update_ratings("model_a", "model_b", "tie")      # tie -> 0.5 each

    # model_a played 3 matches
    assert elo.match_counts["model_a"] == 3

    # model_a won 1 match and tied 1 (1.5 total)
    assert elo.win_counts["model_a"] == 1.5

    # model_b played 2 matches
    assert elo.match_counts["model_b"] == 2


def test_copy():
    """Test that copy creates independent instance."""
    elo1 = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)
    elo1.update_ratings("model_a", "model_b", "model_a")

    elo2 = elo1.copy()

    # Modify elo2
    elo2.update_ratings("model_a", "model_b", "model_b")

    # elo1 should be unchanged
    assert elo1.ratings["model_a"] != elo2.ratings["model_a"]


def test_reset():
    """Test that reset clears all data."""
    elo = EloRatingSystem(initial_rating=1000.0, k_factor=32.0)

    elo.update_ratings("model_a", "model_b", "model_a")
    elo.update_ratings("model_a", "model_c", "model_a")

    assert len(elo.ratings) > 0

    elo.reset()

    assert len(elo.ratings) == 0
    assert len(elo.match_counts) == 0
    assert len(elo.win_counts) == 0


if __name__ == "__main__":
    # Run tests
    pytest.main([__file__, "-v"])

test_filter_empty.py

"""
Regression test for filter_data on empty input (实验 6-6 排行榜).

An empty arena data file (e.g. a failed/truncated download saved as `[]`) used to
crash with ZeroDivisionError at the "After filtering" percentage print.
"""
import pandas as pd
import pytest

from data_loader import filter_data


def test_filter_data_tolerates_empty_dataframe():
    """Empty input no longer raises ZeroDivisionError; returns an empty DataFrame."""
    empty = pd.DataFrame({"model_a": [], "model_b": [], "winner": []})
    result = filter_data(empty)
    assert len(result) == 0


def test_filter_data_normal_case_unchanged():
    """Non-empty input still filters and reports normally."""
    df = pd.DataFrame({
        "model_a": ["a", "b", "a"],
        "model_b": ["b", "a", "c"],
        "winner": ["model_a", "model_b", "tie"],
        "anony": [True, True, False],
    })
    result = filter_data(df, anony_only=True, use_dedup=False)
    assert len(result) == 2   # 非匿名的一条被过滤


if __name__ == "__main__":
    pytest.main([__file__, "-v"])

visualization.py

"""
Visualization tools for Elo leaderboard analysis
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from typing import List, Tuple, Optional
import plotly.graph_objects as go
import plotly.express as px


def plot_leaderboard(leaderboard_data: list, top_n: int = 20, save_path: Optional[str] = None):
    """
    Plot static leaderboard bar chart.

    Args:
        leaderboard_data: List of (model, rating, matches, wins) tuples
        top_n: Number of top models to display
        save_path: If provided, save figure to this path
    """
    # Convert to DataFrame and get top N
    df = pd.DataFrame(leaderboard_data, columns=['model', 'rating', 'matches', 'wins'])
    df = df.head(top_n)

    # Create figure
    fig, ax = plt.subplots(figsize=(12, 8))

    # Create horizontal bar chart
    bars = ax.barh(range(len(df)), df['rating'], color=plt.cm.viridis(np.linspace(0, 1, len(df))))

    # Customize
    ax.set_yticks(range(len(df)))
    ax.set_yticklabels(df['model'])
    ax.set_xlabel('Elo Rating', fontsize=12)
    ax.set_title(f'Model Leaderboard - Top {top_n} Models', fontsize=14, fontweight='bold')
    ax.invert_yaxis()  # Highest rating at top

    # Add value labels on bars
    for i, (rating, matches) in enumerate(zip(df['rating'], df['matches'])):
        ax.text(rating + 10, i, f'{rating:.0f} ({matches} matches)', 
                va='center', fontsize=9)

    ax.grid(axis='x', alpha=0.3)
    plt.tight_layout()

    if save_path:
        plt.savefig(save_path, dpi=300, bbox_inches='tight')
        print(f"Saved leaderboard to {save_path}")

    plt.show()


def plot_win_rate_matrix(win_rate_df: pd.DataFrame, 
                         top_n: int = 15,
                         save_path: Optional[str] = None):
    """
    Plot heatmap of win rate matrix.

    Args:
        win_rate_df: DataFrame with win rates (rows beat columns)
        top_n: Number of models to include
        save_path: If provided, save figure to this path
    """
    # Get top N models by average win rate
    avg_win_rates = win_rate_df.mean(axis=1).sort_values(ascending=False)
    top_models = avg_win_rates.head(top_n).index.tolist()

    # Subset matrix
    subset = win_rate_df.loc[top_models, top_models]

    # Create figure
    fig, ax = plt.subplots(figsize=(14, 12))

    # Plot heatmap
    sns.heatmap(subset, annot=True, fmt='.2f', cmap='RdYlGn', center=0.5,
                vmin=0, vmax=1, square=True, linewidths=0.5,
                cbar_kws={'label': 'Win Rate'}, ax=ax)

    ax.set_title(f'Win Rate Matrix - Top {top_n} Models\n(Row vs Column)', 
                 fontsize=14, fontweight='bold')
    ax.set_xlabel('Opponent (Column)', fontsize=12)
    ax.set_ylabel('Model (Row)', fontsize=12)

    plt.tight_layout()

    if save_path:
        plt.savefig(save_path, dpi=300, bbox_inches='tight')
        print(f"Saved win rate matrix to {save_path}")

    plt.show()


def plot_rating_history(history_df: pd.DataFrame, 
                        models: Optional[List[str]] = None,
                        top_n: int = 10,
                        save_path: Optional[str] = None):
    """
    Plot rating evolution over time for selected models.

    Args:
        history_df: DataFrame with columns: date, model, rating
        models: List of specific models to plot (if None, plot top N)
        top_n: If models not specified, plot top N models by final rating
        save_path: If provided, save figure to this path
    """
    if models is None:
        # Get top N models by final rating
        final_date = history_df['date'].max()
        final_ratings = history_df[history_df['date'] == final_date].nlargest(top_n, 'rating')
        models = final_ratings['model'].tolist()

    # Filter data
    plot_data = history_df[history_df['model'].isin(models)].copy()

    # Create figure
    fig, ax = plt.subplots(figsize=(14, 8))

    # Plot each model
    for model in models:
        model_data = plot_data[plot_data['model'] == model].sort_values('date')
        ax.plot(model_data['date'], model_data['rating'], marker='o', 
                label=model, linewidth=2, markersize=4)

    ax.set_xlabel('Date', fontsize=12)
    ax.set_ylabel('Elo Rating', fontsize=12)
    ax.set_title('Model Rating Evolution Over Time', fontsize=14, fontweight='bold')
    ax.legend(bbox_to_anchor=(1.05, 1), loc='upper left', fontsize=9)
    ax.grid(alpha=0.3)

    plt.xticks(rotation=45)
    plt.tight_layout()

    if save_path:
        plt.savefig(save_path, dpi=300, bbox_inches='tight')
        print(f"Saved rating history to {save_path}")

    plt.show()


def plot_rating_distribution(leaderboard_data: list, save_path: Optional[str] = None):
    """
    Plot distribution of ratings across all models.

    Args:
        leaderboard_data: List of (model, rating, matches, wins) tuples
        save_path: If provided, save figure to this path
    """
    df = pd.DataFrame(leaderboard_data, columns=['model', 'rating', 'matches', 'wins'])

    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))

    # Histogram
    ax1.hist(df['rating'], bins=30, color='steelblue', edgecolor='black', alpha=0.7)
    ax1.axvline(df['rating'].mean(), color='red', linestyle='--', 
                linewidth=2, label=f'Mean: {df["rating"].mean():.1f}')
    ax1.axvline(df['rating'].median(), color='green', linestyle='--', 
                linewidth=2, label=f'Median: {df["rating"].median():.1f}')
    ax1.set_xlabel('Elo Rating', fontsize=12)
    ax1.set_ylabel('Count', fontsize=12)
    ax1.set_title('Rating Distribution', fontsize=13, fontweight='bold')
    ax1.legend()
    ax1.grid(alpha=0.3)

    # Box plot
    ax2.boxplot(df['rating'], vert=True)
    ax2.set_ylabel('Elo Rating', fontsize=12)
    ax2.set_title('Rating Box Plot', fontsize=13, fontweight='bold')
    ax2.grid(alpha=0.3)

    plt.tight_layout()

    if save_path:
        plt.savefig(save_path, dpi=300, bbox_inches='tight')
        print(f"Saved rating distribution to {save_path}")

    plt.show()


def create_interactive_leaderboard(history_df: pd.DataFrame, top_n: int = 15) -> go.Figure:
    """
    Create interactive Plotly visualization of ranking evolution.

    Args:
        history_df: DataFrame with columns: date, model, rating, rank
        top_n: Number of top models to include

    Returns:
        Plotly Figure object
    """
    # Get top N models by final rating
    final_date = history_df['date'].max()
    final_ratings = history_df[history_df['date'] == final_date].nlargest(top_n, 'rating')
    top_models = final_ratings['model'].tolist()

    # Filter data
    plot_data = history_df[history_df['model'].isin(top_models)].copy()

    # Create figure
    fig = go.Figure()

    for model in top_models:
        model_data = plot_data[plot_data['model'] == model].sort_values('date')

        fig.add_trace(go.Scatter(
            x=model_data['date'],
            y=model_data['rating'],
            mode='lines+markers',
            name=model,
            hovertemplate='<b>%{fullData.name}</b><br>' +
                         'Date: %{x}<br>' +
                         'Rating: %{y:.0f}<br>' +
                         '<extra></extra>'
        ))

    fig.update_layout(
        title='Interactive Model Rating Evolution',
        xaxis_title='Date',
        yaxis_title='Elo Rating',
        hovermode='closest',
        height=600,
        legend=dict(
            yanchor="top",
            y=0.99,
            xanchor="left",
            x=0.01
        )
    )

    return fig


def create_rank_evolution_chart(history_df: pd.DataFrame, top_n: int = 15) -> go.Figure:
    """
    Create interactive rank evolution chart (lower rank number is better).

    Args:
        history_df: DataFrame with columns: date, model, rating, rank
        top_n: Number of models to track

    Returns:
        Plotly Figure object
    """
    # Get models that were ever in top N
    models_in_top = history_df[history_df['rank'] <= top_n]['model'].unique()

    # Filter data
    plot_data = history_df[history_df['model'].isin(models_in_top)].copy()

    # Create figure
    fig = go.Figure()

    for model in models_in_top:
        model_data = plot_data[plot_data['model'] == model].sort_values('date')

        fig.add_trace(go.Scatter(
            x=model_data['date'],
            y=model_data['rank'],
            mode='lines+markers',
            name=model,
            hovertemplate='<b>%{fullData.name}</b><br>' +
                         'Date: %{x}<br>' +
                         'Rank: #%{y}<br>' +
                         '<extra></extra>'
        ))

    fig.update_layout(
        title=f'Model Rank Evolution (Top {top_n})',
        xaxis_title='Date',
        yaxis_title='Rank',
        yaxis=dict(autorange='reversed'),  # Lower rank at top
        hovermode='closest',
        height=600,
        legend=dict(
            yanchor="top",
            y=0.99,
            xanchor="right",
            x=0.99
        )
    )

    return fig