"""
AI實(shí)戰(zhàn)項(xiàng)目：垃圾郵件識(shí)別 - 基礎(chǔ)實(shí)現(xiàn)示例
作者：AI教程團(tuán)隊(duì)
"""

import numpy as np
import pandas as pd
from typing import List, Dict, Optional, Tuple
import warnings
warnings.filterwarnings('ignore')


class CoreAIModel:
    """
    AI模型基礎(chǔ)類
    
    這是一個(gè)展示AI實(shí)戰(zhàn)項(xiàng)目：垃圾郵件識(shí)別核心概念的示例類，
    包含了數(shù)據(jù)處理、模型訓(xùn)練、預(yù)測(cè)評(píng)估的完整流程。
    """
    
    def __init__(self, 
                 learning_rate: float = 0.01,
                 epochs: int = 100,
                 batch_size: int = 32):
        """
        初始化模型
        
        Args:
            learning_rate: 學(xué)習(xí)率
            epochs: 訓(xùn)練輪數(shù)
            batch_size: 批量大小
        """
        self.learning_rate = learning_rate
        self.epochs = epochs
        self.batch_size = batch_size
        self.weights = None
        self.bias = None
        self.loss_history = []
        
    def _initialize_parameters(self, n_features: int):
        """初始化模型參數(shù)"""
        np.random.seed(42)
        self.weights = np.random.randn(n_features) * 0.01
        self.bias = 0.0
        
    def _forward(self, X: np.ndarray) -> np.ndarray:
        """前向傳播"""
        return np.dot(X, self.weights) + self.bias
    
    def _compute_loss(self, y_true: np.ndarray, y_pred: np.ndarray) -> float:
        """計(jì)算損失函數(shù)（均方誤差）"""
        return np.mean((y_true - y_pred) ** 2)
    
    def _backward(self, X: np.ndarray, y_true: np.ndarray, y_pred: np.ndarray):
        """反向傳播計(jì)算梯度"""
        m = len(y_true)
        dw = -2/m * np.dot(X.T, (y_true - y_pred))
        db = -2/m * np.sum(y_true - y_pred)
        return dw, db
    
    def fit(self, X: np.ndarray, y: np.ndarray) -> 'CoreAIModel':
        """
        訓(xùn)練模型
        
        Args:
            X: 特征矩陣
            y: 目標(biāo)變量
            
        Returns:
            self: 訓(xùn)練后的模型實(shí)例
        """
        # 初始化參數(shù)
        n_samples, n_features = X.shape
        self._initialize_parameters(n_features)
        
        # 訓(xùn)練循環(huán)
        for epoch in range(self.epochs):
            # 小批量訓(xùn)練
            indices = np.random.permutation(n_samples)
            X_shuffled = X[indices]
            y_shuffled = y[indices]
            
            for i in range(0, n_samples, self.batch_size):
                X_batch = X_shuffled[i:i+self.batch_size]
                y_batch = y_shuffled[i:i+self.batch_size]
                
                # ?前向傳播
                y_pred = self._forward(X_batch)
                
                # 計(jì)算損失
                loss = self._compute_loss(y_batch, y_pred)
                
                # 反向傳播
                dw, db = self._backward(X_batch, y_batch, y_pred)
                
                # 更新參數(shù)
                self.weights -= self.learning_rate * dw
                self.bias -= self.learning_rate * db
            
            # 記錄損失
            if (epoch + 1) % 10 == 0:
                y_pred_full = self._forward(X)
                loss = self._compute_loss(y, y_pred_full)
                self.loss_history.append(loss)
                print(f"Epoch {epoch+1}/{self.epochs}, Loss: {loss:.4f}")
        
        return self
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """
        預(yù)測(cè)
        
        Args:
            X: 特征矩陣
            
        Returns:
            預(yù)測(cè)結(jié)果
        """
        return self._forward(X)
    
    def score(self, X: np.ndarray, y: np.ndarray) -> float:
        """
        計(jì)算R2分?jǐn)?shù)
        
        Args:
            X: 特征矩陣
            y: 真實(shí)值
            
        Returns:
            R2分?jǐn)?shù)
        """
        y_pred = self.predict(X)
        ss_res = np.sum((y - y_pred) ** 2)
        ss_tot = np.sum((y - np.mean(y)) ** 2)
        return 1 - (ss_res / ss_tot)


