# 項目1：分析銷售數(shù)據(jù)
df = pd.read_csv("sales.csv")
df = df.dropna()
df["日期"] = pd.to_datetime(df["日期"])
df = df[df["金額"] > 0]
# ... 分析代碼 ...

# 項目2：分析用戶數(shù)據(jù)
df = pd.read_csv("users.csv")
df = df.dropna()
df["注冊時間"] = pd.to_datetime(df["注冊時間"])
df = df[df["年齡"] > 0]
# ... 又是類似的分析代碼 ...

# 項目3：分析產(chǎn)品數(shù)據(jù)
df = pd.read_csv("products.csv")
df = df.dropna()
df["上架日期"] = pd.to_datetime(df["上架日期"])
df = df[df["價格"] > 0]
# ... 又來一遍 ...

# 不用類的方式：數(shù)據(jù)和操作分開
df = pd.read_csv("data.csv")
# 一堆處理代碼...
# 再一堆處理代碼...
# 變量到處飛，很難追蹤

# 用類的方式：數(shù)據(jù)和操作在一起
processor = DataProcessor("data.csv")
processor.load()
processor.clean()
processor.analyze()
# 邏輯清晰，一目了然

import pandas as pd
import numpy as np

class SimpleAnalyzer:
    """一個簡單的數(shù)據(jù)分析器類"""

    # __init__ 是初始化方法，創(chuàng)建對象時自動調(diào)用
    def __init__(self, name="默認分析器"):
        self.name = name           # 屬性：對象的數(shù)據(jù)
        self.data = None           # 屬性：存放數(shù)據(jù)
        self.report = None         # 屬性：存放分析結(jié)果
        print(f"分析器 '{self.name}' 已創(chuàng)建")

    # 方法：對象能做什么
    def load_data(self, filepath):
        """加載數(shù)據(jù)"""
        self.data = pd.read_csv(filepath)
        print(f"已加載 {len(self.data)} 行數(shù)據(jù)")
        return self.data

    def summary(self):
        """生成數(shù)據(jù)摘要"""
        if self.data is None:
            return "請先加載數(shù)據(jù)"
        result = {
            "行數(shù)": len(self.data),
            "列數(shù)": len(self.data.columns),
            "缺失值": int(self.data.isnull().sum().sum()),
            "重復行": int(self.data.duplicated().sum())
        }
        self.report = result
        return result

    def __str__(self):
        """定義打印對象時的顯示"""
        return f"分析器: {self.name}, 數(shù)據(jù): {len(self.data) if self.data is not None else '未加載'}行"


# 使用示例
# analyzer = SimpleAnalyzer("銷售分析器")
# analyzer.load_data("sales.csv")
# print(analyzer.summary())
# print(analyzer)

import logging
from typing import List, Dict, Optional, Union

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S"
)

