如何优化自定义Perceptron类执行时间，缩小与Scikit-learn版本的差距？

我实现了一个CustomPerceptron类，用scikit-learn的乳腺癌数据集做分类任务，同时用官方的Perceptron完成相同任务，对比后发现自己的代码每次迭代要7-9秒，而scikit-learn版本几乎瞬间完成。

自定义Perceptron代码 #

import numpy as np
import matplotlib.pyplot as plt
import pandas as pd
from sklearn import datasets
from sklearn.model_selection import train_test_split
from sklearn.linear_model import Perceptron
from sklearn.metrics import precision_score
from sklearn.metrics import recall_score
from sklearn.metrics import f1_score
import time

class CustomPerceptron():
    '''
    Parameters:
    n_iterations: int
    Number of epochs the perceptron will run on the dataset
    
    learning_rate: float
    Learning rate to adjust the weights. Value between 0 and 1
    
    random_state: int
    Used in the random number generator to generate different weights every time
    '''
    def __init__(self, n_iterations=100, learning_rate=0.01, random_state=1):
        self.n_iterations = n_iterations
        self.learning_rate = learning_rate
        self.random_state = random_state
 
    
    '''
    Parameters:
    X: Sample dataset
    y: Target values
    '''
    def fit(self, X, y):
        rgen = np.random.RandomState(self.random_state)
        self.weights_ = rgen.normal(loc=0.0, scale=0.01, size=1 + X.shape[1])
        self.errors_ = []
        for _ in range(self.n_iterations):
            errors = 0
            for xi, expected_value in zip(X, y):
                predicted_value = self.predict(xi)
                self.weights_[1:] = self.weights_[1:] + self.learning_rate * (expected_value - predicted_value) * xi
                self.weights_[0] = self.weights_[0] + self.learning_rate * (expected_value - predicted_value) * 1
                update = self.learning_rate * (expected_value - predicted_value)
                errors += int(update != 0.0)
            self.errors_.append(errors)
    
    
    '''
    Parameters:
    X: Sample dataset
    
    Returns:
    Weighted sum of each data point and its corresponding weight with the bias added
    '''
    def net_input(self, X):
            weighted_sum = np.dot(X, self.weights_[1:]) + self.weights_[0]
            return weighted_sum
     
    
    '''
    Parameters:
    X: Sample dataset
    
    Returns:
    nd-array where each element is 1 or 0 depending on the weighted sum
    '''
    def activation_function(self, X):
            weighted_sum = self.net_input(X)
            return np.where(weighted_sum >= 0.0, 1, 0)
     
    
    '''
    Parameters:
    X: Sample dataset
    
    Returns:
    Predicted values on the basis of the output of activation function
    '''
    def predict(self, X):
        return self.activation_function(X)
     
    
    '''
    Parameters:
    X: Sample dataset
    y: Target values
    
    Returns:
    Accuracy of the model on the sample dataset
    '''
    def accuracy(self, X, y):
        misclassified_data_count = 0
        for xi, target in zip(X, y):
            output = self.predict(xi)
            if(target != output):
                misclassified_data_count += 1
        total_data_count = len(X)
        self.score_ = (total_data_count - misclassified_data_count) / total_data_count
        return self.score_
    
    
    '''
    Parameters:
    X: Sample dataset
    y: Target values
    
    Returns:
    A tuple of size 3 where
    index 0: Precision
    index 1: Recall
    index 2: F1 score
    '''
    def metrics(self, X, y):
        TP, TN, FP, FN = 0, 0, 0, 0
        for xi, target in zip(X, y):
            output = self.predict(xi)
            if(target == 1 and output == 1):
                TP += 1
            elif(target == 0 and output == 0):
                TN += 1
            elif(target == 0 and output == 1):
                FP += 1
            elif(target == 1 and output == 0):
                FN += 1
        
        precision = TP / (TP + FP)
        recall = TP / (TP + FN)
        f1_score = (2 * precision * recall) / (precision + recall)
        
        return (precision, recall, f1_score)

测试代码 #

每个感知器运行1000个epochs，我循环执行了100次分类任务：

bc = datasets.load_breast_cancer()
X = bc.data
y = bc.target

accuracy_list = []
our_times = []
precisions = []
recalls = []
f1s = []
sklearn_accuracy_list = []
sklearn_times = []
sklearn_precisions = []
sklearn_recalls = []
sklearn_f1s = []