# 使用示例
if __name__ == "__main__":
    # 生成示例數(shù)據(jù)
    np.random.seed(42)
    X = np.random.randn(1000, 5)
    true_weights = np.array([1.5, -2.0, 0.5, 1.0, -0.5])
    y = np.dot(X, true_weights) + np.random.randn(1000) * 0.1
    
    # 劃分訓(xùn)練集和測(cè)試集
    split = int(0.8 * len(X))
    X_train, X_test = X[:split], X[split:]
    y_train, y_test = y[:split], y[split:]
    
    # 訓(xùn)練模型
    model = CoreAIModel(learning_rate=0.01, epochs=100, batch_size=32)
    model.fit(X_train, y_train)
    
    # 評(píng)估模型
    train_score = model.score(X_train, y_train)
    test_score = model.score(X_test, y_test)
    print(f"\n訓(xùn)練集R2: {train_score:.4f}")
    print(f"測(cè)試集R2: {test_score:.4f}")

"""
AI實(shí)戰(zhàn)項(xiàng)目：垃圾郵件識(shí)別 - 進(jìn)階實(shí)現(xiàn)示例
使用TensorFlow/PyTorch實(shí)現(xiàn)
"""

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import torch
import torch.nn as nn
import torch.optim as optim


# ============== TensorFlow實(shí)現(xiàn) ==============
class TensorFlowModel:
    """TensorFlow版本的模型實(shí)現(xiàn)"""
    
    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):
        """
        初始化TensorFlow模型
        
        Args:
            input_dim: 輸入維度
            hidden_units: 隱藏層單元數(shù)列表
        """
        self.model = self._build_model(input_dim, hidden_units)
        
    def _build_model(self, input_dim: int, hidden_units: List[int]) -> keras.Model:
        """構(gòu)建模型架構(gòu)"""
        inputs = keras.Input(shape=(input_dim,))
        x = inputs
        
        for units in hidden_units:
            x = layers.Dense(units, activation='relu')(x)
            x = layers.BatchNormalization()(x)
            x = layers.Dropout(0.2)(x)
        
        outputs = layers.Dense(1)(x)
        
        model = keras.Model(inputs=inputs, outputs=outputs)
        model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.001),
            loss='mse',
            metrics=['mae']
        )
        
        return model
    
    def train(self, X_train, y_train, X_val, y_val, epochs=100, batch_size=32):
        """訓(xùn)練模型"""
        history = self.model.fit(
            X_train, y_train,
            validation_data=(X_val, y_val),
            epochs=epochs,
            batch_size=batch_size,
            verbose=1
        )
        return history
    
    def predict(self, X):
        """預(yù)測(cè)"""
        return self.model.predict(X)


# ============== PyTorch實(shí)現(xiàn) ==============
class PyTorchModel(nn.Module):
    """PyTorch版本的模型實(shí)現(xiàn)"""
    
    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):
        """
        初始化PyTorch模型
        
        Args:
            input_dim: 輸入維度
            hidden_units: 隱藏層單元數(shù)列表
        """
        super(PyTorchModel, self).__init__()
        
        layers_list = []
        prev_units = input_dim
        
        for units in hidden_units:
            layers_list.append(nn.Linear(prev_units, units))
            layers_list.append(nn.ReLU())
            layers_list.append(nn.BatchNorm1d(units))
            layers_list.append(nn.Dropout(0.2))
            prev_units = units
        
        layers_list.append(nn.Linear(prev_units, 1))
        
        self.network = nn.Sequential(*layers_list)
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """前向傳播"""
        return self.network(x)
    
    def train_model(self, train_loader, val_loader, epochs=100, lr=0.001):
        """訓(xùn)練模型"""
        criterion = nn.MSELoss()
        optimizer = optim.Adam(self.parameters(), lr=lr)
        
        train_losses = []
        val_losses = []
        
        for epoch in range(epochs):
            # 訓(xùn)練階段
            self.train()
            train_loss = 0.0
            for X_batch, y_batch in train_loader:
                optimizer.zero_grad()
                outputs = self(X_batch)
                loss = criterion(outputs, y_batch)
                loss.backward()
                optimizer.step()
                train_loss += loss.item()
            
            # 驗(yàn)證階段
            self.eval()
            val_loss = 0.0
            with torch.no_grad():
                for X_batch, y_batch in val_loader:
                    outputs = self(X_batch)
                    loss = criterion(outputs, y_batch)
                    val_loss += loss.item()
            
            train_losses.append(train_loss / len(train_loader))
            val_losses.append(val_loss / len(val_loader))
            
            if (epoch + 1) % 10 == 0:
                print(f"Epoch {epoch+1}/{epochs}, "
                      f"Train Loss: {train_losses[-1]:.4f}, "
                      f"Val Loss: {val_losses[-1]:.4f}")
        
        return train_losses, val_losses