class DataCleaner:
    """
    通用數(shù)據(jù)清洗器
    功能: 加載、清洗、轉(zhuǎn)換、保存數(shù)據(jù)
    """

    def __init__(self, data: Optional[pd.DataFrame] = None, name: str = "Cleaner"):
        self.name = name
        self.data = data
        self.history = []  # 記錄每一步操作
        self._log = logging.getLogger(self.name)

    def log_action(self, action: str, details: str = ""):
        """記錄操作到歷史日志"""
        entry = {"action": action, "details": details, "rows": len(self.data) if self.data is not None else 0}
        self.history.append(entry)
        self._log.info(f"[{action}] {details}")

    def load_csv(self, filepath: str, **kwargs) -> "DataCleaner":
        """從CSV加載數(shù)據(jù)"""
        try:
            self.data = pd.read_csv(filepath, **kwargs)
            self.log_action("加載", f"從 {filepath} 加載了 {len(self.data)} 行數(shù)據(jù)")
        except FileNotFoundError:
            self._log.error(f"文件不存在: {filepath}")
            raise
        except Exception as e:
            self._log.error(f"加載失敗: {e}")
            raise
        return self  # 返回self支持鏈式調(diào)用

    def load_dict(self, data_dict: Dict) -> "DataCleaner":
        """從字典創(chuàng)建數(shù)據(jù)"""
        self.data = pd.DataFrame(data_dict)
        self.log_action("創(chuàng)建", f"從字典創(chuàng)建了 {len(self.data)} 行數(shù)據(jù)")
        return self

    def remove_duplicates(self, subset: Optional[List[str]] = None) -> "DataCleaner":
        """刪除重復行"""
        before = len(self.data)
        self.data = self.data.drop_duplicates(subset=subset)
        removed = before - len(self.data)
        self.log_action("去重", f"刪除了 {removed} 行重復數(shù)據(jù)")
        return self

    def fill_missing(self, strategy: str = "mean", columns: Optional[List[str]] = None) -> "DataCleaner":
        """
        填充缺失值

        參數(shù):
            strategy: 填充策略 ("mean", "median", "mode", "zero", "forward")
            columns: 指定列，不指定則對所有數(shù)值列操作
        """
        cols = columns or self.data.select_dtypes(include=[np.number]).columns.tolist()
        filled_count = 0

        for col in cols:
            if col not in self.data.columns:
                continue
            missing_before = self.data[col].isnull().sum()

            if strategy == "mean":
                self.data[col] = self.data[col].fillna(self.data[col].mean())
            elif strategy == "median":
                self.data[col] = self.data[col].fillna(self.data[col].median())
            elif strategy == "mode":
                self.data[col] = self.data[col].fillna(self.data[col].mode()[0])
            elif strategy == "zero":
                self.data[col] = self.data[col].fillna(0)
            elif strategy == "forward":
                self.data[col] = self.data[col].ffill()

            filled_count += missing_before

        self.log_action("填充缺失值", f"策略={strategy}, 填充了 {filled_count} 個缺失值")
        return self

    def convert_types(self, type_map: Dict[str, str]) -> "DataCleaner":
        """
        轉(zhuǎn)換列數(shù)據(jù)類型

        參數(shù):
            type_map: {"列名": "目標類型"}，如 {"日期": "datetime", "金額": "float"}
        """
        for col, target_type in type_map.items():
            if col not in self.data.columns:
                self._log.warning(f"列不存在: {col}")
                continue
            try:
                if target_type == "datetime":
                    self.data[col] = pd.to_datetime(self.data[col])
                elif target_type == "int":
                    self.data[col] = pd.to_numeric(self.data[col]).astype("Int64")
                elif target_type == "float":
                    self.data[col] = pd.to_numeric(self.data[col]).astype(float)
                elif target_type == "str":
                    self.data[col] = self.data[col].astype(str)
                elif target_type == "category":
                    self.data[col] = self.data[col].astype("category")
                self.log_action("類型轉(zhuǎn)換", f"{col} -> {target_type}")
            except Exception as e:
                self._log.error(f"類型轉(zhuǎn)換失敗 {col}: {e}")
        return self

    def filter_rows(self, condition) -> "DataCleaner":
        """
        按條件過濾行

        參數(shù):
            condition: 布爾條件，如 df["金額"] > 0
        """
        before = len(self.data)
        self.data = self.data[condition]
        removed = before - len(self.data)
        self.log_action("過濾", f"刪除了 {removed} 行，剩余 {len(self.data)} 行")
        return self

    def rename_columns(self, name_map: Dict[str, str]) -> "DataCleaner":
        """重命名列"""
        self.data = self.data.rename(columns=name_map)
        self.log_action("重命名", f"重命名了 {len(name_map)} 列")
        return self

    def get_summary(self) -> pd.DataFrame:
        """獲取數(shù)據(jù)摘要統(tǒng)計"""
        if self.data is None:
            return pd.DataFrame()
        summary = pd.DataFrame({
            "類型": self.data.dtypes,
            "非空數(shù)": self.data.count(),
            "缺失數(shù)": self.data.isnull().sum(),
            "缺失率": self.data.isnull().sum() / len(self.data),
            "唯一值數(shù)": self.data.nunique()
        })
        # 數(shù)值列額外統(tǒng)計
        num_cols = self.data.select_dtypes(include=[np.number]).columns
        if len(num_cols) > 0:
            stats = self.data[num_cols].describe().T
            stats = stats.rename(columns={
                "mean": "均值", "std": "標準差", "min": "最小值",
                "25%": "25分位", "50%": "中位數(shù)", "75%": "75分位", "max": "最大值"
            })
            summary = summary.join(stats)
        return summary

    def save_csv(self, filepath: str) -> "DataCleaner":
        """保存為CSV"""
        self.data.to_csv(filepath, index=False)
        self.log_action("保存", f"已保存至 {filepath}")
        return self

    def print_history(self):
        """打印操作歷史"""
        print("\n=== 操作歷史 ===")
        for i, entry in enumerate(self.history, 1):
            print(f"  {i}. [{entry['action']}] {entry['details']} (當前: {entry['rows']}行)")

