Python AI基础:Matplotlib与Seaborn数据可视化

"""Python AI基础：Matplotlib与Seaborn数据可视化 - 基础实现示例作者：AI教程团队"""import numpy as npimport pandas as pdfrom typing import List, Dict, Optional, Tupleimport warningswarnings.filterwarnings('ignore')class CoreAIModel:    """    AI模型基础类    这是一个展示Python AI基础：Matplotlib与Seaborn数据可视化核心概念的示例类，    包含了数据处理、模型训练、预测评估的完整流程。    """    def __init__(self,                  learning_rate: float = 0.01,                 epochs: int = 100,                 batch_size: int = 32):        """        初始化模型        Args:            learning_rate: 学习率            epochs: 训练轮数            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):        """初始化模型参数"""        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:        """计算损失函数（均方误差）"""        return np.mean((y_true - y_pred) ** 2)    def _backward(self, X: np.ndarray, y_true: np.ndarray, y_pred: np.ndarray):        """反向传播计算梯度"""        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':        """        训练模型        Args:            X: 特征矩阵            y: 目标变量        Returns:            self: 训练后的模型实例        """        # 初始化参数        n_samples, n_features = X.shape        self._initialize_parameters(n_features)        # 训练循环        for epoch in range(self.epochs):            # 小批量训练            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)                # 计算损失                loss = self._compute_loss(y_batch, y_pred)                # 反向传播                dw, db = self._backward(X_batch, y_batch, y_pred)                # 更新参数                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:        """        预测        Args:            X: 特征矩阵        Returns:            预测结果        """        return self._forward(X)    def score(self, X: np.ndarray, y: np.ndarray) -> float:        """        计算R²分数        Args:            X: 特征矩阵            y: 真实值        Returns:            R²分数        """        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__":    # 生成示例数据    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    # 划分训练集和测试集    split = int(0.8 * len(X))    X_train, X_test = X[:split], X[split:]    y_train, y_test = y[:split], y[split:]    # 训练模型    model = CoreAIModel(learning_rate=0.01, epochs=100, batch_size=32)    model.fit(X_train, y_train)    # 评估模型    train_score = model.score(X_train, y_train)    test_score = model.score(X_test, y_test)    print(f"\n训练集R²: {train_score:.4f}")    print(f"测试集R²: {test_score:.4f}")

"""Python AI基础：Matplotlib与Seaborn数据可视化 - 进阶实现示例使用TensorFlow/PyTorch实现"""import tensorflow as tffrom tensorflow import kerasfrom tensorflow.keras import layersimport torchimport torch.nn as nnimport torch.optim as optim# ============== TensorFlow实现 ==============class TensorFlowModel:    """TensorFlow版本的模型实现"""    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):        """        初始化TensorFlow模型        Args:            input_dim: 输入维度            hidden_units: 隐藏层单元数列表        """        self.model = self._build_model(input_dim, hidden_units)    def _build_model(self, input_dim: int, hidden_units: List[int]) -> keras.Model:        """构建模型架构"""        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):        """训练模型"""        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):        """预测"""        return self.model.predict(X)# ============== PyTorch实现 ==============class PyTorchModel(nn.Module):    """PyTorch版本的模型实现"""    def __init__(self, input_dim: int, hidden_units: List[int] = [64, 32]):        """        初始化PyTorch模型        Args:            input_dim: 输入维度            hidden_units: 隐藏层单元数列表        """        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):        """训练模型"""        criterion = nn.MSELoss()        optimizer = optim.Adam(self.parameters(), lr=lr)        train_losses = []        val_losses = []        for epoch in range(epochs):            # 训练阶段            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()            # 验证阶段            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实现 ===")    tf_model = TensorFlowModel(input_dim=5)    # tf_model.train(X_train, y_train, X_val, y_val)    # PyTorch示例    print("\n=== PyTorch实现 ===")    torch_model = PyTorchModel(input_dim=5)    print(torch_model)

"""数据处理完整流程"""import pandas as pdimport numpy as npfrom sklearn.model_selection import train_test_splitfrom sklearn.preprocessing import StandardScaler, LabelEncoderfrom sklearn.impute import SimpleImputerclass DataProcessor:    """数据处理类"""    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:        """        完整的数据处理流程        Args:            data: 原始数据            target_col: 目标列名            categorical_cols: 类别列名列表            test_size: 测试集比例        Returns:            处理后的训练集和测试集        """        # 1. 分离特征和目标        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. 标准化        X_scaled = self.scaler.fit_transform(X)        # 5. 划分数据集        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__":    # 创建示例数据    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_train.shape}")    print(f"测试集形状: {X_test.shape}")

"""模型评估工具"""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 pltimport seaborn as snsclass ModelEvaluator:    """模型评估类"""    @staticmethod    def evaluate_classification(y_true, y_pred, y_prob=None):        """评估分类模型"""        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):        """评估回归模型"""        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('预测值')        plt.ylabel('真实值')        plt.show()    @staticmethod    def plot_learning_curve(train_losses, val_losses):        """绘制学习曲线"""        plt.figure(figsize=(10, 6))        plt.plot(train_losses, label='训练损失')        plt.plot(val_losses, label='验证损失')        plt.xlabel('Epoch')        plt.ylabel('Loss')        plt.title('学习曲线')        plt.legend()        plt.grid(True)        plt.show()# 使用示例if __name__ == "__main__":    # 分类评估示例    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("分类指标:", cls_metrics)    # 回归评估示例    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("回归指标:", reg_metrics)

# 数据分析完整流程示例import pandas as pdimport numpy as npimport matplotlib.pyplot as pltimport seaborn as snsfrom sklearn.ensemble import RandomForestClassifierfrom sklearn.model_selection import cross_val_scoredef analyze_dataset(data_path: str):    """完整的数据分析流程"""    # 1. 加载数据    data = pd.read_csv(data_path)    print("数据形状:", data.shape)    print("\n数据概览:")    print(data.head())    # 2. 数据基本信息    print("\n数据类型:")    print(data.dtypes)    print("\n缺失值统计:")    print(data.isnull().sum())    # 3. 描述性统计    print("\n描述性统计:")    print(data.describe())    # 4. 可视化分析    # 数值特征分布    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('频数')    plt.tight_layout()    plt.show()    # 5. 相关性分析    plt.figure(figsize=(10, 8))    correlation = data[numeric_cols].corr()    sns.heatmap(correlation, annot=True, cmap='coolwarm', center=0)    plt.title('特征相关性热力图')    plt.show()    return data# 使用示例# data = analyze_dataset('your_data.csv')

# 创建虚拟环境conda create -n ai_env python=3.9conda activate ai_env# 安装核心库pip install numpy pandas matplotlib seabornpip install scikit-learn tensorflow torchpip install jupyter notebook# 验证安装python -c "import tensorflow as tf; print(tf.__version__)"python -c "import torch; print(torch.__version__)"

## AI项目标准目录结构project/├── data/               # 数据目录│   ├── raw/           # 原始数据│   ├── processed/     # 处理后数据│   └── external/      # 外部数据├── notebooks/         # Jupyter笔记本│   └── exploration.ipynb├── src/               # 源代码│   ├── data/         # 数据处理│   ├── features/     # 特征工程│   ├── models/       # 模型定义│   └── utils/        # 工具函数├── tests/             # 测试代码├── configs/           # 配置文件├── requirements.txt   # 依赖列表└── README.md          # 项目说明

"""房价预测完整案例"""import pandas as pdimport numpy as npfrom sklearn.model_selection import train_test_split, GridSearchCVfrom sklearn.preprocessing import StandardScaler, OneHotEncoderfrom sklearn.compose import ColumnTransformerfrom sklearn.pipeline import Pipelinefrom sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressorfrom sklearn.metrics import mean_squared_error, r2_scoreimport matplotlib.pyplot as pltclass HousePricePredictor:    """房价预测器"""    def __init__(self):        self.model = None        self.preprocessor = None    def prepare_data(self, data: pd.DataFrame, target_col: str):        """准备数据"""        X = data.drop(columns=[target_col])        y = data[target_col]        # 识别数值和类别特征        numeric_features = X.select_dtypes(include=[np.number]).columns.tolist()        categorical_features = X.select_dtypes(exclude=[np.number]).columns.tolist()        # 创建预处理器        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):        """训练模型"""        # 创建管道        self.model = Pipeline([            ('preprocessor', self.preprocessor),            ('regressor', GradientBoostingRegressor(                n_estimators=200,                learning_rate=0.1,                max_depth=5,                random_state=42            ))        ])        # 训练        self.model.fit(X_train, y_train)        return self    def evaluate(self, X_test, y_test):        """评估模型"""        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):        """绘制预测结果"""        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('真实价格')        plt.ylabel('预测价格')        plt.title('房价预测结果')        plt.show()# 使用示例if __name__ == "__main__":    # 加载数据（示例）    # 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("评估指标:", metrics)    pass

# 处理数据不平衡的方法from imblearn.over_sampling import SMOTEfrom imblearn.under_sampling import RandomUnderSamplerfrom sklearn.utils.class_weight import compute_class_weight# 方法1：过采样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：类别权重class_weights = compute_class_weight('balanced', classes=np.unique(y), y=y)