逻辑回归

from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
# 生成分类数据
X, y = make_classification(n_samples=114514, n_features=5, n_class=2, n_informative=2, random_state=1919810, n_clusters_per_class=1)

# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(
	X, y, test_size=.2, random_state=1919810
)

from sklearn.linear_model import LogisticRegression  # 直接引入逻辑回归方法
import numpy as np

model = LogisticRegression(
	penalty='None',  # 无正则化，和上文的公式一致，设置为L2可将目标变得强凸
	# C=1.0,   # 只有在启动正则化后才有效，数字越小正则化强度越大
	max_iter=100,  # 最大迭代次数
	solver='saga',  # 梯度下降求解器，模型默认实际上是L-BFGS，更高性能的BFGS法
	random_state=1919810  # 随机种子
)

model.fit(X_train, y_train)

y_pred = model.predict(X_test)  # 预测，输出预测标签值数组
y_prob_pred = model.predict_proba(X_test)  # 输出预测标签概率数组

from pyspark.sql import SparkSession
from pyspark.ml.classification import LogisticRegression


spark = SparkSession.builder \  # 创建Spark会话以与集群资源/本地资源互动
		.appName("LRDemo") \
		.getOrGreate()
		
# 这是一个Estimator子类，能定义模型参数并通过.fit生成带训练参数的Transformer子类
lr = LogisticRegression(
	featuresCol="x_vec",  # 选中特征列
	labelCol="y_vec",  # 选中标签列
	maxIter=100,  # 最大迭代次数
	family="binomial",  # 问题是二分类
	regParam=1  # 正则化强度，作用等同于sklearn的C

)  

# 这是一个Transformer子类，它只能根据Estimator子类给出的训练参数进行推理
model = lr.fit(train_data)  # PySpark会生成新的变量，而非像sklearn那样修改原有变量

predictions = model.transform(test_data)  # 模型推理
# columns = [x_vec, y_vec, prediction]

import numpy as np

# Sigmoid
def sigmoid(z):
	"""
	该函数用于实现h(x)，即Sigmoid函数

	:param z: Sigmoid函数输入值，可以是标量或向量
	:return : Sigmoid函数输出值
	"""
	# np.where有三个参数，[判断条件/布尔值，条件为真的值，条件为假的值]
	return np.where(z >= 0, 
					1 / (1+np.exp(-z)),  # 大于0时使用标准形式 
					np.exp(z)/(1+np.exp(z)))  # 当小于0时使用等级啊变换避免溢出

def logistic_loss(theta, X, y):
	"""
	计算标准的对数损失
	:param theta: 算法参数，应当为数组，尺寸为(特征数, )，这也是我们要更新的部分
	:param X: 数据的特征值，应当为数组，尺寸为(样本数, 特征数)
	:param y: 数据的标签值，应当为数组，尺寸为(样本数)，值必须为0或者1
	:return loss: 逻辑(二元交叉熵/对数)损失值
	"""


	z = X @ theta  # 运算线性组合
	h = sigmoid(z)
	epsilon=1e-15  # 极小数，避免log0
	h = np.clip(h, epsilon, 1-epsilon)  # 裁剪运算结果
	loss = -np.mean(
		y * np.log(h) + (1 - y) *np.log(1 - h)
	)
	return loss

def logistic_loss_reg(theta, X, y, lambda_reg):
	"""
	计算带正则化项的损失
	:param theta: 算法参数
	:param X: 数据特征值
	:param y: 数据标签值
	:param lambda_reg: 正则化强度，直接决定目标函数强凸程度，函数变化为 lambda_reg-强凸函数，取0则不进行正则化
	:return : 带正则化项的对数损失

	"""
	base_loss = logistic_loss(theta, X, y)
	reg_term = (lambda_reg / (2 * m)) * np.sum(theta[1:]**2)  # 正则化需要去除截距项，后同
	return base_loss + reg_term
	

def get_grad(theta, X, y):

	n_samples = len(y)
	z = X @ theta
	h = sigmoid(z)
	gradient = (1/n_samples) * X.T @ (h-y)  # 直接套用公式
	return gradient

def get_grad_reg(theta, X, y, lambda_reg):

	n_samples = len(y)
	gradient = get_grad(theta, X, y)
	reg_gradient[1:] = (lambda_reg / m) * theta[1:]
	return gradient + reg_gradient

def gradient_descent(X, y, theta_init=None, learning_rate=1e-2, n_iter=100, lambda_reg=0):
	"""

	"""

	m, n = X.shape
	if theta_init is None:
		theta = np.zeros()
	else:
		theta = theta_init.copy()

	loss_values = []

	for i in range(n_iter):
		if lambda_reg > 0:
			loss = logistic_loss_reg(theta, X, y, lambda_reg)
			grad = get_grad_reg(theta, X, y, lambda_reg)
		else:
			loss = logistic_loss(theta, X, y)
			grad = get_grad(theta, X, y)
		loss_values.append(loss)
		theta = theta - learning_rate * grad
	return theta, loss_values

class LogisticRegression:
	def __init__(
		self,
		learning_rate=1e-2,
		n_iter=100,
		lambda_reg=0,
		fit_intercept=True
	):
		self.learning_rate = learning_rate
		self.n_iter = n_iter
		self.lambda_reg = lambda_reg
		self.fit_intercept = fit_intercept

		self.theta = None
		self.loss_values = None

	def _add_intercept(self, X):
		m = X.shape[0]
		return np.c_[np.ones(m), X]

	def fit(self, X, y):
		if self.fit_intersept:
			X = self._add_intercept(X)

		self.theta, self.loss_values = gradient_descent(
			X,
			y,
			learning_rate,
			n_iter,
			lambda_reg
		)
		return self

	def predict_proba(self, X):

		if self.theta is None:
			raise ValueError("No parameters fitted yet!")

		if self.fit_intercept:
			X = self._add_inetercept(X)

		return sigmoid(X @ self.theta)