# 问题定义示例：电商推荐系统classProblemDefinition:def__init__(self):# 业务问题self.business_problem="用户购买转化率低，需要提高推荐精准度"# 技术问题self.technical_problem="基于用户行为预测用户可能购买的商品"# 问题类型self.problem_type="推荐系统（分类+排序）"# 成功标准self.success_criteria={"点击率提升":"15%","转化率提升":"10%","推荐准确率":"80%"}# 约束条件self.constraints={"响应时间":"< 100ms","数据隐私":"符合 GDPR 要求","计算资源":"现有服务器配置"}defdefine_features_and_labels(self):"""定义特征和标签"""features={"用户特征":["年龄","性别","购买历史","浏览行为"],"商品特征":["类别","价格","评分","库存"],"上下文特征":["时间","设备","地理位置"]}labels={"主要标签":"是否点击","次要标签":"是否购买","辅助标签":"停留时间"}returnfeatures,labelsdefprint_definition(self):"""打印问题定义"""print("="*50)print("机器学习问题定义")print("="*50)print(f"业务问题：{self.business_problem}")print(f"技术问题：{self.technical_problem}")print(f"问题类型：{self.problem_type}")print("\n成功标准：")formetric,targetinself.success_criteria.items():print(f"  {metric}：{target}")print("\n约束条件：")forconstraint,limitinself.constraints.items():print(f"  {constraint}：{limit}")features,labels=self.define_features_and_labels()print("\n特征定义：")forcategory,itemsinfeatures.items():print(f"  {category}：{', '.join(items)}")print("\n标签定义：")forlabel_type,label_nameinlabels.items():print(f"  {label_type}：{label_name}")# 使用示例problem=ProblemDefinition()problem.print_definition()

# 数据收集示例：模拟多种数据源importpandasaspdimportnumpyasnpfromdatetimeimportdatetime,timedeltaclassDataCollector:def__init__(self):self.collected_data={}defcollect_user_data(self,n_users=1000):"""收集用户数据"""np.random.seed(42)user_data={'user_id':range(1,n_users +1),'age': np.random.randint(18,65,n_users),'gender': np.random.choice(['男','女'],n_users),'city': np.random.choice(['北京','上海','广州','深圳'],n_users),'registration_date':[datetime.now()- timedelta(days=np.random.randint(1,365))for_inrange(n_users)]}self.collected_data['users']=pd.DataFrame(user_data)print(f"收集了 {len(user_data['user_id'])} 条用户数据")returnself.collected_data['users']defcollect_behavior_data(self,n_behaviors=5000):"""收集用户行为数据"""np.random.seed(42)user_ids=np.random.choice(range(1,1001),n_behaviors)product_ids=np.random.choice(range(1,501),n_behaviors)behavior_data={'behavior_id':range(1,n_behaviors +1),'user_id': user_ids,'product_id': product_ids,'behavior_type': np.random.choice(['浏览','点击','加购物车','购买'],n_behaviors,p=[0.4,0.3,0.2,0.1]),'timestamp':[datetime.now()- timedelta(minutes=np.random.randint(1,10080))for_inrange(n_behaviors)],'duration': np.random.exponential(30,n_behaviors)# 停留时间（秒）}self.collected_data['behaviors']=pd.DataFrame(behavior_data)print(f"收集了 {len(behavior_data['behavior_id'])} 条行为数据")returnself.collected_data['behaviors']defcollect_product_data(self,n_products=500):"""收集商品数据"""np.random.seed(42)categories=['电子产品','服装','食品','家居','图书']product_data={'product_id':range(1,n_products +1),'category': np.random.choice(categories,n_products),'price': np.random.uniform(10,1000,n_products),'rating': np.random.uniform(3.0,5.0,n_products),'stock': np.random.randint(0,1000,n_products)}self.collected_data['products']=pd.DataFrame(product_data)print(f"收集了 {len(product_data['product_id'])} 条商品数据")returnself.collected_data['products']defget_data_summary(self):"""获取数据摘要"""print("\n数据收集摘要：")forname,dfinself.collected_data.items():print(f"\n{name} 数据集：")print(f"  形状：{df.shape}")print(f"  列名：{list(df.columns)}")print(f"  缺失值：{df.isnull().sum().sum()}")print(f"  示例数据：")print(df.head(2))# 使用示例collector=DataCollector()collector.collect_user_data()collector.collect_behavior_data()collector.collect_product_data()collector.get_data_summary()