ppn = CustomPerceptron(n_iterations=1000) # 自定义感知器
clf = Perceptron(tol=1e-3, eta0=0.01, random_state=0) # sklearn感知器

for i in range(100):
    # 80-20划分训练测试集
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=i, stratify=y)
    
    start_time = time.time() # 开始计时自定义感知器
    ppn.fit(X_train, y_train) # 拟合训练数据
    accuracy_list.append(ppn.accuracy(X_test, y_test)) # 计算测试集准确率并加入列表
    end_time = time.time() # 结束计时
    our_times.append(end_time - start_time) # 记录单次迭代耗时
    
    start_time = time.time() # 开始计时sklearn感知器
    clf.fit(X_train, y_train) # 拟合训练数据
    sklearn_accuracy_list.append(clf.score(X_test, y_test)) # 计算测试集准确率并加入列表
    end_time = time.time() # 结束计时
    sklearn_times.append(end_time - start_time) # 记录单次迭代耗时

优化方案 #

1. 用NumPy向量化操作替代逐样本循环 #

自定义代码的fit、accuracy、metrics方法都用了Python原生for循环逐样本处理，这是性能瓶颈的核心原因。NumPy的向量化操作基于C实现，速度远快于Python循环。

修改后的核心方法示例：

def fit(self, X, y):
    rgen = np.random.RandomState(self.random_state)
    self.weights_ = rgen.normal(loc=0.0, scale=0.01, size=1 + X.shape[1])
    self.errors_ = []
    
    # 拼接偏置项对应的全1列，统一处理权重和偏置更新
    X_bias = np.hstack([np.ones((X.shape[0], 1)), X])
    for _ in range(self.n_iterations):
        y_pred = self.predict(X)
        error = y - y_pred
        # 批量更新权重：一次矩阵运算完成所有样本的权重调整
        self.weights_ += self.learning_rate * np.dot(X_bias.T, error)
        # 统计误分类数量
        errors = np.sum(error != 0)
        self.errors_.append(errors)

def accuracy(self, X, y):
    y_pred = self.predict(X)
    self.score_ = np.mean(y == y_pred)
    return self.score_

def metrics(self, X, y):
    y_pred = self.predict(X)
    TP = np.sum((y == 1) & (y_pred == 1))
    TN = np.sum((y == 0) & (y_pred == 0))
    FP = np.sum((y == 0) & (y_pred == 1))
    FN = np.sum((y == 1) & (y_pred == 0))
    
    precision = TP / (TP + FP) if (TP + FP) != 0 else 0.0
    recall = TP / (TP + FN) if (TP + FN) != 0 else 0.0
    f1_score = (2 * precision * recall) / (precision + recall) if (precision + recall) != 0 else 0.0
    
    return (precision, recall, f1_score)

2. 添加提前停止机制 #

scikit-learn的Perceptron通过tol参数实现提前停止：当迭代中权重变化小于阈值时，自动终止训练，避免不必要的循环。给自定义感知器添加相同逻辑：

def __init__(self, n_iterations=100, learning_rate=0.01, random_state=1, tol=None):
    self.n_iterations = n_iterations
    self.learning_rate = learning_rate
    self.random_state = random_state
    self.tol = tol  # 新增提前停止阈值

def fit(self, X, y):
    rgen = np.random.RandomState(self.random_state)
    self.weights_ = rgen.normal(loc=0.0, scale=0.01, size=1 + X.shape[1])
    self.errors_ = []
    
    X_bias = np.hstack([np.ones((X.shape[0], 1)), X])
    for epoch in range(self.n_iterations):
        y_pred = self.predict(X)
        error = y - y_pred
        prev_weights = self.weights_.copy()
        self.weights_ += self.learning_rate * np.dot(X_bias.T, error)
        errors = np.sum(error != 0)
        self.errors_.append(errors)
        
        # 检查是否满足提前停止条件
        if self.tol is not None:
            weight_change = np.linalg.norm(self.weights_ - prev_weights)
            if weight_change < self.tol:
                break

3. 数据类型优化 #

将输入数据转换为更高效的数值类型，减少内存占用和计算时间：

X = X.astype(np.float32)
y = y.astype(np.int32)

4. 可选：用Numba JIT编译加速 #

如果仍有性能需求，可使用Numba对核心函数进行即时编译，进一步提升速度：

from numba import jit

@jit(nopython=True)
def net_input(self, X):
    return np.dot(X, self.weights_[1:]) + self.weights_[0]

内容的提问来源于stack exchange，提问作者Moshiur Rahman