Python案例如何用Scikit-learn做在线学习

wen python案例 2026-07-16 245

本文目录导读：

支持增量学习的算法
基础在线学习示例
高级在线学习策略
最佳实践建议

我来详细介绍如何使用Scikit-learn实现在线学习（增量学习）。

支持增量学习的算法

Scikit-learn中支持增量学习的算法主要有：

# 线性模型
from sklearn.linear_model import SGDClassifier, SGDRegressor
from sklearn.linear_model import PassiveAggressiveClassifier, PassiveAggressiveRegressor
# 朴素贝叶斯
from sklearn.naive_bayes import MultinomialNB, BernoulliNB
# 神经网络
from sklearn.neural_network import MLPClassifier
# 预处理
from sklearn.preprocessing import StandardScaler

基础在线学习示例

1 使用SGD分类器进行在线学习

import numpy as np
from sklearn.linear_model import SGDClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
class OnlineLearningExample:
    def __init__(self):
        # 初始化模型
        self.model = SGDClassifier(
            loss='log',  # 逻辑回归损失
            penalty='l2',  # L2正则化
            max_iter=1,  # 每次只迭代一次
            tol=1e-3,
            random_state=42,
            learning_rate='adaptive',  # 自适应学习率
            eta0=0.01  # 初始学习率
        )
    def partial_fit_batch(self, X_batch, y_batch, classes=None):
        """
        增量学习单个批次
        """
        if classes is not None:
            self.model.partial_fit(X_batch, y_batch, classes=classes)
        else:
            self.model.partial_fit(X_batch, y_batch)
    def evaluate(self, X_test, y_test):
        """
        评估模型性能
        """
        predictions = self.model.predict(X_test)
        return accuracy_score(y_test, predictions)
# 生成模拟数据
X, y = make_classification(
    n_samples=10000, 
    n_features=20, 
    n_classes=2, 
    random_state=42
)
# 分割数据
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)
# 创建在线学习实例
online_learner = OnlineLearningExample()
# 分批次进行在线学习
batch_size = 100
n_batches = len(X_train) // batch_size
print("开始在线学习过程...")
for i in range(n_batches):
    # 获取当前批次数据
    start_idx = i * batch_size
    end_idx = (i + 1) * batch_size
    X_batch = X_train[start_idx:end_idx]
    y_batch = y_train[start_idx:end_idx]
    # 第一轮需要指定类别
    if i == 0:
        online_learner.partial_fit_batch(
            X_batch, y_batch, 
            classes=np.unique(y)
        )
    else:
        online_learner.partial_fit_batch(X_batch, y_batch)
    # 每10个批次评估一次
    if (i + 1) % 10 == 0:
        accuracy = online_learner.evaluate(X_test, y_test)
        print(f"Batch {i+1}/{n_batches}, 准确率: {accuracy:.4f}")
# 最终评估
final_accuracy = online_learner.evaluate(X_test, y_test)
print(f"\n最终准确率: {final_accuracy:.4f}")

2 流式数据处理示例

import numpy as np
from sklearn.linear_model import SGDRegressor
from sklearn.preprocessing import StandardScaler
from collections import deque
class StreamingDataProcessor:
    def __init__(self, buffer_size=100):
        self.model = SGDRegressor(
            max_iter=1,
            tol=1e-3,
            random_state=42,
            learning_rate='adaptive'
        )
        self.scaler = StandardScaler()
        self.buffer = deque(maxlen=buffer_size)
        self.is_fitted = False
    def process_stream(self, data_stream):
        """
        处理数据流
        """
        for i, (X, y) in enumerate(data_stream):
            # 添加到缓冲区
            self.buffer.append((X, y))
            # 积累足够数据后开始训练
            if len(self.buffer) >= 50 and not self.is_fitted:
                self._initial_fit()
                self.is_fitted = True
            elif self.is_fitted:
                self._online_update(X, y)
            # 每100个样本输出一次性能
            if (i + 1) % 100 == 0:
                print(f"处理了 {i+1} 个样本")
    def _initial_fit(self):
        """
        初始拟合
        """
        X_batch = np.array([item[0] for item in self.buffer])
        y_batch = np.array([item[1] for item in self.buffer])
        # 标准化
        X_scaled = self.scaler.fit_transform(X_batch)
        # 首次fit需要指定classes
        self.model.partial_fit(X_scaled, y_batch)
    def _online_update(self, X, y):
        """
        在线更新
        """
        X_scaled = self.scaler.transform([X])
        self.model.partial_fit(X_scaled, [y])
# 模拟数据流
def data_stream_generator(n_samples=1000):
    np.random.seed(42)
    for _ in range(n_samples):
        X = np.random.randn(5)  # 5维特征
        y = np.sum(X) + np.random.randn() * 0.1  # 带噪声的目标值
        yield X, y
# 运行流式处理
processor = StreamingDataProcessor(buffer_size=100)
processor.process_stream(data_stream_generator(500))

3 实际应用案例：在线评论情感分析

import numpy as np
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.linear_model import SGDClassifier
class OnlineSentimentAnalyzer:
    def __init__(self):
        # 使用HashingVectorizer避免存储词汇表
        self.vectorizer = HashingVectorizer(
            n_features=2**18,  # 262144个特征
            ngram_range=(1, 2),  # 使用unigrams和bigrams
            alternate_sign=False
        )
        self.classifier = SGDClassifier(
            loss='hinge',  # SVM
            penalty='l2',
            alpha=1e-4,
            max_iter=1,
            tol=1e-3,
            random_state=42
        )
        self.classes = np.array([0, 1])  # 0: 负面, 1: 正面
    def partial_fit(self, texts, labels):
        """
        增量学习新的文本数据
        """
        X = self.vectorizer.transform(texts)
        self.classifier.partial_fit(X, labels, classes=self.classes)
    def predict(self, texts):
        """
        预测情感
        """
        X = self.vectorizer.transform(texts)
        return self.classifier.predict(X)
    def predict_proba(self, texts):
        """
        预测概率
        """
        X = self.vectorizer.transform(texts)
        return self.classifier.predict_proba(X)