# 数据准备示例importpandasaspdimportnumpyasnpfromsklearn.preprocessingimportStandardScaler,LabelEncoderfromsklearn.model_selectionimporttrain_test_splitclassDataPreparer:def__init__(self,data):self.data=data.copy()self.processed_data=Nonedefclean_data(self):"""数据清洗"""print("开始数据清洗...")# 1. 处理缺失值print(f"处理前缺失值数量：{self.data.isnull().sum().sum()}")# 数值列用均值填充numeric_columns=self.data.select_dtypes(include=[np.number]).columnsforcolinnumeric_columns:ifself.data[col].isnull().sum()>0:self.data[col].fillna(self.data[col].mean(),inplace=True)# 类别列用众数填充categorical_columns=self.data.select_dtypes(include=['object']).columnsforcolincategorical_columns:ifself.data[col].isnull().sum()>0:mode_val=self.data[col].mode()[0]self.data[col].fillna(mode_val,inplace=True)print(f"处理后缺失值数量：{self.data.isnull().sum().sum()}")# 2. 处理重复值duplicates_before=self.data.duplicated().sum()self.data.drop_duplicates(inplace=True)duplicates_after=self.data.duplicated().sum()print(f"删除重复值：{duplicates_before - duplicates_after} 条")# 3. 处理异常值（简单方法：使用 IQR）forcolinnumeric_columns:Q1=self.data[col].quantile(0.25)Q3=self.data[col].quantile(0.75)IQR=Q3 - Q1lower_bound=Q1 -1.5* IQRupper_bound=Q3 +1.5* IQRoutliers=((self.data[col]<lower_bound)|(self.data[col]>upper_bound)).sum()ifoutliers>0:# 用边界值替换异常值self.data[col]=self.data[col].clip(lower_bound,upper_bound)print(f"处理 {col} 列的 {outliers} 个异常值")returnself.datadeffeature_engineering(self):"""特征工程"""print("\n开始特征工程...")# 1. 创建新特征（示例）if'price'inself.data.columnsand'rating'inself.data.columns:# 创建性价比特征self.data['price_per_rating']=self.data['price']/self.data['rating']print("创建新特征：price_per_rating")# 2. 特征选择（简单示例：移除低方差特征）numeric_columns=self.data.select_dtypes(include=[np.number]).columnslow_variance_features=[]forcolinnumeric_columns:ifself.data[col].var()<0.01:# 方差阈值low_variance_features.append(col)iflow_variance_features:self.data.drop(columns=low_variance_features,inplace=True)print(f"移除低方差特征：{low_variance_features}")returnself.datadeftransform_data(self):"""数据转换"""print("\n开始数据转换...")# 1. 编码类别变量categorical_columns=self.data.select_dtypes(include=['object']).columnslabel_encoders={}forcolincategorical_columns:le=LabelEncoder()self.data[col]=le.fit_transform(self.data[col])label_encoders[col]=leprint(f"编码类别变量：{col}")# 2. 标准化数值变量numeric_columns=self.data.select_dtypes(include=[np.number]).columnsscaler=StandardScaler()iflen(numeric_columns)>0:self.data[numeric_columns]=scaler.fit_transform(self.data[numeric_columns])print(f"标准化数值变量：{list(numeric_columns)}")returnself.data,label_encoders,scalerdefsplit_data(self,target_column,test_size=0.2,val_size=0.2):"""数据划分"""print(f"\n开始数据划分（测试集比例：{test_size}，验证集比例：{val_size}）...")X=self.data.drop(columns=[target_column])y=self.data[target_column]# 首先分离出测试集X_temp,X_test,y_temp,y_test=train_test_split(X,y,test_size=test_size,random_state=42)# 再从剩余数据中分离出验证集val_size_adjusted=val_size /(1- test_size)X_train,X_val,y_train,y_val=train_test_split(X_temp,y_temp,test_size=val_size_adjusted,random_state=42)print(f"训练集大小：{X_train.shape[0]}")print(f"验证集大小：{X_val.shape[0]}")print(f"测试集大小：{X_test.shape[0]}")return{'X_train': X_train,'y_train': y_train,'X_val': X_val,'y_val': y_val,'X_test': X_test,'y_test': y_test}defprepare_pipeline(self,target_column):"""完整的数据准备流水线"""print("="*50)print("数据准备流水线")print("="*50)# 1. 数据清洗self.clean_data()# 2. 特征工程self.feature_engineering()# 3. 数据转换processed_data,encoders,scaler=self.transform_data()# 4. 数据划分splits=self.split_data(target_column)self.processed_data=processed_datareturnsplits,encoders,scaler# 创建示例数据并演示数据准备np.random.seed(42)sample_data=pd.DataFrame({'age': np.random.randint(18,65,1000),'income': np.random.normal(50000,15000,1000),'gender': np.random.choice(['男','女'],1000),'city': np.random.choice(['北京','上海','广州'],1000),'target': np.random.choice([0,1],1000)})# 添加一些缺失值和异常值sample_data.loc[np.random.choice(1000,50),'income']=np.nansample_data.loc[np.random.choice(1000,20),'age']=np.random.randint(100,150)preparer=DataPreparer(sample_data)splits,encoders,scaler=preparer.prepare_pipeline('target')

