Pandas处理CSV数据的清洗技巧与实践指南

import pandas as pd
import numpy as np
 
# TRAE IDE智能推荐的加载配置
df = pd.read_csv('data.csv', 
                 encoding='utf-8',  # 自动检测编码
                 parse_dates=['date_column'],  # 智能识别日期列
                 dtype={'id': 'int32', 'price': 'float32'},  # 优化内存使用
                 na_values=['NA', 'N/A', '', 'NULL'])  # 统一缺失值标识
 
# 快速数据概览
print(df.info())
print(df.describe())

# 数据质量检查函数
def data_quality_report(df):
    """生成数据质量报告"""
    report = {
        '总行数': len(df),
        '总列数': len(df.columns),
        '缺失值比例': (df.isnull().sum() / len(df) * 100).round(2),
        '重复行数': df.duplicated().sum(),
        '内存使用': df.memory_usage(deep=True).sum() / 1024**2
    }
    return report
 
# TRAE IDE会提示优化内存使用的建议
quality_report = data_quality_report(df)

# 缺失值可视化分析
import matplotlib.pyplot as plt
import seaborn as sns
 
# 使用TRAE IDE智能提示生成可视化代码
plt.figure(figsize=(12, 6))
sns.heatmap(df.isnull(), cbar=True, cmap='viridis')
plt.title('缺失值分布热力图')
plt.tight_layout()
plt.show()
 
# 统计缺失值模式
missing_patterns = df.isnull().sum().sort_values(ascending=False)
missing_percent = (missing_patterns / len(df) * 100)
 
print("缺失值统计：")
print(missing_percent[missing_percent > 0])

# 基于数据类型的智能填充
def smart_fill_missing(df):
    """智能填充缺失值"""
    df_cleaned = df.copy()
    
    for column in df.columns:
        if df[column].dtype in ['int64', 'float64']:
            # 数值型：使用中位数填充
            df_cleaned[column].fillna(df[column].median(), inplace=True)
        elif df[column].dtype == 'object':
            # 分类型：使用众数填充
            mode_value = df[column].mode()
            if len(mode_value) > 0:
                df_cleaned[column].fillna(mode_value[0], inplace=True)
        elif df[column].dtype == 'datetime64[ns]':
            # 日期型：使用前后填充
            df_cleaned[column].fillna(method='ffill', inplace=True)
    
    return df_cleaned
 
# TRAE IDE会提示性能优化建议
df_filled = smart_fill_missing(df)

# 智能数据类型转换
def optimize_data_types(df):
    """优化数据类型以节省内存"""
    df_optimized = df.copy()
    
    for col in df.columns:
        col_type = df[col].dtype
        
        if col_type != 'datetime64[ns]':
            c_min = df[col].min()
            c_max = df[col].max()
            
            if col_type == 'int64':
                if c_min > np.iinfo(np.int8).min and c_max < np.iinfo(np.int8).max:
                    df_optimized[col] = df[col].astype(np.int8)
                elif c_min > np.iinfo(np.int16).min and c_max < np.iinfo(np.int16).max:
                    df_optimized[col] = df[col].astype(np.int16)
                elif c_min > np.iinfo(np.int32).min and c_max < np.iinfo(np.int32).max:
                    df_optimized[col] = df[col].astype(np.int32)
            
            elif col_type == 'float64':
                if c_min > np.finfo(np.float32).min and c_max < np.finfo(np.float32).max:
                    df_optimized[col] = df[col].astype(np.float32)
    
    return df_optimized
 
# TRAE IDE实时显示内存优化效果
print("原始内存使用:", df.memory_usage(deep=True).sum() / 1024**2, "MB")
df_optimized = optimize_data_types(df_filled)
print("优化后内存使用:", df_optimized.memory_usage(deep=True).sum() / 1024**2, "MB")

# 文本数据清洗函数
def clean_text_data(df, text_columns):
    """清洗文本数据"""
    df_cleaned = df.copy()
    
    for col in text_columns:
        if col in df.columns:
            # 去除前后空格
            df_cleaned[col] = df_cleaned[col].str.strip()
            # 统一大小写
            df_cleaned[col] = df_cleaned[col].str.lower()
            # 去除特殊字符
            df_cleaned[col] = df_cleaned[col].str.replace(r'[^\w\s]', '', regex=True)
            # 去除多余空格
            df_cleaned[col] = df_cleaned[col].str.replace(r'\s+', ' ', regex=True)
    
    return df_cleaned
 
# 使用TRAE IDE的智能提示快速应用文本清洗
text_cols = df_optimized.select_dtypes(include=['object']).columns
df_text_cleaned = clean_text_data(df_optimized, text_cols)