# 模擬數(shù)據(jù)
sample_data = {
    "日期": ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-03"],
    "產(chǎn)品": ["A", "B", None, "A", "A"],
    "銷售額": [100, 200, 150, None, 100],
    "數(shù)量": [10, 20, 15, 8, 10],
    "單價": [10.0, 10.0, 10.0, 12.5, 10.0],
}

# 鏈式調(diào)用演示
cleaner = DataCleaner(name="銷售數(shù)據(jù)清洗")
(
    cleaner
    .load_dict(sample_data)
    .remove_duplicates()
    .fill_missing(strategy="mean", columns=["銷售額"])
    .fill_missing(strategy="mode", columns=["產(chǎn)品"])
    .convert_types({"日期": "datetime"})
    .filter_rows(lambda df: df["銷售額"] > 0)
)

print("\n=== 清洗后的數(shù)據(jù) ===")
print(cleaner.data)
print("\n=== 數(shù)據(jù)摘要 ===")
print(cleaner.get_summary())
cleaner.print_history()

class AnalysisPipeline:
    """
    數(shù)據(jù)分析管道
    將數(shù)據(jù)加載、清洗、分析、報告串聯(lián)起來
    """

    def __init__(self, name="分析管道"):
        self.name = name
        self.steps = []         # 管道步驟列表
        self.data = None        # 當前數(shù)據(jù)
        self.results = {}       # 各步驟的結(jié)果
        self._log = logging.getLogger(name)

    def add_step(self, name: str, func, **kwargs):
        """添加一個處理步驟"""
        self.steps.append({"name": name, "func": func, "kwargs": kwargs})
        self._log.info(f"添加步驟: {name}")
        return self

    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        """運行整個管道"""
        self.data = data.copy()
        self._log.info(f"開始運行管道 '{self.name}', 輸入數(shù)據(jù): {len(self.data)} 行")

        for step in self.steps:
            try:
                self._log.info(f"執(zhí)行步驟: {step['name']}")
                self.data = step["func"](self.data, **step["kwargs"])
                self.results[step["name"]] = {
                    "rows": len(self.data),
                    "columns": len(self.data.columns)
                }
                self._log.info(f"  -> 完成: {len(self.data)} 行, {len(self.data.columns)} 列")
            except Exception as e:
                self._log.error(f"步驟 '{step['name']}' 執(zhí)行失敗: {e}")
                raise

        self._log.info(f"管道運行完成!")
        return self.data

    def get_report(self) -> str:
        """生成管道運行報告"""
        report = [f"\n=== 管道運行報告: {self.name} ==="]
        for name, info in self.results.items():
            report.append(f"  {name}: {info['rows']} 行, {info['columns']} 列")
        return "\n".join(report)


# ========== 使用示例 ==========

# 定義各個步驟的處理函數(shù)
def step_remove_duplicates(df, **kwargs):
    before = len(df)
    df = df.drop_duplicates()
    logging.info(f"  去重: {before} -> {len(df)} 行")
    return df

