# sales_analysis_script.py
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from datetime import datetime
import os

# 讀取數(shù)據(jù)
df = pd.read_csv('sales_data.csv')

# 數(shù)據(jù)清洗
df['date'] = pd.to_datetime(df['date'])
df = df.dropna()

# 計算月度銷售統(tǒng)計
df['month'] = df['date'].dt.to_period('M')
monthly_sales = df.groupby('month').agg({
    'sales': ['sum', 'mean', 'std'],
    'profit': ['sum', 'mean']
}).round(2)

# 打印結果
print("月度銷售統(tǒng)計:")
print(monthly_sales)

# 生成圖表
plt.figure(figsize=(12, 6))
plt.subplot(1, 2, 1)
df.groupby('month')['sales'].sum().plot(kind='bar')
plt.title('月度銷售額')
plt.ylabel('銷售額')

plt.subplot(1, 2, 2)
df.groupby('month')['profit'].sum().plot(kind='bar', color='orange')
plt.title('月度利潤')
plt.ylabel('利潤')

plt.tight_layout()
plt.savefig('sales_analysis.png')

# 保存處理后的數(shù)據(jù)
df.to_csv('processed_sales_data.csv', index=False)
print("分析完成！結果已保存。")

# src/sales_analyzer/data_loader.py
"""
數(shù)據(jù)加載模塊
負責從不同源加載銷售數(shù)據(jù)
"""

import pandas as pd
from pathlib import Path
import logging

logger = logging.getLogger(__name__)

class DataLoader:
    """數(shù)據(jù)加載器類"""
    
    def __init__(self, data_dir="data/raw"):
        """
        初始化數(shù)據(jù)加載器
        
        Args:
            data_dir (str): 數(shù)據(jù)目錄路徑
        """
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(parents=True, exist_ok=True)
    
    def load_from_csv(self, file_path, **kwargs):
        """
        從CSV文件加載數(shù)據(jù)
        
        Args:
            file_path (str): CSV文件路徑
            **kwargs: 傳遞給pandas.read_csv的參數(shù)
            
        Returns:
            pd.DataFrame: 加載的數(shù)據(jù)
            
        Raises:
            FileNotFoundError: 當文件不存在時
        """
        file_path = Path(file_path)
        if not file_path.exists():
            logger.error(f"文件不存在: {file_path}")
            raise FileNotFoundError(f"文件不存在: {file_path}")
        
        logger.info(f"正在加載數(shù)據(jù): {file_path}")
        df = pd.read_csv(file_path, **kwargs)
        logger.info(f"成功加載數(shù)據(jù)，形狀: {df.shape}")
        
        return df
    
    def load_from_dict(self, data_dict):
        """
        從字典加載數(shù)據(jù)
        
        Args:
            data_dict (dict): 數(shù)據(jù)字典
            
        Returns:
            pd.DataFrame: 加載的數(shù)據(jù)
        """
        logger.info("從字典加載數(shù)據(jù)")
        df = pd.DataFrame(data_dict)
        logger.info(f"成功從字典加載數(shù)據(jù)，形狀: {df.shape}")
        
        return df
    
    def save_data(self, df, file_path, **kwargs):
        """
        保存數(shù)據(jù)到文件
        
        Args:
            df (pd.DataFrame): 要保存的數(shù)據(jù)框
            file_path (str): 文件路徑
            **kwargs: 傳遞給pandas.to_csv的參數(shù)
        """
        file_path = Path(file_path)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        
        logger.info(f"保存數(shù)據(jù)到: {file_path}")
        df.to_csv(file_path, **kwargs)
        logger.info("數(shù)據(jù)保存成功")

# src/sales_analyzer/data_processor.py
"""
數(shù)據(jù)處理模塊
負責數(shù)據(jù)清洗、轉換和預處理
"""

import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional
import logging

logger = logging.getLogger(__name__)