# 异常值检测函数库
class OutlierDetector:
    """异常值检测器"""
    
    @staticmethod
    def iqr_method(df, column):
        """IQR方法检测异常值"""
        Q1 = df[column].quantile(0.25)
        Q3 = df[column].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR
        
        outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)]
        return outliers, lower_bound, upper_bound
    
    @staticmethod
    def z_score_method(df, column, threshold=3):
        """Z-score方法检测异常值"""
        z_scores = np.abs((df[column] - df[column].mean()) / df[column].std())
        outliers = df[z_scores > threshold]
        return outliers
    
    @staticmethod
    def isolation_forest(df, columns):
        """孤立森林检测异常值"""
        from sklearn.ensemble import IsolationForest
        
        iso_forest = IsolationForest(contamination=0.1, random_state=42)
        outliers = iso_forest.fit_predict(df[columns])
        
        return df[outliers == -1]
 
# TRAE IDE会提示导入必要的库
detector = OutlierDetector()
 
# 检测数值列的异常值
numeric_cols = df_text_cleaned.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
    outliers, lower, upper = detector.iqr_method(df_text_cleaned, col)
    print(f"{col}: 发现 {len(outliers)} 个异常值")
    print(f"正常范围: [{lower:.2f}, {upper:.2f}]")

# 基于业务逻辑的异常值处理
def handle_outliers_smart(df, column, method='cap'):
    """智能处理异常值"""
    df_handled = df.copy()
    
    if method == 'cap':
        # 截断处理
        Q1 = df_handled[column].quantile(0.01)
        Q99 = df_handled[column].quantile(0.99)
        df_handled[column] = df_handled[column].clip(Q1, Q99)
    
    elif method == 'transform':
        # 对数变换
        df_handled[column] = np.log1p(df_handled[column])
    
    elif method == 'remove':
        # 移除异常值
        Q1 = df_handled[column].quantile(0.25)
        Q3 = df_handled[column].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR
        
        mask = (df_handled[column] >= lower_bound) & (df_handled[column] <= upper_bound)
        df_handled = df_handled[mask]
    
    return df_handled
 
# 使用TRAE IDE的智能建议选择合适的处理方法
df_final = df_text_cleaned.copy()
for col in numeric_cols:
    if col in ['price', 'revenue']:  # 业务相关的数值列
        df_final = handle_outliers_smart(df_final, col, method='cap')

# 高级重复数据检测
def detect_duplicates_advanced(df):
    """高级重复数据检测"""
    
    # 完全重复
    exact_duplicates = df[df.duplicated()]
    
    # 关键列重复（业务主键）
    key_columns = ['id', 'email', 'phone']  # 根据业务定义
    available_keys = [col for col in key_columns if col in df.columns]
    
    if available_keys:
        key_duplicates = df[df.duplicated(subset=available_keys, keep=False)]
    else:
        key_duplicates = pd.DataFrame()
    
    # 模糊重复（文本相似）
    text_duplicates = []
    if 'name' in df.columns:
        from difflib import SequenceMatcher
        
        names = df['name'].dropna().unique()
        for i, name1 in enumerate(names):
            for name2 in names[i+1:]:
                similarity = SequenceMatcher(None, name1, name2).ratio()
                if similarity > 0.8:  # 相似度阈值
                    text_duplicates.append({
                        'name1': name1,
                        'name2': name2,
                        'similarity': similarity
                    })
    
    return {
        'exact_duplicates': exact_duplicates,
        'key_duplicates': key_duplicates,
        'text_duplicates': pd.DataFrame(text_duplicates)
    }
 
# TRAE IDE会提示性能优化建议
duplicate_report = detect_duplicates_advanced(df_final)
print(f"完全重复记录: {len(duplicate_report['exact_duplicates'])}")
print(f"关键列重复: {len(duplicate_report['key_duplicates'])}")

# 数据一致性检查函数
def validate_data_consistency(df):
    """验证数据一致性"""
    consistency_issues = []
    
    # 日期一致性检查
    if 'start_date' in df.columns and 'end_date' in df.columns:
        invalid_dates = df[df['start_date'] > df['end_date']]
        if len(invalid_dates) > 0:
            consistency_issues.append(f"发现 {len(invalid_dates)} 条开始日期晚于结束日期的记录")
    
    # 数值范围检查
    if 'age' in df.columns:
        invalid_ages = df[(df['age'] < 0) | (df['age'] > 120)]
        if len(invalid_ages) > 0:
            consistency_issues.append(f"发现 {len(invalid_ages)} 条异常年龄记录")
    
    # 业务逻辑检查
    if 'price' in df.columns and 'discount' in df.columns:
        invalid_discounts = df[df['discount'] > df['price']]
        if len(invalid_discounts) > 0:
            consistency_issues.append(f"发现 {len(invalid_discounts)} 条折扣大于价格的记录")
    
    return consistency_issues
 
