Federated Learning: тренування AI без крадіжки даних — як навчити модель на мільярдах пристроїв

Твій телефон знає, які слова ти часто друкуєш. Твоя лікарня має тисячі медичних знімків. Твій банк бачить всі транзакції. Ці дані могли б натренувати найкращий AI у світі. Але ти не хочеш, щоб Google читав твої повідомлення. Лікарня не може ділитися медичними даними згідно HIPAA. Банк зв'язаний регуляціями PCI DSS.

Federated Learning — революційний підхід до машинного навчання, який дозволяє тренувати модель на всіх цих даних, не збираючи їх в одному місці. Дані залишаються у власників. Моделі — подорожують. Це не компроміс між privacy та якістю моделі. Це краще рішення для обох.

Проблема централізованого ML: чому традиційний підхід не працює

Традиційний підхід до machine learning:

Клієнт 1 → дані →
Клієнт 2 → дані →  Центральний сервер → Модель
Клієнт 3 → дані →

Здається простим, але створює каскад проблем, які стають все серйознішими з кожним роком.

1. Privacy Catastrophe

Всі дані видимі серверу. Один breach — і мільйони records публічні. Equifax (2017) — 147 мільйонів людей. Yahoo — 3 мільярди акаунтів. Це не теорія — це реальність.

2. Bandwidth Nightmare

Передача терабайтів даних по мережі. Медичні зображення, відео, аудіо — все це має фізично переміститися на сервер. Для мільярдів пристроїв це просто неможливо.

3. Regulatory Compliance Violations

GDPR в Європі забороняє передачу персональних даних без explicit consent. HIPAA в США криміналізує неавторизований доступ до медичних даних. CCPA в Каліфорнії дає користувачам право на видалення. Централізація даних порушує все це одночасно.

4. Trust Issues

Клієнти не довіряють серверу. І правильно роблять. Навіть добросовісні компанії можуть бути зламані, продані, або змінити політику privacy.

5. Single Point of Failure

Один витік — і все пропало. Немає можливості відкликати дані після breach.

Federated Learning: парадигма «модель до даних»

Центральний принцип: Не дані до моделі, а модель до даних.

Round 1:
Server: глобальна модель M₀
        ↓ broadcast (тільки weights ~100MB)
┌───────┼───────┐
↓       ↓       ↓
Client1 Client2 Client3
(local  (local  (local
train)  train)  train)
(data   (data   (data
stays!) stays!) stays!)
↓       ↓       ↓
└───────┼───────┘
        ↓ aggregate (тільки gradients)
Server: M₁ = aggregate(M₁¹, M₁², M₁³)

Round 2-N: repeat until convergence

Що передається по мережі:

• Gradient updates (зміни ваг)
• Model weights (параметри моделі)
• НЕ raw data

Що залишається локально:

• Всі training samples
• Personal information
• Sensitive content

FedAvg: базовий алгоритм з глибоким розбором

Federated Averaging (McMahan et al., 2017) — це фундаментальний алгоритм, з якого починається будь-яке вивчення FL.

import numpy as np
import copy
from typing import List, Dict, Tuple