# 使用示例
if __name__ == "__main__":
    # TensorFlow示例
    print("=== TensorFlow實(shí)現(xiàn) ===")
    tf_model = TensorFlowModel(input_dim=5)
    # tf_model.train(X_train, y_train, X_val, y_val)
    
    # PyTorch示例
    print("\n=== PyTorch實(shí)現(xiàn) ===")
    torch_model = PyTorchModel(input_dim=5)
    print(torch_model)

"""
數(shù)據(jù)處理完整流程
"""

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.impute import SimpleImputer


class DataProcessor:
    """數(shù)據(jù)處理類"""
    
    def __init__(self):
        self.scaler = StandardScaler()
        self.label_encoders = {}
        self.imputer = SimpleImputer(strategy='mean')
        
    def process(self, 
                data: pd.DataFrame,
                target_col: str,
                categorical_cols: List[str] = None,
                test_size: float = 0.2) -> Tuple:
        """
        完整的數(shù)據(jù)處理流程
        
        Args:
            data: 原始數(shù)據(jù)
            target_col: 目標(biāo)列名
            categorical_cols: 類別列名列表
            test_size: 測(cè)試集比例
            
        Returns:
            處理后的訓(xùn)練集和測(cè)試集
        """
        # 1. 分離特征和目標(biāo)
        X = data.drop(columns=[target_col])
        y = data[target_col]
        
        # 2. 處理缺失值
        X = pd.DataFrame(
            self.imputer.fit_transform(X.select_dtypes(include=[np.number])),
            columns=X.select_dtypes(include=[np.number]).columns
        )
        
        # 3. 編碼類別特征
        if categorical_cols:
            for col in categorical_cols:
                if col in X.columns:
                    le = LabelEncoder()
                    X[col] = le.fit_transform(X[col].astype(str))
                    self.label_encoders[col] = le
        
        # 4. 標(biāo)準(zhǔn)化
        X_scaled = self.scaler.fit_transform(X)
        
        # 5. 劃分?jǐn)?shù)據(jù)集
        X_train, X_test, y_train, y_test = train_test_split(
            X_scaled, y, test_size=test_size, random_state=42
        )
        
        return X_train, X_test, y_train, y_test


# 使用示例
if __name__ == "__main__":
    # 創(chuàng)建示例數(shù)據(jù)
    data = pd.DataFrame({
        'feature1': np.random.randn(1000),
        'feature2': np.random.randn(1000),
        'feature3': np.random.choice(['A', 'B', 'C'], 1000),
        'target': np.random.randn(1000)
    })
    
    processor = DataProcessor()
    X_train, X_test, y_train, y_test = processor.process(
        data, target_col='target', categorical_cols=['feature3']
    )
    
    print(f"訓(xùn)練集形狀: {X_train.shape}")
    print(f"測(cè)試集形狀: {X_test.shape}")

"""
模型評(píng)估工具
"""

from sklearn.metrics import (
    accuracy_score, precision_score, recall_score, f1_score,
    roc_auc_score, confusion_matrix, classification_report,
    mean_squared_error, mean_absolute_error, r2_score
)
import matplotlib.pyplot as plt
import seaborn as sns


