机器学习——边缘计算与联邦学习

import numpy as np
from sklearn.linear_model import SGDRegressor
from sklearn.datasets import make_regression

# 模拟客户端设备的数量
NUM_CLIENTS = 5

# 创建模拟数据
def create_data(n_samples=100):
    X, y = make_regression(n_samples=n_samples, n_features=10, noise=0.1)
    return X, y

# 模拟客户端本地训练
def local_training(X, y):
    model = SGDRegressor(max_iter=10)
    model.fit(X, y)
    return model.coef_, model.intercept_

# 模拟联邦学习中的全局聚合
def federated_averaging(updates):
    coef_sum = np.sum([coef for coef, _ in updates], axis=0)
    intercept_sum = np.sum([intercept for _, intercept in updates])
    avg_coef = coef_sum / len(updates)
    avg_intercept = intercept_sum / len(updates)
    return avg_coef, avg_intercept

# 模拟联邦学习过程
client_updates = []
for client_id in range(NUM_CLIENTS):
    X, y = create_data()
    coef, intercept = local_training(X, y)
    client_updates.append((coef, intercept))

# 全局聚合
global_coef, global_intercept = federated_averaging(client_updates)
print("Global Model Coefficients:", global_coef)
print("Global Model Intercept:", global_intercept)

import torch
import torch.nn as nn
import torch.optim as optim

# 模拟边缘设备的数量
NUM_EDGE_DEVICES = 3

# 创建简单的线性模型
class SimpleModel(nn.Module):
    def __init__(self):
        super(SimpleModel, self).__init__()
        self.linear = nn.Linear(10, 1)

    def forward(self, x):
        return self.linear(x)

# 模拟边缘设备上的本地训练
def local_training_edge(device_id, data, labels):
    model = SimpleModel()
    criterion = nn.MSELoss()
    optimizer = optim.SGD(model.parameters(), lr=0.01)

    # 本地训练循环
    for epoch in range(5):
        optimizer.zero_grad()
        outputs = model(data)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
    
    # 返回模型参数
    return model.state_dict()

# 模拟数据
edge_data = [torch.randn(100, 10) for _ in range(NUM_EDGE_DEVICES)]
edge_labels = [torch.randn(100, 1) for _ in range(NUM_EDGE_DEVICES)]

# 边缘设备本地训练
model_updates = []
for device_id in range(NUM_EDGE_DEVICES):
    update = local_training_edge(device_id, edge_data[device_id], edge_labels[device_id])
    model_updates.append(update)

# 模拟参数聚合
def aggregate_model_updates(updates):
    aggregated_state_dict = updates[0]
    for key in aggregated_state_dict.keys():
        for i in range(1, len(updates)):
            aggregated_state_dict[key] += updates[i][key]
        aggregated_state_dict[key] = aggregated_state_dict[key] / len(updates)
    return aggregated_state_dict

# 聚合边缘设备的模型更新
global_model_state_dict = aggregate_model_updates(model_updates)