# 模拟在线评论流
class CommentStream:
    def __init__(self):
        self.comments = [
            ("这个产品非常好！", 1),
            ("太差了，不推荐", 0),
            ("还不错，性价比高", 1),
            ("质量很差，很快坏了", 0),
            ("物流很快，好评！", 1),
            ("一般般吧，凑合用", 0),
            ("服务态度很好", 1),
            ("完全不行，退货了", 0),
            ("比想象中要好", 1),
            ("有点失望，一般", 0),
        ] * 10  # 重复10次模拟流数据
    def __iter__(self):
        return iter(self.comments)
# 使用示例
analyzer = OnlineSentimentAnalyzer()
stream = CommentStream()
print("开始在线情感分析学习...")
batch_size = 5
batch_texts = []
batch_labels = []
for i, (text, label) in enumerate(stream):
    batch_texts.append(text)
    batch_labels.append(label)
    # 每积累5个样本进行一次更新
    if (i + 1) % batch_size == 0:
        analyzer.partial_fit(batch_texts, batch_labels)
        batch_texts = []
        batch_labels = []
        # 测试当前模型
        if (i + 1) % 20 == 0:  # 每20个样本测试一次
            test_texts = ["非常好用", "质量太差了", "还不错", "很失望"]
            predictions = analyzer.predict(test_texts)
            print(f"\n处理 {i+1} 个样本后:")
            for text, pred in zip(test_texts, predictions):
                sentiment = "正面" if pred == 1 else "负面"
                print(f"  '{text}' -> {sentiment}")

高级在线学习策略

1 自适应学习率调整

from sklearn.linear_model import SGDClassifier
import numpy as np
class AdaptiveOnlineLearner:
    def __init__(self):
        self.model = SGDClassifier(
            learning_rate='adaptive',
            eta0=0.01,
            power_t=0.5,
            max_iter=1
        )
        self.performance_history = []
        self.learning_rate = 0.01
    def adaptive_update(self, X_batch, y_batch, validation_set=None):
        """
        自适应更新模型
        """
        # 更新模型
        self.model.partial_fit(X_batch, y_batch)
        # 如果有验证集，评估性能
        if validation_set is not None:
            X_val, y_val = validation_set
            score = self.model.score(X_val, y_val)
            self.performance_history.append(score)
            # 根据性能变化调整学习策略
            if len(self.performance_history) > 5:
                recent_performance = self.performance_history[-5:]
                if len(set(recent_performance)) == 1:
                    # 性能稳定，可以尝试增加学习率
                    self.model.eta0 *= 1.1

2 动态特征选择

from sklearn.linear_model import SGDClassifier
from sklearn.feature_selection import SelectFromModel
import numpy as np
class OnlineFeatureSelector:
    def __init__(self, n_features_to_select=10):
        self.sgd = SGDClassifier(max_iter=1, tol=1e-3)
        self.selector = None
        self.n_features_to_select = n_features_to_select
        self.buffer_X = []
        self.buffer_y = []
    def partial_fit(self, X, y):
        """
        带特征选择的增量学习
        """
        self.buffer_X.append(X)
        self.buffer_y.append(y)
        # 每积累100个样本更新一次特征选择
        if len(self.buffer_X) >= 100:
            X_stack = np.vstack(self.buffer_X)
            y_stack = np.hstack(self.buffer_y)
            # 训练临时模型用于特征选择
            temp_model = SGDClassifier(max_iter=100)
            temp_model.fit(X_stack, y_stack)
            # 特征选择
            self.selector = SelectFromModel(
                temp_model, 
                max_features=self.n_features_to_select
            )
            self.selector.fit(X_stack, y_stack)
            # 清空缓冲区
            self.buffer_X = []
            self.buffer_y = []
        # 使用选择后的特征进行训练
        if self.selector is not None:
            X_selected = self.selector.transform(X.reshape(1, -1))
            self.sgd.partial_fit(X_selected, [y], classes=[0, 1])

最佳实践建议

class OnlineLearningBestPractices:
    """
    在线学习最佳实践
    """
    @staticmethod
    def data_augmentation(X, y):
        """
        数据增强防止过拟合
        """
        from sklearn.utils import shuffle
        # 添加噪声
        noise = np.random.normal(0, 0.01, X.shape)
        X_augmented = X + noise
        return X_augmented, y
    @staticmethod
    def concept_drift_detection(model, X, y, window_size=100):
        """
        概念漂移检测
        """
        predictions = model.predict(X)
        recent_accuracy = np.mean(predictions == y)
        # 如果最近准确率下降明显，可能需要重训练
        if recent_accuracy < 0.6:  # 阈值可调整
            print("检测到概念漂移，建议模型重置")
            return True
        return False
    @staticmethod
    def ensemble_online_learning(X, y, n_models=3):
        """
        集成在线学习
        """
        models = [
            SGDClassifier(max_iter=1, random_state=i) 
            for i in range(n_models)
        ]
        predictions = []
        for model in models:
            model.partial_fit(X, y, classes=[0, 1])
            predictions.append(model.predict(X))
        # 投票集成
        final_prediction = np.round(np.mean(predictions, axis=0))
        return final_prediction