class DataProcessor:
    """數(shù)據(jù)處理器類"""
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        """
        初始化數(shù)據(jù)處理器
        
        Args:
            config (dict, optional): 處理配置
        """
        self.config = config or {}
        self._validate_config()
    
    def _validate_config(self):
        """驗證配置參數(shù)"""
        required_params = ['date_column', 'value_columns']
        for param in required_params:
            if param not in self.config:
                raise ValueError(f"缺少必要配置參數(shù): {param}")
    
    def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        數(shù)據(jù)清洗
        
        Args:
            df (pd.DataFrame): 原始數(shù)據(jù)
            
        Returns:
            pd.DataFrame: 清洗后的數(shù)據(jù)
        """
        logger.info("開始數(shù)據(jù)清洗")
        
        # 創(chuàng)建副本，避免修改原始數(shù)據(jù)
        cleaned_df = df.copy()
        
        # 處理日期列
        date_column = self.config.get('date_column', 'date')
        if date_column in cleaned_df.columns:
            cleaned_df[date_column] = pd.to_datetime(
                cleaned_df[date_column], errors='coerce'
            )
        
        # 處理缺失值
        value_columns = self.config.get('value_columns', [])
        for column in value_columns:
            if column in cleaned_df.columns:
                # 數(shù)值列用中位數(shù)填充
                if pd.api.types.is_numeric_dtype(cleaned_df[column]):
                    median_value = cleaned_df[column].median()
                    cleaned_df[column] = cleaned_df[column].fillna(median_value)
                # 分類列用眾數(shù)填充
                else:
                    mode_value = cleaned_df[column].mode()[0] if not cleaned_df[column].mode().empty else 'Unknown'
                    cleaned_df[column] = cleaned_df[column].fillna(mode_value)
        
        # 刪除仍然包含缺失值的行
        initial_shape = cleaned_df.shape
        cleaned_df = cleaned_df.dropna()
        final_shape = cleaned_df.shape
        
        rows_removed = initial_shape[0] - final_shape[0]
        if rows_removed > 0:
            logger.warning(f"刪除了 {rows_removed} 行包含缺失值的數(shù)據(jù)")
        
        logger.info(f"數(shù)據(jù)清洗完成，最終形狀: {cleaned_df.shape}")
        return cleaned_df
    
    def add_time_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        添加時間特征
        
        Args:
            df (pd.DataFrame): 輸入數(shù)據(jù)
            
        Returns:
            pd.DataFrame: 添加了時間特征的數(shù)據(jù)
        """
        logger.info("添加時間特征")
        
        enhanced_df = df.copy()
        date_column = self.config.get('date_column', 'date')
        
        if date_column in enhanced_df.columns:
            # 添加各種時間維度特征
            enhanced_df['year'] = enhanced_df[date_column].dt.year
            enhanced_df['month'] = enhanced_df[date_column].dt.month
            enhanced_df['quarter'] = enhanced_df[date_column].dt.quarter
            enhanced_df['day_of_week'] = enhanced_df[date_column].dt.dayofweek
            enhanced_df['is_weekend'] = enhanced_df['day_of_week'].isin([5, 6]).astype(int)
            
            # 添加月份名稱
            enhanced_df['month_name'] = enhanced_df[date_column].dt.strftime('%B')
        
        logger.info("時間特征添加完成")
        return enhanced_df
    
    def calculate_aggregations(self, df: pd.DataFrame, 
                             groupby_columns: List[str],
                             aggregation_rules: Dict[str, Any]) -> pd.DataFrame:
        """
        計算聚合統(tǒng)計
        
        Args:
            df (pd.DataFrame): 輸入數(shù)據(jù)
            groupby_columns (list): 分組列
            aggregation_rules (dict): 聚合規(guī)則
            
        Returns:
            pd.DataFrame: 聚合結果
        """
        logger.info(f"計算聚合統(tǒng)計，分組列: {groupby_columns}")
        
        # 驗證分組列是否存在
        for column in groupby_columns:
            if column not in df.columns:
                raise ValueError(f"分組列不存在: {column}")
        
        # 執(zhí)行聚合
        aggregated_df = df.groupby(groupby_columns).agg(aggregation_rules)
        
        # 扁平化多級列名
        if isinstance(aggregated_df.columns, pd.MultiIndex):
            aggregated_df.columns = ['_'.join(col).strip() for col in aggregated_df.columns.values]
        
        logger.info(f"聚合計算完成，結果形狀: {aggregated_df.shape}")
        return aggregated_df.reset_index()

