Python房屋价格预测实战：数据预处理与模型构建全流程

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
 
# TRAE IDE智能提示：自动导入缺失的依赖包
class DataQualityAnalyzer:
    def __init__(self, df):
        self.df = df
        self.quality_report = {}
    
    def comprehensive_analysis(self):
        """全面数据质量分析"""
        print("=== 数据质量评估报告 ===")
        
        # 基础统计信息
        print(f"数据集维度: {self.df.shape}")
        print(f"内存占用: {self.df.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
        
        # 缺失值分析
        missing_analysis = self._analyze_missing_values()
        self.quality_report['missing'] = missing_analysis
        
        # 异常值检测
        outlier_analysis = self._detect_outliers_iqr()
        self.quality_report['outliers'] = outlier_analysis
        
        # 数据类型优化
        dtype_optimization = self._optimize_dtypes()
        self.quality_report['dtypes'] = dtype_optimization
        
        return self.quality_report
    
    def _analyze_missing_values(self):
        """智能缺失值分析"""
        missing_info = pd.DataFrame({
            'column': self.df.columns,
            'missing_count': self.df.isnull().sum(),
            'missing_percentage': (self.df.isnull().sum() / len(self.df)) * 100,
            'data_type': self.df.dtypes
        })
        
        # 识别缺失模式
        missing_patterns = self.df.isnull().sum().sort_values(ascending=False)
        high_missing = missing_patterns[missing_patterns > len(self.df) * 0.3]
        
        print(f"高缺失值特征 (>30%): {list(high_missing.index)}")
        return missing_info
    
    def _detect_outliers_iqr(self, threshold=1.5):
        """基于IQR的异常值检测"""
        numeric_cols = self.df.select_dtypes(include=[np.number]).columns
        outlier_summary = {}
        
        for col in numeric_cols:
            Q1 = self.df[col].quantile(0.25)
            Q3 = self.df[col].quantile(0.75)
            IQR = Q3 - Q1
            
            lower_bound = Q1 - threshold * IQR
            upper_bound = Q3 + threshold * IQR
            
            outliers = self.df[(self.df[col] < lower_bound) | (self.df[col] > upper_bound)]
            outlier_summary[col] = {
                'count': len(outliers),
                'percentage': (len(outliers) / len(self.df)) * 100,
                'bounds': (lower_bound, upper_bound)
            }
        
        return outlier_summary
    
    def _optimize_dtypes(self):
        """数据类型优化以节省内存"""
        memory_before = self.df.memory_usage(deep=True).sum() / 1024**2
        
        # 整数类型优化
        int_cols = self.df.select_dtypes(include=['int64']).columns
        for col in int_cols:
            col_min = self.df[col].min()
            col_max = self.df[col].max()
            
            if col_min >= 0:  # 无符号整数
                if col_max < 255:
                    self.df[col] = self.df[col].astype(np.uint8)
                elif col_max < 65535:
                    self.df[col] = self.df[col].astype(np.uint16)
                elif col_max < 4294967295:
                    self.df[col] = self.df[col].astype(np.uint32)
            else:  # 有符号整数
                if col_min > np.iinfo(np.int8).min and col_max < np.iinfo(np.int8).max:
                    self.df[col] = self.df[col].astype(np.int8)
                elif col_min > np.iinfo(np.int16).min and col_max < np.iinfo(np.int16).max:
                    self.df[col] = self.df[col].astype(np.int16)
                elif col_min > np.iinfo(np.int32).min and col_max < np.iinfo(np.int32).max:
                    self.df[col] = self.df[col].astype(np.int32)
        
        # 浮点数类型优化
        float_cols = self.df.select_dtypes(include=['float64']).columns
        for col in float_cols:
            self.df[col] = self.df[col].astype(np.float32)
        
        memory_after = self.df.memory_usage(deep=True).sum() / 1024**2
        memory_saved = memory_before - memory_after
        
        return {
            'memory_before_mb': memory_before,
            'memory_after_mb': memory_after,
            'memory_saved_mb': memory_saved,
            'saving_percentage': (memory_saved / memory_before) * 100
        }
 
# 使用示例
# df = pd.read_csv('house_prices.csv')
# analyzer = DataQualityAnalyzer(df)
# quality_report = analyzer.comprehensive_analysis()

from sklearn.preprocessing import StandardScaler, RobustScaler, QuantileTransformer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
import category_encoders as ce
 
class FeatureEngineering:
    """高级特征工程处理器"""
    
    def __init__(self, target_col='SalePrice'):
        self.target_col = target_col
        self.feature_importance = {}
        self.transformation_pipelines = {}
    
    def create_mathematical_features(self, df):
        """数学变换特征构造"""
        df_fe = df.copy()
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        numeric_cols = [col for col in numeric_cols if col != self.target_col]
        
        # 多项式特征
        for col in ['OverallQual', 'GrLivArea', 'TotalBsmtSF']:
            if col in numeric_cols:
                df_fe[f'{col}_squared'] = df_fe[col] ** 2
                df_fe[f'{col}_log'] = np.log1p(df_fe[col])
        
        # 交互特征
        if 'OverallQual' in numeric_cols and 'GrLivArea' in numeric_cols:
            df_fe['Qual_Area_Interaction'] = df_fe['OverallQual'] * df_fe['GrLivArea']
        
        if 'TotalBsmtSF' in numeric_cols and '1stFlrSF' in numeric_cols:
            df_fe['TotalSF'] = df_fe['TotalBsmtSF'] + df_fe['1stFlrSF'] + df_fe.get('2ndFlrSF', 0)
        
        # 比率特征
        if 'GrLivArea' in numeric_cols and 'TotRmsAbvGrd' in numeric_cols:
            df_fe['Area_per_Room'] = df_fe['GrLivArea'] / (df_fe['TotRmsAbvGrd'] + 1)
        
        return df_fe
    
    def create_temporal_features(self, df):
        """时间相关特征构造"""
        df_fe = df.copy()
        
        if 'YearBuilt' in df.columns and 'YearRemodAdd' in df.columns:
            df_fe['House_Age'] = df_fe['YrSold'] - df_fe['YearBuilt']
            df_fe['Remod_Age'] = df_fe['YrSold'] - df_fe['YearRemodAdd']
            df_fe['Years_Since_Remod'] = df_fe['YearRemodAdd'] - df_fe['YearBuilt']
        
        if 'GarageYrBlt' in df.columns:
            df_fe['Garage_Age'] = df_fe['YrSold'] - df_fe['GarageYrBlt']
        
        return df_fe
    
    def encode_categorical_features(self, df, target=None):
        """高级分类特征编码"""
        df_fe = df.copy()
        categorical_cols = df.select_dtypes(include=['object']).columns
        
        # 目标编码（需要目标变量）
        if target is not None and self.target_col in df.columns:
            target_encoder = ce.TargetEncoder(cols=categorical_cols, smoothing=20)
            df_encoded = target_encoder.fit_transform(df_fe[categorical_cols], df[self.target_col])
            
            # 重命名编码后的列
            for col in categorical_cols:
                df_fe[f'{col}_target_encoded'] = df_encoded[col]
        
        # 频率编码
        for col in categorical_cols:
            freq_map = df_fe[col].value_counts().to_dict()
            df_fe[f'{col}_frequency'] = df_fe[col].map(freq_map)
        
        # 独热编码（针对低基数分类变量）
        low_cardinality_cols = [col for col in categorical_cols 
                               if df_fe[col].nunique() <= 10]
        
        if low_cardinality_cols:
            df_fe = pd.get_dummies(df_fe, columns=low_cardinality_cols, drop_first=True)
        
        return df_fe
    
    def create_domain_specific_features(self, df):
        """领域特定特征构造"""
        df_fe = df.copy()
        
        # 房屋质量相关
        if 'OverallQual' in df.columns and 'OverallCond' in df.columns:
            df_fe['OverallScore'] = df_fe['OverallQual'] * df_fe['OverallCond']
        
        # 地下室相关
        basement_cols = ['TotalBsmtSF', 'BsmtFullBath', 'BsmtHalfBath']
        if all(col in df.columns for col in basement_cols):
            df_fe['BsmtBath'] = df_fe['BsmtFullBath'] + 0.5 * df_fe['BsmtHalfBath']
            df_fe['BsmtSF_per_Bath'] = df_fe['TotalBsmtSF'] / (df_fe['BsmtBath'] + 1)
        
        # 车库相关
        garage_cols = ['GarageCars', 'GarageArea']
        if all(col in df.columns for col in garage_cols):
            df_fe['GarageArea_per_Car'] = df_fe['GarageArea'] / (df_fe['GarageCars'] + 1)
        
        # 门廊相关
        porch_cols = ['OpenPorchSF', 'EnclosedPorch', '3SsnPorch', 'ScreenPorch']
        existing_porches = [col for col in porch_cols if col in df.columns]
        if existing_porches:
            df_fe['TotalPorchSF'] = df_fe[existing_porches].sum(axis=1)
        
        return df_fe
    
    def feature_selection_rfe(self, X, y, n_features_to_select=50):
        """递归特征消除特征选择"""
        from sklearn.feature_selection import RFE
        from sklearn.ensemble import RandomForestRegressor
        
        estimator = RandomForestRegressor(n_estimators=100, random_state=42)
        selector = RFE(estimator, n_features_to_select=n_features_to_select, step=10)
        
        X_selected = selector.fit_transform(X, y)
        selected_features = X.columns[selector.support_]
        
        self.feature_importance['rfe_ranking'] = dict(zip(X.columns, selector.ranking_))
        
        return pd.DataFrame(X_selected, columns=selected_features, index=X.index)

from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import Ridge, Lasso, ElasticNet
from sklearn.svm import SVR
from xgboost import XGBRegressor
from lightgbm import LGBMRegressor
from sklearn.model_selection import cross_val_score, KFold
import optuna
 
class AdvancedModeling:
    """高级模型构建器"""
    
    def __init__(self, random_state=42):
        self.random_state = random_state
        self.models = {}
        self.model_performance = {}
        self.best_model = None
        self.stacking_model = None
    
    def define_base_models(self):
        """定义基础模型集合"""
        self.models = {
            'ridge': Ridge(alpha=10, random_state=self.random_state),
            'lasso': Lasso(alpha=0.001, random_state=self.random_state),
            'elastic_net': ElasticNet(alpha=0.001, l1_ratio=0.5, random_state=self.random_state),
            'svr': SVR(kernel='rbf', C=100, gamma=0.1),
            'rf': RandomForestRegressor(
                n_estimators=100,
                max_depth=20,
                min_samples_split=2,
                min_samples_leaf=1,
                random_state=self.random_state
            ),
            'gb': GradientBoostingRegressor(
                n_estimators=100,
                learning_rate=0.1,
                max_depth=3,
                random_state=self.random_state
            ),
            'xgb': XGBRegressor(
                n_estimators=100,
                learning_rate=0.1,
                max_depth=5,
                random_state=self.random_state
            ),
            'lgb': LGBMRegressor(
                n_estimators=100,
                learning_rate=0.1,
                max_depth=5,
                random_state=self.random_state
            )
        }
        
        return self.models
    
    def evaluate_models(self, X, y, cv_folds=5):
        """评估所有基础模型"""
        kfold = KFold(n_splits=cv_folds, shuffle=True, random_state=self.random_state)
        
        for name, model in self.models.items():
            print(f"正在评估模型: {name}")
            
            # 计算交叉验证分数
            scores = cross_val_score(model, X, y, cv=kfold, scoring='neg_mean_squared_error')
            rmse_scores = np.sqrt(-scores)
            
            self.model_performance[name] = {
                'mean_rmse': rmse_scores.mean(),
                'std_rmse': rmse_scores.std(),
                'rmse_scores': rmse_scores
            }
            
            print(f"{name} - RMSE: {rmse_scores.mean():.4f} (+/- {rmse_scores.std():.4f})")
        
        return self.model_performance
    
    def hyperparameter_optimization(self, X, y, model_name, n_trials=100):
        """使用Optuna进行超参数优化"""
        
        def objective(trial):
            if model_name == 'xgb':
                params = {
                    'n_estimators': trial.suggest_int('n_estimators', 50, 300),
                    'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
                    'max_depth': trial.suggest_int('max_depth', 3, 10),
                    'min_child_weight': trial.suggest_int('min_child_weight', 1, 10),
                    'subsample': trial.suggest_float('subsample', 0.6, 1.0),
                    'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
                    'reg_alpha': trial.suggest_float('reg_alpha', 0, 1),
                    'reg_lambda': trial.suggest_float('reg_lambda', 0, 1)
                }
                model = XGBRegressor(**params, random_state=self.random_state)
            
            elif model_name == 'lgb':
                params = {
                    'n_estimators': trial.suggest_int('n_estimators', 50, 300),
                    'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
                    'max_depth': trial.suggest_int('max_depth', 3, 10),
                    'num_leaves': trial.suggest_int('num_leaves', 20, 100),
                    'min_child_samples': trial.suggest_int('min_child_samples', 5, 30),
                    'subsample': trial.suggest_float('subsample', 0.6, 1.0),
                    'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
                    'reg_alpha': trial.suggest_float('reg_alpha', 0, 1),
                    'reg_lambda': trial.suggest_float('reg_lambda', 0, 1)
                }
                model = LGBMRegressor(**params, random_state=self.random_state)
            
            else:
                return 0
            
            scores = cross_val_score(model, X, y, cv=5, scoring='neg_mean_squared_error')
            return np.sqrt(-scores).mean()
        
        study = optuna.create_study(direction='minimize')
        study.optimize(objective, n_trials=n_trials)
        
        print(f"{model_name} 最佳参数: {study.best_params}")
        print(f"最佳 RMSE: {study.best_value:.4f}")
        
        return study.best_params
    
    def build_stacking_ensemble(self, X, y):
        """构建堆叠集成模型"""
        from sklearn.ensemble import StackingRegressor
        from sklearn.linear_model import LinearRegression
        
        # 选择表现最好的几个模型作为基础模型
        best_models = sorted(self.model_performance.items(), 
                             key=lambda x: x[1]['mean_rmse'])[:4]
        
        base_models = [(name, self.models[name]) for name, _ in best_models]
        
        # 使用线性回归作为元模型
        stacking_model = StackingRegressor(
            estimators=base_models,
            final_estimator=LinearRegression(),
            cv=5,
            n_jobs=-1
        )
        
        self.stacking_model = stacking_model
        return stacking_model

from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
from sklearn.model_selection import learning_curve, validation_curve
import matplotlib.pyplot as plt
 
class ComprehensiveEvaluator:
    """综合模型评估器"""
    
    def __init__(self, model, X_train, X_test, y_train, y_test):
        self.model = model
        self.X_train = X_train
        self.X_test = X_test
        self.y_train = y_train
        self.y_test = y_test
        self.predictions = {}
        self.metrics = {}
    
    def evaluate_all_metrics(self):
        """计算所有相关指标"""
        # 训练集预测
        y_train_pred = self.model.predict(self.X_train)
        y_test_pred = self.model.predict(self.X_test)
        
        self.predictions = {
            'train': y_train_pred,
            'test': y_test_pred
        }
        
        # 计算各种指标
        for dataset, y_true, y_pred in [('train', self.y_train, y_train_pred), 
                                         ('test', self.y_test, y_test_pred)]:
            
            rmse = np.sqrt(mean_squared_error(y_true, y_pred))
            mae = mean_absolute_error(y_true, y_pred)
            r2 = r2_score(y_true, y_pred)
            
            # 自定义指标
            mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100
            
            self.metrics[dataset] = {
                'RMSE': rmse,
                'MAE': mae,
                'R²': r2,
                'MAPE': mape
            }
        
        return self.metrics
    
    def plot_learning_curves(self):
        """绘制学习曲线"""
        train_sizes, train_scores, val_scores = learning_curve(
            self.model, self.X_train, self.y_train, 
            cv=5, scoring='neg_mean_squared_error',
            train_sizes=np.linspace(0.1, 1.0, 10)
        )
        
        train_rmse = np.sqrt(-train_scores)
        val_rmse = np.sqrt(-val_scores)
        
        plt.figure(figsize=(10, 6))
        plt.plot(train_sizes, train_rmse.mean(axis=1), 'o-', label='Training RMSE')
        plt.plot(train_sizes, val_rmse.mean(axis=1), 'o-', label='Validation RMSE')
        plt.fill_between(train_sizes, train_rmse.mean(axis=1) - train_rmse.std(axis=1),
                         train_rmse.mean(axis=1) + train_rmse.std(axis=1), alpha=0.1)
        plt.fill_between(train_sizes, val_rmse.mean(axis=1) - val_rmse.std(axis=1),
                         val_rmse.mean(axis=1) + val_rmse.std(axis=1), alpha=0.1)
        
        plt.xlabel('Training Set Size')
        plt.ylabel('RMSE')
        plt.title('Learning Curves')
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.show()
    
    def plot_residual_analysis(self):
        """残差分析"""
        y_test_pred = self.predictions['test']
        residuals = self.y_test - y_test_pred
        
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        # 残差散点图
        axes[0, 0].scatter(y_test_pred, residuals, alpha=0.6)
        axes[0, 0].axhline(y=0, color='r', linestyle='--')
        axes[0, 0].set_xlabel('Predicted Values')
        axes[0, 0].set_ylabel('Residuals')
        axes[0, 0].set_title('Residual Plot')
        
        # 残差直方图
        axes[0, 1].hist(residuals, bins=30, alpha=0.7, density=True)
        axes[0, 1].set_xlabel('Residuals')
        axes[0, 1].set_ylabel('Density')
        axes[0, 1].set_title('Residual Distribution')
        
        # Q-Q图
        from scipy import stats
        stats.probplot(residuals, dist="norm", plot=axes[1, 0])
        axes[1, 0].set_title('Q-Q Plot')
        
        # 预测值vs实际值
        axes[1, 1].scatter(self.y_test, y_test_pred, alpha=0.6)
        axes[1, 1].plot([self.y_test.min(), self.y_test.max()], 
                       [self.y_test.min(), self.y_test.max()], 'r--', lw=2)
        axes[1, 1].set_xlabel('Actual Values')
        axes[1, 1].set_ylabel('Predicted Values')
        axes[1, 1].set_title('Actual vs Predicted')
        
        plt.tight_layout()
        plt.show()
    
    def cross_validate_model(self, cv_folds=10):
        """交叉验证"""
        from sklearn.model_selection import cross_validate
        
        scoring = ['neg_mean_squared_error', 'neg_mean_absolute_error', 'r2']
        
        cv_results = cross_validate(
            self.model, self.X_train, self.y_train,
            cv=cv_folds, scoring=scoring,
            return_train_score=True
        )
        
        cv_summary = {}
        for metric in scoring:
            test_scores = cv_results[f'test_{metric}']
            train_scores = cv_results[f'train_{metric}']
            
            if metric == 'neg_mean_squared_error':
                test_scores = np.sqrt(-test_scores)
                train_scores = np.sqrt(-train_scores)
                metric_name = 'RMSE'
            elif metric == 'neg_mean_absolute_error':
                test_scores = -test_scores
                train_scores = -train_scores
                metric_name = 'MAE'
            else:
                metric_name = 'R²'
            
            cv_summary[metric_name] = {
                'test_mean': test_scores.mean(),
                'test_std': test_scores.std(),
                'train_mean': train_scores.mean(),
                'train_std': train_scores.std()
            }
        
        return cv_summary

# TRAE IDE性能监控示例
class TRAEPerformanceMonitor:
    """TRAE IDE性能监控集成"""
    
    def __init__(self):
        self.metrics = {
            'memory_usage': [],
            'execution_time': [],
            'cpu_utilization': []
        }
    
    def profile_data_operation(self, operation_func, *args, **kwargs):
        """分析数据操作的性能"""
        import tracemalloc
        import time
        
        # 开始内存跟踪
        tracemalloc.start()
        start_time = time.time()
        
        # 执行操作
        result = operation_func(*args, **kwargs)
        
        # 记录性能指标
        current, peak = tracemalloc.get_traced_memory()
        execution_time = time.time() - start_time
        
        self.metrics['memory_usage'].append({
            'current_mb': current / 1024 / 1024,
            'peak_mb': peak / 1024 / 1024,
            'operation': operation_func.__name__
        })
        
        self.metrics['execution_time'].append({
            'time_seconds': execution_time,
            'operation': operation_func.__name__
        })
        
        tracemalloc.stop()
        
        # TRAE IDE会自动分析这些指标并提供优化建议
        return result