class ModelEvaluator:
    """模型評(píng)估類"""
    
    @staticmethod
    def evaluate_classification(y_true, y_pred, y_prob=None):
        """評(píng)估分類模型"""
        metrics = {
            'accuracy': accuracy_score(y_true, y_pred),
            'precision': precision_score(y_true, y_pred, average='weighted'),
            'recall': recall_score(y_true, y_pred, average='weighted'),
            'f1': f1_score(y_true, y_pred, average='weighted')
        }
        
        if y_prob is not None:
            metrics['roc_auc'] = roc_auc_score(y_true, y_prob, multi_class='ovr')
        
        return metrics
    
    @staticmethod
    def evaluate_regression(y_true, y_pred):
        """評(píng)估回歸模型"""
        return {
            'mse': mean_squared_error(y_true, y_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)
        }
    
    @staticmethod
    def plot_confusion_matrix(y_true, y_pred, labels=None):
        """繪制混淆矩陣"""
        cm = confusion_matrix(y_true, y_pred)
        plt.figure(figsize=(8, 6))
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
                    xticklabels=labels, yticklabels=labels)
        plt.title('混淆矩陣')
        plt.xlabel('預(yù)測(cè)值')
        plt.ylabel('真實(shí)值')
        plt.show()
    
    @staticmethod
    def plot_learning_curve(train_losses, val_losses):
        """繪制學(xué)習(xí)曲線"""
        plt.figure(figsize=(10, 6))
        plt.plot(train_losses, label='訓(xùn)練損失')
        plt.plot(val_losses, label='驗(yàn)證損失')
        plt.xlabel('Epoch')
        plt.ylabel('Loss')
        plt.title('學(xué)習(xí)曲線')
        plt.legend()
        plt.grid(True)
        plt.show()


# 使用示例
if __name__ == "__main__":
    # 分類評(píng)估示例
    y_true_cls = [0, 1, 0, 1, 0, 1, 0, 0, 1, 1]
    y_pred_cls = [0, 1, 0, 0, 0, 1, 1, 0, 1, 1]
    
    cls_metrics = ModelEvaluator.evaluate_classification(y_true_cls, y_pred_cls)
    print("分類指標(biāo):", cls_metrics)
    
    # 回歸評(píng)估示例
    y_true_reg = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
    y_pred_reg = np.array([1.1, 1.9, 3.2, 3.8, 5.1])
    
    reg_metrics = ModelEvaluator.evaluate_regression(y_true_reg, y_pred_reg)
    print("回歸指標(biāo):", reg_metrics)

# 數(shù)據(jù)分析完整流程示例
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score