def step_fill_missing(df, strategy="mean", columns=None, **kwargs):
    cols = columns or df.select_dtypes(include=[np.number]).columns.tolist()
    for col in cols:
        if col in df.columns:
            if strategy == "mean":
                df[col] = df[col].fillna(df[col].mean())
            elif strategy == "median":
                df[col] = df[col].fillna(df[col].median())
    logging.info(f"  填充缺失值: 策略={strategy}")
    return df

def step_convert_dates(df, columns=None, **kwargs):
    cols = columns or []
    for col in cols:
        if col in df.columns:
            df[col] = pd.to_datetime(df[col])
    logging.info(f"  日期轉(zhuǎn)換: {len(cols)} 列")
    return df

def step_add_features(df, **kwargs):
    """添加衍生特征"""
    num_cols = df.select_dtypes(include=[np.number]).columns
    for col in num_cols:
        df[f"{col}_標準化"] = (df[col] - df[col].mean()) / df[col].std()
    logging.info(f"  添加特征: {len(num_cols)} 個標準化列")
    return df

def step_filter_valid(df, **kwargs):
    """過濾有效數(shù)據(jù)"""
    before = len(df)
    # 這里可以根據(jù)業(yè)務(wù)規(guī)則過濾
    df = df.dropna(subset=df.select_dtypes(include=[np.number]).columns[:1])
    logging.info(f"  過濾: {before} -> {len(df)} 行")
    return df

# 創(chuàng)建并運行管道
sample_data = pd.DataFrame({
    "日期": ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-02"],
    "A": [100, 200, None, 200],
    "B": [50, 100, 75, 100],
    "C": ["x", "y", "z", "y"]
})

pipeline = AnalysisPipeline("銷售數(shù)據(jù)處理管道")
(
    pipeline
    .add_step("去重", step_remove_duplicates)
    .add_step("填充缺失值", step_fill_missing, strategy="mean")
    .add_step("日期轉(zhuǎn)換", step_convert_dates, columns=["日期"])
    .add_step("特征工程", step_add_features)
    .add_step("數(shù)據(jù)過濾", step_filter_valid)
)

result = pipeline.run(sample_data)
print(pipeline.get_report())
print("\n最終數(shù)據(jù):")
print(result)

import logging
import os

def setup_logger(name: str, log_file: str = None, level: int = logging.INFO) -> logging.Logger:
    """
    配置一個帶文件輸出的日志器

    參數(shù):
        name: 日志器名稱
        log_file: 日志文件路徑（可選）
        level: 日志級別
    """
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 避免重復添加handler
    if logger.handlers:
        return logger

    # 格式化器
    formatter = logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    # 控制臺輸出
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

    # 文件輸出（如果指定了文件）
    if log_file:
        os.makedirs(os.path.dirname(log_file) if os.path.dirname(log_file) else ".", exist_ok=True)
        file_handler = logging.FileHandler(log_file, encoding="utf-8")
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)
        logger.info(f"日志文件: {log_file}")

    return logger

# 使用示例
logger = setup_logger("數(shù)據(jù)分析", log_file="analysis.log")

logger.debug("這是一條調(diào)試信息（通常不顯示）")
logger.info("數(shù)據(jù)處理開始")
logger.warning("發(fā)現(xiàn)3個缺失值，已填充")
logger.error("文件讀取失敗，使用默認值")

class DataProcessingError(Exception):
    """自定義異常類"""
    def __init__(self, message: str, detail: str = ""):
        super().__init__(message)
        self.detail = detail
        self.message = message

    def __str__(self):
        if self.detail:
            return f"{self.message} - {self.detail}"
        return self.message


def safe_divide(a, b, default=0):
    """安全的除法運算"""
    try:
        return a / b
    except ZeroDivisionError:
        return default
    except TypeError:
        return default


