项目1: 回归分析实战¶
难度: ⭐⭐ 中等 时间: 8-12小时 涉及知识: 线性回归、多项式回归、正则化、模型评估
📖 项目概述¶
项目背景¶
回归分析是机器学习中最基础也是最重要的任务之一，用于预测连续值。广泛应用于房价预测、股票价格预测、销量预测等场景。本项目将带你从零开始构建一个完整的回归分析系统。
项目目标¶
构建一个完整的回归分析系统，能够： - 处理和分析数据 - 实现多种回归算法 - 进行特征工程 - 评估模型性能 - 可视化结果 - 部署模型为Web服务
技术栈¶
- 机器学习框架: scikit-learn
- 数据处理: pandas, numpy
- 可视化: matplotlib, seaborn
- Web框架: Streamlit
🏗️ 项目结构¶
Text Only
regression-analysis/
├── data/ # 数据目录
│ ├── raw/ # 原始数据
│ ├── processed/ # 处理后数据
│ └── features/ # 特征数据
├── models/ # 模型目录
│ ├── __init__.py
│ ├── linear.py # 线性回归
│ ├── polynomial.py # 多项式回归
│ └── regularized.py # 正则化回归
├── utils/ # 工具函数
│ ├── __init__.py
│ ├── data_preprocessing.py # 数据预处理
│ ├── feature_engineering.py # 特征工程
│ ├── model_evaluation.py # 模型评估
│ └── visualization.py # 可视化
├── train.py # 训练脚本
├── evaluate.py # 评估脚本
├── app.py # Web应用
├── config.py # 配置文件
└── requirements.txt # 依赖文件
🎯 核心功能¶
1. 数据预处理¶
- 数据加载: 加载CSV/Excel数据
- 缺失值处理: 填充或删除缺失值
- 异常值检测: 识别和处理异常值
- 数据标准化: 标准化/归一化
2. 特征工程¶
- 特征选择: 选择重要特征
- 特征变换: 对数变换、Box-Cox变换
- 多项式特征: 生成交互项
- 特征编码: 独热编码、标签编码
3. 回归模型¶
- 线性回归: 基础线性回归
- 多项式回归: 非线性关系建模
- Ridge回归: L2正则化
- Lasso回归: L1正则化
- ElasticNet: L1+L2正则化
4. 模型评估¶
- MSE: 均方误差
- RMSE: 均方根误差
- MAE: 平均绝对误差
- R²: 决定系数
- 交叉验证: K折交叉验证
5. 结果可视化¶
- 散点图: 实际值vs预测值
- 残差图: 残差分析
- 学习曲线: 训练过程可视化
- 特征重要性: 特征重要性可视化
💻 代码实现¶
1. 配置文件 (config.py)¶
Python
"""
回归分析配置文件
"""
from dataclasses import dataclass
@dataclass # @dataclass自动生成__init__等方法
class Config:
"""配置类"""
# 数据配置
data_dir: str = "./data"
train_size: float = 0.8
random_state: int = 42
# 特征工程配置
polynomial_degree: int = 2
feature_selection_method: str = "all" # all, k_best, rfe
n_features: int = 10
# 模型配置
model_type: str = "linear" # linear, ridge, lasso, elasticnet
alpha: float = 1.0 # 正则化系数
l1_ratio: float = 0.5 # ElasticNet的L1比例
# 训练配置
cv_folds: int = 5
scoring: str = "neg_mean_squared_error"
# 可视化配置
figsize: tuple[int, int] = (12, 8)
style: str = "seaborn-v0_8" # matplotlib 3.6+ 中 "seaborn" 已弃用
# 保存配置
model_dir: str = "./models"
plot_dir: str = "./plots"
config = Config()
2. 数据预处理 (utils/data_preprocessing.py)¶
Python
"""
数据预处理
"""
from typing import Tuple
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.impute import SimpleImputer
class DataPreprocessor:
"""数据预处理器"""
def __init__(self, config): # __init__构造方法，创建对象时自动调用
"""初始化预处理器"""
self.config = config
self.scaler = None
self.imputer = None
def load_data(self, filepath) -> pd.DataFrame:
"""
加载数据（支持文件路径或Streamlit UploadedFile对象）
Args:
filepath: 文件路径(str)或Streamlit UploadedFile对象
Returns:
数据框
"""
# 兼容Streamlit的UploadedFile对象（使用.name属性获取文件名）
filename = filepath.name if hasattr(filepath, 'name') else filepath
if filename.endswith('.csv'):
return pd.read_csv(filepath)
elif filename.endswith('.xlsx'):
return pd.read_excel(filepath)
else:
raise ValueError("不支持的文件格式")
def handle_missing_values(
self,
df: pd.DataFrame,
strategy: str = "mean",
) -> pd.DataFrame:
"""
处理缺失值
Args:
df: 数据框
strategy: 填充策略 (mean, median, most_frequent)
Returns:
处理后的数据框
"""
self.imputer = SimpleImputer(strategy=strategy)
df_imputed = pd.DataFrame(
self.imputer.fit_transform(df),
columns=df.columns,
index=df.index,
)
return df_imputed
def detect_outliers(
self,
df: pd.DataFrame,
column: str,
method: str = "iqr",
threshold: float = 1.5,
) -> pd.Series:
"""
检测异常值
Args:
df: 数据框
column: 列名
method: 检测方法 (iqr, zscore)
threshold: 阈值
Returns:
异常值布尔序列
"""
if method == "iqr":
Q1 = df[column].quantile(0.25)
Q3 = df[column].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - threshold * IQR
upper_bound = Q3 + threshold * IQR
return (df[column] < lower_bound) | (df[column] > upper_bound)
elif method == "zscore":
z_scores = np.abs((df[column] - df[column].mean()) / df[column].std())
return z_scores > threshold
def remove_outliers(
self,
df: pd.DataFrame,
column: str,
method: str = "iqr",
threshold: float = 1.5,
) -> pd.DataFrame:
"""
移除异常值
Args:
df: 数据框
column: 列名
method: 检测方法
threshold: 阈值
Returns:
处理后的数据框
"""
outliers = self.detect_outliers(df, column, method, threshold)
return df[~outliers]
def normalize_data(
self,
df: pd.DataFrame,
method: str = "standard",
) -> pd.DataFrame:
"""
标准化数据
Args:
df: 数据框
method: 标准化方法 (standard, minmax)
Returns:
标准化后的数据框
"""
if method == "standard":
self.scaler = StandardScaler()
elif method == "minmax":
self.scaler = MinMaxScaler()
else:
raise ValueError("不支持的标准化方法")
df_normalized = pd.DataFrame(
self.scaler.fit_transform(df),
columns=df.columns,
index=df.index,
)
return df_normalized
def split_data(
self,
X: pd.DataFrame,
y: pd.Series,
) -> Tuple:
"""
划分数据集
Args:
X: 特征
y: 目标
Returns:
训练集和测试集
"""
return train_test_split(
X, y,
train_size=self.config.train_size,
random_state=self.config.random_state,
)
3. 线性回归模型 (models/linear.py)¶
Python
"""
线性回归模型
"""
import numpy as np
from sklearn.linear_model import LinearRegression, Ridge, Lasso, ElasticNet
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
from sklearn.model_selection import cross_val_score
class LinearRegressionModel:
"""线性回归模型"""
def __init__(self, model_type="linear", **kwargs): # *args接收任意位置参数，**kwargs接收任意关键字参数
"""
初始化模型
Args:
model_type: 模型类型
**kwargs: 模型参数
"""
self.model_type = model_type
if model_type == "linear":
self.model = LinearRegression(**kwargs)
elif model_type == "ridge":
self.model = Ridge(**kwargs)
elif model_type == "lasso":
self.model = Lasso(**kwargs)
elif model_type == "elasticnet":
self.model = ElasticNet(**kwargs)
else:
raise ValueError(f"不支持的模型类型: {model_type}")
def fit(self, X_train, y_train):
"""
训练模型
Args:
X_train: 训练特征
y_train: 训练目标
"""
self.model.fit(X_train, y_train)
def predict(self, X):
"""
预测
Args:
X: 特征
Returns:
预测值
"""
return self.model.predict(X)
def evaluate(self, X_test, y_test):
"""
评估模型
Args:
X_test: 测试特征
y_test: 测试目标
Returns:
评估指标字典
"""
y_pred = self.predict(X_test)
metrics = {
"mse": mean_squared_error(y_test, y_pred),
"rmse": np.sqrt(mean_squared_error(y_test, y_pred)),
"mae": mean_absolute_error(y_test, y_pred),
"r2": r2_score(y_test, y_pred),
}
return metrics
def cross_validate(self, X, y, cv=5):
"""
交叉验证
Args:
X: 特征
y: 目标
cv: 折数
Returns:
交叉验证分数
"""
scores = cross_val_score(
self.model, X, y,
cv=cv,
scoring="neg_mean_squared_error",
)
return -scores
def get_coefficients(self):
"""
获取系数
Returns:
系数数组
"""
if hasattr(self.model, 'coef_'): # hasattr检查对象是否有指定属性
return self.model.coef_
else:
return None
def get_intercept(self):
"""
获取截距
Returns:
截距值
"""
if hasattr(self.model, 'intercept_'):
return self.model.intercept_
else:
return None
4. 多项式回归 (models/polynomial.py)¶
Python
"""
多项式回归
"""
import numpy as np
from sklearn.preprocessing import PolynomialFeatures
from sklearn.linear_model import LinearRegression
from sklearn.pipeline import Pipeline
class PolynomialRegression:
"""多项式回归"""
def __init__(self, degree=2):
"""
初始化模型
Args:
degree: 多项式次数
"""
self.degree = degree
self.model = Pipeline([
('poly', PolynomialFeatures(degree=degree)),
('linear', LinearRegression()),
])
def fit(self, X_train, y_train):
"""
训练模型
Args:
X_train: 训练特征
y_train: 训练目标
"""
self.model.fit(X_train, y_train)
def predict(self, X):
"""
预测
Args:
X: 特征
Returns:
预测值
"""
return self.model.predict(X)
def evaluate(self, X_test, y_test):
"""
评估模型
Args:
X_test: 测试特征
y_test: 测试目标
Returns:
评估指标字典
"""
y_pred = self.predict(X_test)
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
metrics = {
"mse": mean_squared_error(y_test, y_pred),
"rmse": np.sqrt(mean_squared_error(y_test, y_pred)),
"mae": mean_absolute_error(y_test, y_pred),
"r2": r2_score(y_test, y_pred),
}
return metrics
5. 可视化工具 (utils/visualization.py)¶
Python
"""
可视化工具
"""
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
def plot_actual_vs_predicted(
y_true: np.ndarray,
y_pred: np.ndarray,
figsize: tuple[int, int] = (10, 6),
):
"""
绘制实际值vs预测值散点图
Args:
y_true: 实际值
y_pred: 预测值
figsize: 图形大小
"""
plt.figure(figsize=figsize)
# 绘制散点图
plt.scatter(y_true, y_pred, alpha=0.5)
# 绘制对角线
min_val = min(y_true.min(), y_pred.min())
max_val = max(y_true.max(), y_pred.max())
plt.plot([min_val, max_val], [min_val, max_val], 'r--', lw=2)
plt.xlabel('实际值', fontsize=12)
plt.ylabel('预测值', fontsize=12)
plt.title('实际值 vs 预测值', fontsize=14)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
def plot_residuals(
y_true: np.ndarray,
y_pred: np.ndarray,
figsize: tuple[int, int] = (10, 6),
):
"""
绘制残差图
Args:
y_true: 实际值
y_pred: 预测值
figsize: 图形大小
"""
residuals = y_true - y_pred
plt.figure(figsize=figsize)
plt.subplot(1, 2, 1)
plt.scatter(y_pred, residuals, alpha=0.5)
plt.axhline(y=0, color='r', linestyle='--')
plt.xlabel('预测值', fontsize=12)
plt.ylabel('残差', fontsize=12)
plt.title('残差 vs 预测值', fontsize=14)
plt.grid(True, alpha=0.3)
plt.subplot(1, 2, 2)
plt.hist(residuals, bins=30, edgecolor='black', alpha=0.7)
plt.xlabel('残差', fontsize=12)
plt.ylabel('频数', fontsize=12)
plt.title('残差分布', fontsize=14)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
def plot_feature_importance(
feature_names: list,
importance: np.ndarray,
top_n: int = 10,
figsize: tuple[int, int] = (10, 6),
):
"""
绘制特征重要性
Args:
feature_names: 特征名称列表
importance: 重要性数组
top_n: 显示前N个特征
figsize: 图形大小
"""
# 排序
indices = np.argsort(importance)[::-1][:top_n]
plt.figure(figsize=figsize)
plt.bar(range(len(indices)), importance[indices])
plt.xticks(range(len(indices)), [feature_names[i] for i in indices], rotation=45) # 列表推导式，简洁创建列表
plt.xlabel('特征', fontsize=12)
plt.ylabel('重要性', fontsize=12)
plt.title(f'前{top_n}个重要特征', fontsize=14)
plt.tight_layout()
plt.show()
def plot_learning_curve(
train_scores: np.ndarray,
val_scores: np.ndarray,
figsize: tuple[int, int] = (10, 6),
):
"""
绘制学习曲线
Args:
train_scores: 训练分数
val_scores: 验证分数
figsize: 图形大小
"""
plt.figure(figsize=figsize)
epochs = range(1, len(train_scores) + 1)
plt.plot(epochs, train_scores, 'b-', label='训练集')
plt.plot(epochs, val_scores, 'r-', label='验证集')
plt.xlabel('Epoch', fontsize=12)
plt.ylabel('分数', fontsize=12)
plt.title('学习曲线', fontsize=14)
plt.legend(fontsize=12)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
6. Streamlit应用 (app.py)¶
Python
"""
回归分析Web应用
"""
import streamlit as st
import pandas as pd
import numpy as np
import joblib
import os
from config import config
from utils.data_preprocessing import DataPreprocessor
from models.linear import LinearRegressionModel
from utils.visualization import plot_actual_vs_predicted, plot_residuals
# 页面配置
st.set_page_config(
page_title="回归分析系统",
page_icon="📊",
layout="wide"
)
# 标题
st.title("📊 回归分析系统")
st.markdown("---")
# 侧边栏
st.sidebar.header("模型设置")
# 模型选择
model_type = st.sidebar.selectbox(
"选择模型",
["linear", "ridge", "lasso", "elasticnet"],
format_func=lambda x: { # lambda匿名函数，简洁定义单行函数
"linear": "线性回归",
"ridge": "Ridge回归",
"lasso": "Lasso回归",
"elasticnet": "ElasticNet",
}[x],
)
# 正则化系数
if model_type in ["ridge", "lasso", "elasticnet"]:
alpha = st.sidebar.slider(
"正则化系数 (α)",
min_value=0.01,
max_value=10.0,
value=1.0,
step=0.1,
)
else:
alpha = 1.0
# ElasticNet的L1比例
if model_type == "elasticnet":
l1_ratio = st.sidebar.slider(
"L1比例",
min_value=0.0,
max_value=1.0,
value=0.5,
step=0.1,
)
else:
l1_ratio = 0.5
# 主界面
col1, col2 = st.columns(2)
with col1:
st.subheader("数据上传")
# 上传数据
uploaded_file = st.file_uploader(
"上传CSV或Excel文件",
type=['csv', 'xlsx'],
)
if uploaded_file is not None:
# 加载数据
preprocessor = DataPreprocessor(config)
df = preprocessor.load_data(uploaded_file)
st.write("数据预览:")
st.dataframe(df.head())
# 选择目标列
target_column = st.selectbox(
"选择目标列",
df.columns,
)
# 选择特征列
feature_columns = st.multiselect(
"选择特征列",
[col for col in df.columns if col != target_column],
default=[col for col in df.columns if col != target_column][:5],
)
if st.button("训练模型", type="primary"):
if not feature_columns:
st.warning("请至少选择一个特征列")
else:
# 准备数据
X = df[feature_columns]
y = df[target_column]
# 划分数据
X_train, X_test, y_train, y_test = preprocessor.split_data(X, y)
# 训练模型
with st.spinner("正在训练模型..."):
if model_type == "elasticnet":
model = LinearRegressionModel(
model_type=model_type,
alpha=alpha,
l1_ratio=l1_ratio,
)
else:
model = LinearRegressionModel(
model_type=model_type,
alpha=alpha,
)
model.fit(X_train, y_train)
# 评估模型
metrics = model.evaluate(X_test, y_test)
# 显示结果
with col2:
st.subheader("模型评估")
# 显示指标
st.metric("MSE", f"{metrics['mse']:.4f}")
st.metric("RMSE", f"{metrics['rmse']:.4f}")
st.metric("MAE", f"{metrics['mae']:.4f}")
st.metric("R²", f"{metrics['r2']:.4f}")
# 预测
y_pred = model.predict(X_test)
# 可视化
st.subheader("结果可视化")
# 实际值vs预测值
plot_actual_vs_predicted(y_test.values, y_pred)
# 残差图
plot_residuals(y_test.values, y_pred)
# 特征重要性
if model.get_coefficients() is not None:
st.subheader("特征重要性")
importance = np.abs(model.get_coefficients())
plot_feature_importance(
feature_columns,
importance,
top_n=min(10, len(feature_columns)),
)
🧪 测试方法¶
1. 单元测试¶
Python
"""
单元测试示例
"""
import pytest
import numpy as np
from models.linear import LinearRegressionModel
def test_linear_regression():
"""测试线性回归"""
# 创建模型
model = LinearRegressionModel(model_type="linear")
# 创建测试数据
X_train = np.random.randn(100, 5)
y_train = X_train @ np.array([1, 2, 3, 4, 5]) + np.random.randn(100) * 0.1 # np.array创建NumPy数组
X_test = np.random.randn(20, 5)
y_test = X_test @ np.array([1, 2, 3, 4, 5]) + np.random.randn(20) * 0.1
# 训练
model.fit(X_train, y_train)
# 预测
y_pred = model.predict(X_test)
# 验证
assert y_pred.shape == y_test.shape
metrics = model.evaluate(X_test, y_test)
assert metrics['r2'] > 0.5 # R²应该大于0.5
print("✓ 线性回归测试通过")
2. 集成测试¶
Python
"""
集成测试示例
"""
def test_regression_pipeline():
"""测试回归流程"""
from config import config
from utils.data_preprocessing import DataPreprocessor
# 创建预处理器
preprocessor = DataPreprocessor(config)
# 创建测试数据
df = pd.DataFrame({
'feature1': np.random.randn(100),
'feature2': np.random.randn(100),
'feature3': np.random.randn(100),
'target': np.random.randn(100),
})
# 处理数据
X = df[['feature1', 'feature2', 'feature3']]
y = df['target']
X_train, X_test, y_train, y_test = preprocessor.split_data(X, y)
# 训练模型
model = LinearRegressionModel(model_type="linear")
model.fit(X_train, y_train)
# 评估
metrics = model.evaluate(X_test, y_test)
# 验证
assert 'mse' in metrics
assert 'rmse' in metrics
assert 'mae' in metrics
assert 'r2' in metrics
print("✓ 回归流程测试通过")
📊 扩展建议¶
1. 功能扩展¶
- 时间序列: ARIMA, Prophet
- 集成方法: 随机森林, XGBoost
- 贝叶斯回归: 贝叶斯线性回归
- Gaussian Process: 高斯过程回归
2. 性能优化¶
- 特征选择: 递归特征消除
- 超参数调优: GridSearch, RandomSearch
- 模型融合: Stacking, Blending
- 自动ML: Auto-sklearn
3. 部署优化¶
- 模型导出: ONNX格式
- API服务: FastAPI
- 批量预测: 批量处理
- 缓存机制: 结果缓存
📚 学习收获¶
完成本项目后，你将掌握：
- ✅ 回归分析原理和算法
- ✅ 特征工程技术
- ✅ 正则化方法
- ✅ 模型评估指标
- ✅ 可视化技巧
- ✅ Streamlit Web应用开发
- ✅ 完整的回归分析项目开发
🔗 参考资源¶
项目完成时间: 8-12小时 难度等级: ⭐⭐ 中等 推荐指数: ⭐⭐⭐⭐⭐