Python机器学习项目完整工作流：从数据到部署
·
目录
『宝藏代码胶囊开张啦！』—— 我的 CodeCapsule 来咯！✨
写代码不再头疼！我的新站点 CodeCapsule 主打一个 “白菜价”+“量身定制”！无论是卡脖子的毕设/课设/文献复现，需要灵光一现的算法改进，还是想给项目加个“外挂”，这里都有便宜又好用的代码方案等你发现！低成本，高适配，助你轻松通关！速来围观 👉 CodeCapsule官网
Python机器学习项目完整工作流：从数据到部署
1. 引言：机器学习项目的完整生命周期
1.1 机器学习项目的重要性与挑战
在当今数据驱动的世界中，机器学习已经成为企业获得竞争优势的关键技术。根据Gartner的预测，到2025年，超过75%的企业将从试点机器学习项目转向系统化部署。然而，大多数机器学习项目在实践中面临严峻挑战：
- 高失败率：约85%的机器学习项目未能成功部署到生产环境
- 技术债务：快速迭代导致代码质量下降，维护成本增加
- 数据漂移：生产环境中的数据分布随时间变化，模型性能衰减
- 团队协作：数据科学家、工程师和业务人员之间的沟通障碍
一个结构化的完整工作流能够显著提高项目成功率。根据Google的研究，采用系统化MLOps实践的组织，其机器学习项目的成功率提高了3-5倍。
1.2 完整工作流的价值
完整的机器学习工作流不仅仅关注模型精度，更强调端到端的可重复性、可维护性和业务价值。通过标准化流程，团队能够：
- 减少70%的重复工作
- 提高模型部署速度50%以上
- 降低生产环境故障率60%
- 增强模型的可解释性和可信度
2. 项目规划与业务理解
2.1 定义项目目标与成功指标
在开始任何技术工作之前，必须明确业务目标和成功标准：
class ProjectDefiner:
"""项目定义器：明确机器学习项目目标和约束"""
def __init__(self):
self.project_scope = {}
self.success_metrics = {}
self.constraints = {}
def define_business_problem(self, problem_statement, business_impact):
"""定义业务问题和影响"""
self.project_scope = {
'problem_statement': problem_statement,
'business_impact': business_impact,
'stakeholders': [],
'timeline': {},
'budget': None
}
print("🎯 业务问题定义:")
print(f" 问题: {problem_statement}")
print(f" 业务影响: {business_impact}")
return self.project_scope
def set_success_metrics(self, technical_metrics, business_metrics):
"""设置成功指标"""
self.success_metrics = {
'technical': technical_metrics, # 如准确率、AUC等
'business': business_metrics # 如收入提升、成本节约等
}
print("\n📊 成功指标定义:")
print(" 技术指标:", technical_metrics)
print(" 业务指标:", business_metrics)
return self.success_metrics
def identify_constraints(self, data_constraints, technical_constraints, business_constraints):
"""识别项目约束"""
self.constraints = {
'data': data_constraints, # 数据可用性、质量等
'technical': technical_constraints, # 计算资源、延迟等
'business': business_constraints # 合规性、时间等
}
print("\n⚡ 项目约束:")
print(" 数据约束:", data_constraints)
print(" 技术约束:", technical_constraints)
print(" 业务约束:", business_constraints)
return self.constraints
def create_project_charter(self):
"""创建项目章程文档"""
charter = {
'project_title': '客户流失预测系统',
'version': '1.0',
'created_date': '2024-01-15',
'project_scope': self.project_scope,
'success_metrics': self.success_metrics,
'constraints': self.constraints,
'assumptions': [
'历史数据能够代表未来模式',
'特征在预测期间保持可用',
'业务环境相对稳定'
],
'risks': [
'数据质量问题可能影响模型性能',
'生产环境数据分布可能发生变化',
'模型可解释性可能影响业务采纳'
]
}
print("\n📋 项目章程摘要:")
for key, value in charter.items():
if key not in ['project_scope', 'success_metrics', 'constraints']:
print(f" {key}: {value}")
return charter
# 示例：定义客户流失预测项目
def define_churn_prediction_project():
"""定义客户流失预测项目"""
definer = ProjectDefiner()
# 定义业务问题
definer.define_business_problem(
problem_statement="预测哪些客户可能在接下来30天内流失",
business_impact="减少客户流失，预计每年节省500万收入"
)
# 设置成功指标
definer.set_success_metrics(
technical_metrics={'accuracy': 0.85, 'precision': 0.80, 'recall': 0.75},
business_metrics={'churn_reduction': 0.15, 'roi': 3.5}
)
# 识别约束
definer.identify_constraints(
data_constraints=['历史数据2年', '数据更新频率每日', 'GDPR合规'],
technical_constraints=['预测延迟<1秒', '模型大小<500MB', 'API可用性99.9%'],
business_constraints=['预算10万', '3个月交付', '可解释性要求高']
)
# 创建项目章程
charter = definer.create_project_charter()
return definer, charter
if __name__ == "__main__":
project_definer, project_charter = define_churn_prediction_project()
2.2 数据需求分析与项目计划
class DataRequirementsAnalyzer:
"""数据需求分析器"""
def __init__(self):
self.data_sources = []
self.feature_requirements = {}
self.data_quality_standards = {}
def analyze_data_sources(self, internal_sources, external_sources):
"""分析数据源"""
self.data_sources = {
'internal': internal_sources, # 内部数据库、数据仓库等
'external': external_sources # 第三方API、公开数据集等
}
print("📁 数据源分析:")
print(" 内部数据源:", internal_sources)
print(" 外部数据源:", external_sources)
return self.data_sources
def define_feature_requirements(self, required_features, optional_features):
"""定义特征需求"""
self.feature_requirements = {
'required': required_features, # 必需特征
'optional': optional_features # 可选特征
}
print("\n🎯 特征需求:")
print(" 必需特征:", required_features)
print(" 可选特征:", optional_features)
return self.feature_requirements
def set_data_quality_standards(self, completeness, accuracy, consistency, timeliness):
"""设置数据质量标准"""
self.data_quality_standards = {
'completeness': completeness, # 完整性标准
'accuracy': accuracy, # 准确性标准
'consistency': consistency, # 一致性标准
'timeliness': timeliness # 及时性标准
}
print("\n🔍 数据质量标准:")
for standard, value in self.data_quality_standards.items():
print(f" {standard}: {value}")
return self.data_quality_standards
def create_data_plan(self):
"""创建数据计划"""
data_plan = {
'data_collection_strategy': {
'etl_pipeline': '每日批处理',
'real_time_sources': ['用户行为流'],
'data_validation': '自动数据质量检查'
},
'feature_store_requirements': {
'online_features': ['最近购买金额', '活跃天数'],
'offline_features': ['历史总消费', '平均订单价值']
},
'privacy_and_compliance': {
'data_retention': '24个月',
'pii_handling': '加密存储',
'access_control': '基于角色的权限'
}
}
print("\n📊 数据计划摘要:")
for category, details in data_plan.items():
print(f" {category}:")
for key, value in details.items():
print(f" - {key}: {value}")
return data_plan
class ProjectPlanner:
"""项目计划器"""
def __init__(self):
self.phases = {}
self.milestones = {}
self.resource_plan = {}
def create_project_phases(self):
"""创建项目阶段"""
self.phases = {
'phase_1': {
'name': '数据准备与探索',
'duration': '2周',
'deliverables': ['数据质量报告', '探索性分析'],
'dependencies': []
},
'phase_2': {
'name': '特征工程与建模',
'duration': '3周',
'deliverables': ['特征管道', '基准模型'],
'dependencies': ['phase_1']
},
'phase_3': {
'name': '模型优化与验证',
'duration': '2周',
'deliverables': ['优化模型', 'A/B测试计划'],
'dependencies': ['phase_2']
},
'phase_4': {
'name': '部署与监控',
'duration': '3周',
'deliverables': ['生产API', '监控仪表板'],
'dependencies': ['phase_3']
}
}
print("📅 项目阶段计划:")
for phase_id, phase_info in self.phases.items():
print(f" {phase_id}: {phase_info['name']} ({phase_info['duration']})")
return self.phases
def set_milestones(self):
"""设置关键里程碑"""
self.milestones = {
'milestone_1': {
'name': '数据准备完成',
'date': '2024-02-01',
'acceptance_criteria': ['数据质量>95%', '特征定义完成']
},
'milestone_2': {
'name': '基准模型建立',
'date': '2024-02-22',
'acceptance_criteria': ['准确率>80%', '业务验证通过']
},
'milestone_3': {
'name': '生产部署',
'date': '2024-03-15',
'acceptance_criteria': ['API性能达标', '监控系统就绪']
}
}
print("\n🎯 关键里程碑:")
for milestone_id, milestone_info in self.milestones.items():
print(f" {milestone_id}: {milestone_info['name']} ({milestone_info['date']})")
return self.milestones
# 运行项目规划
def run_project_planning():
"""运行完整的项目规划"""
print("🚀 开始机器学习项目规划")
print("=" * 60)
# 数据需求分析
data_analyzer = DataRequirementsAnalyzer()
data_analyzer.analyze_data_sources(
internal_sources=['CRM系统', '交易数据库', '用户行为日志'],
external_sources=['经济指标API', '行业基准数据']
)
data_analyzer.define_feature_requirements(
required_features=['购买频率', '客单价', '服务使用时长'],
optional_features=['社交媒体活跃度', '客户满意度评分']
)
data_analyzer.set_data_quality_standards(
completeness=0.95,
accuracy=0.98,
consistency=0.90,
timeliness='T+1'
)
data_plan = data_analyzer.create_data_plan()
# 项目计划
project_planner = ProjectPlanner()
phases = project_planner.create_project_phases()
milestones = project_planner.set_milestones()
return data_analyzer, project_planner
if __name__ == "__main__":
data_analyzer, project_planner = run_project_planning()
3. 数据收集与探索性分析
3.1 数据收集与质量检查
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
import warnings
warnings.filterwarnings('ignore')
class DataCollector:
"""数据收集器：模拟真实业务数据收集过程"""
def __init__(self, random_state=42):
self.random_state = random_state
self.raw_data = None
self.data_quality_report = {}
def generate_sample_churn_data(self, n_samples=10000):
"""生成模拟客户流失数据"""
np.random.seed(self.random_state)
# 生成分类数据
X, y = make_classification(
n_samples=n_samples,
n_features=12,
n_informative=8,
n_redundant=2,
n_clusters_per_class=1,
flip_y=0.05,
random_state=self.random_state
)
# 创建有意义的特征名称
feature_names = [
'tenure_months', # 使用月数
'monthly_charges', # 月费用
'total_charges', # 总费用
'contract_type', # 合同类型
'payment_method', # 支付方式
'paperless_billing', # 电子账单
'tech_support', # 技术支持
'online_security', # 在线安全
'online_backup', # 在线备份
'device_protection', # 设备保护
'streaming_tv', # 流媒体TV
'streaming_movies' # 流媒体电影
]
# 创建DataFrame
df = pd.DataFrame(X, columns=feature_names)
# 调整特征范围使其更符合业务实际
df['tenure_months'] = (df['tenure_months'] * 20 + 1).astype(int) # 1-60个月
df['monthly_charges'] = (df['monthly_charges'] * 50 + 20).round(2) # 20-120美元
df['total_charges'] = (df['total_charges'] * 1000 + 100).round(2) # 100-2000美元
# 添加分类变量
df['contract_type'] = np.random.choice(['Monthly', 'Yearly', 'Two-year'], n_samples)
df['payment_method'] = np.random.choice(['Bank transfer', 'Credit card', 'Electronic check'], n_samples)
# 添加目标变量
df['churn'] = y
# 添加一些缺失值和异常值以模拟真实数据
self._add_realistic_noise(df)
self.raw_data = df
print(f"✅ 生成模拟数据完成: {df.shape[0]} 行, {df.shape[1]} 列")
return df
def _add_realistic_noise(self, df):
"""添加现实的噪声"""
# 添加缺失值 (5%)
for col in ['total_charges', 'tech_support', 'online_security']:
mask = np.random.random(len(df)) < 0.05
df.loc[mask, col] = np.nan
# 添加异常值 (2%)
for col in ['monthly_charges', 'total_charges']:
outlier_mask = np.random.random(len(df)) < 0.02
df.loc[outlier_mask, col] = df[col] * 10
# 添加数据不一致性
inconsistency_mask = (df['tenure_months'] < 3) & (df['total_charges'] > 1000)
df.loc[inconsistency_mask, 'total_charges'] = df.loc[inconsistency_mask, 'monthly_charges'] * df.loc[inconsistency_mask, 'tenure_months']
def perform_data_quality_check(self, df):
"""执行数据质量检查"""
print("\n🔍 执行数据质量检查...")
quality_metrics = {}
# 1. 完整性检查
completeness = {}
for col in df.columns:
missing_count = df[col].isna().sum()
missing_percentage = (missing_count / len(df)) * 100
completeness[col] = {
'missing_count': missing_count,
'missing_percentage': missing_percentage,
'status': 'GOOD' if missing_percentage < 5 else 'WARNING' if missing_percentage < 20 else 'CRITICAL'
}
quality_metrics['completeness'] = completeness
# 2. 唯一性检查
uniqueness = {}
for col in df.columns:
unique_count = df[col].nunique()
unique_percentage = (unique_count / len(df)) * 100
uniqueness[col] = {
'unique_count': unique_count,
'unique_percentage': unique_percentage
}
quality_metrics['uniqueness'] = uniqueness
# 3. 有效性检查（数据类型和范围）
validity = {}
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
validity[col] = {
'min': df[col].min(),
'max': df[col].max(),
'mean': df[col].mean(),
'std': df[col].std(),
'zeros': (df[col] == 0).sum(),
'negatives': (df[col] < 0).sum()
}
quality_metrics['validity'] = validity
# 4. 一致性检查
consistency_issues = []
# 检查总费用是否合理
expected_total = df['monthly_charges'] * df['tenure_months']
inconsistency = abs(df['total_charges'] - expected_total) > 10
if inconsistency.sum() > 0:
consistency_issues.append(f"总费用不一致: {inconsistency.sum()} 条记录")
quality_metrics['consistency_issues'] = consistency_issues
self.data_quality_report = quality_metrics
# 打印质量报告
self._print_quality_report(quality_metrics)
return quality_metrics
def _print_quality_report(self, quality_metrics):
"""打印质量报告"""
print("\n📊 数据质量报告:")
print("=" * 50)
# 完整性总结
print("\n1. 完整性分析:")
critical_issues = 0
for col, metrics in quality_metrics['completeness'].items():
if metrics['status'] != 'GOOD':
print(f" ⚠️ {col}: {metrics['missing_percentage']:.1f}% 缺失 ({metrics['status']})")
if metrics['status'] == 'CRITICAL':
critical_issues += 1
if critical_issues == 0:
print(" ✅ 无严重缺失值问题")
# 有效性总结
print("\n2. 有效性分析:")
numeric_issues = 0
for col, metrics in quality_metrics['validity'].items():
if metrics['negatives'] > 0 and col in ['monthly_charges', 'total_charges']:
print(f" ⚠️ {col}: 发现 {metrics['negatives']} 个负值")
numeric_issues += 1
if numeric_issues == 0:
print(" ✅ 无数值范围问题")
# 一致性总结
print("\n3. 一致性分析:")
if quality_metrics['consistency_issues']:
for issue in quality_metrics['consistency_issues']:
print(f" ⚠️ {issue}")
else:
print(" ✅ 无数据一致性问题")
# 总体评估
total_issues = critical_issues + numeric_issues + len(quality_metrics['consistency_issues'])
if total_issues == 0:
print("\n🎉 数据质量优秀，可以进入下一步")
elif total_issues <= 3:
print(f"\nℹ️ 发现 {total_issues} 个问题，需要在预处理阶段解决")
else:
print(f"\n🚨 发现 {total_issues} 个问题，需要重点关注数据质量")
# 运行数据收集和质量检查
def run_data_collection_and_quality():
"""运行数据收集和质量检查流程"""
print("🚀 开始数据收集与质量检查")
print("=" * 50)
# 收集数据
collector = DataCollector()
raw_data = collector.generate_sample_churn_data(10000)
# 数据质量检查
quality_report = collector.perform_data_quality_check(raw_data)
return collector, raw_data, quality_report
if __name__ == "__main__":
data_collector, raw_data, quality_report = run_data_collection_and_quality()
3.2 探索性数据分析
class DataExplorer:
"""数据探索器：执行全面的探索性数据分析"""
def __init__(self, df):
self.df = df
self.insights = {}
def perform_comprehensive_eda(self):
"""执行全面的探索性数据分析"""
print("\n🔍 开始探索性数据分析...")
# 1. 基础统计
self._basic_statistics()
# 2. 目标变量分析
self._target_variable_analysis()
# 3. 特征分布分析
self._feature_distribution_analysis()
# 4. 相关性分析
self._correlation_analysis()
# 5. 多变量分析
self._multivariate_analysis()
# 6. 数据洞察总结
self._summarize_insights()
return self.insights
def _basic_statistics(self):
"""基础统计分析"""
print("\n1. 📈 基础统计信息:")
print(f" 数据集形状: {self.df.shape}")
print(f" 内存使用: {self.df.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
# 数据类型分布
dtype_counts = self.df.dtypes.value_counts()
print(" 数据类型分布:")
for dtype, count in dtype_counts.items():
print(f" - {dtype}: {count} 列")
self.insights['basic_stats'] = {
'shape': self.df.shape,
'memory_mb': self.df.memory_usage(deep=True).sum() / 1024**2,
'dtype_distribution': dtype_counts.to_dict()
}
def _target_variable_analysis(self):
"""目标变量分析"""
if 'churn' not in self.df.columns:
return
print("\n2. 🎯 目标变量分析 (Churn):")
churn_counts = self.df['churn'].value_counts()
churn_percentage = self.df['churn'].value_counts(normalize=True) * 100
print(f" 流失分布:")
for value, (count, percentage) in enumerate(zip(churn_counts, churn_percentage)):
status = "流失" if value == 1 else "留存"
print(f" - {status}: {count} 客户 ({percentage:.1f}%)")
# 可视化目标变量分布
plt.figure(figsize=(10, 6))
plt.subplot(1, 2, 1)
churn_counts.plot(kind='bar', color=['lightblue', 'lightcoral'])
plt.title('客户流失分布')
plt.xlabel('流失状态')
plt.ylabel('客户数量')
plt.xticks(ticks=[0, 1], labels=['留存', '流失'], rotation=0)
plt.subplot(1, 2, 2)
plt.pie(churn_counts, labels=['留存', '流失'], autopct='%1.1f%%',
colors=['lightblue', 'lightcoral'])
plt.title('客户流失比例')
plt.tight_layout()
plt.show()
self.insights['target_analysis'] = {
'churn_rate': churn_percentage[1],
'class_imbalance': churn_percentage[1] / churn_percentage[0]
}
def _feature_distribution_analysis(self):
"""特征分布分析"""
print("\n3. 📊 特征分布分析:")
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
categorical_cols = self.df.select_dtypes(include=['object']).columns
print(f" 数值特征 ({len(numeric_cols)}): {list(numeric_cols)}")
print(f" 分类特征 ({len(categorical_cols)}): {list(categorical_cols)}")
# 数值特征分布可视化
if len(numeric_cols) > 0:
n_numeric = len(numeric_cols)
n_cols = 3
n_rows = (n_numeric + n_cols - 1) // n_cols
fig, axes = plt.subplots(n_rows, n_cols, figsize=(15, 5*n_rows))
axes = axes.flatten()
for i, col in enumerate(numeric_cols):
if i < len(axes):
self.df[col].hist(bins=30, ax=axes[i], alpha=0.7, color='skyblue')
axes[i].set_title(f'{col} 分布')
axes[i].set_xlabel(col)
axes[i].set_ylabel('频数')
# 隐藏多余的子图
for i in range(n_numeric, len(axes)):
axes[i].set_visible(False)
plt.tight_layout()
plt.show()
# 分类特征分布
if len(categorical_cols) > 0:
fig, axes = plt.subplots(1, len(categorical_cols), figsize=(15, 5))
if len(categorical_cols) == 1:
axes = [axes]
for i, col in enumerate(categorical_cols):
if i < len(axes):
value_counts = self.df[col].value_counts()
axes[i].bar(value_counts.index, value_counts.values, color='lightgreen')
axes[i].set_title(f'{col} 分布')
axes[i].set_xlabel(col)
axes[i].set_ylabel('频数')
axes[i].tick_params(axis='x', rotation=45)
plt.tight_layout()
plt.show()
self.insights['feature_distributions'] = {
'numeric_features': len(numeric_cols),
'categorical_features': len(categorical_cols)
}
def _correlation_analysis(self):
"""相关性分析"""
print("\n4. 🔗 特征相关性分析:")
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) > 1:
# 计算相关性矩阵
correlation_matrix = self.df[numeric_cols].corr()
# 可视化相关性热力图
plt.figure(figsize=(12, 10))
sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm', center=0,
fmt='.2f', square=True, cbar_kws={'shrink': 0.8})
plt.title('特征相关性热力图')
plt.tight_layout()
plt.show()
# 找出与目标变量最相关的特征
if 'churn' in numeric_cols:
churn_correlations = correlation_matrix['churn'].sort_values(ascending=False)
print(" 与流失最相关的特征:")
for feature, corr in churn_correlations.items():
if feature != 'churn' and abs(corr) > 0.1:
direction = "正相关" if corr > 0 else "负相关"
print(f" - {feature}: {corr:.3f} ({direction})")
self.insights['correlation_analysis'] = {
'highly_correlated_features': self._find_highly_correlated(correlation_matrix)
}
def _find_highly_correlated(self, corr_matrix, threshold=0.8):
"""找出高度相关的特征对"""
highly_correlated = []
for i in range(len(corr_matrix.columns)):
for j in range(i+1, len(corr_matrix.columns)):
if abs(corr_matrix.iloc[i, j]) > threshold:
highly_correlated.append((
corr_matrix.columns[i],
corr_matrix.columns[j],
corr_matrix.iloc[i, j]
))
return highly_correlated
def _multivariate_analysis(self):
"""多变量分析"""
print("\n5. 📊 多变量分析:")
# 数值特征 vs 目标变量
if 'churn' in self.df.columns:
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
important_numeric = [col for col in numeric_cols if col != 'churn'][:4]
if len(important_numeric) > 0:
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
axes = axes.flatten()
for i, col in enumerate(important_numeric):
if i < len(axes):
# 箱线图显示不同流失状态的分布
data_to_plot = [self.df[self.df['churn'] == 0][col],
self.df[self.df['churn'] == 1][col]]
axes[i].boxplot(data_to_plot, labels=['留存', '流失'])
axes[i].set_title(f'{col} vs 流失状态')
axes[i].set_ylabel(col)
plt.tight_layout()
plt.show()
# 分类特征 vs 目标变量
categorical_cols = self.df.select_dtypes(include=['object']).columns
if len(categorical_cols) > 0 and 'churn' in self.df.columns:
for col in categorical_cols[:2]: # 只显示前两个分类特征
cross_tab = pd.crosstab(self.df[col], self.df['churn'], normalize='index')
cross_tab.plot(kind='bar', figsize=(10, 6), color=['lightblue', 'lightcoral'])
plt.title(f'{col} 与流失率关系')
plt.xlabel(col)
plt.ylabel('比例')
plt.legend(['留存', '流失'])
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()
def _summarize_insights(self):
"""总结数据洞察"""
print("\n6. 💡 关键数据洞察总结:")
print("=" * 50)
insights = []
# 从分析中提取洞察
if 'target_analysis' in self.insights:
churn_rate = self.insights['target_analysis']['churn_rate']
insights.append(f"• 整体流失率: {churn_rate:.1f}%")
if churn_rate > 20:
insights.append("• ⚠️ 流失率较高，需要重点关注")
elif churn_rate < 5:
insights.append("• ✅ 流失率在健康范围内")
if 'correlation_analysis' in self.insights:
high_corr_pairs = self.insights['correlation_analysis']['highly_correlated_features']
if high_corr_pairs:
insights.append("• 🔍 发现高度相关的特征对，考虑特征选择")
# 数据质量问题洞察
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
if self.df[col].isna().sum() > 0:
insights.append(f"• 🚨 {col} 存在缺失值需要处理")
break
# 业务洞察建议
insights.extend([
"• 💡 建议对数值特征进行标准化处理",
"• 💡 分类特征需要进行编码转换",
"• 💡 考虑创建交互特征和多项式特征"
])
for insight in insights:
print(f" {insight}")
self.insights['summary'] = insights
# 运行探索性数据分析
def run_exploratory_data_analysis(df):
"""运行完整的探索性数据分析"""
print("🚀 开始探索性数据分析")
print("=" * 50)
explorer = DataExplorer(df)
insights = explorer.perform_comprehensive_eda()
return explorer, insights
if __name__ == "__main__":
# 假设已经有数据
data_collector = DataCollector()
sample_data = data_collector.generate_sample_churn_data(5000)
explorer, insights = run_exploratory_data_analysis(sample_data)
4. 特征工程与数据预处理
4.1 数据清洗与特征转换
from sklearn.preprocessing import StandardScaler, LabelEncoder, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
import pandas as pd
import numpy as np
class DataPreprocessor:
"""数据预处理器：执行数据清洗和基本特征工程"""
def __init__(self, df):
self.df = df.copy()
self.preprocessing_steps = {}
self.transformed_df = None
def handle_missing_values(self, strategy='auto'):
"""处理缺失值"""
print("🔧 处理缺失值...")
missing_before = self.df.isnull().sum().sum()
if strategy == 'auto':
# 自动选择策略
for col in self.df.columns:
if self.df[col].isnull().sum() > 0:
if self.df[col].dtype in ['object']:
# 分类变量用众数填充
self.df[col].fillna(self.df[col].mode()[0] if not self.df[col].mode().empty else 'Unknown', inplace=True)
else:
# 数值变量用中位数填充
self.df[col].fillna(self.df[col].median(), inplace=True)
missing_after = self.df.isnull().sum().sum()
print(f" 处理前缺失值: {missing_before}")
print(f" 处理后缺失值: {missing_after}")
self.preprocessing_steps['missing_values'] = {
'strategy': strategy,
'missing_before': missing_before,
'missing_after': missing_after
}
return self.df
def handle_outliers(self, method='iqr'):
"""处理异常值"""
print("🔧 处理异常值...")
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
outlier_count_before = 0
outlier_count_after = 0
for col in numeric_cols:
if col == 'churn': # 跳过目标变量
continue
Q1 = self.df[col].quantile(0.25)
Q3 = self.df[col].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers_before = ((self.df[col] < lower_bound) | (self.df[col] > upper_bound)).sum()
outlier_count_before += outliers_before
if method == 'cap':
# 缩尾处理
self.df[col] = np.where(self.df[col] < lower_bound, lower_bound, self.df[col])
self.df[col] = np.where(self.df[col] > upper_bound, upper_bound, self.df[col])
elif method == 'remove':
# 移除异常值
self.df = self.df[~((self.df[col] < lower_bound) | (self.df[col] > upper_bound))]
outliers_after = ((self.df[col] < lower_bound) | (self.df[col] > upper_bound)).sum()
outlier_count_after += outliers_after
print(f" 处理前异常值: {outlier_count_before}")
print(f" 处理后异常值: {outlier_count_after}")
self.preprocessing_steps['outliers'] = {
'method': method,
'outliers_before': outlier_count_before,
'outliers_after': outlier_count_after
}
return self.df
def encode_categorical_variables(self, encoding_strategy='auto'):
"""编码分类变量"""
print("🔧 编码分类变量...")
categorical_cols = self.df.select_dtypes(include=['object']).columns
if encoding_strategy == 'auto':
for col in categorical_cols:
unique_count = self.df[col].nunique()
if unique_count <= 5:
# 使用One-Hot编码
dummies = pd.get_dummies(self.df[col], prefix=col)
self.df = pd.concat([self.df, dummies], axis=1)
self.df.drop(col, axis=1, inplace=True)
print(f" {col}: One-Hot编码 ({unique_count}个类别)")
else:
# 使用标签编码或频率编码
if self.df[col].dtype == 'object':
# 频率编码
freq_encoding = self.df[col].value_counts().to_dict()
self.df[f'{col}_freq'] = self.df[col].map(freq_encoding)
self.df.drop(col, axis=1, inplace=True)
print(f" {col}: 频率编码 ({unique_count}个类别)")
self.preprocessing_steps['categorical_encoding'] = {
'strategy': encoding_strategy,
'columns_encoded': list(categorical_cols)
}
return self.df
def create_new_features(self):
"""创建新特征"""
print("🔧 创建新特征...")
# 基于业务逻辑创建特征
if 'tenure_months' in self.df.columns and 'monthly_charges' in self.df.columns:
# 客户生命周期价值
self.df['customer_lifetime_value'] = self.df['tenure_months'] * self.df['monthly_charges']
if 'tenure_months' in self.df.columns:
# 客户分段
self.df['tenure_segment'] = pd.cut(self.df['tenure_months'],
bins=[0, 12, 24, 60, np.inf],
labels=['New', 'Regular', 'Loyal', 'VIP'])
# 创建交互特征
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) >= 2:
# 选择最重要的两个数值特征创建交互项
col1, col2 = numeric_cols[0], numeric_cols[1]
self.df[f'{col1}_x_{col2}'] = self.df[col1] * self.df[col2]
new_features_count = len([col for col in self.df.columns if col not in self.preprocessing_steps.get('original_columns', [])])
print(f" 创建了 {new_features_count} 个新特征")
self.preprocessing_steps['feature_engineering'] = {
'new_features_created': new_features_count
}
return self.df
def scale_numerical_features(self):
"""标准化数值特征"""
print("🔧 标准化数值特征...")
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
# 移除目标变量（如果有）
if 'churn' in numeric_cols:
numeric_cols = numeric_cols.drop('churn')
if len(numeric_cols) > 0:
scaler = StandardScaler()
self.df[numeric_cols] = scaler.fit_transform(self.df[numeric_cols])
print(f" 标准化了 {len(numeric_cols)} 个数值特征")
self.preprocessing_steps['feature_scaling'] = {
'columns_scaled': list(numeric_cols),
'method': 'StandardScaler'
}
return self.df
def execute_full_preprocessing(self):
"""执行完整的预处理流程"""
print("🚀 开始完整的数据预处理流程")
print("=" * 50)
# 保存原始列名用于跟踪
self.preprocessing_steps['original_columns'] = list(self.df.columns)
# 执行预处理步骤
self.handle_missing_values()
self.handle_outliers()
self.encode_categorical_variables()
self.create_new_features()
self.scale_numerical_features()
self.transformed_df = self.df
print("\n✅ 数据预处理完成!")
print(f" 最终数据集形状: {self.df.shape}")
return self.df
class FeatureSelector:
"""特征选择器：选择最重要的特征"""
def __init__(self, df, target_column='churn'):
self.df = df
self.target_column = target_column
self.selected_features = []
self.feature_importance = {}
def correlation_based_selection(self, threshold=0.05):
"""基于相关性的特征选择"""
print("🔍 基于相关性的特征选择...")
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
if self.target_column in numeric_cols:
correlations = self.df[numeric_cols].corr()[self.target_column].abs()
selected_features = correlations[correlations > threshold].index.tolist()
selected_features.remove(self.target_column) # 移除目标变量
self.selected_features = selected_features
self.feature_importance = correlations[selected_features].to_dict()
print(f" 选择了 {len(selected_features)} 个特征 (阈值: {threshold})")
print(" 重要特征:", selected_features)
return self.selected_features
def tree_based_selection(self, n_features=15):
"""基于树模型的特征选择"""
from sklearn.ensemble import RandomForestClassifier
print("🔍 基于树模型的特征选择...")
X = self.df.drop(self.target_column, axis=1)
y = self.df[self.target_column]
# 只处理数值特征
numeric_cols = X.select_dtypes(include=[np.number]).columns
X_numeric = X[numeric_cols]
rf = RandomForestClassifier(n_estimators=100, random_state=42)
rf.fit(X_numeric, y)
# 获取特征重要性
importance_scores = rf.feature_importances_
feature_importance_df = pd.DataFrame({
'feature': numeric_cols,
'importance': importance_scores
}).sort_values('importance', ascending=False)
# 选择最重要的特征
top_features = feature_importance_df.head(n_features)['feature'].tolist()
self.selected_features = top_features
self.feature_importance = feature_importance_df.set_index('feature')['importance'].to_dict()
print(f" 选择了 {len(top_features)} 个最重要的特征")
print(" 特征重要性排名:")
for i, (feature, importance) in enumerate(feature_importance_df.head(10).values):
print(f" {i+1}. {feature}: {importance:.4f}")
return self.selected_features
def get_final_features(self, method='tree', **kwargs):
"""获取最终特征集合"""
if method == 'correlation':
return self.correlation_based_selection(**kwargs)
elif method == 'tree':
return self.tree_based_selection(**kwargs)
else:
# 返回所有特征（除了目标变量）
all_features = self.df.columns.tolist()
if self.target_column in all_features:
all_features.remove(self.target_column)
return all_features
# 运行完整的数据预处理和特征工程
def run_feature_engineering_pipeline(raw_df):
"""运行特征工程完整流程"""
print("🚀 开始特征工程与数据预处理")
print("=" * 50)
# 数据预处理
preprocessor = DataPreprocessor(raw_df)
processed_df = preprocessor.execute_full_preprocessing()
# 特征选择
feature_selector = FeatureSelector(processed_df)
selected_features = feature_selector.get_final_features(method='tree', n_features=15)
# 创建最终数据集
final_features = selected_features + ['churn']
final_df = processed_df[final_features]
print(f"\n🎯 最终特征集: {len(selected_features)} 个特征")
print(" 特征列表:", selected_features)
return preprocessor, feature_selector, final_df
if __name__ == "__main__":
# 生成示例数据并运行特征工程
data_collector = DataCollector()
raw_data = data_collector.generate_sample_churn_data(5000)
preprocessor, feature_selector, final_data = run_feature_engineering_pipeline(raw_data)
5. 模型训练与优化
5.1 模型训练与交叉验证
from sklearn.model_selection import cross_val_score, StratifiedKFold
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score
import matplotlib.pyplot as plt
import seaborn as sns
class ModelTrainer:
"""模型训练器：训练和评估多个机器学习模型"""
def __init__(self, X, y):
self.X = X
self.y = y
self.models = {}
self.results = {}
self.best_model = None
def initialize_models(self):
"""初始化多个机器学习模型"""
self.models = {
'Logistic Regression': LogisticRegression(random_state=42, max_iter=1000),
'Random Forest': RandomForestClassifier(n_estimators=100, random_state=42),
'Gradient Boosting': GradientBoostingClassifier(n_estimators=100, random_state=42),
'SVM': SVC(probability=True, random_state=42)
}
print("🤖 初始化的模型:")
for name, model in self.models.items():
print(f" - {name}")
return self.models
def train_and_evaluate_models(self, cv_folds=5):
"""训练并评估所有模型"""
print("\n🚀 开始模型训练与评估...")
# 设置交叉验证
cv = StratifiedKFold(n_splits=cv_folds, shuffle=True, random_state=42)
for name, model in self.models.items():
print(f"\n🔍 训练 {name}...")
# 交叉验证
cv_scores = cross_val_score(model, self.X, self.y, cv=cv, scoring='roc_auc')
# 训练最终模型
model.fit(self.X, self.y)
# 预测
y_pred = model.predict(self.X)
y_pred_proba = model.predict_proba(self.X)[:, 1]
# 计算指标
train_auc = roc_auc_score(self.y, y_pred_proba)
self.results[name] = {
'model': model,
'cv_mean_auc': cv_scores.mean(),
'cv_std_auc': cv_scores.std(),
'train_auc': train_auc,
'cv_scores': cv_scores,
'predictions': y_pred,
'probabilities': y_pred_proba
}
print(f" ✅ 交叉验证 AUC: {cv_scores.mean():.4f} (±{cv_scores.std():.4f})")
print(f" ✅ 训练集 AUC: {train_auc:.4f}")
return self.results
def compare_models(self):
"""比较所有模型的性能"""
print("\n🏆 模型性能比较")
print("=" * 60)
print(f"{'Model':<20} {'CV AUC Mean':<12} {'CV AUC Std':<12} {'Train AUC':<12}")
print("-" * 60)
best_score = -1
best_model_name = None
for name, result in self.results.items():
cv_mean = result['cv_mean_auc']
cv_std = result['cv_std_auc']
train_auc = result['train_auc']
print(f"{name:<20} {cv_mean:<12.4f} {cv_std:<12.4f} {train_auc:<12.4f}")
if cv_mean > best_score:
best_score = cv_mean
best_model_name = name
print("-" * 60)
print(f"🎯 最佳模型: {best_model_name} (CV AUC: {best_score:.4f})")
self.best_model = {
'name': best_model_name,
'model': self.results[best_model_name]['model'],
'score': best_score
}
return self.best_model
def visualize_model_comparison(self):
"""可视化模型比较结果"""
model_names = list(self.results.keys())
cv_scores = [self.results[name]['cv_mean_auc'] for name in model_names]
cv_stds = [self.results[name]['cv_std_auc'] for name in model_names]
plt.figure(figsize=(10, 6))
y_pos = np.arange(len(model_names))
plt.barh(y_pos, cv_scores, xerr=cv_stds, align='center', alpha=0.7, color='skyblue')
plt.yticks(y_pos, model_names)
plt.xlabel('交叉验证 AUC 分数')
plt.title('模型性能比较')
# 添加数值标签
for i, v in enumerate(cv_scores):
plt.text(v + 0.01, i, f'{v:.3f}', va='center')
plt.tight_layout()
plt.show()
def get_best_model(self):
"""获取最佳模型"""
if self.best_model is None:
self.compare_models()
return self.best_model
class HyperparameterOptimizer:
"""超参数优化器"""
def __init__(self, model, param_grid):
self.model = model
self.param_grid = param_grid
self.best_params = None
self.best_score = None
def optimize_with_gridsearch(self, X, y, cv=5):
"""使用网格搜索优化超参数"""
from sklearn.model_selection import GridSearchCV
print(f"🔧 对 {self.model.__class__.__name__} 进行超参数优化...")
grid_search = GridSearchCV(
estimator=self.model,
param_grid=self.param_grid,
cv=cv,
scoring='roc_auc',
n_jobs=-1,
verbose=1
)
grid_search.fit(X, y)
self.best_params = grid_search.best_params_
self.best_score = grid_search.best_score_
best_model = grid_search.best_estimator_
print(f"✅ 最佳参数: {self.best_params}")
print(f"✅ 最佳分数: {self.best_score:.4f}")
return best_model
# 运行模型训练流程
def run_model_training_pipeline(X, y):
"""运行完整的模型训练流程"""
print("🚀 开始模型训练流程")
print("=" * 50)
# 初始化训练器
trainer = ModelTrainer(X, y)
trainer.initialize_models()
results = trainer.train_and_evaluate_models(cv_folds=5)
best_model = trainer.compare_models()
trainer.visualize_model_comparison()
return trainer, best_model
# 超参数优化示例
def optimize_random_forest(X, y):
"""优化随机森林超参数"""
rf = RandomForestClassifier(random_state=42)
param_grid = {
'n_estimators': [50, 100, 200],
'max_depth': [10, 20, None],
'min_samples_split': [2, 5, 10],
'min_samples_leaf': [1, 2, 4]
}
optimizer = HyperparameterOptimizer(rf, param_grid)
optimized_model = optimizer.optimize_with_gridsearch(X, y, cv=5)
return optimizer, optimized_model
if __name__ == "__main__":
# 假设已经有预处理好的数据
data_collector = DataCollector()
raw_data = data_collector.generate_sample_churn_data(5000)
# 预处理数据
preprocessor = DataPreprocessor(raw_data)
processed_data = preprocessor.execute_full_preprocessing()
# 准备特征和目标
X = processed_data.drop('churn', axis=1)
y = processed_data['churn']
# 训练模型
trainer, best_model = run_model_training_pipeline(X, y)
# 超参数优化
print("\n" + "="*50)
print("开始超参数优化...")
optimizer, optimized_rf = optimize_random_forest(X, y)
5.2 模型评估与解释
from sklearn.metrics import precision_recall_curve, roc_curve, auc
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
class ModelEvaluator:
"""模型评估器：全面评估模型性能"""
def __init__(self, model, X, y, model_name="Model"):
self.model = model
self.X = X
self.y = y
self.model_name = model_name
self.predictions = None
self.probabilities = None
def comprehensive_evaluation(self):
"""执行全面评估"""
print(f"📊 对 {self.model_name} 进行全面评估")
print("=" * 50)
# 预测
self.predictions = self.model.predict(self.X)
self.probabilities = self.model.predict_proba(self.X)[:, 1]
# 1. 基础指标
self._calculate_basic_metrics()
# 2. 可视化分析
self._create_evaluation_visualizations()
# 3. 业务指标
self._calculate_business_metrics()
print("\n✅ 模型评估完成!")
def _calculate_basic_metrics(self):
"""计算基础指标"""
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
accuracy = accuracy_score(self.y, self.predictions)
precision = precision_score(self.y, self.predictions)
recall = recall_score(self.y, self.predictions)
f1 = f1_score(self.y, self.predictions)
auc_score = roc_auc_score(self.y, self.probabilities)
print("\n📈 基础性能指标:")
print(f" 准确率 (Accuracy): {accuracy:.4f}")
print(f" 精确率 (Precision): {precision:.4f}")
print(f" 召回率 (Recall): {recall:.4f}")
print(f" F1分数: {f1:.4f}")
print(f" AUC分数: {auc_score:.4f}")
# 分类报告
print("\n📋 详细分类报告:")
print(classification_report(self.y, self.predictions, target_names=['非流失', '流失']))
def _create_evaluation_visualizations(self):
"""创建评估可视化"""
fig, axes = plt.subplots(2, 2, figsize=(15, 12))
fig.suptitle(f'{self.model_name} 模型评估', fontsize=16, fontweight='bold')
# 1. 混淆矩阵
cm = confusion_matrix(self.y, self.predictions)
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', ax=axes[0, 0])
axes[0, 0].set_title('混淆矩阵')
axes[0, 0].set_xlabel('预测标签')
axes[0, 0].set_ylabel('真实标签')
axes[0, 0].set_xticklabels(['非流失', '流失'])
axes[0, 0].set_yticklabels(['非流失', '流失'])
# 2. ROC曲线
fpr, tpr, _ = roc_curve(self.y, self.probabilities)
roc_auc = auc(fpr, tpr)
axes[0, 1].plot(fpr, tpr, color='darkorange', lw=2, label=f'ROC曲线 (AUC = {roc_auc:.2f})')
axes[0, 1].plot([0, 1], [0, 1], color='navy', lw=2, linestyle='--', label='随机分类器')
axes[0, 1].set_xlim([0.0, 1.0])
axes[0, 1].set_ylim([0.0, 1.05])
axes[0, 1].set_xlabel('假正率')
axes[0, 1].set_ylabel('真正率')
axes[0, 1].set_title('ROC曲线')
axes[0, 1].legend(loc="lower right")
axes[0, 1].grid(True, alpha=0.3)
# 3. 精确率-召回率曲线
precision_vals, recall_vals, _ = precision_recall_curve(self.y, self.probabilities)
axes[1, 0].plot(recall_vals, precision_vals, color='green', lw=2)
axes[1, 0].set_xlabel('召回率')
axes[1, 0].set_ylabel('精确率')
axes[1, 0].set_title('精确率-召回率曲线')
axes[1, 0].grid(True, alpha=0.3)
# 4. 概率分布
probabilities_df = pd.DataFrame({
'probability': self.probabilities,
'actual': self.y
})
for actual_value in [0, 1]:
subset = probabilities_df[probabilities_df['actual'] == actual_value]
label = '非流失' if actual_value == 0 else '流失'
axes[1, 1].hist(subset['probability'], bins=30, alpha=0.6, label=label)
axes[1, 1].set_xlabel('预测概率')
axes[1, 1].set_ylabel('频数')
axes[1, 1].set_title('预测概率分布')
axes[1, 1].legend()
axes[1, 1].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
def _calculate_business_metrics(self):
"""计算业务指标"""
# 假设的业务价值
acquisition_cost = 200 # 获取新客户成本
lifetime_value = 1000 # 客户生命周期价值
# 计算混淆矩阵元素
tn, fp, fn, tp = confusion_matrix(self.y, self.predictions).ravel()
# 业务指标计算
total_customers = len(self.y)
actual_churn = self.y.sum()
predicted_churn = self.predictions.sum()
# 成本效益分析
false_positive_cost = fp * 50 # 误报成本（不必要的保留措施）
false_negative_cost = fn * (lifetime_value - acquisition_cost) # 漏报成本（失去客户）
true_positive_benefit = tp * (lifetime_value - acquisition_cost - 50) # 正确预测流失的收益
total_net_benefit = true_positive_benefit - false_positive_cost - false_negative_cost
print("\n💼 业务影响分析:")
print(f" 总客户数: {total_customers}")
print(f" 实际流失客户: {actual_churn}")
print(f" 预测流失客户: {predicted_churn}")
print(f" 正确预测流失: {tp}")
print(f" 漏报流失: {fn}")
print(f" 误报流失: {fp}")
print(f" 净业务收益: ${total_net_benefit:,.2f}")
class ModelInterpreter:
"""模型解释器：解释模型预测"""
def __init__(self, model, feature_names):
self.model = model
self.feature_names = feature_names
def analyze_feature_importance(self):
"""分析特征重要性"""
if hasattr(self.model, 'feature_importances_'):
importances = self.model.feature_importances_
feature_importance_df = pd.DataFrame({
'feature': self.feature_names,
'importance': importances
}).sort_values('importance', ascending=False)
print("\n🔍 特征重要性分析:")
print(feature_importance_df.head(10))
# 可视化特征重要性
plt.figure(figsize=(10, 8))
top_features = feature_importance_df.head(15)
plt.barh(top_features['feature'], top_features['importance'], color='lightcoral')
plt.xlabel('特征重要性')
plt.title('Top 15 特征重要性')
plt.gca().invert_yaxis()
plt.tight_layout()
plt.show()
return feature_importance_df
else:
print("该模型不支持特征重要性分析")
return None
# 运行模型评估和解释
def run_model_evaluation_pipeline(model, X, y, feature_names, model_name="Model"):
"""运行完整的模型评估流程"""
print("🚀 开始模型评估与解释")
print("=" * 50)
# 模型评估
evaluator = ModelEvaluator(model, X, y, model_name)
evaluator.comprehensive_evaluation()
# 模型解释
interpreter = ModelInterpreter(model, feature_names)
feature_importance = interpreter.analyze_feature_importance()
return evaluator, interpreter
if __name__ == "__main__":
# 假设已经有训练好的模型和数据
data_collector = DataCollector()
raw_data = data_collector.generate_sample_churn_data(5000)
preprocessor = DataPreprocessor(raw_data)
processed_data = preprocessor.execute_full_preprocessing()
X = processed_data.drop('churn', axis=1)
y = processed_data['churn']
# 训练一个模型
from sklearn.ensemble import RandomForestClassifier
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
# 评估模型
evaluator, interpreter = run_model_evaluation_pipeline(
model, X, y, list(X.columns), "随机森林"
)
6. 模型部署与生产化
6.1 创建部署就绪的模型管道
import joblib
import json
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
import pandas as pd
import numpy as np
class ModelDeploymentPipeline:
"""模型部署管道：创建生产就绪的机器学习管道"""
def __init__(self, model, preprocessor, feature_names):
self.model = model
self.preprocessor = preprocessor
self.feature_names = feature_names
self.pipeline = None
self.metadata = {}
def create_production_pipeline(self):
"""创建生产环境管道"""
print("🔧 创建生产环境模型管道...")
# 创建完整的预处理和建模管道
self.pipeline = Pipeline([
('preprocessor', self.preprocessor),
('model', self.model)
])
# 创建管道元数据
self.metadata = {
'model_type': type(self.model).__name__,
'feature_names': self.feature_names,
'created_date': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S'),
'version': '1.0.0',
'input_schema': self._create_input_schema()
}
print("✅ 生产管道创建完成")
return self.pipeline
def _create_input_schema(self):
"""创建输入数据模式"""
schema = {
'required_features': self.feature_names,
'data_types': {
'tenure_months': 'numeric',
'monthly_charges': 'numeric',
'total_charges': 'numeric',
'contract_type': 'categorical',
'payment_method': 'categorical'
},
'constraints': {
'tenure_months': {'min': 0, 'max': 72},
'monthly_charges': {'min': 0, 'max': 200}
}
}
return schema
def save_pipeline(self, model_path='model.pkl', metadata_path='model_metadata.json'):
"""保存管道和元数据"""
if self.pipeline is None:
print("❌ 请先创建管道")
return False
try:
# 保存模型管道
joblib.dump(self.pipeline, model_path)
print(f"✅ 模型管道保存至: {model_path}")
# 保存元数据
with open(metadata_path, 'w') as f:
json.dump(self.metadata, f, indent=2)
print(f"✅ 模型元数据保存至: {metadata_path}")
return True
except Exception as e:
print(f"❌ 保存失败: {e}")
return False
def load_pipeline(self, model_path='model.pkl', metadata_path='model_metadata.json'):
"""加载管道和元数据"""
try:
# 加载模型管道
self.pipeline = joblib.load(model_path)
print(f"✅ 模型管道从 {model_path} 加载")
# 加载元数据
with open(metadata_path, 'r') as f:
self.metadata = json.load(f)
print(f"✅ 模型元数据从 {metadata_path} 加载")
return True
except Exception as e:
print(f"❌ 加载失败: {e}")
return False
def validate_input(self, input_data):
"""验证输入数据"""
if not isinstance(input_data, dict):
raise ValueError("输入数据必须是字典格式")
# 检查必需特征
missing_features = set(self.metadata['input_schema']['required_features']) - set(input_data.keys())
if missing_features:
raise ValueError(f"缺少必需特征: {missing_features}")
# 验证数据类型和范围
for feature, value in input_data.items():
if feature in self.metadata['input_schema']['data_types']:
expected_type = self.metadata['input_schema']['data_types'][feature]
if expected_type == 'numeric' and not isinstance(value, (int, float)):
raise ValueError(f"特征 {feature} 应该是数值类型")
# 检查数值范围
if feature in self.metadata['input_schema']['constraints']:
constraints = self.metadata['input_schema']['constraints'][feature]
if 'min' in constraints and value < constraints['min']:
raise ValueError(f"特征 {feature} 值 {value} 小于最小值 {constraints['min']}")
if 'max' in constraints and value > constraints['max']:
raise ValueError(f"特征 {feature} 值 {value} 大于最大值 {constraints['max']}")
print("✅ 输入数据验证通过")
return True
def predict(self, input_data):
"""使用管道进行预测"""
if self.pipeline is None:
raise ValueError("请先加载或创建管道")
# 验证输入
self.validate_input(input_data)
# 转换为DataFrame
input_df = pd.DataFrame([input_data])
try:
# 进行预测
prediction = self.pipeline.predict(input_df)[0]
probability = self.pipeline.predict_proba(input_df)[0]
result = {
'prediction': int(prediction),
'probability': float(probability[1]), # 流失概率
'confidence': 'high' if max(probability) > 0.8 else 'medium' if max(probability) > 0.6 else 'low',
'timestamp': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')
}
print(f"✅ 预测完成: 流失概率 {result['probability']:.3f}")
return result
except Exception as e:
print(f"❌ 预测失败: {e}")
raise
class ModelMonitoring:
"""模型监控：监控生产环境模型性能"""
def __init__(self, model_pipeline):
self.pipeline = model_pipeline
self.performance_history = []
self.data_drift_detector = DataDriftDetector()
def log_prediction(self, input_data, prediction, actual_result=None):
"""记录预测结果"""
log_entry = {
'timestamp': pd.Timestamp.now(),
'input_data': input_data,
'prediction': prediction,
'actual_result': actual_result
}
self.performance_history.append(log_entry)
# 保持最近1000条记录
if len(self.performance_history) > 1000:
self.performance_history = self.performance_history[-1000:]
def calculate_performance_metrics(self):
"""计算性能指标"""
if len(self.performance_history) < 100:
return None
# 获取有实际结果的记录
evaluated_predictions = [log for log in self.performance_history if log['actual_result'] is not None]
if len(evaluated_predictions) == 0:
return None
actuals = [log['actual_result'] for log in evaluated_predictions]
predictions = [log['prediction']['prediction'] for log in evaluated_predictions]
probabilities = [log['prediction']['probability'] for log in evaluated_predictions]
from sklearn.metrics import accuracy_score, precision_score, recall_score
metrics = {
'accuracy': accuracy_score(actuals, predictions),
'precision': precision_score(actuals, predictions, zero_division=0),
'recall': recall_score(actuals, predictions, zero_division=0),
'sample_size': len(evaluated_predictions),
'last_updated': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')
}
return metrics
def check_data_drift(self, reference_data, current_data):
"""检查数据漂移"""
drift_report = self.data_drift_detector.detect_drift(reference_data, current_data)
if drift_report['has_drift']:
print(f"🚨 检测到数据漂移: {drift_report['drift_score']:.3f}")
return True
else:
print("✅ 数据分布稳定")
return False
class DataDriftDetector:
"""数据漂移检测器"""
def detect_drift(self, reference_data, current_data, threshold=0.1):
"""检测数据漂移"""
from scipy import stats
drift_report = {
'has_drift': False,
'drift_score': 0,
'feature_drifts': {}
}
numeric_columns = reference_data.select_dtypes(include=[np.number]).columns
for col in numeric_columns:
# KS检验检测分布变化
statistic, p_value = stats.ks_2samp(reference_data[col].dropna(), current_data[col].dropna())
drift_report['feature_drifts'][col] = {
'ks_statistic': statistic,
'p_value': p_value,
'has_drift': p_value < 0.05
}
if p_value < 0.05:
drift_report['has_drift'] = True
drift_report['drift_score'] += statistic
drift_report['drift_score'] /= len(numeric_columns)
drift_report['has_drift'] = drift_report['drift_score'] > threshold
return drift_report
# 创建和测试部署管道
def create_deployment_pipeline():
"""创建完整的部署管道"""
print("🚀 创建模型部署管道")
print("=" * 50)
# 生成示例数据
data_collector = DataCollector()
raw_data = data_collector.generate_sample_churn_data(1000)
# 创建预处理管道
numeric_features = ['tenure_months', 'monthly_charges', 'total_charges']
categorical_features = ['contract_type', 'payment_method']
numeric_transformer = Pipeline(steps=[
('imputer', SimpleImputer(strategy='median')),
('scaler', StandardScaler())
])
categorical_transformer = Pipeline(steps=[
('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
('onehot', OneHotEncoder(handle_unknown='ignore'))
])
preprocessor = ColumnTransformer(
transformers=[
('num', numeric_transformer, numeric_features),
('cat', categorical_transformer, categorical_features)
])
# 训练模型
from sklearn.ensemble import RandomForestClassifier
model = RandomForestClassifier(n_estimators=100, random_state=42)
X = raw_data[numeric_features + categorical_features]
y = raw_data['churn']
# 创建完整管道
full_pipeline = Pipeline([
('preprocessor', preprocessor),
('model', model)
])
full_pipeline.fit(X, y)
# 创建部署管道
deployment_pipeline = ModelDeploymentPipeline(
model=model,
preprocessor=preprocessor,
feature_names=numeric_features + categorical_features
)
deployment_pipeline.pipeline = full_pipeline
deployment_pipeline.create_production_pipeline()
# 保存管道
deployment_pipeline.save_pipeline()
# 测试预测
test_input = {
'tenure_months': 24,
'monthly_charges': 65.50,
'total_charges': 1572.00,
'contract_type': 'Yearly',
'payment_method': 'Credit card'
}
print("\n🔮 测试预测:")
prediction = deployment_pipeline.predict(test_input)
print(f" 预测结果: {prediction}")
return deployment_pipeline
if __name__ == "__main__":
deployment_pipeline = create_deployment_pipeline()
6.2 创建REST API服务
from flask import Flask, request, jsonify
import pandas as pd
import joblib
import os
class MLModelAPI:
"""机器学习模型API服务"""
def __init__(self, model_path='model.pkl', metadata_path='model_metadata.json'):
self.app = Flask(__name__)
self.model_pipeline = None
self.metadata = None
self.model_path = model_path
self.metadata_path = metadata_path
self._setup_routes()
self._load_model()
def _load_model(self):
"""加载模型和元数据"""
try:
if os.path.exists(self.model_path):
self.model_pipeline = joblib.load(self.model_path)
print(f"✅ 模型从 {self.model_path} 加载成功")
else:
print(f"❌ 模型文件 {self.model_path} 不存在")
return False
if os.path.exists(self.metadata_path):
import json
with open(self.metadata_path, 'r') as f:
self.metadata = json.load(f)
print(f"✅ 元数据从 {self.metadata_path} 加载成功")
else:
print(f"⚠️ 元数据文件 {self.metadata_path} 不存在")
return True
except Exception as e:
print(f"❌ 加载模型失败: {e}")
return False
def _setup_routes(self):
"""设置API路由"""
@self.app.route('/health', methods=['GET'])
def health_check():
"""健康检查端点"""
return jsonify({
'status': 'healthy',
'model_loaded': self.model_pipeline is not None,
'timestamp': pd.Timestamp.now().isoformat()
})
@self.app.route('/predict', methods=['POST'])
def predict():
"""预测端点"""
try:
# 获取输入数据
input_data = request.get_json()
if not input_data:
return jsonify({'error': '没有提供输入数据'}), 400
# 验证输入数据
if self.metadata:
required_features = self.metadata['input_schema']['required_features']
missing_features = set(required_features) - set(input_data.keys())
if missing_features:
return jsonify({
'error': f'缺少必需特征: {list(missing_features)}',
'required_features': required_features
}), 400
# 转换为DataFrame并进行预测
input_df = pd.DataFrame([input_data])
# 进行预测
prediction = self.model_pipeline.predict(input_df)[0]
probability = self.model_pipeline.predict_proba(input_df)[0]
# 准备响应
response = {
'prediction': int(prediction),
'probability': float(probability[1]),
'confidence': 'high' if max(probability) > 0.8 else 'medium' if max(probability) > 0.6 else 'low',
'timestamp': pd.Timestamp.now().isoformat(),
'model_version': self.metadata.get('version', 'unknown') if self.metadata else 'unknown'
}
return jsonify(response)
except Exception as e:
return jsonify({'error': f'预测失败: {str(e)}'}), 500
@self.app.route('/model_info', methods=['GET'])
def model_info():
"""模型信息端点"""
if self.metadata:
return jsonify(self.metadata)
else:
return jsonify({'error': '模型元数据不可用'}), 404
def run(self, host='0.0.0.0', port=5000, debug=False):
"""运行API服务"""
if self.model_pipeline is None:
print("❌ 无法启动API: 模型未加载")
return
print(f"🚀 启动机器学习API服务: http://{host}:{port}")
print(" 可用端点:")
print(" - GET /health : 健康检查")
print(" - POST /predict : 模型预测")
print(" - GET /model_info : 模型信息")
self.app.run(host=host, port=port, debug=debug)
# 客户端测试类
class ModelAPIClient:
"""模型API客户端"""
def __init__(self, base_url='http://localhost:5000'):
self.base_url = base_url
self.session = None
def health_check(self):
"""健康检查"""
import requests
try:
response = requests.get(f'{self.base_url}/health')
return response.json()
except requests.exceptions.ConnectionError:
return {'error': '无法连接到API服务'}
def predict(self, input_data):
"""发送预测请求"""
import requests
try:
response = requests.post(
f'{self.base_url}/predict',
json=input_data,
headers={'Content-Type': 'application/json'}
)
return response.json()
except requests.exceptions.ConnectionError:
return {'error': '无法连接到API服务'}
def get_model_info(self):
"""获取模型信息"""
import requests
try:
response = requests.get(f'{self.base_url}/model_info')
return response.json()
except requests.exceptions.ConnectionError:
return {'error': '无法连接到API服务'}
# 示例使用
def demo_api_usage():
"""演示API使用"""
print("🚀 演示机器学习API使用")
print("=" * 50)
# 创建客户端
client = ModelAPIClient()
# 健康检查
print("1. 🔍 健康检查:")
health = client.health_check()
print(f" 响应: {health}")
# 获取模型信息
print("\n2. 📋 模型信息:")
model_info = client.get_model_info()
if 'error' not in model_info:
print(f" 模型版本: {model_info.get('version', 'unknown')}")
print(f" 特征数量: {len(model_info.get('input_schema', {}).get('required_features', []))}")
else:
print(f" 错误: {model_info['error']}")
# 测试预测
print("\n3. 🔮 测试预测:")
test_data = {
'tenure_months': 24,
'monthly_charges': 65.50,
'total_charges': 1572.00,
'contract_type': 'Yearly',
'payment_method': 'Credit card'
}
prediction = client.predict(test_data)
print(f" 输入数据: {test_data}")
print(f" 预测结果: {prediction}")
# 创建简单的部署脚本
def create_deployment_script():
"""创建部署脚本"""
script_content = '''#!/bin/bash
# 机器学习模型部署脚本
echo "🚀 开始部署机器学习模型..."
# 检查Python环境
if ! command -v python &> /dev/null; then
echo "❌ Python未安装"
exit 1
fi
# 检查依赖
echo "📦 安装依赖..."
pip install -r requirements.txt
# 检查模型文件
if [ ! -f "model.pkl" ]; then
echo "❌ 模型文件 model.pkl 不存在"
exit 1
fi
# 启动API服务
echo "🌐 启动API服务..."
python api_service.py &
# 等待服务启动
sleep 5
# 测试服务
echo "🔍 测试服务..."
curl -s http://localhost:5000/health | python -m json.tool
echo "✅ 部署完成！服务运行在 http://localhost:5000"
'''
with open('deploy.sh', 'w') as f:
f.write(script_content)
print("✅ 部署脚本创建完成: deploy.sh")
if __name__ == "__main__":
# 注意：在生产环境中，应该使用WSGI服务器如Gunicorn
# 这里使用Flask开发服务器仅用于演示
# 创建部署管道（如果还没有）
deployment_pipeline = create_deployment_pipeline()
# 启动API服务
api = MLModelAPI()
# 在后台线程中运行API（用于演示）
import threading
api_thread = threading.Thread(target=api.run, kwargs={'debug': False, 'port': 5000})
api_thread.daemon = True
api_thread.start()
# 等待服务启动
import time
time.sleep(2)
# 演示API使用
demo_api_usage()
# 创建部署脚本
create_deployment_script()
print("\n🎉 模型部署完成！")
print(" 运行 './deploy.sh' 来部署到生产环境")
7. 完整项目代码整合
7.1 端到端项目实现
# complete_ml_pipeline.py
"""
完整的机器学习项目工作流：从数据到部署
作者：机器学习工程师
描述：客户流失预测项目的完整实现
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
import joblib
import json
import warnings
warnings.filterwarnings('ignore')
class CompleteMLWorkflow:
"""完整的机器学习工作流"""
def __init__(self):
self.raw_data = None
self.processed_data = None
self.model = None
self.pipeline = None
self.results = {}
def run_complete_workflow(self, sample_size=10000):
"""运行完整的工作流"""
print("🚀 开始完整的机器学习工作流")
print("=" * 60)
# 1. 数据收集与生成
self.data_collection(sample_size)
# 2. 数据探索与分析
self.data_exploration()
# 3. 数据预处理
self.data_preprocessing()
# 4. 特征工程
self.feature_engineering()
# 5. 模型训练
self.model_training()
# 6. 模型评估
self.model_evaluation()
# 7. 模型部署准备
self.deployment_preparation()
print("\n🎉 完整工作流执行完成！")
self._print_summary()
def data_collection(self, sample_size):
"""步骤1: 数据收集"""
print("\n1. 📊 数据收集")
print("-" * 40)
# 生成模拟数据（在实际项目中替换为真实数据收集）
np.random.seed(42)
# 创建模拟的客户流失数据
data = {
'customer_id': range(1, sample_size + 1),
'tenure_months': np.random.randint(1, 73, sample_size),
'monthly_charges': np.random.uniform(20, 120, sample_size).round(2),
'total_charges': np.random.uniform(100, 2000, sample_size).round(2),
'contract_type': np.random.choice(['Monthly', 'Yearly', 'Two-year'], sample_size),
'payment_method': np.random.choice(['Bank transfer', 'Credit card', 'Electronic check'], sample_size),
'paperless_billing': np.random.choice([0, 1], sample_size),
'tech_support': np.random.choice([0, 1], sample_size),
'online_security': np.random.choice([0, 1], sample_size)
}
self.raw_data = pd.DataFrame(data)
# 创建目标变量（基于业务规则）
churn_probability = (
0.1 + # 基础流失率
0.3 * (self.raw_data['contract_type'] == 'Monthly').astype(int) + # 月付合同更容易流失
0.2 * (self.raw_data['tenure_months'] < 12).astype(int) + # 新客户更容易流失
0.1 * (self.raw_data['monthly_charges'] > 80).astype(int) # 高费用客户更容易流失
)
self.raw_data['churn'] = np.random.binomial(1, churn_probability)
print(f"✅ 生成 {sample_size} 条客户记录")
print(f" 流失率: {self.raw_data['churn'].mean():.2%}")
print(f" 特征数量: {len(self.raw_data.columns) - 1}") # 减去目标变量
self.results['data_collection'] = {
'sample_size': sample_size,
'churn_rate': self.raw_data['churn'].mean(),
'features_count': len(self.raw_data.columns) - 1
}
def data_exploration(self):
"""步骤2: 数据探索"""
print("\n2. 🔍 数据探索")
print("-" * 40)
# 基础统计
print("📈 基础统计:")
print(f" 数据集形状: {self.raw_data.shape}")
print(f" 缺失值数量: {self.raw_data.isnull().sum().sum()}")
# 目标变量分析
churn_counts = self.raw_data['churn'].value_counts()
print(f"🎯 目标变量分布:")
print(f" 留存客户: {churn_counts[0]} ({churn_counts[0]/len(self.raw_data):.1%})")
print(f" 流失客户: {churn_counts[1]} ({churn_counts[1]/len(self.raw_data):.1%})")
# 特征相关性
numeric_cols = self.raw_data.select_dtypes(include=[np.number]).columns
if 'churn' in numeric_cols and len(numeric_cols) > 1:
correlations = self.raw_data[numeric_cols].corr()['churn'].abs().sort_values(ascending=False)
top_correlations = correlations[1:4] # 排除目标变量本身
print("🔗 与流失最相关的特征:")
for feature, corr in top_correlations.items():
print(f" - {feature}: {corr:.3f}")
self.results['data_exploration'] = {
'dataset_shape': self.raw_data.shape,
'missing_values': self.raw_data.isnull().sum().sum(),
'class_balance': churn_counts[1] / churn_counts[0]
}
def data_preprocessing(self):
"""步骤3: 数据预处理"""
print("\n3. 🔧 数据预处理")
print("-" * 40)
# 复制数据
self.processed_data = self.raw_data.copy()
# 处理缺失值
missing_before = self.processed_data.isnull().sum().sum()
numeric_cols = self.processed_data.select_dtypes(include=[np.number]).columns
categorical_cols = self.processed_data.select_dtypes(include=['object']).columns
for col in numeric_cols:
self.processed_data[col].fillna(self.processed_data[col].median(), inplace=True)
for col in categorical_cols:
self.processed_data[col].fillna(self.processed_data[col].mode()[0], inplace=True)
missing_after = self.processed_data.isnull().sum().sum()
print(f"✅ 缺失值处理: {missing_before} → {missing_after}")
self.results['data_preprocessing'] = {
'missing_values_before': missing_before,
'missing_values_after': missing_after
}
def feature_engineering(self):
"""步骤4: 特征工程"""
print("\n4. 🎯 特征工程")
print("-" * 40)
# 创建新特征
new_features_created = 0
# 客户价值相关特征
if 'tenure_months' in self.processed_data.columns and 'monthly_charges' in self.processed_data.columns:
self.processed_data['customer_lifetime_value'] = (
self.processed_data['tenure_months'] * self.processed_data['monthly_charges']
)
new_features_created += 1
# 客户分段
if 'tenure_months' in self.processed_data.columns:
self.processed_data['tenure_segment'] = pd.cut(
self.processed_data['tenure_months'],
bins=[0, 12, 24, 48, np.inf],
labels=['New', 'Regular', 'Loyal', 'VIP']
)
new_features_created += 1
# 费用比率特征
if 'monthly_charges' in self.processed_data.columns and 'total_charges' in self.processed_data.columns:
self.processed_data['charge_ratio'] = (
self.processed_data['monthly_charges'] / self.processed_data['total_charges']
).replace([np.inf, -np.inf], 0)
new_features_created += 1
print(f"✅ 创建了 {new_features_created} 个新特征")
self.results['feature_engineering'] = {
'new_features_created': new_features_created
}
def model_training(self):
"""步骤5: 模型训练"""
print("\n5. 🤖 模型训练")
print("-" * 40)
# 准备特征和目标变量
features_to_use = [
'tenure_months', 'monthly_charges', 'total_charges',
'contract_type', 'payment_method', 'paperless_billing',
'tech_support', 'online_security',
'customer_lifetime_value', 'charge_ratio'
]
X = self.processed_data[features_to_use]
y = self.processed_data['churn']
# 分割数据
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
print(f" 训练集: {X_train.shape[0]} 样本")
print(f" 测试集: {X_test.shape[0]} 样本")
# 创建预处理管道
numeric_features = ['tenure_months', 'monthly_charges', 'total_charges',
'customer_lifetime_value', 'charge_ratio']
categorical_features = ['contract_type', 'payment_method']
numeric_transformer = Pipeline(steps=[
('scaler', StandardScaler())
])
categorical_transformer = Pipeline(steps=[
('onehot', OneHotEncoder(handle_unknown='ignore', drop='first'))
])
preprocessor = ColumnTransformer(
transformers=[
('num', numeric_transformer, numeric_features),
('cat', categorical_transformer, categorical_features)
])
# 创建完整管道
self.pipeline = Pipeline(steps=[
('preprocessor', preprocessor),
('classifier', RandomForestClassifier(
n_estimators=100,
max_depth=10,
random_state=42,
class_weight='balanced'
))
])
# 训练模型
print(" 训练模型中...")
self.pipeline.fit(X_train, y_train)
# 在测试集上评估
y_pred = self.pipeline.predict(X_test)
y_pred_proba = self.pipeline.predict_proba(X_test)[:, 1]
test_accuracy = (y_pred == y_test).mean()
test_auc = roc_auc_score(y_test, y_pred_proba)
print(f"✅ 模型训练完成")
print(f" 测试集准确率: {test_accuracy:.4f}")
print(f" 测试集AUC: {test_auc:.4f}")
self.model = self.pipeline.named_steps['classifier']
self.results['model_training'] = {
'test_accuracy': test_accuracy,
'test_auc': test_auc,
'features_used': len(features_to_use)
}
def model_evaluation(self):
"""步骤6: 模型评估"""
print("\n6. 📊 模型评估")
print("-" * 40)
# 准备数据
features_to_use = [
'tenure_months', 'monthly_charges', 'total_charges',
'contract_type', 'payment_method', 'paperless_billing',
'tech_support', 'online_security',
'customer_lifetime_value', 'charge_ratio'
]
X = self.processed_data[features_to_use]
y = self.processed_data['churn']
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# 交叉验证
cv_scores = cross_val_score(self.pipeline, X_train, y_train, cv=5, scoring='roc_auc')
print(f"📈 交叉验证性能:")
print(f" AUC分数: {cv_scores.mean():.4f} (±{cv_scores.std() * 2:.4f})")
# 特征重要性
if hasattr(self.model, 'feature_importances_'):
feature_names = (
numeric_features +
list(self.pipeline.named_steps['preprocessor']
.named_transformers_['cat']
.named_steps['onehot']
.get_feature_names_out(categorical_features))
)
importances = self.model.feature_importances_
top_features = pd.DataFrame({
'feature': feature_names,
'importance': importances
}).sort_values('importance', ascending=False).head(5)
print("🔍 最重要的特征:")
for _, row in top_features.iterrows():
print(f" - {row['feature']}: {row['importance']:.4f}")
self.results['model_evaluation'] = {
'cv_auc_mean': cv_scores.mean(),
'cv_auc_std': cv_scores.std(),
'top_features': top_features.to_dict('records') if 'top_features' in locals() else []
}
def deployment_preparation(self):
"""步骤7: 部署准备"""
print("\n7. 🚀 部署准备")
print("-" * 40)
# 保存模型
model_filename = 'customer_churn_model.pkl'
joblib.dump(self.pipeline, model_filename)
# 保存元数据
metadata = {
'model_type': 'RandomForestClassifier',
'version': '1.0.0',
'created_date': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S'),
'performance': self.results['model_training'],
'features_used': [
'tenure_months', 'monthly_charges', 'total_charges',
'contract_type', 'payment_method', 'paperless_billing',
'tech_support', 'online_security',
'customer_lifetime_value', 'charge_ratio'
]
}
with open('model_metadata.json', 'w') as f:
json.dump(metadata, f, indent=2)
print(f"✅ 模型保存为: {model_filename}")
print(f"✅ 元数据保存为: model_metadata.json")
# 创建预测函数示例
self._create_prediction_example()
self.results['deployment'] = {
'model_saved': model_filename,
'metadata_saved': 'model_metadata.json'
}
def _create_prediction_example(self):
"""创建预测示例"""
print("\n🔮 预测示例:")
# 示例客户数据
example_customer = {
'tenure_months': 15,
'monthly_charges': 75.50,
'total_charges': 1132.50,
'contract_type': 'Monthly',
'payment_method': 'Electronic check',
'paperless_billing': 1,
'tech_support': 0,
'online_security': 0,
'customer_lifetime_value': 1132.50,
'charge_ratio': 0.0667
}
# 进行预测
try:
prediction = self.pipeline.predict(pd.DataFrame([example_customer]))[0]
probability = self.pipeline.predict_proba(pd.DataFrame([example_customer]))[0][1]
status = "高风险流失" if prediction == 1 else "低风险流失"
print(f" 示例客户预测:")
print(f" - 流失风险: {status}")
print(f" - 流失概率: {probability:.1%}")
print(f" - 建议措施: {'立即干预' if prediction == 1 else '常规维护'}")
except Exception as e:
print(f" 预测示例失败: {e}")
def _print_summary(self):
"""打印工作流总结"""
print("\n" + "=" * 60)
print("📋 工作流执行总结")
print("=" * 60)
for step, results in self.results.items():
print(f"\n{step.replace('_', ' ').title()}:")
for key, value in results.items():
if isinstance(value, float):
print(f" {key}: {value:.4f}")
else:
print(f" {key}: {value}")
print("\n🎯 下一步建议:")
print(" 1. 在真实数据上验证模型性能")
print(" 2. 设置模型监控和定期重训练")
print(" 3. 集成到业务系统中")
print(" 4. 建立A/B测试框架")
# 运行完整工作流
if __name__ == "__main__":
# 设置更好的可视化样式
plt.style.use('seaborn-v0_8')
sns.set_palette("husl")
# 创建工作流实例
workflow = CompleteMLWorkflow()
# 运行完整工作流
workflow.run_complete_workflow(sample_size=5000)
print("\n" + "=" * 60)
print("🎉 恭喜！你已经完成了从数据到部署的完整机器学习工作流！")
print("=" * 60)
8. 代码自查清单
8.1 完整项目质量检查
在部署机器学习项目之前，请进行全面的质量检查：
数据质量检查
- 数据收集完整，覆盖了所有必要的业务场景
- 数据清洗流程完善，处理了缺失值、异常值和重复值
- 特征工程创造了有业务意义的衍生特征
- 数据分割正确，训练集和测试集没有数据泄露
模型质量检查
- 选择了适合业务问题的算法和评估指标
- 进行了超参数优化，模型性能达到业务要求
- 交叉验证结果显示模型稳定性良好
- 模型可解释性满足业务需求
代码质量检查
- 代码结构清晰，模块化程度高
- 错误处理完善，有适当的异常捕获
- 日志记录详细，便于调试和监控
- 代码注释充分，特别是复杂业务逻辑
部署就绪检查
- 模型管道包含完整的数据预处理步骤
- API接口设计合理，输入输出格式规范
- 模型版本管理和元数据完整
- 监控和日志系统就绪
8.2 生产环境检查清单
class ProductionReadinessChecker:
"""生产环境就绪检查器"""
def __init__(self, workflow):
self.workflow = workflow
self.check_results = {}
def perform_production_checks(self):
"""执行生产环境就绪检查"""
print("🔍 执行生产环境就绪检查")
print("=" * 50)
checks = [
("模型性能检查", self.check_model_performance),
("代码质量检查", self.check_code_quality),
("数据管道检查", self.check_data_pipeline),
("API就绪检查", self.check_api_readiness),
("监控就绪检查", self.check_monitoring_readiness),
("安全合规检查", self.check_security_compliance)
]
all_passed = True
for check_name, check_function in checks:
print(f"\n📋 {check_name}:")
try:
result = check_function()
self.check_results[check_name] = result
if result['passed']:
print(" ✅ 通过")
else:
print(f" ❌ 未通过: {result.get('message', '')}")
all_passed = False
except Exception as e:
print(f" ⚠️ 检查失败: {e}")
all_passed = False
# 总体评估
print("\n" + "=" * 50)
if all_passed:
print("🎉 所有检查通过！项目已准备好部署到生产环境")
else:
print("🚨 存在未通过的检查项，请在部署前解决")
return all_passed
def check_model_performance(self):
"""检查模型性能"""
training_results = self.workflow.results.get('model_training', {})
evaluation_results = self.workflow.results.get('model_evaluation', {})
checks = {
'test_auc': training_results.get('test_auc', 0) > 0.7,
'cv_stability': evaluation_results.get('cv_auc_std', 1) < 0.05,
'feature_importance': len(evaluation_results.get('top_features', [])) > 0
}
passed = all(checks.values())
message = ""
if not checks['test_auc']:
message = "模型AUC分数低于0.7，建议优化"
elif not checks['cv_stability']:
message = "交叉验证稳定性不足，模型可能过拟合"
return {'passed': passed, 'message': message, 'details': checks}
def check_code_quality(self):
"""检查代码质量"""
# 这里可以集成代码质量工具如pylint、black等
checks = {
'module_structure': True, # 假设已检查
'error_handling': True,
'logging': True,
'documentation': True
}
passed = all(checks.values())
message = "建议使用自动化代码质量工具进行详细检查"
return {'passed': passed, 'message': message, 'details': checks}
def check_data_pipeline(self):
"""检查数据管道"""
checks = {
'preprocessing_in_pipeline': self.workflow.pipeline is not None,
'feature_consistency': True,
'data_validation': True
}
passed = all(checks.values())
message = "数据管道就绪"
return {'passed': passed, 'message': message, 'details': checks}
def check_api_readiness(self):
"""检查API就绪状态"""
checks = {
'model_serialization': True,
'api_endpoints': True,
'input_validation': True
}
passed = all(checks.values())
message = "API接口设计完成"
return {'passed': passed, 'message': message, 'details': checks}
def check_monitoring_readiness(self):
"""检查监控就绪状态"""
checks = {
'performance_metrics': True,
'data_drift_detection': False, # 需要实现
'model_decay_monitoring': False # 需要实现
}
passed = all(checks.values())
message = "基础监控就绪，建议实现数据漂移检测"
return {'passed': passed, 'message': message, 'details': checks}
def check_security_compliance(self):
"""检查安全合规"""
checks = {
'data_privacy': True,
'api_security': True,
'compliance_documentation': False # 需要补充
}
passed = all(checks.values())
message = "基础安全措施就绪，需要完善合规文档"
return {'passed': passed, 'message': message, 'details': checks}
# 运行生产就绪检查
if __name__ == "__main__":
# 假设已经有完整的工作流实例
workflow = CompleteMLWorkflow()
workflow.run_complete_workflow(sample_size=1000) # 用小样本快速测试
# 执行生产就绪检查
checker = ProductionReadinessChecker(workflow)
production_ready = checker.perform_production_checks()
9. 总结与最佳实践
9.1 关键成功因素
通过这个完整的机器学习工作流，我们展示了从数据到部署的全过程。关键的成功因素包括：
- 业务理解优先：在开始任何技术工作前，深入理解业务问题和目标
- 迭代开发：采用敏捷方法，快速原型、测试和迭代
- 自动化管道：建立可重复的数据处理和模型训练管道
- 全面监控：监控模型性能、数据质量和业务影响
- 团队协作：数据科学家、工程师和业务人员的紧密合作
9.2 持续改进建议
机器学习项目不是一次性的工作，而是需要持续改进的过程：
9.3 推荐工具和框架
- 数据处理：Pandas, NumPy, Dask
- 机器学习：Scikit-learn, XGBoost, LightGBM
- 深度学习：TensorFlow, PyTorch
- 部署服务：Flask, FastAPI, Docker
- 工作流管理：MLflow, Kubeflow, Airflow
- 监控告警：Prometheus, Grafana, Evidently
9.4 结语
完整的机器学习工作流是将数据科学转化为业务价值的关键。通过标准化的工作流程，组织能够：
- 提高项目成功率：减少85%的项目失败风险
- 加速价值实现：将模型部署时间从数月缩短到数周
- 降低维护成本：通过自动化管道减少人工干预
- 增强业务信任：通过透明和可解释的流程建立信任
记住，优秀的机器学习项目不仅仅是技术实现，更是技术、业务和流程的完美结合。开始你的第一个完整机器学习项目，体验从数据到价值的完整旅程！
下一步行动：选择一个真实的业务问题，应用这个完整工作流，开始你的机器学习项目实践！
码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践
更多推荐
8
0- 0
已为社区贡献3条内容
所有评论(0)