def analyze_dataset(data_path: str):
    """完整的數(shù)據(jù)分析流程"""
    
    # 1. 加載數(shù)據(jù)
    data = pd.read_csv(data_path)
    print("數(shù)據(jù)形狀:", data.shape)
    print("\n數(shù)據(jù)概覽:")
    print(data.head())
    
    # 2. 數(shù)據(jù)基本信息
    print("\n數(shù)據(jù)類型:")
    print(data.dtypes)
    
    print("\n缺失值統(tǒng)計(jì):")
    print(data.isnull().sum())
    
    # 3. 描述性統(tǒng)計(jì)
    print("\n描述性統(tǒng)計(jì):")
    print(data.describe())
    
    # 4. 可視化分析
    # 數(shù)值特征分布
    numeric_cols = data.select_dtypes(include=[np.number]).columns
    fig, axes = plt.subplots(2, 2, figsize=(12, 10))
    
    for i, col in enumerate(numeric_cols[:4]):
        ax = axes[i//2, i%2]
        data[col].hist(ax=ax, bins=30, edgecolor='black')
        ax.set_title(f'{col}分布')
        ax.set_xlabel(col)
        ax.set_ylabel('頻數(shù)')
    
    plt.tight_layout()
    plt.show()
    
    # 5. 相關(guān)性分析
    plt.figure(figsize=(10, 8))
    correlation = data[numeric_cols].corr()
    sns.heatmap(correlation, annot=True, cmap='coolwarm', center=0)
    plt.title('特征相關(guān)性熱力圖')
    plt.show()
    
    return data


# 使用示例
# data = analyze_dataset('your_data.csv')

# 創(chuàng)建虛擬環(huán)境
conda create -n ai_env python=3.9
conda activate ai_env

# 安裝核心庫(kù)
pip install numpy pandas matplotlib seaborn
pip install scikit-learn tensorflow torch
pip install jupyter notebook

# 驗(yàn)證安裝
python -c "import tensorflow as tf; print(tf.__version__)"
python -c "import torch; print(torch.__version__)"

"""
房?jī)r(jià)預(yù)測(cè)完整案例
"""

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.metrics import mean_squared_error, r2_score
import matplotlib.pyplot as plt


class HousePricePredictor:
    """房?jī)r(jià)預(yù)測(cè)器"""
    
    def __init__(self):
        self.model = None
        self.preprocessor = None
        
    def prepare_data(self, data: pd.DataFrame, target_col: str):
        """準(zhǔn)備數(shù)據(jù)"""
        X = data.drop(columns=[target_col])
        y = data[target_col]
        
        # 識(shí)別數(shù)值和類別特征
        numeric_features = X.select_dtypes(include=[np.number]).columns.tolist()
        categorical_features = X.select_dtypes(exclude=[np.number]).columns.tolist()
        
        # 創(chuàng)建預(yù)處理器
        self.preprocessor = ColumnTransformer(
            transformers=[
                ('num', StandardScaler(), numeric_features),
                ('cat', OneHotEncoder(handle_unknown='ignore'), categorical_features)
            ]
        )
        
        return train_test_split(X, y, test_size=0.2, random_state=42)
    
    def train(self, X_train, y_train):
        """訓(xùn)練模型"""
        # 創(chuàng)建管道
        self.model = Pipeline([
            ('preprocessor', self.preprocessor),
            ('regressor', GradientBoostingRegressor(
                n_estimators=200,
                learning_rate=0.1,
                max_depth=5,
                random_state=42
            ))
        ])
        
        # 訓(xùn)練
        self.model.fit(X_train, y_train)
        
        return self
    
    def evaluate(self, X_test, y_test):
        """評(píng)估模型"""
        y_pred = self.model.predict(X_test)
        
        metrics = {
            'RMSE': np.sqrt(mean_squared_error(y_test, y_pred)),
            'MAE': mean_absolute_error(y_test, y_pred),
            'R2': r2_score(y_test, y_pred)
        }
        
        return metrics, y_pred
    
    def plot_predictions(self, y_test, y_pred):
        """繪制預(yù)測(cè)結(jié)果"""
        plt.figure(figsize=(10, 6))
        plt.scatter(y_test, y_pred, alpha=0.5)
        plt.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()], 'r--')
        plt.xlabel('真實(shí)價(jià)格')
        plt.ylabel('預(yù)測(cè)價(jià)格')
        plt.title('房?jī)r(jià)預(yù)測(cè)結(jié)果')
        plt.show()


# 使用示例
if __name__ == "__main__":
    # 加載數(shù)據(jù)（示例）
    # data = pd.read_csv('house_prices.csv')
    # predictor = HousePricePredictor()
    # X_train, X_test, y_train, y_test = predictor.prepare_data(data, 'price')
    # predictor.train(X_train, y_train)
    # metrics, y_pred = predictor.evaluate(X_test, y_test)
    # print("評(píng)估指標(biāo):", metrics)
    pass

# 處理數(shù)據(jù)不平衡的方法
from imblearn.over_sampling import SMOTE
from imblearn.under_sampling import RandomUnderSampler
from sklearn.utils.class_weight import compute_class_weight

# 方法1：過(guò)采樣
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X, y)

# 方法2：欠采樣
undersampler = RandomUnderSampler(random_state=42)
X_resampled, y_resampled = undersampler.fit_resample(X, y)

# 方法3：類別權(quán)重
class_weights = compute_class_weight('balanced', classes=np.unique(y), y=y)