def load_data_safely(filepath: str) -> pd.DataFrame:
    """安全地加載數(shù)據(jù)文件"""
    if not os.path.exists(filepath):
        raise DataProcessingError("文件不存在", f"路徑: {filepath}")

    if not filepath.endswith((".csv", ".xlsx", ".parquet")):
        raise DataProcessingError("不支持的文件格式", f"文件: {filepath}")

    try:
        if filepath.endswith(".csv"):
            df = pd.read_csv(filepath)
        elif filepath.endswith(".xlsx"):
            df = pd.read_excel(filepath)
        elif filepath.endswith(".parquet"):
            df = pd.read_parquet(filepath)
        return df
    except pd.errors.EmptyDataError:
        raise DataProcessingError("文件為空", f"文件: {filepath}")
    except Exception as e:
        raise DataProcessingError("數(shù)據(jù)加載失敗", str(e))

def run_analysis_with_safety(filepath: str):
    """帶完整異常處理的分析流程"""
    logger = setup_logger("安全分析")

    try:
        # 1. 加載數(shù)據(jù)
        logger.info("正在加載數(shù)據(jù)...")
        df = load_data_safely(filepath)
        logger.info(f"成功加載 {len(df)} 行數(shù)據(jù)")

        # 2. 數(shù)據(jù)驗證
        required_cols = ["日期", "銷售額"]
        missing = [col for col in required_cols if col not in df.columns]
        if missing:
            raise DataProcessingError("缺少必要列", f"缺失: {missing}")

        # 3. 數(shù)據(jù)清洗
        logger.info("正在清洗數(shù)據(jù)...")
        df["銷售額"] = pd.to_numeric(df["銷售額"], errors="coerce")
        n_missing = df["銷售額"].isnull().sum()
        if n_missing > len(df) * 0.5:
            logger.warning(f"超過50%的銷售額缺失({n_missing}行)，分析結(jié)果可能不可靠")

        df = df.dropna(subset=["銷售額"])
        df = df[df["銷售額"] > 0]
        logger.info(f"清洗后剩余 {len(df)} 行有效數(shù)據(jù)")

        # 4. 分析計算
        logger.info("正在計算統(tǒng)計指標...")
        summary = {
            "總銷售額": df["銷售額"].sum(),
            "平均銷售額": df["銷售額"].mean(),
            "最大銷售額": df["銷售額"].max(),
            "記錄數(shù)": len(df)
        }

        logger.info("分析完成!")
        for key, value in summary.items():
            logger.info(f"  {key}: {value}")

        return summary

    except DataProcessingError as e:
        logger.error(f"數(shù)據(jù)處理錯誤: {e}")
        return None
    except Exception as e:
        logger.error(f"未知錯誤: {e}")
        return None
    finally:
        logger.info("分析流程結(jié)束")  # finally總是執(zhí)行

# 不好的做法：硬編碼
db_host = "192.168.1.100"    # 換環(huán)境就要改代碼
db_port = 5432
output_dir = "/data/reports"
threshold = 0.85

# 好的做法：配置文件
# config.yaml
# database:
#   host: "192.168.1.100"
#   port: 5432
# paths:
#   output: "/data/reports"
# analysis:
#   threshold: 0.85

import json
import os

class ConfigManager:
    """配置文件管理器"""

    def __init__(self, config_path: str = "config.json"):
        self.config_path = config_path
        self.config = {}
        self.load()

    def load(self):
        """加載配置文件"""
        if os.path.exists(self.config_path):
            with open(self.config_path, "r", encoding="utf-8") as f:
                self.config = json.load(f)
            logging.info(f"已加載配置: {self.config_path}")
        else:
            logging.warning(f"配置文件不存在: {self.config_path}，使用默認配置")
            self.config = self._get_defaults()
            self.save()

    def _get_defaults(self):
        """默認配置"""
        return {
            "data": {
                "source_dir": "./data",
                "output_dir": "./output",
                "file_format": "csv"
            },
            "analysis": {
                "outlier_threshold": 3.0,
                "missing_fill_strategy": "mean",
                "date_columns": ["日期", "時間", "timestamp"]
            },
            "report": {
                "title": "數(shù)據(jù)分析報告",
                "author": "數(shù)據(jù)分析系統(tǒng)",
                "format": "html"
            },
            "logging": {
                "level": "INFO",
                "file": "analysis.log"
            }
        }

    def get(self, key_path: str, default=None):
        """
        獲取配置值（支持嵌套路徑）
        例: config.get("data.source_dir")
        """
        keys = key_path.split(".")
        value = self.config
        for key in keys:
            if isinstance(value, dict) and key in value:
                value = value[key]
            else:
                return default
        return value

    def set(self, key_path: str, value):
        """設(shè)置配置值"""
        keys = key_path.split(".")
        config = self.config
        for key in keys[:-1]:
            if key not in config:
                config[key] = {}
            config = config[key]
        config[keys[-1]] = value

    def save(self):
        """保存配置到文件"""
        with open(self.config_path, "w", encoding="utf-8") as f:
            json.dump(self.config, f, indent=2, ensure_ascii=False)
        logging.info(f"配置已保存: {self.config_path}")