# 模型训练示例fromsklearn.linear_modelimportLogisticRegression,LinearRegressionfromsklearn.ensembleimportRandomForestClassifier,RandomForestRegressorfromsklearn.svmimportSVC,SVRfromsklearn.metricsimportaccuracy_score,mean_squared_error,classification_reportclassModelTrainer:def__init__(self):self.models={}self.trained_models={}defregister_model(self,name,model,problem_type):"""注册模型"""self.models[name]={'model': model,'problem_type': problem_type}print(f"注册模型：{name}（{problem_type}）")deftrain_single_model(self,name,X_train,y_train):"""训练单个模型"""ifnamenotinself.models:raiseValueError(f"模型 {name} 未注册")model_info=self.models[name]model=model_info['model']print(f"\n训练模型：{name}")model.fit(X_train,y_train)self.trained_models[name]=modelprint(f"模型 {name} 训练完成")returnmodeldeftrain_all_models(self,X_train,y_train):"""训练所有注册的模型"""print("\n开始训练所有模型...")fornameinself.models.keys():try:self.train_single_model(name,X_train,y_train)exceptExceptionase:print(f"训练模型 {name} 时出错：{e}")returnself.trained_modelsdefevaluate_models(self,X_test,y_test):"""评估所有训练好的模型"""print("\n模型评估结果：")print("-"*50)results={}forname,modelinself.trained_models.items():problem_type=self.models[name]['problem_type']# 预测y_pred=model.predict(X_test)# 根据问题类型选择评估指标ifproblem_type=='classification':accuracy=accuracy_score(y_test,y_pred)results[name]={'accuracy': accuracy}print(f"{name}: 准确率 = {accuracy:.4f}")# 详细报告print(classification_report(y_test,y_pred))elifproblem_type=='regression':mse=mean_squared_error(y_test,y_pred)rmse=np.sqrt(mse)results[name]={'mse': mse,'rmse': rmse}print(f"{name}: MSE = {mse:.4f}, RMSE = {rmse:.4f}")print("-"*50)returnresultsdefget_best_model(self,results,metric='accuracy'):"""获取最佳模型"""ifnotresults:returnNonebest_model_name=max(results.keys(),key=lambdax: results[x].get(metric,0))best_score=results[best_model_name][metric]print(f"\n最佳模型：{best_model_name}（{metric} = {best_score:.4f}）")returnbest_model_name,self.trained_models[best_model_name]# 使用示例trainer=ModelTrainer()# 注册不同类型的模型trainer.register_model('逻辑回归',LogisticRegression(random_state=42),'classification')trainer.register_model('随机森林',RandomForestClassifier(n_estimators=100,random_state=42),'classification')trainer.register_model('支持向量机',SVC(random_state=42),'classification')# 创建训练数据X_train=splits['X_train']y_train=splits['y_train']X_test=splits['X_test']y_test=splits['y_test']# 训练所有模型trained_models=trainer.train_all_models(X_train,y_train)# 评估模型results=trainer.evaluate_models(X_test,y_test)# 获取最佳模型best_name,best_model=trainer.get_best_model(results)

