Claude Code多语言支持：Python、JavaScript、Go等全栈开发能力深度解析

# 项目结构
"""
ecommerce-analysis/
├── data/
│   ├── raw/                    # 原始数据
│   ├── processed/              # 处理后数据
│   └── external/               # 外部数据源
├── notebooks/
│   ├── 01_data_exploration.ipynb
│   ├── 02_sales_analysis.ipynb
│   ├── 03_customer_analysis.ipynb
│   └── 04_predictive_modeling.ipynb
├── src/
│   ├── data/
│   │   ├── __init__.py
│   │   ├── data_loader.py
│   │   ├── data_cleaner.py
│   │   └── data_validator.py
│   ├── features/
│   │   ├── __init__.py
│   │   ├── engineering.py
│   │   └── selection.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── baseline.py
│   │   ├── ensemble.py
│   │   └── neural_network.py
│   ├── visualization/
│   │   ├── __init__.py
│   │   ├── plots.py
│   │   └── dashboard.py
│   └── utils/
│       ├── __init__.py
│       ├── config.py
│       └── helpers.py
├── tests/
├── requirements.txt
├── setup.py
└── README.md
"""

# src/data/data_loader.py
import pandas as pd
import numpy as np
from pathlib import Path
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class DataConfig:
    """数据配置类"""
    raw_data_path: Path
    processed_data_path: Path
    date_columns: List[str]
    categorical_columns: List[str]
    numerical_columns: List[str]
    target_column: str

class EcommerceDataLoader:
    """电商数据加载器"""
    
    def __init__(self, config: DataConfig):
        self.config = config
        self.raw_data: Optional[pd.DataFrame] = None
        self.processed_data: Optional[pd.DataFrame] = None
    
    def load_raw_data(self, file_path: str) -> pd.DataFrame:
        """
        加载原始数据
        
        Args:
            file_path: 数据文件路径
            
        Returns:
            pd.DataFrame: 原始数据
        """
        try:
            logger.info(f"正在加载数据: {file_path}")
            
            # 根据文件类型选择加载方法
            if file_path.endswith('.csv'):
                data = pd.read_csv(file_path)
            elif file_path.endswith('.xlsx'):
                data = pd.read_excel(file_path)
            elif file_path.endswith('.json'):
                data = pd.read_json(file_path)
            else:
                raise ValueError(f"不支持的文件格式: {file_path}")
            
            logger.info(f"成功加载数据: {data.shape[0]} 行, {data.shape[1]} 列")
            self.raw_data = data
            return data
            
        except Exception as e:
            logger.error(f"数据加载失败: {str(e)}")
            raise
    
    def get_data_info(self, data: pd.DataFrame) -> Dict:
        """
        获取数据概览信息
        
        Args:
            data: 数据框
            
        Returns:
            Dict: 数据信息字典
        """
        info = {
            'shape': data.shape,
            'columns': data.columns.tolist(),
            'dtypes': data.dtypes.to_dict(),
            'missing_values': data.isnull().sum().to_dict(),
            'duplicates': data.duplicated().sum(),
            'memory_usage': data.memory_usage(deep=True).sum() / 1024**2,  # MB
        }
        
        # 数值列统计
        numerical_stats = data.select_dtypes(include=[np.number]).describe()
        info['numerical_stats'] = numerical_stats.to_dict()
        
        # 分类列统计
        categorical_stats = {}
        for col in data.select_dtypes(include=['object']).columns:
            categorical_stats[col] = {
                'unique_count': data[col].nunique(),
                'top_values': data[col].value_counts().head().to_dict()
            }
        info['categorical_stats'] = categorical_stats
        
        return info
    
    def validate_data_quality(self, data: pd.DataFrame) -> Dict:
        """
        数据质量验证
        
        Args:
            data: 数据框
            
        Returns:
            Dict: 质量检查结果
        """
        quality_report = {
            'completeness': {},
            'consistency': {},
            'validity': {},
            'accuracy': {}
        }
        
        # 完整性检查
        missing_ratio = data.isnull().sum() / len(data)
        quality_report['completeness'] = {
            'missing_ratio': missing_ratio.to_dict(),
            'high_missing_columns': missing_ratio[missing_ratio > 0.3].index.tolist()
        }
        
        # 一致性检查
        quality_report['consistency'] = {
            'duplicate_rows': data.duplicated().sum(),
            'duplicate_ratio': data.duplicated().sum() / len(data)
        }
        
        # 有效性检查
        validity_issues = {}
        
        # 检查数值列的异常值
        for col in data.select_dtypes(include=[np.number]).columns:
            Q1 = data[col].quantile(0.25)
            Q3 = data[col].quantile(0.75)
            IQR = Q3 - Q1
            outliers = data[(data[col] < Q1 - 1.5 * IQR) | (data[col] > Q3 + 1.5 * IQR)]
            validity_issues[col] = {
                'outlier_count': len(outliers),
                'outlier_ratio': len(outliers) / len(data)
            }
        
        quality_report['validity'] = validity_issues
        
        return quality_report

