from __future__ import annotations import copy import json import random from pathlib import Path import numpy as np import torch import trackio from model import FederatedMLP, parameter_count from safetensors.torch import save_file from sklearn.datasets import load_digits from sklearn.metrics import accuracy_score from sklearn.model_selection import train_test_split from torch.nn import functional as F from torch.utils.data import DataLoader, TensorDataset PROJECT_DIR = Path(__file__).resolve().parent ARTIFACT_DIR = PROJECT_DIR / "artifacts" / "fedavg-global" CLIENTS = 5 def seed_everything(seed: int) -> None: random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) def dirichlet_partition( labels: np.ndarray, clients: int = CLIENTS, alpha: float = 0.25, seed: int = 2026, ) -> list[np.ndarray]: rng = np.random.default_rng(seed) assignments = [[] for _ in range(clients)] for label in sorted(np.unique(labels)): indices = np.flatnonzero(labels == label) rng.shuffle(indices) proportions = rng.dirichlet(np.full(clients, alpha)) boundaries = (np.cumsum(proportions)[:-1] * len(indices)).astype(int) for client, shard in enumerate(np.split(indices, boundaries)): assignments[client].extend(shard.tolist()) return [np.asarray(sorted(indices), dtype=np.int64) for indices in assignments] def train_local( global_model: FederatedMLP, dataset: TensorDataset, *, local_epochs: int, learning_rate: float, ) -> tuple[dict[str, torch.Tensor], float]: model = copy.deepcopy(global_model) loader = DataLoader(dataset, batch_size=48, shuffle=True) optimizer = torch.optim.SGD( model.parameters(), lr=learning_rate, momentum=0.85, ) losses = [] for _ in range(local_epochs): model.train() for pixels, labels in loader: logits = model(pixels) loss = F.cross_entropy(logits, labels) optimizer.zero_grad(set_to_none=True) loss.backward() optimizer.step() losses.append(loss.item()) return ( {key: value.detach().clone() for key, value in model.state_dict().items()}, float(np.mean(losses)), ) def fedavg( states: list[dict[str, torch.Tensor]], weights: list[int], ) -> dict[str, torch.Tensor]: total = sum(weights) averaged = {} for key in states[0]: averaged[key] = sum( state[key] * (weight / total) for state, weight in zip(states, weights, strict=True) ) return averaged def update_disagreement( global_state: dict[str, torch.Tensor], client_states: list[dict[str, torch.Tensor]], ) -> float: distances = [] for state in client_states: squared = sum( torch.sum((state[key] - global_state[key]) ** 2).item() for key in global_state ) distances.append(squared**0.5) return float(np.mean(distances)) @torch.inference_mode() def evaluate(model: FederatedMLP, dataset: TensorDataset) -> float: model.eval() loader = DataLoader(dataset, batch_size=256, shuffle=False) labels, predictions = [], [] for pixels, targets in loader: predictions.extend(model(pixels).argmax(dim=1).tolist()) labels.extend(targets.tolist()) return float(accuracy_score(labels, predictions)) def centralized_control( dataset: TensorDataset, test: TensorDataset, epochs: int, ) -> tuple[FederatedMLP, float]: seed_everything(2026) model = FederatedMLP() loader = DataLoader( dataset, batch_size=48, shuffle=True, generator=torch.Generator().manual_seed(2026), ) optimizer = torch.optim.SGD(model.parameters(), lr=0.06, momentum=0.85) for _ in range(epochs): model.train() for pixels, labels in loader: loss = F.cross_entropy(model(pixels), labels) optimizer.zero_grad(set_to_none=True) loss.backward() optimizer.step() return model, evaluate(model, test) def main() -> None: seed_everything(2026) digits = load_digits() pixels = digits.data.astype(np.float32) / 16.0 labels = digits.target.astype(np.int64) train_indices, test_indices = train_test_split( np.arange(len(labels)), test_size=0.25, stratify=labels, random_state=2026, ) train_pixels = torch.from_numpy(pixels[train_indices]) train_labels = torch.from_numpy(labels[train_indices]) test_dataset = TensorDataset( torch.from_numpy(pixels[test_indices]), torch.from_numpy(labels[test_indices]), ) client_indices = dirichlet_partition(labels[train_indices]) client_datasets = [ TensorDataset(train_pixels[indices], train_labels[indices]) for indices in client_indices ] distributions = [ np.bincount(train_labels[indices].numpy(), minlength=10).tolist() for indices in client_indices ] rounds = 30 global_model = FederatedMLP() history = [] trackio.init( project="federated-digits", name="fedavg-dirichlet-alpha-0.25", config={ "parameters": parameter_count(global_model), "clients": CLIENTS, "rounds": rounds, "local_epochs": 2, "dirichlet_alpha": 0.25, "client_sizes": [len(dataset) for dataset in client_datasets], }, ) for round_index in range(1, rounds + 1): global_state = { key: value.detach().clone() for key, value in global_model.state_dict().items() } client_states = [] client_losses = [] for dataset in client_datasets: state, loss = train_local( global_model, dataset, local_epochs=2, learning_rate=0.06, ) client_states.append(state) client_losses.append(loss) disagreement = update_disagreement(global_state, client_states) global_model.load_state_dict( fedavg(client_states, [len(dataset) for dataset in client_datasets]) ) accuracy = evaluate(global_model, test_dataset) record = { "round": round_index, "global_test_accuracy": accuracy, "mean_client_loss": float(np.mean(client_losses)), "client_update_disagreement": disagreement, } history.append(record) trackio.log(record) trackio.finish() full_train = TensorDataset(train_pixels, train_labels) centralized_model, centralized_accuracy = centralized_control( full_train, test_dataset, epochs=rounds * 2, ) ARTIFACT_DIR.mkdir(parents=True, exist_ok=True) save_file(global_model.state_dict(), ARTIFACT_DIR / "model.safetensors") results = { "model": "Federated Digits FedAvg", "parameters": parameter_count(global_model), "clients": CLIENTS, "dirichlet_alpha": 0.25, "client_sizes": [len(dataset) for dataset in client_datasets], "client_class_distributions": distributions, "rounds": rounds, "local_epochs_per_round": 2, "initial_test_accuracy": history[0]["global_test_accuracy"], "final_test_accuracy": history[-1]["global_test_accuracy"], "best_test_accuracy": max(record["global_test_accuracy"] for record in history), "centralized_control_accuracy": centralized_accuracy, "federated_centralized_gap": ( history[-1]["global_test_accuracy"] - centralized_accuracy ), "history": history, } (ARTIFACT_DIR / "evaluation.json").write_text( json.dumps(results, indent=2), encoding="utf-8", ) print(json.dumps(results, indent=2)) if __name__ == "__main__": main()