# 使用示例
# config = ConfigManager()
# print(config.get("data.source_dir"))           # 獲取值
# config.set("analysis.outlier_threshold", 2.5)  # 設(shè)置值
# config.save()                                   # 保存

# ========== 綜合實戰(zhàn)：模塊化數(shù)據(jù)處理系統(tǒng) ==========

import os
import json
from datetime import datetime

class DataAnalysisProject:
    """
    模塊化的數(shù)據(jù)分析項目
    整合了清洗器、管道、日志、配置等功能
    """

    def __init__(self, project_name: str, config_path: str = None):
        self.project_name = project_name
        self.start_time = datetime.now()
        self.logger = setup_logger(project_name, log_file=f"{project_name}.log")

        # 加載配置
        self.config = ConfigManager(config_path) if config_path else ConfigManager()

        self.logger.info(f"項目 '{project_name}' 已啟動")

    def run(self, data: pd.DataFrame) -> Dict:
        """運行完整分析流程"""
        self.logger.info("=" * 50)
        self.logger.info("開始執(zhí)行分析流程")
        self.logger.info("=" * 50)

        try:
            # Step 1: 數(shù)據(jù)概覽
            self.logger.info("Step 1: 數(shù)據(jù)概覽")
            overview = self._data_overview(data)

            # Step 2: 數(shù)據(jù)清洗
            self.logger.info("Step 2: 數(shù)據(jù)清洗")
            clean_data = self._clean_data(data)

            # Step 3: 數(shù)據(jù)分析
            self.logger.info("Step 3: 數(shù)據(jù)分析")
            analysis_results = self._analyze_data(clean_data)

            # Step 4: 生成報告
            self.logger.info("Step 4: 生成報告")
            report = self._generate_report(overview, analysis_results)

            # Step 5: 輸出結(jié)果
            self.logger.info("Step 5: 輸出結(jié)果")
            output_dir = self.config.get("data.output_dir", "./output")
            os.makedirs(output_dir, exist_ok=True)
            clean_data.to_csv(f"{output_dir}/cleaned_data.csv", index=False)

            with open(f"{output_dir}/analysis_report.json", "w", encoding="utf-8") as f:
                json.dump(report, f, indent=2, ensure_ascii=False, default=str)

            elapsed = (datetime.now() - self.start_time).total_seconds()
            self.logger.info(f"項目完成! 耗時: {elapsed:.1f}秒")

            return report

        except Exception as e:
            self.logger.error(f"分析流程失敗: {e}")
            raise

    def _data_overview(self, df: pd.DataFrame) -> Dict:
        """數(shù)據(jù)概覽"""
        return {
            "rows": len(df),
            "columns": len(df.columns),
            "column_names": list(df.columns),
            "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
            "missing_values": int(df.isnull().sum().sum()),
            "duplicate_rows": int(df.duplicated().sum())
        }

    def _clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """數(shù)據(jù)清洗"""
        cleaner = DataCleaner(name=f"{self.project_name}_清洗")
        cleaner.data = df.copy()

        # 去重
        cleaner.remove_duplicates()

        # 填充缺失值
        strategy = self.config.get("analysis.missing_fill_strategy", "mean")
        cleaner.fill_missing(strategy=strategy)

        # 轉(zhuǎn)換日期列
        date_cols = self.config.get("analysis.date_columns", [])
        type_map = {col: "datetime" for col in date_cols if col in df.columns}
        if type_map:
            cleaner.convert_types(type_map)

        return cleaner.data

    def _analyze_data(self, df: pd.DataFrame) -> Dict:
        """數(shù)據(jù)分析"""
        num_cols = df.select_dtypes(include=[np.number]).columns.tolist()
        results = {}

        for col in num_cols:
            results[col] = {
                "count": int(df[col].count()),
                "mean": round(float(df[col].mean()), 2),
                "std": round(float(df[col].std()), 2),
                "min": round(float(df[col].min()), 2),
                "max": round(float(df[col].max()), 2),
                "median": round(float(df[col].median()), 2)
            }

        return results

    def _generate_report(self, overview: Dict, analysis: Dict) -> Dict:
        """生成報告"""
        report = {
            "project": self.project_name,
            "timestamp": datetime.now().isoformat(),
            "data_overview": overview,
            "statistics": analysis,
            "title": self.config.get("report.title", "分析報告"),
            "author": self.config.get("report.author", "數(shù)據(jù)分析系統(tǒng)")
        }
        return report