# 使用TRAE IDE的智能提示完善验证逻辑
consistency_issues = validate_data_consistency(df_final)
for issue in consistency_issues:
    print(f"⚠️ {issue}")

# 使用TRAE IDE智能生成的完整数据清洗管道
class DataCleaningPipeline:
    """数据清洗管道"""
    
    def __init__(self, df):
        self.original_df = df.copy()
        self.cleaned_df = df.copy()
        self.report = {}
    
    def run_full_cleaning(self):
        """运行完整的数据清洗流程"""
        print("🚀 开始数据清洗流程...")
        
        # 步骤1：数据质量评估
        print("📊 步骤1: 数据质量评估")
        self.report['original_shape'] = self.cleaned_df.shape
        self.report['missing_values'] = self.cleaned_df.isnull().sum().sum()
        
        # 步骤2：处理缺失值
        print("🔧 步骤2: 处理缺失值")
        self.cleaned_df = smart_fill_missing(self.cleaned_df)
        
        # 步骤3：数据类型优化
        print("⚙️ 步骤3: 数据类型优化")
        self.cleaned_df = optimize_data_types(self.cleaned_df)
        
        # 步骤4：文本清洗
        print("📝 步骤4: 文本数据清洗")
        text_cols = self.cleaned_df.select_dtypes(include=['object']).columns
        self.cleaned_df = clean_text_data(self.cleaned_df, text_cols)
        
        # 步骤5：异常值处理
        print("📈 步骤5: 异常值处理")
        numeric_cols = self.cleaned_df.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            if col in ['price', 'revenue', 'sales']:
                self.cleaned_df = handle_outliers_smart(self.cleaned_df, col, method='cap')
        
        # 步骤6：重复数据处理
        print("🔄 步骤6: 重复数据处理")
        initial_rows = len(self.cleaned_df)
        self.cleaned_df = self.cleaned_df.drop_duplicates()
        self.report['duplicates_removed'] = initial_rows - len(self.cleaned_df)
        
        # 步骤7：数据一致性验证
        print("✅ 步骤7: 数据一致性验证")
        self.report['consistency_issues'] = validate_data_consistency(self.cleaned_df)
        
        # 生成最终报告
        self.generate_cleaning_report()
        
        return self.cleaned_df
    
    def generate_cleaning_report(self):
        """生成清洗报告"""
        self.report['final_shape'] = self.cleaned_df.shape
        self.report['memory_reduction'] = (
            self.original_df.memory_usage(deep=True).sum() - 
            self.cleaned_df.memory_usage(deep=True).sum()
        ) / 1024**2
        
        print("\n📋 数据清洗完成报告:")
        print(f"原始数据形状: {self.report['original_shape']}")
        print(f"清洗后形状: {self.report['final_shape']}")
        print(f"内存节省: {self.report['memory_reduction']:.2f} MB")
        print(f"处理缺失值: {self.report['missing_values']}")
        print(f"移除重复行: {self.report['duplicates_removed']}")
 
# 在TRAE IDE中运行完整流程
pipeline = DataCleaningPipeline(df)
cleaned_data = pipeline.run_full_cleaning()

# 保存清洗后的数据
def save_cleaned_data(df, original_filename):
    """保存清洗后的数据，自动生成版本信息"""
    import datetime
    
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    base_name = original_filename.split('.')[0]
    
    # 生成带时间戳的文件名
    cleaned_filename = f"{base_name}_cleaned_{timestamp}.csv"
    
    # 保存清洗后的数据
    df.to_csv(cleaned_filename, index=False, encoding='utf-8')
    
    # 生成数据清洗日志
    log_filename = f"{base_name}_cleaning_log_{timestamp}.txt"
    with open(log_filename, 'w', encoding='utf-8') as f:
        f.write(f"数据清洗日志 - {datetime.datetime.now()}\n")
        f.write(f"原始文件: {original_filename}\n")
        f.write(f"清洗后文件: {cleaned_filename}\n")
        f.write(f"数据形状变化: {df.shape}\n")
    
    print(f"✅ 数据已保存至: {cleaned_filename}")
    print(f"📋 清洗日志已生成: {log_filename}")
 
# TRAE IDE会自动跟踪文件变更
save_cleaned_data(cleaned_data, 'data.csv')