# src/data/data_cleaner.py
class EcommerceDataCleaner:
    """电商数据清洗器"""
    
    def __init__(self, config: DataConfig):
        self.config = config
    
    def clean_data(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        执行完整的数据清洗流程
        
        Args:
            data: 原始数据
            
        Returns:
            pd.DataFrame: 清洗后的数据
        """
        logger.info("开始数据清洗...")
        
        # 创建数据副本
        cleaned_data = data.copy()
        
        # 1. 处理缺失值
        cleaned_data = self._handle_missing_values(cleaned_data)
        
        # 2. 处理重复值
        cleaned_data = self._handle_duplicates(cleaned_data)
        
        # 3. 数据类型转换
        cleaned_data = self._convert_data_types(cleaned_data)
        
        # 4. 处理异常值
        cleaned_data = self._handle_outliers(cleaned_data)
        
        # 5. 标准化文本数据
        cleaned_data = self._standardize_text(cleaned_data)
        
        # 6. 创建派生特征
        cleaned_data = self._create_derived_features(cleaned_data)
        
        logger.info(f"数据清洗完成: {cleaned_data.shape}")
        return cleaned_data
    
    def _handle_missing_values(self, data: pd.DataFrame) -> pd.DataFrame:
        """处理缺失值"""
        logger.info("处理缺失值...")
        
        # 数值列：使用中位数填充
        numerical_columns = data.select_dtypes(include=[np.number]).columns
        for col in numerical_columns:
            if data[col].isnull().sum() > 0:
                median_value = data[col].median()
                data[col].fillna(median_value, inplace=True)
                logger.info(f"列 {col} 使用中位数 {median_value} 填充缺失值")
        
        # 分类列：使用众数填充
        categorical_columns = data.select_dtypes(include=['object']).columns
        for col in categorical_columns:
            if data[col].isnull().sum() > 0:
                mode_value = data[col].mode()[0] if not data[col].mode().empty else 'Unknown'
                data[col].fillna(mode_value, inplace=True)
                logger.info(f"列 {col} 使用众数 {mode_value} 填充缺失值")
        
        return data
    
    def _handle_duplicates(self, data: pd.DataFrame) -> pd.DataFrame:
        """处理重复值"""
        logger.info("处理重复值...")
        
        initial_count = len(data)
        data_deduplicated = data.drop_duplicates()
        removed_count = initial_count - len(data_deduplicated)
        
        if removed_count > 0:
            logger.info(f"移除了 {removed_count} 行重复数据")
        
        return data_deduplicated
    
    def _convert_data_types(self, data: pd.DataFrame) -> pd.DataFrame:
        """转换数据类型"""
        logger.info("转换数据类型...")
        
        # 转换日期列
        for col in self.config.date_columns:
            if col in data.columns:
                data[col] = pd.to_datetime(data[col], errors='coerce')
                logger.info(f"列 {col} 转换为日期类型")
        
        # 转换分类列
        for col in self.config.categorical_columns:
            if col in data.columns:
                data[col] = data[col].astype('category')
                logger.info(f"列 {col} 转换为分类类型")
        
        return data
    
    def _handle_outliers(self, data: pd.DataFrame) -> pd.DataFrame:
        """处理异常值"""
        logger.info("处理异常值...")
        
        for col in self.config.numerical_columns:
            if col in data.columns:
                # 使用IQR方法检测异常值
                Q1 = data[col].quantile(0.25)
                Q3 = data[col].quantile(0.75)
                IQR = Q3 - Q1
                
                lower_bound = Q1 - 1.5 * IQR
                upper_bound = Q3 + 1.5 * IQR
                
                # 记录异常值数量
                outliers = data[(data[col] < lower_bound) | (data[col] > upper_bound)]
                if len(outliers) > 0:
                    logger.info(f"列 {col} 发现 {len(outliers)} 个异常值")
                    
                    # 使用边界值替换异常值
                    data[col] = data[col].clip(lower=lower_bound, upper=upper_bound)
        
        return data
    
    def _standardize_text(self, data: pd.DataFrame) -> pd.DataFrame:
        """标准化文本数据"""
        logger.info("标准化文本数据...")
        
        text_columns = data.select_dtypes(include=['object']).columns
        
        for col in text_columns:
            if col not in self.config.date_columns:
                # 转换为小写
                data[col] = data[col].astype(str).str.lower()
                # 去除前后空格
                data[col] = data[col].str.strip()
                # 替换多个空格为单个空格
                data[col] = data[col].str.replace(r'\s+', ' ', regex=True)
        
        return data
    
    def _create_derived_features(self, data: pd.DataFrame) -> pd.DataFrame:
        """创建派生特征"""
        logger.info("创建派生特征...")
        
        # 示例：如果有订单日期，创建时间相关特征
        if 'order_date' in data.columns:
            data['order_year'] = data['order_date'].dt.year
            data['order_month'] = data['order_date'].dt.month
            data['order_day_of_week'] = data['order_date'].dt.dayofweek
            data['order_quarter'] = data['order_date'].dt.quarter
        
        # 示例：如果有价格和数量，计算总金额
        if 'price' in data.columns and 'quantity' in data.columns:
            data['total_amount'] = data['price'] * data['quantity']
        
        # 示例：如果有客户ID，计算客户订单频率
        if 'customer_id' in data.columns and 'order_date' in data.columns:
            customer_order_counts = data.groupby('customer_id')['order_date'].count()
            data['customer_order_frequency'] = data['customer_id'].map(customer_order_counts)
        
        return data