# 模型评估示例importmatplotlib.pyplotaspltfromsklearn.metricsimport(accuracy_score,precision_score,recall_score,f1_score,confusion_matrix,roc_curve,auc)classModelEvaluator:def__init__(self):self.evaluation_results={}defevaluate_classification(self,y_true,y_pred,y_prob=None,model_name="Model"):"""评估分类模型"""results={}# 基本指标results['accuracy']=accuracy_score(y_true,y_pred)results['precision']=precision_score(y_true,y_pred,average='weighted')results['recall']=recall_score(y_true,y_pred,average='weighted')results['f1']=f1_score(y_true,y_pred,average='weighted')print(f"\n{model_name} 分类评估结果：")print(f"准确率：{results['accuracy']:.4f}")print(f"精确率：{results['precision']:.4f}")print(f"召回率：{results['recall']:.4f}")print(f"F1 分数：{results['f1']:.4f}")# 混淆矩阵cm=confusion_matrix(y_true,y_pred)print(f"\n混淆矩阵：")print(cm)# ROC 曲线（如果有概率预测）ify_probisnotNoneandlen(np.unique(y_true))==2:fpr,tpr,thresholds=roc_curve(y_true,y_prob[:,1])roc_auc=auc(fpr,tpr)results['roc_auc']=roc_auc# 绘制 ROC 曲线plt.figure(figsize=(8,6))plt.plot(fpr,tpr,color='darkorange',lw=2,label=f'ROC 曲线 (AUC = {roc_auc:.2f})')plt.plot([0,1],[0,1],color='navy',lw=2,linestyle='--')plt.xlim([0.0,1.0])plt.ylim([0.0,1.05])plt.xlabel('假正率')plt.ylabel('真正率')plt.title(f'{model_name} ROC 曲线')plt.legend(loc="lower right")plt.grid(True)plt.show()self.evaluation_results[model_name]=resultsreturnresultsdefevaluate_regression(self,y_true,y_pred,model_name="Model"):"""评估回归模型"""results={}# 基本指标mse=np.mean((y_true - y_pred)**2)rmse=np.sqrt(mse)mae=np.mean(np.abs(y_true - y_pred))# R² 分数ss_res=np.sum((y_true - y_pred)**2)ss_tot=np.sum((y_true - np.mean(y_true))**2)r2=1-(ss_res / ss_tot)results['mse']=mseresults['rmse']=rmseresults['mae']=maeresults['r2']=r2print(f"\n{model_name} 回归评估结果：")print(f"均方误差 (MSE)：{mse:.4f}")print(f"均方根误差 (RMSE)：{rmse:.4f}")print(f"平均绝对误差 (MAE)：{mae:.4f}")print(f"R² 分数：{r2:.4f}")# 绘制预测 vs 真实值plt.figure(figsize=(8,6))plt.scatter(y_true,y_pred,alpha=0.6)plt.plot([y_true.min(),y_true.max()],[y_true.min(),y_true.max()],'r--',lw=2)plt.xlabel('真实值')plt.ylabel('预测值')plt.title(f'{model_name} 预测 vs 真实值')plt.grid(True)plt.show()self.evaluation_results[model_name]=resultsreturnresultsdefcompare_models(self):"""比较所有评估过的模型"""ifnotself.evaluation_results:print("没有可比较的模型评估结果")returnprint("\n模型比较：")print("-"*50)# 创建比较表格comparison_data=[]formodel_name,resultsinself.evaluation_results.items():row=[model_name]formetric,valueinresults.items():row.append(f"{value:.4f}")comparison_data.append(row)# 打印表格headers=["模型名称"]+list(self.evaluation_results.values())[0].keys()print("\t".join(headers))forrowincomparison_data:print("\t".join(row))# 使用示例evaluator=ModelEvaluator()# 评估分类模型y_pred_class=best_model.predict(X_test)y_prob_class=best_model.predict_proba(X_test)evaluator.evaluate_classification(y_test,y_pred_class,y_prob_class,"最佳分类模型")# 比较所有模型evaluator.compare_models()

