{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.6.6","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":3004,"databundleVersionId":861823,"sourceType":"competition"}],"dockerImageVersionId":25114,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import numpy as np\nimport pandas as pd\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport torch.optim as optim\nfrom torch.utils.data import TensorDataset, DataLoader\nfrom sklearn.model_selection import train_test_split\nfrom torchvision import transforms\nfrom PIL import Image\n\n# 检查GPU可用性并设置设备\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(f\"Using device: {device}\")\n\n# 1. 数据加载与预处理\ndef load_data():\n    # 加载原始数据\n    train = pd.read_csv(\"../input/train.csv\", dtype=np.float32)\n    final_test = pd.read_csv(\"../input/test.csv\", dtype=np.float32)\n    \n    # 分离特征和标签\n    targets_np = train.label.values\n    features_np = train.loc[:, train.columns != 'label'].values/255  # 归一化\n    \n    # 划分训练集和验证集\n    features_train, features_test, target_train, target_test = train_test_split(\n        features_np, targets_np, test_size=0.2, random_state=42)\n    \n    # 转换为PyTorch张量\n    featuresTrain = torch.from_numpy(features_train)\n    targetsTrain = torch.from_numpy(target_train).type(torch.LongTensor)\n    featuresTest = torch.from_numpy(features_test)\n    targetsTest = torch.from_numpy(target_test).type(torch.LongTensor)\n    \n    return featuresTrain, targetsTrain, featuresTest, targetsTest\n\n# 2. 数据增强类\nclass MNISTTransform:\n    def __init__(self):\n        self.transform = transforms.Compose([\n            lambda x: Image.fromarray(x.view(28, 28).numpy()),\n            transforms.RandomRotation(10),\n            transforms.RandomAffine(degrees=0, translate=(0.1, 0.1)),\n            transforms.ToTensor(),\n            transforms.Normalize((0.5,), (0.5,))\n        ])\n    \n    def __call__(self, x):\n        return self.transform(x)\n\n# 3. 网络模型\nclass ImprovedClassifier(nn.Module):\n    def __init__(self):\n        super().__init__()\n        # 增强的CNN结构\n        self.conv_block1 = nn.Sequential(\n            nn.Conv2d(1, 32, 3, padding=1),\n            nn.BatchNorm2d(32),\n            nn.ReLU(),\n            nn.Conv2d(32, 32, 3, padding=1),\n            nn.BatchNorm2d(32),\n            nn.ReLU(),\n            nn.MaxPool2d(2),\n            nn.Dropout(0.25)\n        )\n        self.conv_block2 = nn.Sequential(\n            nn.Conv2d(32, 64, 3, padding=1),\n            nn.BatchNorm2d(64),\n            nn.ReLU(),\n            nn.Conv2d(64, 64, 3, padding=1),\n            nn.BatchNorm2d(64),\n            nn.ReLU(),\n            nn.MaxPool2d(2),\n            nn.Dropout(0.25)\n        )\n        self.fc_block = nn.Sequential(\n            nn.Linear(64*7*7, 512),\n            nn.BatchNorm1d(512),\n            nn.ReLU(),\n            nn.Dropout(0.5),\n            nn.Linear(512, 256),\n            nn.BatchNorm1d(256),\n            nn.ReLU(),\n            nn.Dropout(0.5),\n            nn.Linear(256, 10)\n        )\n\n    def forward(self, x):\n        x = x.view(-1, 1, 28, 28)\n        x = self.conv_block1(x)\n        x = self.conv_block2(x)\n        x = x.view(x.size(0), -1)\n        return self.fc_block(x)\n\n# 4. 训练函数\ndef train_model():\n    # 加载数据\n    featuresTrain, targetsTrain, featuresTest, targetsTest = load_data()\n    \n    # 创建数据集\n    train_dataset = TensorDataset(featuresTrain, targetsTrain)\n    test_dataset = TensorDataset(featuresTest, targetsTest)\n    \n    # 数据加载器\n    batch_size = 256\n    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, pin_memory=True)\n    test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, pin_memory=True)\n    \n    # 初始化模型并移至GPU\n    model = ImprovedClassifier().to(device)\n    \n    # 损失函数\n    criterion = nn.CrossEntropyLoss()\n\n    \n    # 使用SGD优化器\n    optimizer = optim.SGD(model.parameters(), \n                         lr=0.01,         # 初始学习率较大，适合SGD\n                         momentum=0.9,     # 添加动量\n                          nesterov=True,   # 启用Nesterov加速\n                         weight_decay=0.01)  # L2正则化\n    \n    \n    \n    # 数据增强\n    transform = MNISTTransform()\n    \n    epochs = 30  # SGD通常需要更多epoch\n    train_losses, test_losses = [], []\n\n    # 学习率调度器（更激进的学习率衰减）\n    scheduler = optim.lr_scheduler.CosineAnnealingLR(\n        optimizer, T_max=epochs, eta_min=1e-5)  # 改用余弦退火学习率调度\n    \n    for e in range(epochs):\n        model.train()\n        running_loss = 0\n        \n        for images, labels in train_loader:\n            # 将数据移至GPU并数据增强（未做）\n            # images = torch.stack([transform(img) for img in images]).to(device, non_blocking=True)\n            images = torch.stack([img for img in images]).to(device, non_blocking=True)\n            labels = labels.to(device, non_blocking=True)\n            \n            optimizer.zero_grad()\n            log_ps = model(images)\n            loss = criterion(log_ps, labels)\n            loss.backward()\n            optimizer.step()\n            \n            running_loss += loss.item()\n        \n        # 更新学习率\n        scheduler.step()\n        \n        # 验证阶段\n        test_loss = 0\n        accuracy = 0\n        model.eval()\n        \n        with torch.no_grad():\n            for images, labels in test_loader:\n                images = images.view(-1, 1, 28, 28).to(device, non_blocking=True)\n                labels = labels.to(device, non_blocking=True)\n                \n                log_ps = model(images)\n                test_loss += criterion(log_ps, labels)\n                \n                ps = torch.exp(log_ps)\n                top_p, top_class = ps.topk(1, dim=1)\n                equals = top_class == labels.view(*top_class.shape)\n                accuracy += torch.mean(equals.type(torch.FloatTensor))\n        \n        train_loss = running_loss/len(train_loader)\n        test_loss = test_loss/len(test_loader)\n        test_acc = accuracy/len(test_loader)\n        \n        train_losses.append(train_loss)\n        test_losses.append(test_loss)\n        \n        print(f\"Epoch: {e+1}/{epochs}.. \"\n              f\"Train Loss: {train_loss:.3f}.. \"\n              f\"Test Loss: {test_loss:.3f}.. \"\n              f\"Test Accuracy: {test_acc:.3f}.. \"\n              f\"LR: {scheduler.get_lr()[0]:.6f}\")\n    \n    return model, train_losses, test_losses\n\n# 5. 主函数\nif __name__ == \"__main__\":\n    # 训练模型\n    model, train_losses, test_losses = train_model()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-05-12T16:38:22.247660Z","iopub.execute_input":"2025-05-12T16:38:22.247947Z","iopub.status.idle":"2025-05-12T16:39:33.697625Z","shell.execute_reply.started":"2025-05-12T16:38:22.247905Z","shell.execute_reply":"2025-05-12T16:39:33.696889Z"}},"outputs":[],"execution_count":null}]}