class FederatedAveragingServer:
    """
    Сервер для Federated Averaging.
    Координує тренування без доступу до даних клієнтів.
    """

    def __init__(self, model, num_rounds: int = 100,
                 clients_per_round: int = 10,
                 local_epochs: int = 5,
                 learning_rate: float = 0.01):
        self.global_model = model
        self.num_rounds = num_rounds
        self.clients_per_round = clients_per_round
        self.local_epochs = local_epochs
        self.lr = learning_rate
        self.history = {'loss': [], 'accuracy': []}

    def aggregate_weights(self, client_weights: List[Dict],
                          client_sizes: List[int]) -> Dict:
        """
        Weighted average of client model weights.
        Більші datasets мають більший вплив.
        """
        total_samples = sum(client_sizes)

        # Ініціалізуємо aggregated weights нулями
        aggregated = {}
        for key in client_weights[0].keys():
            aggregated[key] = np.zeros_like(client_weights[0][key])

        # Weighted sum
        for weights, size in zip(client_weights, client_sizes):
            weight_factor = size / total_samples
            for key in weights.keys():
                aggregated[key] += weights[key] * weight_factor

        return aggregated

    def select_clients(self, clients: List, k: int) -> List:
        """
        Випадковий вибір клієнтів для раунду.
        В реальності враховується availability, battery level, network.
        """
        return np.random.choice(clients, size=min(k, len(clients)),
                                replace=False).tolist()

    def train_round(self, clients: List) -> Tuple[Dict, float]:
        """
        Один раунд федеративного навчання.
        """
        # 1. Вибираємо підмножину клієнтів
        selected = self.select_clients(clients, self.clients_per_round)

        # 2. Розсилаємо поточні глобальні ваги
        global_weights = self.global_model.get_weights()

        # 3. Клієнти тренуються локально
        client_weights = []
        client_sizes = []
        client_losses = []

        for client in selected:
            # Клієнт отримує копію глобальної моделі
            local_weights, local_size, local_loss = client.local_train(
                global_weights,
                epochs=self.local_epochs,
                lr=self.lr
            )
            client_weights.append(local_weights)
            client_sizes.append(local_size)
            client_losses.append(local_loss)

        # 4. Агрегуємо результати
        new_global_weights = self.aggregate_weights(client_weights, client_sizes)

        # 5. Оновлюємо глобальну модель
        self.global_model.set_weights(new_global_weights)

        avg_loss = np.average(client_losses, weights=client_sizes)
        return new_global_weights, avg_loss

    def train(self, clients: List) -> Dict:
        """
        Повний цикл федеративного навчання.
        """
        for round_num in range(self.num_rounds):
            weights, loss = self.train_round(clients)
            self.history['loss'].append(loss)

            if round_num % 10 == 0:
                print(f"Round {round_num}: Loss = {loss:.4f}")

        return self.history