# 模型部署示例importpickleimportjsonfromdatetimeimportdatetimeclassModelDeployer:def__init__(self):self.deployed_models={}self.deployment_logs=[]defsave_model(self,model,model_name,filepath=None):"""保存模型"""iffilepathisNone:filepath=f"{model_name}.pkl"withopen(filepath,'wb')asf:pickle.dump(model,f)print(f"模型 {model_name} 已保存到 {filepath}")# 记录部署日志log_entry={'timestamp':datetime.now().isoformat(),'action':'save_model','model_name': model_name,'filepath': filepath}self.deployment_logs.append(log_entry)returnfilepathdefload_model(self,model_name,filepath):"""加载模型"""withopen(filepath,'rb')asf:model=pickle.load(f)self.deployed_models[model_name]=modelprint(f"模型 {model_name} 已从 {filepath} 加载")# 记录部署日志log_entry={'timestamp':datetime.now().isoformat(),'action':'load_model','model_name': model_name,'filepath': filepath}self.deployment_logs.append(log_entry)returnmodeldefcreate_prediction_service(self,model_name,encoders=None,scaler=None):"""创建预测服务"""ifmodel_namenotinself.deployed_models:raiseValueError(f"模型 {model_name} 未部署")model=self.deployed_models[model_name]defpredict_service(input_data):"""预测服务函数"""try:# 数据预处理ifencoders:forcol,encoderinencoders.items():ifcolininput_data.columns:input_data[col]=encoder.transform(input_data[col])ifscaler:numeric_cols=input_data.select_dtypes(include=['number']).columnsinput_data[numeric_cols]=scaler.transform(input_data[numeric_cols])# 预测prediction=model.predict(input_data)# 如果是分类模型，也返回概率ifhasattr(model,'predict_proba'):probability=model.predict_proba(input_data)return{'prediction': prediction.tolist(),'probability': probability.tolist(),'status':'success','timestamp':datetime.now().isoformat()}else:return{'prediction': prediction.tolist(),'status':'success','timestamp':datetime.now().isoformat()}exceptExceptionase:return{'error':str(e),'status':'error','timestamp':datetime.now().isoformat()}# 记录服务创建日志log_entry={'timestamp':datetime.now().isoformat(),'action':'create_service','model_name': model_name}self.deployment_logs.append(log_entry)returnpredict_servicedefmonitor_model(self,model_name,input_data,true_labels=None):"""监控模型性能"""ifmodel_namenotinself.deployed_models:raiseValueError(f"模型 {model_name} 未部署")predict_service=self.create_prediction_service(model_name)# 获取预测结果result=predict_service(input_data)# 监控信息monitoring_info={'timestamp':datetime.now().isoformat(),'model_name': model_name,'input_shape': input_data.shape,'prediction_count':len(result.get('prediction',[])),'status': result.get('status','unknown')}# 如果有真实标签，计算性能指标iftrue_labelsisnotNoneand'prediction'inresult:predictions=result['prediction']iflen(predictions)==len(true_labels):accuracy=accuracy_score(true_labels,predictions)monitoring_info['accuracy']=accuracyprint("模型监控信息：")forkey,valueinmonitoring_info.items():print(f"  {key}: {value}")returnmonitoring_infodefget_deployment_logs(self):"""获取部署日志"""returnself.deployment_logs# 使用示例deployer=ModelDeployer()# 保存最佳模型model_path=deployer.save_model(best_model,"best_classification_model")# 加载模型deployer.load_model("best_classification_model",model_path)# 创建预测服务prediction_service=deployer.create_prediction_service("best_classification_model",encoders,scaler)# 使用预测服务test_input=X_test.head(5)prediction_result=prediction_service(test_input)print("\n预测结果：")print(json.dumps(prediction_result,indent=2,ensure_ascii=False))# 监控模型deployer.monitor_model("best_classification_model",test_input,y_test.head(5).values)

