演習：PyTorchで分類モデルを実装しよう

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader, random_split
import numpy as np

np.random.seed(42)
torch.manual_seed(42)

# === データ生成 ===
n = 5000
X = np.random.randn(n, 10).astype(np.float32)
# 離反ラベル（非線形なパターン）
z = 0.5 * X[:, 0] - 0.3 * X[:, 1] + 0.8 * X[:, 2] * X[:, 3] - 0.4 * X[:, 4] ** 2
prob = 1 / (1 + np.exp(-z + 1.5))  # 離反率を約8%に調整
y = (np.random.random(n) < prob).astype(np.float32)
print(f"離反率: {y.mean():.3f}")

# === Dataset 定義 ===
class ChurnDataset(Dataset):
    def __init__(self, features, labels):
        self.features = torch.tensor(features, dtype=torch.float32)
        self.labels = torch.tensor(labels, dtype=torch.float32).unsqueeze(1)

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, idx):
        return self.features[idx], self.labels[idx]

# === 分割 ===
dataset = ChurnDataset(X, y)
train_size = int(0.6 * len(dataset))
val_size = int(0.2 * len(dataset))
test_size = len(dataset) - train_size - val_size

train_ds, val_ds, test_ds = random_split(
    dataset, [train_size, val_size, test_size],
    generator=torch.Generator().manual_seed(42)
)

# === DataLoader ===
train_loader = DataLoader(train_ds, batch_size=64, shuffle=True)
val_loader = DataLoader(val_ds, batch_size=128, shuffle=False)
test_loader = DataLoader(test_ds, batch_size=128, shuffle=False)

print(f"学習: {len(train_ds)}, 検証: {len(val_ds)}, テスト: {len(test_ds)}")

# バッチ確認
for X_batch, y_batch in train_loader:
    print(f"バッチ shape: X={X_batch.shape}, y={y_batch.shape}")
    break

import matplotlib.pyplot as plt

# === モデル定義 ===
class ChurnNet(nn.Module):
    def __init__(self, input_dim=10):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(input_dim, 128),
            nn.BatchNorm1d(128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 64),
            nn.BatchNorm1d(64),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(64, 32),
            nn.BatchNorm1d(32),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(32, 1)
        )

    def forward(self, x):
        return self.network(x)

# === 学習準備 ===
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = ChurnNet().to(device)
criterion = nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-4)

# === Early Stopping ===
class EarlyStopping:
    def __init__(self, patience=10):
        self.patience = patience
        self.counter = 0
        self.best_loss = float('inf')
        self.best_state = None

    def __call__(self, val_loss, model):
        if val_loss < self.best_loss - 1e-4:
            self.best_loss = val_loss
            self.counter = 0
            self.best_state = {k: v.clone() for k, v in model.state_dict().items()}
            return False
        self.counter += 1
        if self.counter >= self.patience:
            model.load_state_dict(self.best_state)
            return True
        return False

early_stopping = EarlyStopping(patience=15)

# === 学習ループ ===
history = {'train_loss': [], 'val_loss': []}
num_epochs = 100

for epoch in range(num_epochs):
    # 学習
    model.train()
    train_loss = 0.0
    for X_batch, y_batch in train_loader:
        X_batch, y_batch = X_batch.to(device), y_batch.to(device)
        optimizer.zero_grad()
        outputs = model(X_batch)
        loss = criterion(outputs, y_batch)
        loss.backward()
        optimizer.step()
        train_loss += loss.item()

    # 検証
    model.eval()
    val_loss = 0.0
    with torch.no_grad():
        for X_batch, y_batch in val_loader:
            X_batch, y_batch = X_batch.to(device), y_batch.to(device)
            outputs = model(X_batch)
            loss = criterion(outputs, y_batch)
            val_loss += loss.item()

    avg_train = train_loss / len(train_loader)
    avg_val = val_loss / len(val_loader)
    history['train_loss'].append(avg_train)
    history['val_loss'].append(avg_val)

    if (epoch + 1) % 10 == 0:
        print(f"Epoch [{epoch+1}/{num_epochs}] Train: {avg_train:.4f}, Val: {avg_val:.4f}")

    if early_stopping(avg_val, model):
        print(f"Early Stopping at epoch {epoch+1}")
        break

# === 学習曲線 ===
plt.figure(figsize=(8, 4))
plt.plot(history['train_loss'], label='Train Loss')
plt.plot(history['val_loss'], label='Val Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('学習曲線')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

from sklearn.metrics import classification_report, confusion_matrix, f1_score

model.eval()
all_preds = []
all_labels = []

with torch.no_grad():
    for X_batch, y_batch in test_loader:
        X_batch = X_batch.to(device)
        outputs = model(X_batch)
        probs = torch.sigmoid(outputs).cpu().numpy()
        all_preds.extend(probs.flatten())
        all_labels.extend(y_batch.numpy().flatten())

all_preds = np.array(all_preds)
all_labels = np.array(all_labels)
y_pred = (all_preds > 0.5).astype(int)

print("=== テストデータ評価 ===")
print(classification_report(all_labels, y_pred, target_names=['継続', '離反']))

cm = confusion_matrix(all_labels, y_pred)
print(f"\n混同行列:")
print(f"  {'':>10} 予測:継続  予測:離反")
print(f"  実際:継続  {cm[0,0]:>6}    {cm[0,1]:>6}")
print(f"  実際:離反  {cm[1,0]:>6}    {cm[1,1]:>6}")

f1 = f1_score(all_labels, y_pred)
print(f"\nF1スコア: {f1:.4f}")