案例引入
某银行A与某互联网公司B达成了企业级的合作。互联网公司A与银行B有着一大部分重合的用户，A有着客户上网行为等特征信息。B有着客户的存贷情况等特征信息以及客户的标签信息——客户的还贷情况(Y)。B希望能够将他所独有的特征信息与A所独有的特征信息相结合，训练出一个更强大的识别客户信用风险的模型，但由于不同行业之间的行政手续，用户数据隐私安全等因素，企业A，B无法直接互通数据，联邦学习应运而生。
联邦学习概述
联邦学习的定义
联邦学习旨在建立一个基于分布数据集的联邦学习模型。在模型训练的过程中，模型相关的信息能够在各方之间交换（或者是以加密形式交换），但原始数据不能。这一交换不会暴露每个站点上数据的任何受保护的隐私部分。已训练好的联邦学习模型可以置于联邦学习系统的各参与方，也可以在多方之间共享。
设有N位参与方
协作使用各自的训练数据集
来训练机器学习模型。传统的方法是将所有的数据
的机器学习过程。
设
的性能度量。在使用安全的联邦学习在分布式数据源上构建机器学习模型时，我们允许在保护用户隐私的情况下，联邦学习模型的性能略低于集中型模型的性能。
即为允许的性能损失。
联邦学习的分类
根据联邦学习所使用数据在各参与方的不同分布情况，我们可以将联邦学习划分为三类：横向联邦学习（Horizontal Federated Learning, HFL）、纵向联邦学习（Vertical Federated Learning, VFL）和联邦迁移学习（Federated Transfer Learning, FTL）。下面是这三种类型联邦学习所针对的不同数据分布情况：
· 横向联邦学习：不同参与方的数据有较大的特征的重叠（横向），但数据样本（纵向），即特征所属的样本的重叠度不高。例如，联邦学习的参与方是两家服务于不同区域市场的银行，他们所服务的客户群体差别较大，但客户的特征可能会因为相似的商业模式而重叠度较高。
· 纵向联邦学习：不同参与方的数据样本有较大的重叠，但样本特征的重叠度不高。例如，两家公司（银行和电子商务公司）向客户提供不同的服务，拥有客户不同方面的数据，但他们所服务的客户群体有较大的重叠。
· 联邦迁移学习：不同参与方的数据在特征和样本维度重叠度都不是非常高。
纵向联邦学习算法
纵向联邦学习算法有利于各企业之间建立合作，使用各自的特有数据，共同建立更加强大的模型。本篇将着重介绍一种基于加法同态加密的纵向联邦学习算法。
应用情景
细化开头的案例，企业B 有特征X3 和Y（标签），可独立建模，企业A 有特征X1、X2，缺乏Y，无法独立建模，现在企业A，B 合作，建立联合模型，显然效果会超过企业B单边数据建模。
但两方之间如何合作来共同训练一个模型呢？以逻辑回归为例，一个经典的逻辑回归的损失函数和梯度公式如下所示：
可以看到，梯度的计算离不开特征数据（x）和标签数据（y）。因此，一种最直接的数据交互方向就是其中一方将自己独有的数据直接以明文的方式发送给对方，由对方计算出梯度后再返回。但这样的交互方式会产生信息的泄露，其中一方会获得全部的信息，这显然是不符合规范的。
既然明文的传输不行，一种解决思路就是将需要的数据以密文的形式发送，但这又会产生另一个问题，其中一方获得另一方的密文数据后无法解密，又如何进行计算呢？这时就需要引入同态加密算法。
同态加密算法简介
由于篇幅所限，这里将只介绍同态加密算法的作用，而不介绍其具体细节。
同态加密（Homomorphic Encryption）是一种特殊的加密方法，允许对密文进行处理得到仍然是加密的结果，即对密文直接进行处理，跟对明文进行处理后再对处理结果加密，得到的结果相同。从抽象代数的角度讲，保持了同态性。
假设存在两个数x、y，OP(x,y)表示x与y之间的一种操作运算（加、减、乘、除、指数……）。E(x)表示对x的加密操作，D(x)表示对x的解密操作，则当某种加密算法对某个操作OP满足同态性时，表达式如下：
根据算法所能支持的操作运算的范围和次数的大小，可以将同态加密算法分为部分同态加密算法(PHE)、些许同态加密算法(SHE)和全同态加密算法(FHE)，其支持的运算范围与次数依次扩大。本文之后的纵向联邦学习算法将基于Paillier算法实现，它是一种部分同态加密算法，支持加法以及与常数的乘法运算。下面我将基于Python的phe库演示Paillier算法的作用。
phe库需要安装
fromphe import paillier
生成公钥与私钥
public_key, private_key = paillier.generate_paillier_keypair()
需要加密的数据
secret_number_list = [3.141592653, 300, -4.6e-12]
公钥加密
encrypted_number_list = [public_key.encrypt(x) forx in secret_number_list]
私钥解密
[private_key.decrypt(x) forx in encrypted_number_list]
支持加减法以及与常数的乘除法
a, b, c = encrypted_number_list
a_plus_5 =a + 5 #= a + 5
print("a + 5 =",private_key.decrypt(a_plus_5))
a_plus_b =a + b #= a + b
print("a + b =",private_key.decrypt(a_plus_b))
a_times_3_5 =a * 3.5 #= a * 3.5
print("a * 3.5 =",private_key.decrypt(a_times_3_5))
a_minus_1 =a - 1 #= a + (-1)
print("a - 1=",private_key.decrypt(a_minus_1))
a_div_minus_3_1 =a / -3.1 #= a * (-1/3.1)
print("a / -3.1 =",private_key.decrypt(a_div_minus_3_1))
a_minus_b =a - b #= a + (b*-1)
print("a - b =",private_key.decrypt(a_minus_b))
若一些函数内部的逻辑是加法或者是与常数的乘法，同样支持。
importnumpy as np
enc_mean = np.mean(encrypted_number_list)
enc_dot = np.dot(encrypted_number_list, [2, -400.1, 5318008])
print("enc_mean:", private_key.decrypt(enc_mean))
print("enc_dot:", private_key.decrypt(enc_dot))
算法流程
逻辑回归的损失和梯度的公式中包含着指数运算，因此，如果要用Paillier算法进行加密，需要对原公式进行一定的改造，使其仅用加法和乘法来表示。将指数运算改造为加法与乘法运算的一个常用方法就是用泰勒展开来进行近似。
最终得到的转化后的梯度矩阵的上半部分就是参与方A更新其参数需要的梯度（其中包含了正则项），下半部分对应B。我们的目标是希望参与方A、B能够尽量地进行单独的计算，再通过加密信息的交互获得各自的梯度计算结果，因此我们需要对计算的任务进行一定的划分，可以采用以下的一种设计流程。
在每一轮参数更新中，各参与方需要按序进行如下的计算和交互：
参与方A和B各自初始化自己的参数，参与方C生成秘钥对并分发公钥给A和B。
-
参与方A计算图片.png
，使用公钥加密后发送给A。
-
此时A和B能各自计算图片.png
（[[x]]表示x的同态加密形式）。
-
A和B需要加密的梯度发送给C来进行解密，但为了避免C直接获得梯度信息，A和B可以将梯度加上一个随机数图片.png
再发送给C。C获得加密梯度进行后进行解密再返还A和B。
- A和B只需要再减去之间加的随机数就能获得真实的梯度，更新其参数。
代码实现
下面我们将基于Python代码来实现这整个算法流程。为了更清晰地展现算法的流程，将极度简化交互流程的实现。
导入所需模块
import math
importnumpy as np
fromphe import paillier
importpandas as pd
fromsklearn import datasets
from sklearn.datasets import load_diabetes
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.utils import shuffle
各参与方的定义
设置参与方的父类，各参与方都需要保存模型的参数、一些中间计算结果以及与其他参与方的连接状况。
class Client:
def __init__(self, config):
## 模型参数
self.config = config
## 中间计算结果
self.data = {}
## 与其他节点的连接状况
self.other_client = {}
## 与其他参与方建立连接
def connect(self, client_name, target_client):
self.other_client[client_name] = target_client
## 向特定参与方发送数据
def send_data(self, data, target_client):
target_client.data.update(data)
参与方A在训练过程中仅提供特征数据。
class ClientA(Client):
def __init__(self, X, config):
super().__init__(config)
self.X = X
self.weights = np.zeros(X.shape[1])
def compute_z_a(self):
z_a = np.dot(self.X, self.weights)
return z_a
## 加密梯度的计算，对应step4
def compute_encrypted_dJ_a(self, encrypted_u):
encrypted_dJ_a = self.X.T.dot(encrypted_u) + self.config['lambda'] * self.weights
return encrypted_dJ_a
##参数的更新
def update_weight(self, dJ_a):
self.weights = self.weights - self.config["lr"] *dJ_a / len(self.X)
return
## A: step2
def task_1(self, client_B_name):
dt = self.data
assert "public_key" in dt.keys(), "Error: 'public_key' from C in step 1 not successfully received."
public_key = dt['public_key']
z_a = self.compute_z_a()
u_a = 0.25 * z_a
z_a_square =z_a ** 2
encrypted_u_a = np.asarray([public_key.encrypt(x) forx in u_a])
encrypted_z_a_square = np.asarray([public_key.encrypt(x) forx in z_a_square])
dt.update({"encrypted_u_a": encrypted_u_a})
data_to_B = {"encrypted_u_a": encrypted_u_a, "encrypted_z_a_square": encrypted_z_a_square}
self.send_data(data_to_B, self.other_client[client_B_name])
## A: step3、4
def task_2(self, client_C_name):
dt = self.data
assert "encrypted_u_b" in dt.keys(), "Error: 'encrypted_u_b' from B in step 1 not successfully received."
encrypted_u_b = dt['encrypted_u_b']
encrypted_u =encrypted_u_b + dt['encrypted_u_a']
encrypted_dJ_a = self.compute_encrypted_dJ_a(encrypted_u)
mask = np.random.rand(len(encrypted_dJ_a))
encrypted_masked_dJ_a =encrypted_dJ_a + mask
dt.update({"mask": mask})
data_to_C = {'encrypted_masked_dJ_a': encrypted_masked_dJ_a}
self.send_data(data_to_C, self.other_client[client_C_name])
## A: step6
def task_3(self):
dt = self.data
assert "masked_dJ_a" in dt.keys(), "Error: 'masked_dJ_a' from C in step 2 not successfully received."
masked_dJ_a = dt['masked_dJ_a']
dJ_a =masked_dJ_a - dt['mask']
self.update_weight(dJ_a)
print(f"A weight: {self.weights}")
return
参与方B在训练过程中既提供特征数据，又提供标签数据。
class ClientB(Client):
def __init__(self, X, y, config):
super().__init__(config)
self.X = X
self.y = y
self.weights = np.zeros(X.shape[1])
self.data = {}
def compute_u_b(self):
z_b = np.dot(self.X, self.weights)
u_b = 0.25 *z_b - self.y + 0.5
return z_b, u_b
def compute_encrypted_dJ_b(self, encrypted_u):
encrypted_dJ_b = self.X.T.dot(encrypted_u) + self.config['lambda'] * self.weights
return encrypted_dJ_b
def update_weight(self, dJ_b):
self.weights = self.weights - self.config["lr"] *dJ_b / len(self.X)
## B: step2
def task_1(self, client_A_name):
try:
dt = self.data
assert "public_key" in dt.keys(), "Error: 'public_key' from C in step 1 not successfully received."
public_key = dt['public_key']
exceptException as e:
print("B step 1 exception: %s" % e)
try:
z_b, u_b = self.compute_u_b()
encrypted_u_b = np.asarray([public_key.encrypt(x) forx in u_b])
dt.update({"encrypted_u_b": encrypted_u_b})
dt.update({"z_b": z_b})
exceptException as e:
print("Wrong 1 in B: %s" % e)
data_to_A= {"encrypted_u_b": encrypted_u_b}
self.send_data(data_to_A, self.other_client[client_A_name])
## B: step3、4
def task_2(self,client_C_name):
try:
dt = self.data
assert "encrypted_u_a" in dt.keys(), "Error: 'encrypt_u_a' from A in step 1 not successfully received."
encrypted_u_a = dt['encrypted_u_a']
encrypted_u =encrypted_u_a + dt['encrypted_u_b']
encrypted_dJ_b = self.compute_encrypted_dJ_b(encrypted_u)
mask = np.random.rand(len(encrypted_dJ_b))
encrypted_masked_dJ_b =encrypted_dJ_b + mask
dt.update({"mask": mask})
exceptException as e:
print("B step 2 exception: %s" % e)
try:
assert "encrypted_z_a_square" in dt.keys(), "Error: 'encrypted_z_a_square' from A in step 1 not successfully received."
encrypted_z = 4*encrypted_u_a + dt['z_b']
encrypted_loss = np.sum((0.5-self.y)encrypted_z + 0.125dt["encrypted_z_a_square"] + 0.125dt["z_b"] * (encrypted_z+4encrypted_u_a))
exceptException as e:
print("B step 2 exception: %s" % e)
data_to_C = {"encrypted_masked_dJ_b": encrypted_masked_dJ_b, "encrypted_loss": encrypted_loss}
self.send_data(data_to_C, self.other_client[client_C_name])
## B: step6
def task_3(self):
try:
dt = self.data
assert "masked_dJ_b" in dt.keys(), "Error: 'masked_dJ_b' from C in step 2 not successfully received."
masked_dJ_b = dt['masked_dJ_b']
dJ_b =masked_dJ_b - dt['mask']
self.update_weight(dJ_b)
exceptException as e:
print("A step 3 exception: %s" % e)
print(f"B weight: {self.weights}")
return
参与方C在整个训练过程中主要的作用就是分发秘钥，以及最后的对A和B加密梯度的解密。
class ClientC(Client):
"""
Client C as trusted dealer.
"""
def __init__(self, A_d_shape, B_d_shape, config):
super().__init__(config)
self.A_data_shape = A_d_shape
self.B_data_shape = B_d_shape
self.public_key = None
self.private_key = None
## 保存训练中的损失值（泰展开近似）
self.loss = []
## C: step1
def task_1(self, client_A_name, client_B_name):
try:
public_key, private_key = paillier.generate_paillier_keypair()
self.public_key = public_key
self.private_key = private_key
exceptException as e:
print("C step 1 error 1: %s" % e)
data_to_AB = {"public_key": public_key}
self.send_data(data_to_AB, self.other_client[client_A_name])
self.send_data(data_to_AB, self.other_client[client_B_name])
return
## C: step5
def task_2(self, client_A_name, client_B_name):
try:
dt = self.data
assert "encrypted_masked_dJ_a" in dt.keys() and "encrypted_masked_dJ_b" in dt.keys(), "Error: 'masked_dJ_a' from A or 'masked_dJ_b' from B in step 2 not successfully received."
encrypted_masked_dJ_a = dt['encrypted_masked_dJ_a']
encrypted_masked_dJ_b = dt['encrypted_masked_dJ_b']
masked_dJ_a = np.asarray([self.private_key.decrypt(x) forx in encrypted_masked_dJ_a])
masked_dJ_b = np.asarray([self.private_key.decrypt(x) forx in encrypted_masked_dJ_b])
exceptException as e:
print("C step 2 exception: %s" % e)
try:
assert "encrypted_loss" in dt.keys(), "Error: 'encrypted_loss' from B in step 2 not successfully received."
encrypted_loss = dt['encrypted_loss']
loss = self.private_key.decrypt(encrypted_loss) / self.A_data_shape[0] + math.log(2)
print("******loss: ", loss, "******")
self.loss.append(loss)
exceptException as e:
print("C step 2 exception: %s" % e)
data_to_A = {"masked_dJ_a": masked_dJ_a}
data_to_B = {"masked_dJ_b": masked_dJ_b}
self.send_data(data_to_A, self.other_client[client_A_name])
self.send_data(data_to_B, self.other_client[client_B_name])
return
模拟数据的生成
这里将基于sklearn中的乳腺癌数据集生成一组模拟数据，参与方A获得部分特征数据，参与方B获得部分特征数据与标签数据。
def load_data():
# 加载数据
breast = load_breast_cancer()
# 数据拆分
X_train, X_test, y_train, y_test = train_test_split(breast.data, breast.target, random_state=1)
# 数据标准化
std = StandardScaler()
X_train = std.fit_transform(X_train)
X_test = std.transform(X_test)
return X_train, y_train, X_test, y_test
将特征分配给A和B
def vertically_partition_data(X, X_test, A_idx, B_idx):
"""
Vertically partition feature for party A and B
:param X: train feature
:param X_test: test feature
:param A_idx: feature index of party A
:param B_idx: feature index of party B
:return: train data for A, B; test data for A, B
"""
XA = X[:, A_idx]
XB = X[:, B_idx]
XB = np.c_[np.ones(X.shape[0]), XB]
XA_test = X_test[:, A_idx]
XB_test = X_test[:, B_idx]
XB_test = np.c_[np.ones(XB_test.shape[0]), XB_test]
return XA, XB, XA_test, XB_test
训练流程的实现
def vertical_logistic_regression(X, y, X_test, y_test, config):
"""
Start the processes of the three clients: A, B and C.
:param X: features of the training dataset
:param y: labels of the training dataset
:param X_test: features of the test dataset
:param y_test: labels of the test dataset
:param config: the config dict
:return: True
"""
## 获取数据
XA, XB, XA_test, XB_test = vertically_partition_data(X, X_test, config['A_idx'], config['B_idx'])
print('XA:',XA.shape, ' XB:',XB.shape)
## 各参与方的初始化
client_A = ClientA(XA, config)
print("Client_A successfully initialized.")
client_B = ClientB(XB, y, config)
print("Client_B successfully initialized.")
client_C = ClientC(XA.shape, XB.shape, config)
print("Client_C successfully initialized.")
## 各参与方之间连接的建立
client_A.connect("B", client_B)
client_A.connect("C", client_C)
client_B.connect("A", client_A)
client_B.connect("C", client_C)
client_C.connect("A", client_A)
client_C.connect("B", client_B)
## 训练
fori in range(config['n_iter']):
client_C.task_1("A", "B")
client_A.task_1("B")
client_B.task_1("A")
client_A.task_2("C")
client_B.task_2("C")
client_C.task_2("A", "B")
client_A.task_3()
client_B.task_3()
print("All process done.")
return True
config = {
'n_iter': 100,
'lambda': 10,
'lr': 0.05,
'A_idx': [10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29],
'B_idx': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
}
X, y, X_test, y_test = load_data()
vertical_logistic_regression(X, y, X_test, y_test, config)
训练效果
为测试该纵向联邦学习算法的训练效果。可以设置普通的集中式训练的逻辑回归算法作为对照组，基于乳腺癌数据集，使用相同的训练集数据及相同的逻辑回归模型来进行训练，观察其损失值的下降曲线以及在相同测试集上的预测准确率。
以下是两种情况下，训练的损失值的下降情况：
各曲线代表的情形：
Logistic: 普通逻辑回归的损失值变化曲线，使用的是正常的损失函数
Taylor_Logistic: 普通逻辑回归的损失值变化曲线，使用的是泰勒展开拟合的损失函数
Taylor_Taylor：纵向逻辑回归的损失值变化曲线，使用的是泰勒展开拟合的损失函数
以下是在sklearn中不同数据集上，普通逻辑回归与纵向逻辑回归的训练结果的正确率及AUC的差异，其中rows代表样本数量，feat代表特征数量，logistic代表集中式逻辑回归的训练结果，Vertical代表纵向联邦学习算法的训练效果。
由训练结果的比较可以看到，与普通的逻辑回归相比，该纵向逻辑回归算法在保证各方数据隐私性的同时，在实验数据集上能够达到不错的训练效果。
参考文献
[1] Yang Q , Liu Y , Chen T , et al. Federated Machine Learning: Concept and Applications[J]. ACM Transactions on Intelligent Systems and Technology, 2019, 10(2):1-19.
[2] Hardy S , Henecka W , Ivey-Law H , et al. Private federated learning on vertically partitioned data via entity resolution and additively homomorphic encryption[J]. 2017.