# src/sales_analyzer/analyzer.py
"""
分析模塊
負責執(zhí)行各種數(shù)據(jù)分析任務
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Any, Tuple
import logging
from scipy import stats

logger = logging.getLogger(__name__)

class SalesAnalyzer:
    """銷售分析器類"""
    
    def __init__(self, config: Dict[str, Any] = None):
        """
        初始化分析器
        
        Args:
            config (dict): 分析配置
        """
        self.config = config or {}
        self.results = {}
    
    def calculate_basic_statistics(self, df: pd.DataFrame, 
                                 value_columns: List[str]) -> Dict[str, Any]:
        """
        計算基本統(tǒng)計量
        
        Args:
            df (pd.DataFrame): 輸入數(shù)據(jù)
            value_columns (list): 數(shù)值列名列表
            
        Returns:
            dict: 統(tǒng)計結果
        """
        logger.info("計算基本統(tǒng)計量")
        
        statistics = {}
        for column in value_columns:
            if column not in df.columns:
                logger.warning(f"列不存在，跳過: {column}")
                continue
            
            if pd.api.types.is_numeric_dtype(df[column]):
                stats_data = {
                    'count': df[column].count(),
                    'mean': df[column].mean(),
                    'std': df[column].std(),
                    'min': df[column].min(),
                    '25%': df[column].quantile(0.25),
                    '50%': df[column].quantile(0.50),
                    '75%': df[column].quantile(0.75),
                    'max': df[column].max(),
                    'variance': df[column].var(),
                    'skewness': df[column].skew(),
                    'kurtosis': df[column].kurtosis()
                }
                statistics[column] = {k: round(v, 4) if isinstance(v, (int, float)) else v 
                                    for k, v in stats_data.items()}
        
        self.results['basic_statistics'] = statistics
        logger.info("基本統(tǒng)計量計算完成")
        return statistics
    
    def analyze_trends(self, df: pd.DataFrame, 
                      date_column: str,
                      value_column: str,
                      freq: str = 'M') -> Dict[str, Any]:
        """
        分析趨勢
        
        Args:
            df (pd.DataFrame): 輸入數(shù)據(jù)
            date_column (str): 日期列名
            value_column (str): 數(shù)值列名
            freq (str): 時間頻率（M-月，W-周，D-天）
            
        Returns:
            dict: 趨勢分析結果
        """
        logger.info(f"分析趨勢: {value_column} vs {date_column}")
        
        if date_column not in df.columns or value_column not in df.columns:
            raise ValueError("必要的列不存在")
        
        # 確保日期列是datetime類型
        analysis_df = df.copy()
        analysis_df[date_column] = pd.to_datetime(analysis_df[date_column])
        
        # 設置日期索引并重采樣
        analysis_df = analysis_df.set_index(date_column)
        time_series = analysis_df[value_column].resample(freq).sum()
        
        # 計算趨勢指標
        trend_analysis = {
            'time_series': time_series,
            'total': time_series.sum(),
            'average': time_series.mean(),
            'growth_rate': self._calculate_growth_rate(time_series),
            'seasonality': self._detect_seasonality(time_series),
            'trend_strength': self._calculate_trend_strength(time_series)
        }
        
        self.results['trend_analysis'] = trend_analysis
        logger.info("趨勢分析完成")
        return trend_analysis
    
    def _calculate_growth_rate(self, time_series: pd.Series) -> float:
        """計算增長率"""
        if len(time_series) < 2:
            return 0.0
        
        first_value = time_series.iloc[0]
        last_value = time_series.iloc[-1]
        
        if first_value == 0:
            return 0.0
        
        return (last_value - first_value) / first_value
    
    def _detect_seasonality(self, time_series: pd.Series) -> Dict[str, Any]:
        """檢測季節(jié)性"""
        if len(time_series) < 12:  # 至少需要一年數(shù)據(jù)
            return {'has_seasonality': False, 'strength': 0}
        
        # 簡單的季節(jié)性檢測（實際項目中可以使用更復雜的方法）
        seasonal_variance = time_series.groupby(time_series.index.month).var().mean()
        total_variance = time_series.var()
        
        strength = seasonal_variance / total_variance if total_variance > 0 else 0
        
        return {
            'has_seasonality': strength > 0.1,
            'strength': round(strength, 4)
        }
    
    def _calculate_trend_strength(self, time_series: pd.Series) -> float:
        """計算趨勢強度"""
        if len(time_series) < 2:
            return 0.0
        
        # 使用Spearman相關系數(shù)衡量趨勢強度
        x = np.arange(len(time_series))
        correlation, _ = stats.spearmanr(x, time_series.values)
        
        return abs(correlation) if not np.isnan(correlation) else 0.0
    
    def correlation_analysis(self, df: pd.DataFrame, 
                           numeric_columns: List[str]) -> pd.DataFrame:
        """
        相關性分析
        
        Args:
            df (pd.DataFrame): 輸入數(shù)據(jù)
            numeric_columns (list): 數(shù)值列名列表
            
        Returns:
            pd.DataFrame: 相關性矩陣
        """
        logger.info("執(zhí)行相關性分析")
        
        # 選擇數(shù)值列
        numeric_df = df[numeric_columns].select_dtypes(include=[np.number])
        
        if numeric_df.empty:
            logger.warning("沒有可用的數(shù)值列進行相關性分析")
            return pd.DataFrame()
        
        # 計算相關性矩陣
        correlation_matrix = numeric_df.corr()
        
        self.results['correlation_matrix'] = correlation_matrix
        logger.info("相關性分析完成")
        return correlation_matrix
    
    def generate_report(self) -> str:
        """
        生成分析報告
        
        Returns:
            str: 格式化報告
        """
        logger.info("生成分析報告")
        
        report_lines = ["銷售分析報告", "=" * 50]
        
        if 'basic_statistics' in self.results:
            report_lines.append("\n基本統(tǒng)計量:")
            report_lines.append("-" * 30)
            for column, stats in self.results['basic_statistics'].items():
                report_lines.append(f"\n{column}:")
                for stat_name, value in stats.items():
                    report_lines.append(f"  {stat_name}: {value}")
        
        if 'trend_analysis' in self.results:
            trend = self.results['trend_analysis']
            report_lines.append(f"\n趨勢分析:")
            report_lines.append("-" * 30)
            report_lines.append(f"總銷售額: {trend['total']:,.2f}")
            report_lines.append(f"平均月銷售額: {trend['average']:,.2f}")
            report_lines.append(f"增長率: {trend['growth_rate']:.2%}")
            report_lines.append(f"趨勢強度: {trend['trend_strength']:.4f}")
        
        report = "\n".join(report_lines)
        return report

# src/sales_analyzer/visualizer.py
"""
可視化模塊
負責生成各種圖表和可視化結果
"""

import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import numpy as np
from pathlib import Path
from typing import Dict, List, Any, Optional
import logging

logger = logging.getLogger(__name__)

class SalesVisualizer:
    """銷售數(shù)據(jù)可視化器類"""
    
    def __init__(self, style: str = 'seaborn'):
        """
        初始化可視化器
        
        Args:
            style (str): 圖表樣式
        """
        self.style = style
        self.set_style(style)
    
    def set_style(self, style: str):
        """
        設置圖表樣式
        
        Args:
            style (str): 樣式名稱
        """
        plt.style.use(style)
        sns.set_palette("husl")
    
    def create_sales_trend_chart(self, time_series: pd.Series, 
                               title: str = "銷售趨勢",
                               save_path: Optional[str] = None) -> plt.Figure:
        """
        創(chuàng)建銷售趨勢圖
        
        Args:
            time_series (pd.Series): 時間序列數(shù)據(jù)
            title (str): 圖表標題
            save_path (str, optional): 保存路徑
            
        Returns:
            plt.Figure: 圖表對象
        """
        logger.info("創(chuàng)建銷售趨勢圖")
        
        fig, ax = plt.subplots(figsize=(12, 6))
        
        # 繪制趨勢線
        ax.plot(time_series.index, time_series.values, 
               marker='o', linewidth=2, markersize=4)
        
        # 添加趨勢線
        if len(time_series) > 1:
            z = np.polyfit(range(len(time_series)), time_series.values, 1)
            p = np.poly1d(z)
            ax.plot(time_series.index, p(range(len(time_series))), 
                   'r--', alpha=0.7, label='趨勢線')
        
        ax.set_title(title, fontsize=16, fontweight='bold')
        ax.set_xlabel('時間')
        ax.set_ylabel('銷售額')
        ax.grid(True, alpha=0.3)
        ax.legend()
        
        # 格式化y軸標簽
        ax.yaxis.set_major_formatter(
            plt.FuncFormatter(lambda x, p: f'{x:,.0f}')
        )
        
        plt.xticks(rotation=45)
        plt.tight_layout()
        
        if save_path:
            self._save_figure(fig, save_path)
        
        return fig
    
    def create_comparison_chart(self, data: Dict[str, pd.Series],
                              title: str = "指標對比",
                              chart_type: str = 'bar',
                              save_path: Optional[str] = None) -> plt.Figure:
        """
        創(chuàng)建對比圖表
        
        Args:
            data (dict): 數(shù)據(jù)字典
            title (str): 圖表標題
            chart_type (str): 圖表類型（bar, line, area）
            save_path (str, optional): 保存路徑
            
        Returns:
            plt.Figure: 圖表對象
        """
        logger.info("創(chuàng)建對比圖表")
        
        fig, ax = plt.subplots(figsize=(12, 6))
        
        if chart_type == 'bar':
            # 創(chuàng)建分組柱狀圖
            df = pd.DataFrame(data)
            df.plot(kind='bar', ax=ax, width=0.8)
        elif chart_type == 'line':
            for name, series in data.items():
                ax.plot(series.index, series.values, marker='o', label=name)
            ax.legend()
        elif chart_type == 'area':
            df = pd.DataFrame(data)
            df.plot(kind='area', ax=ax, alpha=0.7)
        
        ax.set_title(title, fontsize=16, fontweight='bold')
        ax.set_xlabel('時間周期')
        ax.set_ylabel('數(shù)值')
        ax.grid(True, alpha=0.3)
        
        # 格式化y軸標簽
        ax.yaxis.set_major_formatter(
            plt.FuncFormatter(lambda x, p: f'{x:,.0f}')
        )
        
        plt.xticks(rotation=45)
        plt.tight_layout()
        
        if save_path:
            self._save_figure(fig, save_path)
        
        return fig
    
    def create_correlation_heatmap(self, correlation_matrix: pd.DataFrame,
                                 title: str = "相關性熱力圖",
                                 save_path: Optional[str] = None) -> plt.Figure:
        """
        創(chuàng)建相關性熱力圖
        
        Args:
            correlation_matrix (pd.DataFrame): 相關性矩陣
            title (str): 圖表標題
            save_path (str, optional): 保存路徑
            
        Returns:
            plt.Figure: 圖表對象
        """
        logger.info("創(chuàng)建相關性熱力圖")
        
        fig, ax = plt.subplots(figsize=(10, 8))
        
        # 創(chuàng)建熱力圖
        mask = np.triu(np.ones_like(correlation_matrix, dtype=bool))
        sns.heatmap(correlation_matrix, mask=mask, annot=True, fmt='.2f',
                   cmap='coolwarm', center=0, square=True, ax=ax,
                   cbar_kws={"shrink": .8})
        
        ax.set_title(title, fontsize=16, fontweight='bold')
        
        plt.tight_layout()
        
        if save_path:
            self._save_figure(fig, save_path)
        
        return fig
    
    def create_dashboard(self, analysis_results: Dict[str, Any],
                        save_path: Optional[str] = None) -> plt.Figure:
        """
        創(chuàng)建分析儀表板
        
        Args:
            analysis_results (dict): 分析結果
            save_path (str, optional): 保存路徑
            
        Returns:
            plt.Figure: 儀表板圖表
        """
        logger.info("創(chuàng)建分析儀表板")
        
        fig = plt.figure(figsize=(15, 10))
        
        # 創(chuàng)建2x2的子圖布局
        gs = fig.add_gridspec(2, 2)
        
        # 趨勢圖
        ax1 = fig.add_subplot(gs[0, :])
        if 'trend_analysis' in analysis_results:
            trend_data = analysis_results['trend_analysis']['time_series']
            ax1.plot(trend_data.index, trend_data.values, 
                    marker='o', linewidth=2, color='blue')
            ax1.set_title('銷售趨勢', fontweight='bold')
            ax1.grid(True, alpha=0.3)
        
        # 月度對比圖
        ax2 = fig.add_subplot(gs[1, 0])
        if 'basic_statistics' in analysis_results:
            stats = analysis_results['basic_statistics']
            months = list(stats.keys())[:6]  # 顯示前6個月
            values = [stats[month]['mean'] for month in months]
            ax2.bar(months, values, color='lightblue')
            ax2.set_title('月度平均銷售', fontweight='bold')
            ax2.tick_params(axis='x', rotation=45)
        
        # 分布圖
        ax3 = fig.add_subplot(gs[1, 1])
        if 'trend_analysis' in analysis_results:
            trend_data = analysis_results['trend_analysis']['time_series']
            ax3.hist(trend_data.values, bins=10, alpha=0.7, color='green')
            ax3.set_title('銷售分布', fontweight='bold')
            ax3.set_xlabel('銷售額')
            ax3.set_ylabel('頻次')
        
        plt.tight_layout()
        
        if save_path:
            self._save_figure(fig, save_path)
        
        return fig
    
    def _save_figure(self, fig: plt.Figure, save_path: str):
        """
        保存圖表
        
        Args:
            fig (plt.Figure): 圖表對象
            save_path (str): 保存路徑
        """
        path = Path(save_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        
        fig.savefig(path, dpi=300, bbox_inches='tight', 
                   facecolor='white', edgecolor='none')
        logger.info(f"圖表已保存: {save_path}")

# src/sales_analyzer/main.py
"""
主程序模塊
提供統(tǒng)一的命令行接口和主要功能入口
"""

import argparse
import logging
import sys
from pathlib import Path
from .data_loader import DataLoader
from .data_processor import DataProcessor
from .analyzer import SalesAnalyzer
from .visualizer import SalesVisualizer

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('sales_analysis.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

logger = logging.getLogger(__name__)

class SalesAnalysisApp:
    """銷售分析應用程序"""
    
    def __init__(self, config_path: str = None):
        """
        初始化應用程序
        
        Args:
            config_path (str, optional): 配置文件路徑
        """
        self.config = self._load_config(config_path)
        self.setup_components()
    
    def _load_config(self, config_path: str) -> dict:
        """
        加載配置
        
        Args:
            config_path (str): 配置文件路徑
            
        Returns:
            dict: 配置字典
        """
        # 這里可以擴展為從JSON/YAML文件加載配置
        base_config = {
            'data_loader': {
                'data_dir': 'data/raw'
            },
            'data_processor': {
                'date_column': 'date',
                'value_columns': ['sales', 'profit', 'quantity']
            },
            'analyzer': {
                'numeric_columns': ['sales', 'profit', 'quantity']
            },
            'output': {
                'reports_dir': 'reports',
                'images_dir': 'images'
            }
        }
        return base_config
    
    def setup_components(self):
        """設置各個組件"""
        # 初始化各個模塊
        self.data_loader = DataLoader(
            self.config['data_loader']['data_dir']
        )
        self.data_processor = DataProcessor(
            self.config['data_processor']
        )
        self.analyzer = SalesAnalyzer(
            self.config['analyzer']
        )
        self.visualizer = SalesVisualizer()
        
        # 創(chuàng)建輸出目錄
        Path(self.config['output']['reports_dir']).mkdir(exist_ok=True)
        Path(self.config['output']['images_dir']).mkdir(exist_ok=True)
    
    def run_analysis(self, input_file: str, output_prefix: str = "sales_analysis"):
        """
        運行完整分析流程
        
        Args:
            input_file (str): 輸入文件路徑
            output_prefix (str): 輸出文件前綴
        """
        logger.info(f"開始分析流程，輸入文件: {input_file}")
        
        try:
            # 1. 加載數(shù)據(jù)
            raw_data = self.data_loader.load_from_csv(input_file)
            logger.info(f"原始數(shù)據(jù)形狀: {raw_data.shape}")
            
            # 2. 數(shù)據(jù)處理
            cleaned_data = self.data_processor.clean_data(raw_data)
            enhanced_data = self.data_processor.add_time_features(cleaned_data)
            
            # 保存處理后的數(shù)據(jù)
            processed_file = f"data/processed/{output_prefix}_processed.csv"
            self.data_loader.save_data(enhanced_data, processed_file)
            
            # 3. 數(shù)據(jù)分析
            # 基本統(tǒng)計
            basic_stats = self.analyzer.calculate_basic_statistics(
                enhanced_data, 
                self.config['data_processor']['value_columns']
            )
            
            # 趨勢分析
            trend_analysis = self.analyzer.analyze_trends(
                enhanced_data,
                'date',
                'sales'
            )
            
            # 相關性分析
            correlation_matrix = self.analyzer.correlation_analysis(
                enhanced_data,
                self.config['analyzer']['numeric_columns']
            )
            
            # 4. 生成報告和可視化
            # 文本報告
            report = self.analyzer.generate_report()
            report_file = f"{self.config['output']['reports_dir']}/{output_prefix}_report.txt"
            with open(report_file, 'w', encoding='utf-8') as f:
                f.write(report)
            logger.info(f"分析報告已保存: {report_file}")
            
            # 可視化圖表
            self._create_visualizations(
                enhanced_data, 
                self.analyzer.results,
                output_prefix
            )
            
            logger.info("分析流程完成")
            
        except Exception as e:
            logger.error(f"分析過程中發(fā)生錯誤: {e}")
            raise
    
    def _create_visualizations(self, data: pd.DataFrame, 
                             results: dict, output_prefix: str):
        """
        創(chuàng)建可視化圖表
        
        Args:
            data (pd.DataFrame): 處理后的數(shù)據(jù)
            results (dict): 分析結果
            output_prefix (str): 輸出文件前綴
        """
        images_dir = self.config['output']['images_dir']
        
        # 趨勢圖
        if 'trend_analysis' in results:
            trend_fig = self.visualizer.create_sales_trend_chart(
                results['trend_analysis']['time_series'],
                "月度銷售趨勢",
                f"{images_dir}/{output_prefix}_trend.png"
            )
            plt.close(trend_fig)
        
        # 相關性熱力圖
        if 'correlation_matrix' in results and not results['correlation_matrix'].empty:
            heatmap_fig = self.visualizer.create_correlation_heatmap(
                results['correlation_matrix'],
                "銷售指標相關性",
                f"{images_dir}/{output_prefix}_correlation.png"
            )
            plt.close(heatmap_fig)
        
        # 分析儀表板
        dashboard_fig = self.visualizer.create_dashboard(
            results,
            f"{images_dir}/{output_prefix}_dashboard.png"
        )
        plt.close(dashboard_fig)

def main():
    """主函數(shù)"""
    parser = argparse.ArgumentParser(description='銷售數(shù)據(jù)分析工具')
    parser.add_argument('input_file', help='輸入數(shù)據(jù)文件路徑')
    parser.add_argument('-o', '--output', default='sales_analysis',
                       help='輸出文件前綴')
    parser.add_argument('--config', help='配置文件路徑')
    
    args = parser.parse_args()
    
    # 創(chuàng)建并運行應用
    app = SalesAnalysisApp(args.config)
    app.run_analysis(args.input_file, args.output)

if __name__ == "__main__":
    main()

pandas>=1.5.0
numpy>=1.21.0
matplotlib>=3.5.0
seaborn>=0.11.0
scipy>=1.7.0
pathlib2>=2.3.0; python_version < '3.4'

from setuptools import setup, find_packages

with open("README.md", "r", encoding="utf-8") as fh:
    long_description = fh.read()

with open("requirements.txt", "r", encoding="utf-8") as fh:
    requirements = [line.strip() for line in fh if line.strip() and not line.startswith("#")]

setup(
    name="sales-analyzer",
    version="0.1.0",
    author="Your Name",
    author_email="your.email@example.com",
    description="A modular sales data analysis tool",
    long_description=long_description,
    long_description_content_type="text/markdown",
    packages=find_packages(where="src"),
    package_dir={"": "src"},
    classifiers=[
        "Development Status :: 3 - Alpha",
        "Intended Audience :: Developers",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",
        "Programming Language :: Python :: 3",
        "Programming Language :: Python :: 3.8",
        "Programming Language :: Python :: 3.9",
        "Programming Language :: Python :: 3.10",
    ],
    python_requires=">=3.8",
    install_requires=requirements,
    entry_points={
        "console_scripts": [
            "sales-analyzer=sales_analyzer.main:main",
        ],
    },
)

[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "sales-analyzer"
version = "0.1.0"
description = "A modular sales data analysis tool"
authors = [
    {name = "Your Name", email = "your.email@example.com"}
]
readme = "README.md"
license = {text = "MIT"}
classifiers = [
    "Development Status :: 3 - Alpha",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.8",
    "Programming Language :: Python :: 3.9",
    "Programming Language :: Python :: 3.10",
]
requires-python = ">=3.8"
dependencies = [
    "pandas>=1.5.0",
    "numpy>=1.21.0",
    "matplotlib>=3.5.0",
    "seaborn>=0.11.0",
    "scipy>=1.7.0",
]

[project.scripts]
sales-analyzer = "sales_analyzer.main:main"

[tool.setuptools.packages.find]
where = ["src"]

# tests/test_data_loader.py
import pytest
import pandas as pd
from pathlib import Path
from sales_analyzer.data_loader import DataLoader

class TestDataLoader:
    """測試數(shù)據(jù)加載器"""
    
    def setup_method(self):
        """測試設置"""
        self.loader = DataLoader()
        self.test_data = {
            'date': ['2023-01-01', '2023-01-02', '2023-01-03'],
            'sales': [100, 150, 200],
            'profit': [10, 15, 20]
        }
    
    def test_load_from_dict(self):
        """測試從字典加載數(shù)據(jù)"""
        df = self.loader.load_from_dict(self.test_data)
        
        assert isinstance(df, pd.DataFrame)
        assert df.shape == (3, 3)
        assert list(df.columns) == ['date', 'sales', 'profit']
    
    def test_save_and_load_csv(self, tmp_path):
        """測試保存和加載CSV文件"""
        # 創(chuàng)建測試數(shù)據(jù)
        df = pd.DataFrame(self.test_data)
        test_file = tmp_path / "test_data.csv"
        
        # 測試保存
        self.loader.save_data(df, test_file)
        assert test_file.exists()
        
        # 測試加載
        loaded_df = self.loader.load_from_csv(test_file)
        assert loaded_df.shape == df.shape
        assert list(loaded_df.columns) == list(df.columns)

from sales_analyzer import SalesAnalysisApp

# 創(chuàng)建應用實例
app = SalesAnalysisApp()

# 運行分析
app.run_analysis("data/raw/sales_data.csv", "my_analysis")

# 安裝包
pip install -e .

# 運行分析
sales-analyzer data/raw/sales_data.csv -o my_analysis

# 使用配置文件
sales-analyzer data/raw/sales_data.csv --config config.json

# 由于代碼量較大，這里提供的是項目結構的完整實現(xiàn)
# 各個模塊的代碼已在前面各節(jié)中詳細展示

"""
完整的銷售分析項目結構：

sales_analyzer/
├── src/
│   └── sales_analyzer/
│       ├── __init__.py
│       ├── data_loader.py      # 數(shù)據(jù)加載模塊
│       ├── data_processor.py   # 數(shù)據(jù)處理模塊  
│       ├── analyzer.py         # 分析模塊
│       ├── visualizer.py       # 可視化模塊
│       └── main.py            # 主程序入口
├── tests/                      # 測試目錄
├── docs/                       # 文檔目錄
├── data/                       # 數(shù)據(jù)目錄
├── examples/                   # 使用示例
├── requirements.txt            # 依賴列表
├── setup.py                   # 安裝配置
├── pyproject.toml             # 項目配置
└── README.md                  # 項目說明
"""

# 所有模塊的具體實現(xiàn)請參考前面各節(jié)的代碼
# 這里強調項目結構的完整性和各模塊的職責分離