ARENA Score

ARENA Score (Adaptive Review and Evaluation using Novel Aggregation Score) is a novel client evaluation and weighted aggregation algorithm for Federated Learning.

🎯 Key Features

• Adaptive Client Evaluation: Dynamically assesses client update quality using multiple metrics
• Robust Aggregation: Filters out unreliable/malicious client contributions
• Anomaly Detection: Detects Byzantine clients using KL divergence and cosine similarity
• Gradient Recycling: Maintains momentum by reusing successful updates from missing clients
• Model Agnostic: Works with any model that implements the simple weight interface

📊 ARENA Score Formula

S_j = α(t) × max(0, ΔAcc_j) + γ(t) × CS_j + η × SN_j/(1 + SN_j)

Where:

• ΔAcc - Local accuracy improvement
• CS - Cosine similarity between client update and global direction
• SN - Spectral norm: log(1 + σ_max(W_j))
• α(t), γ(t) - Adaptive time-varying weights

🚀 Installation

pip install arena-score

For Numba acceleration (optional):

pip install arena-score[accelerate]

📝 Quick Start

from arena_score import ARENAScoreClient, ARENAScoreServer, run_arena_score

# Create clients with your model
clients = [
    ARENAScoreClient(
        client_id=i,
        model=your_model.copy(),  # Model with get_weights/set_weights
        X=X_train[i],
        y=y_train[i],
        local_epochs=5,
        batch_size=32
    )
    for i in range(n_clients)
]

# Run ARENA Score federated learning
history = run_arena_score(
    global_model=your_model,
    clients=clients,
    n_rounds=10,
    eval_data=(X_test, y_test)
)

print(f"Final accuracy: {history['test_accuracy'][-1]:.4f}")

🔧 Advanced Usage

Custom Server Configuration

from arena_score import ARENAScoreServer

server = ARENAScoreServer(
    global_model=model,
    alpha_0=0.7,           # Initial accuracy weight
    alpha_min=0.3,         # Minimum accuracy weight
    gamma_0=0.3,           # Initial cosine similarity weight
    gamma_min=0.5,         # Minimum cosine similarity weight
    eta=0.5,               # Spectral norm coefficient
    lambda_decay=0.1,      # Decay rate for adaptive weights
    anomaly_threshold=-0.5, # CS threshold for anomaly detection
    kl_threshold=0.5,      # KL divergence threshold
    enable_gradient_recycling=True
)

Computing ARENA Score Directly

from arena_score import compute_arena_score

score = compute_arena_score(
    delta_acc=0.05,      # Accuracy improvement
    cosine_sim=0.8,      # Cosine similarity
    spectral_norm=2.5,   # Spectral norm
    alpha_t=0.5,         # Current alpha weight
    gamma_t=0.4          # Current gamma weight
)

Utility Functions

from arena_score import (
    compute_cosine_similarity,
    compute_spectral_norm,
    compute_kl_divergence,
    flatten_weights,
    unflatten_weights
)

# Compute cosine similarity between weight vectors
cs = compute_cosine_similarity(weights_a, weights_b)

# Compute spectral norm of weights
sn = compute_spectral_norm(weights_dict)

📈 Model Interface

Your model must implement these methods:

class YourModel:
    def get_weights(self) -> Dict[str, np.ndarray]:
        """Return model weights as a dictionary."""
        pass
    
    def set_weights(self, weights: Dict[str, np.ndarray]):
        """Set model weights from a dictionary."""
        pass
    
    def forward(self, X: np.ndarray) -> np.ndarray:
        """Forward pass, returns predictions."""
        pass
    
    def backward(self, X, y, y_pred) -> Dict[str, np.ndarray]:
        """Backward pass, returns gradients."""
        pass
    
    def update_weights(self, gradients: Dict[str, np.ndarray]):
        """Update weights using gradients."""
        pass
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Make predictions (0/1 for classification)."""
        pass
    
    def predict_proba(self, X: np.ndarray) -> np.ndarray:
        """Return probability predictions."""
        pass
    
    def copy(self) -> 'YourModel':
        """Return a deep copy of the model."""
        pass

📜 License

MIT License - see LICENSE for details.