# 完整的机器学习流程示例classMLProjectPipeline:def__init__(self):self.data_collector=DataCollector()self.data_preparer=Noneself.model_trainer=ModelTrainer()self.model_evaluator=ModelEvaluator()self.model_deployer=ModelDeployer()defrun_complete_pipeline(self,target_column):"""运行完整的机器学习流水线"""print("="*60)print("机器学习项目完整流程")print("="*60)# 1. 数据收集print("\n第1步：数据收集")print("-"*30)user_data=self.data_collector.collect_user_data(1000)behavior_data=self.data_collector.collect_behavior_data(5000)# 合并数据（简化示例）merged_data=pd.merge(user_data,behavior_data,on='user_id',how='inner')# 创建目标变量（示例：是否购买）merged_data['purchased']=(merged_data['behavior_type']=='购买').astype(int)# 2. 数据准备print("\n第2步：数据准备")print("-"*30)# 选择特征列feature_columns=['age','gender','city','duration']ifall(colinmerged_data.columnsforcolinfeature_columns):data_for_ml=merged_data[feature_columns +['purchased']].copy()# 处理类别变量data_for_ml['gender']=data_for_ml['gender'].map({'男':0,'女':1})data_for_ml['city']=data_for_ml['city'].map({'北京':0,'上海':1,'广州':2})# 数据准备self.data_preparer=DataPreparer(data_for_ml)splits,encoders,scaler=self.data_preparer.prepare_pipeline('purchased')# 3. 模型训练print("\n第3步：模型训练")print("-"*30)# 注册模型self.model_trainer.register_model('逻辑回归',LogisticRegression(random_state=42),'classification')self.model_trainer.register_model('随机森林',RandomForestClassifier(n_estimators=100,random_state=42),'classification')# 训练模型trained_models=self.model_trainer.train_all_models(splits['X_train'],splits['y_train'])# 4. 模型评估print("\n第4步：模型评估")print("-"*30)results=self.model_trainer.evaluate_models(splits['X_test'],splits['y_test'])best_name,best_model=self.model_trainer.get_best_model(results)# 5. 模型部署print("\n第5步：模型部署")print("-"*30)# 保存模型model_path=self.model_deployer.save_model(best_model,"production_model")# 创建预测服务prediction_service=self.model_deployer.create_prediction_service("production_model")# 测试预测服务test_input=splits['X_test'].head(3)prediction_result=prediction_service(test_input)print("\n预测服务测试结果：")print(json.dumps(prediction_result,indent=2,ensure_ascii=False))print("\n"+"="*60)print("机器学习项目流程完成！")print("="*60)return{'data': data_for_ml,'splits': splits,'best_model': best_model,'best_model_name': best_name,'evaluation_results': results,'prediction_service': prediction_service}else:print("数据列不完整，无法继续流程")returnNone# 运行完整流程pipeline=MLProjectPipeline()project_results=pipeline.run_complete_pipeline('purchased')