class FederatedClient:
    """
    Клієнт федеративного навчання.
    Має локальні дані та виконує тренування на них.
    """

    def __init__(self, client_id: str, local_data, local_labels, model_fn):
        self.client_id = client_id
        self.local_data = local_data
        self.local_labels = local_labels
        self.model = model_fn()
        self.data_size = len(local_data)

    def local_train(self, global_weights: Dict,
                    epochs: int, lr: float) -> Tuple[Dict, int, float]:
        """
        Локальне тренування на даних клієнта.
        Дані НІКОЛИ не покидають цей метод.
        """
        # Завантажуємо глобальні ваги
        self.model.set_weights(copy.deepcopy(global_weights))

        # Тренуємо локально
        total_loss = 0
        for epoch in range(epochs):
            # Shuffle local data
            indices = np.random.permutation(self.data_size)
            shuffled_data = self.local_data[indices]
            shuffled_labels = self.local_labels[indices]

            # Mini-batch training
            batch_size = 32
            for i in range(0, self.data_size, batch_size):
                batch_x = shuffled_data[i:i+batch_size]
                batch_y = shuffled_labels[i:i+batch_size]

                loss = self.model.train_step(batch_x, batch_y, lr)
                total_loss += loss

        avg_loss = total_loss / (epochs * (self.data_size // batch_size))

        # Повертаємо ТІЛЬКИ ваги, не дані
        return self.model.get_weights(), self.data_size, avg_loss

Differential Privacy: коли градієнтів недостатньо

Проблема: Навіть gradients можуть витікати інформацію про training data.

Membership Inference Attack:

class MembershipInferenceAttack:
    """
    Атака для визначення, чи був конкретний sample
    в training data моделі.
    """

    def __init__(self, target_model, shadow_models):
        self.target = target_model
        self.shadows = shadow_models
        self.attack_model = self._train_attack_model()

    def _train_attack_model(self):
        """
        Тренуємо класифікатор на shadow models.
        Вхід: confidence scores моделі
        Вихід: member / non-member
        """
        features = []
        labels = []

        for shadow in self.shadows:
            # Отримуємо predictions для training data (members)
            member_preds = shadow.predict(shadow.train_data)
            features.extend(member_preds)
            labels.extend([1] * len(member_preds))

            # Отримуємо predictions для test data (non-members)
            non_member_preds = shadow.predict(shadow.test_data)
            features.extend(non_member_preds)
            labels.extend([0] * len(non_member_preds))

        return train_classifier(features, labels)

    def attack(self, sample) -> float:
        """
        Повертає ймовірність того, що sample був у training data.
        """
        prediction = self.target.predict([sample])
        return self.attack_model.predict_proba([prediction])[0][1]


class GradientLeakageAttack:
    """
    Deep Leakage from Gradients (Zhu et al., 2019)
    Відновлення training data з градієнтів.
    """

    def __init__(self, model, gradient):
        self.model = model
        self.target_gradient = gradient

    def reconstruct(self, num_iterations=1000):
        """
        Ітеративно відновлюємо вхідні дані.
        """
        # Починаємо з випадкових даних
        dummy_data = torch.randn_like(self.model.expected_input)
        dummy_data.requires_grad = True

        optimizer = torch.optim.LBFGS([dummy_data])

        for i in range(num_iterations):
            def closure():
                optimizer.zero_grad()

                # Обчислюємо градієнт для dummy даних
                dummy_gradient = self.model.compute_gradient(dummy_data)

                # Мінімізуємо різницю між градієнтами
                loss = sum([
                    ((dummy_g - target_g) ** 2).sum()
                    for dummy_g, target_g in zip(dummy_gradient, self.target_gradient)
                ])

                loss.backward()
                return loss

            optimizer.step(closure)

        return dummy_data.detach()

Рішення: Differential Privacy (DP)

import torch
import numpy as np
from typing import List, Tuple

class DifferentiallyPrivateSGD:
    """
    DP-SGD: Gradient clipping + Gaussian noise.
    Математична гарантія (ε, δ)-differential privacy.
    """

    def __init__(self, model,
                 max_grad_norm: float = 1.0,  # C
                 noise_multiplier: float = 1.0,  # σ
                 delta: float = 1e-5):
        self.model = model
        self.max_grad_norm = max_grad_norm
        self.noise_multiplier = noise_multiplier
        self.delta = delta

        # Privacy accountant для трекінгу витраченого бюджету
        self.privacy_spent = 0.0

    def clip_gradients(self, gradients: List[torch.Tensor]) -> List[torch.Tensor]:
        """
        Per-sample gradient clipping.
        Обмежує вплив одного sample на модель.
        """
        clipped = []
        for grad in gradients:
            grad_norm = torch.norm(grad)
            clip_factor = min(1.0, self.max_grad_norm / (grad_norm + 1e-8))
            clipped.append(grad * clip_factor)
        return clipped

    def add_noise(self, gradients: List[torch.Tensor],
                  batch_size: int) -> List[torch.Tensor]:
        """
        Додаємо калібрований Gaussian noise.
        Noise scale = σ * C / batch_size
        """
        noised = []
        noise_scale = self.noise_multiplier * self.max_grad_norm / batch_size

        for grad in gradients:
            noise = torch.randn_like(grad) * noise_scale
            noised.append(grad + noise)

        return noised

    def private_step(self, batch_data, batch_labels) -> float:
        """
        Один крок DP-SGD.
        """
        batch_size = len(batch_data)

        # 1. Обчислюємо per-sample gradients
        per_sample_grads = []
        for x, y in zip(batch_data, batch_labels):
            grad = self.model.compute_gradient(x.unsqueeze(0), y.unsqueeze(0))
            per_sample_grads.append(grad)

        # 2. Clip кожен gradient окремо
        clipped_grads = [self.clip_gradients(g) for g in per_sample_grads]

        # 3. Aggregate (mean)
        aggregated = []
        for i in range(len(clipped_grads[0])):
            layer_grads = [g[i] for g in clipped_grads]
            aggregated.append(torch.stack(layer_grads).mean(dim=0))

        # 4. Add noise
        noised_grads = self.add_noise(aggregated, batch_size)

        # 5. Apply to model
        self.model.apply_gradients(noised_grads)

        # Update privacy budget
        self._update_privacy_accountant(batch_size)

        return self.model.current_loss

    def _update_privacy_accountant(self, batch_size: int):
        """
        Moments accountant для tight privacy bounds.
        """
        # Simplified: в реальності використовують RDP composition
        q = batch_size / self.total_dataset_size
        sigma = self.noise_multiplier

        # Approximate privacy cost per step
        step_epsilon = q * np.sqrt(2 * np.log(1.25 / self.delta)) / sigma
        self.privacy_spent += step_epsilon

    def get_privacy_spent(self) -> Tuple[float, float]:
        """
        Повертає (epsilon, delta) — скільки privacy витрачено.
        """
        return (self.privacy_spent, self.delta)

Non-IID Data: головний виклик федеративного навчання

Проблема: У реальному світі дані на клієнтах не є IID (Independent and Identically Distributed).

class NonIIDDataDistribution:
    """
    Моделювання різних типів non-IID розподілів.
    """

    @staticmethod
    def label_skew(data, labels, num_clients: int,
                   classes_per_client: int = 2):
        """
        Кожен клієнт має тільки певні класи.
        Приклад: User A бачить тільки котів і собак,
                 User B — тільки машини і літаки.
        """
        num_classes = len(np.unique(labels))
        client_data = [[] for _ in range(num_clients)]
        client_labels = [[] for _ in range(num_clients)]

        # Розподіляємо класи по клієнтах
        class_assignments = {}
        for c in range(num_classes):
            # Кожен клас присвоюється кільком клієнтам
            assigned_clients = np.random.choice(
                num_clients,
                size=num_clients // num_classes * classes_per_client,
                replace=False
            )
            class_assignments[c] = assigned_clients

        # Розподіляємо дані
        for i, (x, y) in enumerate(zip(data, labels)):
            eligible_clients = class_assignments[y]
            chosen = np.random.choice(eligible_clients)
            client_data[chosen].append(x)
            client_labels[chosen].append(y)

        return client_data, client_labels

    @staticmethod
    def quantity_skew(data, labels, num_clients: int,
                      alpha: float = 0.5):
        """
        Dirichlet distribution — різна кількість даних у клієнтів.
        alpha → 0: екстремальний дисбаланс
        alpha → ∞: рівномірний розподіл
        """
        proportions = np.random.dirichlet([alpha] * num_clients)
        client_sizes = (proportions * len(data)).astype(int)

        # Забезпечуємо, що кожен клієнт має хоча б 1 sample
        client_sizes = np.maximum(client_sizes, 1)

        indices = np.random.permutation(len(data))
        client_data = []
        client_labels = []

        start = 0
        for size in client_sizes:
            end = min(start + size, len(data))
            client_indices = indices[start:end]
            client_data.append(data[client_indices])
            client_labels.append(labels[client_indices])
            start = end

        return client_data, client_labels

    @staticmethod
    def feature_skew(data, labels, num_clients: int):
        """
        Різні feature distributions на клієнтах.
        Приклад: різні камери, освітлення, кути.
        """
        client_data = []
        client_labels = []

        for i in range(num_clients):
            # Кожен клієнт має своє "викривлення" даних
            brightness = np.random.uniform(0.7, 1.3)
            contrast = np.random.uniform(0.8, 1.2)
            noise_level = np.random.uniform(0, 0.1)

            # Застосовуємо трансформації
            transformed = data * brightness * contrast
            transformed += np.random.randn(*data.shape) * noise_level

            # Випадкова підмножина
            indices = np.random.choice(len(data), size=len(data)//num_clients)
            client_data.append(transformed[indices])
            client_labels.append(labels[indices])

        return client_data, client_labels

FedProx: рішення для Non-IID

FedProx (Li et al., 2020) додає proximal term до локального loss:

class FedProxClient:
    """
    FedProx = FedAvg + proximal regularization.
    Зменшує drift від глобальної моделі при non-IID даних.
    """

    def __init__(self, model, mu: float = 0.01):
        self.model = model
        self.mu = mu  # Proximal coefficient

    def local_train(self, global_weights: Dict,
                    epochs: int, lr: float) -> Tuple[Dict, int, float]:
        """
        Локальне тренування з proximal term.
        """
        self.model.set_weights(copy.deepcopy(global_weights))
        global_weights_tensor = self._dict_to_tensor(global_weights)

        for epoch in range(epochs):
            for batch_x, batch_y in self.data_loader:
                # Standard loss
                predictions = self.model(batch_x)
                task_loss = self.criterion(predictions, batch_y)

                # Proximal term: ||w - w_global||²
                current_weights = self._dict_to_tensor(self.model.get_weights())
                proximal_loss = (self.mu / 2) * torch.sum(
                    (current_weights - global_weights_tensor) ** 2
                )

                # Total loss
                total_loss = task_loss + proximal_loss

                # Backprop
                self.optimizer.zero_grad()
                total_loss.backward()
                self.optimizer.step()

        return self.model.get_weights(), len(self.dataset), total_loss.item()


class ScaffoldClient:
    """
    SCAFFOLD: Stochastic Controlled Averaging for FL.
    Variance reduction через control variates.
    """

    def __init__(self, model):
        self.model = model
        self.control_variate = None  # c_i
        self.server_control = None   # c

    def local_train(self, global_weights: Dict, server_control: Dict,
                    epochs: int, lr: float) -> Tuple[Dict, Dict, int]:
        """
        SCAFFOLD local update з control variate correction.
        """
        self.server_control = server_control
        self.model.set_weights(copy.deepcopy(global_weights))

        if self.control_variate is None:
            self.control_variate = {k: torch.zeros_like(v)
                                   for k, v in global_weights.items()}

        initial_weights = copy.deepcopy(global_weights)

        for epoch in range(epochs):
            for batch_x, batch_y in self.data_loader:
                # Compute gradient
                predictions = self.model(batch_x)
                loss = self.criterion(predictions, batch_y)
                loss.backward()

                # SCAFFOLD correction
                for name, param in self.model.named_parameters():
                    if param.grad is not None:
                        # g_i - c_i + c
                        param.grad.data += (
                            self.server_control[name] -
                            self.control_variate[name]
                        )

                self.optimizer.step()

        # Update local control variate
        final_weights = self.model.get_weights()
        new_control = {}
        for name in initial_weights.keys():
            # c_i^+ = c_i - c + (w^t - w^{t+1}) / (K * η)
            new_control[name] = (
                self.control_variate[name] - self.server_control[name] +
                (initial_weights[name] - final_weights[name]) / (epochs * lr)
            )

        delta_control = {
            name: new_control[name] - self.control_variate[name]
            for name in new_control.keys()
        }
        self.control_variate = new_control

        return final_weights, delta_control, len(self.dataset)

Secure Aggregation: криптографічний захист

Проблема: Навіть із DP, сервер бачить індивідуальні client updates.

Secure Aggregation: Сервер бачить ТІЛЬКИ суму, не окремі значення.

import secrets
from typing import Dict, List
import numpy as np

class SecureAggregationProtocol:
    """
    Secure Aggregation через pairwise masking.
    Сервер отримує тільки aggregate, не individual updates.
    """

    def __init__(self, num_clients: int, key_size: int = 256):
        self.num_clients = num_clients
        self.key_size = key_size
        self.shared_keys = {}  # Pairwise shared secrets

    def setup_phase(self, clients: List[str]):
        """
        Клієнти обмінюються ключами (Diffie-Hellman).
        """
        # Генеруємо pairwise shared secrets
        for i, client_i in enumerate(clients):
            for j, client_j in enumerate(clients):
                if i < j:
                    # В реальності — DH key exchange
                    shared_secret = secrets.token_bytes(self.key_size // 8)
                    key = (client_i, client_j)
                    self.shared_keys[key] = shared_secret

    def _generate_mask(self, seed: bytes, shape: tuple) -> np.ndarray:
        """
        Детерміновано генеруємо маску з shared secret.
        """
        rng = np.random.default_rng(int.from_bytes(seed[:8], 'big'))
        return rng.standard_normal(shape)

    def client_mask(self, client_id: str, update: np.ndarray,
                    all_clients: List[str]) -> np.ndarray:
        """
        Клієнт маскує свій update.
        Сума масок = 0 (вони попарно скасовуються).
        """
        masked_update = update.copy()

        for other_client in all_clients:
            if other_client == client_id:
                continue

            # Знаходимо shared key
            if client_id < other_client:
                key = (client_id, other_client)
                sign = 1
            else:
                key = (other_client, client_id)
                sign = -1

            if key in self.shared_keys:
                mask = self._generate_mask(self.shared_keys[key], update.shape)
                masked_update += sign * mask

        return masked_update

    def server_aggregate(self, masked_updates: List[np.ndarray]) -> np.ndarray:
        """
        Сервер просто сумує — маски скасовуються.
        """
        # Маски попарно рівні з протилежними знаками
        # Σ masks = 0
        # Тому Σ masked_updates = Σ original_updates
        return np.sum(masked_updates, axis=0)


class HomomorphicAggregation:
    """
    Альтернатива: Homomorphic Encryption.
    Сервер обчислює на зашифрованих даних.
    """

    def __init__(self, key_size: int = 2048):
        # В реальності: Paillier, CKKS, або BFV схеми
        self.public_key = None
        self.private_key = None
        self._generate_keys(key_size)

    def _generate_keys(self, key_size: int):
        """Генеруємо ключі для гомоморфного шифрування."""
        # Simplified — в реальності використовуйте tenseal, seal-python
        pass

    def encrypt_update(self, update: np.ndarray) -> 'EncryptedTensor':
        """Клієнт шифрує свій update."""
        return self._encrypt(update, self.public_key)

    def aggregate_encrypted(self,
                           encrypted_updates: List['EncryptedTensor']
                           ) -> 'EncryptedTensor':
        """
        Сервер сумує ЗАШИФРОВАНІ updates.
        Гомоморфна властивість: E(a) + E(b) = E(a + b)
        """
        result = encrypted_updates[0]
        for enc_update in encrypted_updates[1:]:
            result = result + enc_update  # Homomorphic addition
        return result

    def decrypt_aggregate(self,
                         encrypted_sum: 'EncryptedTensor') -> np.ndarray:
        """Тільки власник private key може розшифрувати."""
        return self._decrypt(encrypted_sum, self.private_key)

Практична реалізація з Flower Framework

Flower — найпопулярніший production-ready FL framework.

import flwr as fl
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from collections import OrderedDict
from typing import Dict, List, Tuple

# Define model
class CNNModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(1, 32, 3, padding=1)
        self.conv2 = nn.Conv2d(32, 64, 3, padding=1)
        self.pool = nn.MaxPool2d(2, 2)
        self.fc1 = nn.Linear(64 * 7 * 7, 128)
        self.fc2 = nn.Linear(128, 10)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.5)

    def forward(self, x):
        x = self.pool(self.relu(self.conv1(x)))
        x = self.pool(self.relu(self.conv2(x)))
        x = x.view(-1, 64 * 7 * 7)
        x = self.relu(self.fc1(x))
        x = self.dropout(x)
        return self.fc2(x)


class FlowerClient(fl.client.NumPyClient):
    """
    Production-ready Flower client.
    """

    def __init__(self, model: nn.Module, trainloader: DataLoader,
                 testloader: DataLoader, device: str = "cpu"):
        self.model = model.to(device)
        self.trainloader = trainloader
        self.testloader = testloader
        self.device = device

    def get_parameters(self, config: Dict) -> List[np.ndarray]:
        """Повертаємо параметри моделі як numpy arrays."""
        return [val.cpu().numpy() for _, val in self.model.state_dict().items()]

    def set_parameters(self, parameters: List[np.ndarray]):
        """Завантажуємо параметри в модель."""
        params_dict = zip(self.model.state_dict().keys(), parameters)
        state_dict = OrderedDict({k: torch.tensor(v) for k, v in params_dict})
        self.model.load_state_dict(state_dict, strict=True)

    def fit(self, parameters: List[np.ndarray],
            config: Dict) -> Tuple[List[np.ndarray], int, Dict]:
        """
        Локальне тренування.
        Викликається сервером кожен раунд.
        """
        self.set_parameters(parameters)

        epochs = config.get("local_epochs", 1)
        lr = config.get("learning_rate", 0.01)

        # Train
        self.model.train()
        optimizer = torch.optim.SGD(self.model.parameters(), lr=lr)
        criterion = nn.CrossEntropyLoss()

        total_loss = 0
        for epoch in range(epochs):
            for batch in self.trainloader:
                images, labels = batch[0].to(self.device), batch[1].to(self.device)
                optimizer.zero_grad()
                loss = criterion(self.model(images), labels)
                loss.backward()
                optimizer.step()
                total_loss += loss.item()

        return (
            self.get_parameters(config={}),
            len(self.trainloader.dataset),
            {"loss": total_loss / len(self.trainloader)}
        )

    def evaluate(self, parameters: List[np.ndarray],
                 config: Dict) -> Tuple[float, int, Dict]:
        """Оцінка на тестових даних."""
        self.set_parameters(parameters)

        self.model.eval()
        criterion = nn.CrossEntropyLoss()

        total_loss = 0
        correct = 0
        total = 0

        with torch.no_grad():
            for batch in self.testloader:
                images, labels = batch[0].to(self.device), batch[1].to(self.device)
                outputs = self.model(images)
                loss = criterion(outputs, labels)
                total_loss += loss.item()

                _, predicted = torch.max(outputs.data, 1)
                total += labels.size(0)
                correct += (predicted == labels).sum().item()

        accuracy = correct / total
        avg_loss = total_loss / len(self.testloader)

        return avg_loss, len(self.testloader.dataset), {"accuracy": accuracy}


# Custom aggregation strategy
class DPFedAvg(fl.server.strategy.FedAvg):
    """
    FedAvg з Differential Privacy на сервері.
    """

    def __init__(self, noise_multiplier: float = 0.1,
                 max_grad_norm: float = 1.0, **kwargs):
        super().__init__(**kwargs)
        self.noise_multiplier = noise_multiplier
        self.max_grad_norm = max_grad_norm

    def aggregate_fit(self, server_round: int,
                      results: List[Tuple[fl.server.client_proxy.ClientProxy,
                                         fl.common.FitRes]],
                      failures: List) -> Tuple[fl.common.Parameters, Dict]:
        """Агрегація з DP noise."""

        # Standard FedAvg aggregation
        aggregated_parameters, metrics = super().aggregate_fit(
            server_round, results, failures
        )

        if aggregated_parameters is not None:
            # Add DP noise
            ndarrays = fl.common.parameters_to_ndarrays(aggregated_parameters)
            noised_ndarrays = []

            for arr in ndarrays:
                noise = np.random.normal(
                    0,
                    self.noise_multiplier * self.max_grad_norm,
                    arr.shape
                )
                noised_ndarrays.append(arr + noise)

            aggregated_parameters = fl.common.ndarrays_to_parameters(noised_ndarrays)

        return aggregated_parameters, metrics


# Server startup
def start_server():
    strategy = DPFedAvg(
        fraction_fit=0.3,  # 30% clients per round
        fraction_evaluate=0.2,
        min_fit_clients=2,
        min_evaluate_clients=2,
        min_available_clients=5,
        noise_multiplier=0.1,
        max_grad_norm=1.0
    )

    fl.server.start_server(
        server_address="0.0.0.0:8080",
        config=fl.server.ServerConfig(num_rounds=50),
        strategy=strategy
    )


# Client startup
def start_client(client_id: int):
    model = CNNModel()
    trainloader, testloader = load_data_partition(client_id)

    client = FlowerClient(model, trainloader, testloader)

    fl.client.start_numpy_client(
        server_address="localhost:8080",
        client=client
    )

Real-World Deployments: хто вже використовує FL

1. Google Gboard (2017+)

• Next-word prediction на мільярдах devices
• On-device training у background
• Model updates завантажуються через WiFi вночі
• Privacy: Google ніколи не бачить, що ви друкуєте

2. Apple (2019+)

• Siri voice recognition improvement
• QuickType keyboard suggestions
• Differential privacy on device + secure aggregation
• Повна приватність для користувачів

3. NVIDIA FLARE (Healthcare)

• Mammography AI: 20 hospitals collaboration
• Без sharing patient data (HIPAA compliant)
• 30% improvement в rare cancer detection
• Кожна лікарня зберігає свої дані локально

4. WeBank (Finance)

• Fraud detection across institutions
• Regulatory compliant cross-bank ML
• Федеративна feature engineering
• Конкуренти співпрацюють без sharing даних

Бенчмарки: FedAvg vs FedProx vs SCAFFOLD

| Метрика | FedAvg | FedProx (μ=0.01) | SCAFFOLD |

|---------|--------|------------------|----------|

| IID Accuracy | 97.2% | 97.1% | 97.3% |

| Non-IID Accuracy | 89.4% | 93.1% | 95.2% |

| Rounds to converge (IID) | 50 | 48 | 35 |

| Rounds to converge (Non-IID) | 150 | 85 | 60 |

| Communication overhead | 1x | 1x | 2x |

| Client computation | 1x | 1.1x | 1.2x |

Ідеї для дослідження

Для бакалавра:

• Implement FedAvg на MNIST/CIFAR-10
• Порівняння IID vs non-IID settings
• Вимірювання communication cost
• Візуалізація convergence curves

Для магістра:

• Додавання Differential Privacy до FL
• Personalized Federated Learning (local fine-tuning)
• Gradient compression techniques (Top-K, Random-K)
• Cross-device vs cross-silo FL порівняння

Для PhD:

• Novel aggregation algorithms для extreme non-IID
• Theoretical privacy guarantees з moments accountant
• Byzantine-robust aggregation (захист від malicious clients)
• Federated Learning для foundation models (LLM)

Висновок: чому FL — це майбутнє

GDPR в Європі. CCPA в Каліфорнії. HIPAA для медицини. LGPD в Бразилії. Регуляції стають жорсткішими з кожним роком. Централізувати дані — все складніше юридично, етично, технічно.

Federated Learning — це не компроміс. Це краще рішення:

• Більше даних — бо люди погоджуються ділитися
• Більше privacy — бо дані не витікають
• Compliance by design — бо дані не покидають юрисдикцію
• Менше ризику — бо немає central point of failure

Хто освоїть FL зараз — буде тренувати моделі на даних, до яких конкуренти просто не мають доступу. Якщо ви плануєте дослідження в цій галузі, команда SKP-Degree на skp-degree.com.ua готова допомогти з реалізацією федеративних систем будь-якої складності. Для консультацій пишіть у Telegram: @kursovi_diplomy.

Federated Learning, differential privacy, приватність даних, GDPR compliance, Flower framework, PySyft, розподілене машинне навчання, edge AI — це основні концепції для дипломної чи магістерської роботи з децентралізованого штучного інтелекту та privacy-preserving ML.