# ========== 使用示例 ==========

# 創(chuàng)建模擬數(shù)據(jù)
sample_df = pd.DataFrame({
    "日期": ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05",
             "2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05"],
    "銷售額": [100, 200, 150, None, 180, 220, 190, 210, 170, 230],
    "利潤": [30, 60, 45, 50, 55, 70, 58, 65, 52, 75],
    "客戶數(shù)": [10, 25, 18, 20, 22, 30, 24, 28, 21, 32],
    "產(chǎn)品": ["A", "B", "A", "B", "A", "B", "A", "B", "A", "B"]
})

# 運行項目
project = DataAnalysisProject("銷售分析項目")
# report = project.run(sample_df)

# 參考答案

class EnhancedCleaner(DataCleaner):
    """增強版數(shù)據(jù)清洗器"""

    def normalize_columns(self, columns: Optional[List[str]] = None) -> "EnhancedCleaner":
        """最大-最小歸一化"""
        cols = columns or self.data.select_dtypes(include=[np.number]).columns.tolist()
        for col in cols:
            if col in self.data.columns:
                min_val = self.data[col].min()
                max_val = self.data[col].max()
                if max_val > min_val:
                    self.data[col] = (self.data[col] - min_val) / (max_val - min_val)
        self.log_action("歸一化", f"歸一化了 {len(cols)} 列")
        return self

    def outlier_removal(self, columns: Optional[List[str]] = None) -> "EnhancedCleaner":
        """使用IQR方法移除異常值"""
        cols = columns or self.data.select_dtypes(include=[np.number]).columns.tolist()
        before = len(self.data)

        for col in cols:
            if col in self.data.columns:
                q1 = self.data[col].quantile(0.25)
                q3 = self.data[col].quantile(0.75)
                iqr = q3 - q1
                lower = q1 - 1.5 * iqr
                upper = q3 + 1.5 * iqr
                self.data = self.data[(self.data[col] >= lower) & (self.data[col] <= upper)]

        removed = before - len(self.data)
        self.log_action("異常值移除", f"使用IQR方法移除了 {removed} 行")
        return self

    def export_report(self, filepath: str) -> str:
        """導出清洗報告為JSON"""
        report = {
            "name": self.name,
            "final_rows": len(self.data),
            "columns": list(self.data.columns),
            "dtypes": {col: str(dtype) for col, dtype in self.data.dtypes.items()},
            "missing_values": {col: int(self.data[col].isnull().sum()) for col in self.data.columns},
            "history": self.history
        }
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(report, f, indent=2, ensure_ascii=False, default=str)
        self.log_action("導出報告", f"已導出至 {filepath}")
        return filepath