The Clothing Sale

from pathlib import Path
import random
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader
from torchvision import transforms
from sklearn.model_selection import train_test_split
from sklearn.metrics import confusion_matrix, classification_report, ConfusionMatrixDisplay
import kagglehub

folder = Path(kagglehub.competition_download("bdaio-2024-the-clothing-sale"))
train_df = pd.read_csv(folder / "mnist_train.csv")
test_df = pd.read_csv(folder / "mnist_test.csv")

pixel_cols = [f"pixel{i}" for i in range(1, 785)]
class_names = ["T-shirt/top", "Trouser", "Pullover", "Dress", "Coat",
               "Sandal", "Shirt", "Sneaker", "Bag", "Ankle boot"]

df = train_df
PIXEL_COLS = pixel_cols
CLASS_NAMES = class_names

output_dir = Path("submissions")
output_dir.mkdir(exist_ok=True)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("Device:", device)
print("Training images:", len(train_df), "Test images:", len(test_df))

# 1. Dataset structure, quality checks, and exact duplicates
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

eda_train_pixels = df[PIXEL_COLS].to_numpy(dtype=np.uint8).reshape(-1, 28, 28)
eda_test_pixels = test_df[PIXEL_COLS].to_numpy(dtype=np.uint8).reshape(-1, 28, 28)

summary = pd.DataFrame({
    'Train': [len(df), len(PIXEL_COLS), df.isna().sum().sum(), df['ID'].duplicated().sum(),
              df.duplicated(subset=PIXEL_COLS).sum()],
    'Test': [len(test_df), len(PIXEL_COLS), test_df.isna().sum().sum(),
             test_df['ID'].duplicated().sum(), test_df.duplicated(subset=PIXEL_COLS).sum()]
}, index=['Images', 'Pixel features', 'Missing values', 'Duplicate IDs', 'Duplicate images'])
display(summary)

assert df['label'].dropna().isin(range(10)).all(), 'Unexpected class labels'
assert df[PIXEL_COLS].notna().all().all() and test_df[PIXEL_COLS].notna().all().all()
assert eda_train_pixels.min() >= 0 and eda_train_pixels.max() <= 255
assert eda_test_pixels.min() >= 0 and eda_test_pixels.max() <= 255

train_image_keys = {img.tobytes() for img in eda_train_pixels}
train_test_overlap = sum(img.tobytes() in train_image_keys for img in eda_test_pixels)
print(f'Pixel range: train [{eda_train_pixels.min()}, {eda_train_pixels.max()}] | '
      f'test [{eda_test_pixels.min()}, {eda_test_pixels.max()}]')
print(f'Exact competition train/test image matches: {train_test_overlap}/{len(test_df)}')

Pixel range: train [0, 255] | test [0, 255]
Exact competition train/test image matches: 0/2000

# 2. Class balance in labeled training data
class_counts = df['label'].value_counts().reindex(range(10), fill_value=0).sort_index()
class_table = pd.DataFrame({
    'Class': CLASS_NAMES, 'Count': class_counts.to_numpy(),
    'Share (%)': (100 * class_counts / len(df)).round(2).to_numpy()
})
display(class_table)
fig, ax = plt.subplots(figsize=(11, 3.5))
ax.bar(CLASS_NAMES, class_counts.to_numpy())
ax.set(title='Training samples per class', ylabel='Count')
ax.tick_params(axis='x', rotation=40)
plt.tight_layout()
plt.show()

	Class	Count	Share (%)
0	T-shirt/top	1003	10.03
1	Trouser	972	9.72
2	Pullover	969	9.69
3	Dress	1037	10.37
4	Coat	1014	10.14
5	Sandal	961	9.61
6	Shirt	1044	10.44
7	Sneaker	1003	10.03
8	Bag	943	9.43
9	Ankle boot	1054	10.54

# 3. Two representative images from each class
rng = np.random.default_rng(42)
fig, axes = plt.subplots(2, 10, figsize=(16, 4.1))
train_labels = df['label'].to_numpy()
for label in range(10):
    candidates = np.flatnonzero(train_labels == label)
    if len(candidates) == 0:
        for row in range(2): axes[row, label].axis('off')
        continue
    picks = rng.choice(candidates, size=2, replace=len(candidates) < 2)
    for row, sample_i in enumerate(picks):
        axes[row, label].imshow(eda_train_pixels[sample_i], cmap='gray', vmin=0, vmax=255)
        axes[row, label].axis('off')
        if row == 0: axes[row, label].set_title(CLASS_NAMES[label], fontsize=9)
fig.suptitle('Examples from the labeled training set', fontsize=12)
plt.tight_layout()
plt.show()

# 5. Average image for each category (visual similarity / class confusion)
fig, axes = plt.subplots(2, 5, figsize=(10.5, 5.4))
for i, ax in enumerate(axes.flat):
    imgs = eda_train_pixels[train_labels == i]
    if len(imgs):
        ax.imshow(imgs.mean(axis=0), cmap='gray', vmin=0, vmax=255)
    ax.set_title(CLASS_NAMES[i], fontsize=10)
    ax.axis('off')
plt.suptitle('Mean training image by class', y=1.01)
plt.tight_layout()
plt.show()

images = train_df[pixel_cols].to_numpy(dtype=np.float32).reshape(-1, 1, 28, 28) / 255
labels = train_df["label"].to_numpy(dtype=np.int64)

x_train, x_val, y_train, y_val = train_test_split(
    images, labels, test_size=0.2, random_state=42, stratify=labels
)
x_train = torch.from_numpy(x_train)
y_train = torch.from_numpy(y_train)
x_val = torch.from_numpy(x_val)
y_val = torch.from_numpy(y_val)
x_test = torch.from_numpy(
    test_df[pixel_cols].to_numpy(dtype=np.float32).reshape(-1, 1, 28, 28) / 255
)

class FashionDataset(Dataset):
    def __init__(self, images, labels, augmentation=None):
        self.images = images
        self.labels = labels
        self.augmentation = augmentation

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

    def __getitem__(self, index):
        image = self.images[index]
        if self.augmentation is not None:
            image = self.augmentation(image)
        return image, self.labels[index]


def predict(model, images):
    model.eval()
    predictions = []
    loader = DataLoader(images, batch_size=256, shuffle=False)
    with torch.no_grad():
        for batch in loader:
            scores = model(batch.to(device))
            predictions.append(torch.softmax(scores, dim=1).cpu().numpy())
    return np.concatenate(predictions)


def train_model(architecture, name, epochs, augmentation=None,
                optimizer="Adam", cosine=False, seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

    model = architecture().to(device)
    loader = DataLoader(
        FashionDataset(x_train, y_train, augmentation),
        batch_size=64, shuffle=True
    )
    if optimizer == "AdamW":
        opt = torch.optim.AdamW(model.parameters(), lr=0.001, weight_decay=1e-4)
    else:
        opt = torch.optim.Adam(model.parameters(), lr=0.001)

    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
        opt, T_max=epochs, eta_min=1e-5
    ) if cosine else None

    best_accuracy = -1
    best_epoch = 0
    checkpoint = output_dir / f"{name}.pth"

    for epoch in range(epochs):
        model.train()
        total_loss = 0
        for batch, targets in loader:
            batch, targets = batch.to(device), targets.to(device)
            opt.zero_grad()
            outputs = model(batch)
            loss = nn.functional.cross_entropy(outputs, targets)
            loss.backward()
            opt.step()
            total_loss += loss.item() * len(targets)

        if scheduler is not None:
            scheduler.step()

        accuracy = (predict(model, x_val).argmax(axis=1) == y_val.numpy()).mean()
        if accuracy > best_accuracy:
            best_accuracy = accuracy
            best_epoch = epoch + 1
            torch.save(model.state_dict(), checkpoint)

        if (epoch + 1) % 10 == 0 or epoch == 0 or epoch + 1 == epochs:
            print(f"{name}: epoch {epoch + 1}/{epochs}, "
                  f"loss {total_loss / len(y_train):.4f}, val {accuracy:.4f}")

    model.load_state_dict(torch.load(checkpoint, map_location=device, weights_only=True))
    return model, best_accuracy, best_epoch


scores = {}
submission_files = []

def submit(model_name, probabilities, val_accuracy):
    path = output_dir / f"{model_name}.csv"
    pd.DataFrame({"ID": test_df["ID"], "label": probabilities.argmax(axis=1)}).to_csv(
        path, index=False
    )
    submission_files.append(path)
    scores[model_name] = val_accuracy
    print(f"Saved {path} (validation: {val_accuracy:.4f})")


class BiggerCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, 32, 3, padding=1), nn.BatchNorm2d(32), nn.ReLU(),
            nn.Conv2d(32, 32, 3, padding=1), nn.BatchNorm2d(32), nn.ReLU(),
            nn.MaxPool2d(2), nn.Dropout2d(0.10),
            nn.Conv2d(32, 64, 3, padding=1), nn.BatchNorm2d(64), nn.ReLU(),
            nn.Conv2d(64, 64, 3, padding=1), nn.BatchNorm2d(64), nn.ReLU(),
            nn.MaxPool2d(2), nn.Dropout2d(0.15),
            nn.Conv2d(64, 128, 3, padding=1), nn.BatchNorm2d(128), nn.ReLU(),
            nn.MaxPool2d(2)
        )
        self.classifier = nn.Sequential(
            nn.Flatten(), nn.Linear(128 * 3 * 3, 128), nn.ReLU(),
            nn.Dropout(0.4), nn.Linear(128, 10)
        )

    def forward(self, x):
        return self.classifier(self.features(x))


original_aug = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),
    transforms.RandomAffine(degrees=10, translate=(0.08, 0.08), scale=(0.95, 1.05)),
    transforms.RandomErasing(p=0.15, scale=(0.02, 0.08), ratio=(0.5, 2.0))
])

light_aug = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),
    transforms.RandomAffine(degrees=6, translate=(0.04, 0.04), scale=(0.97, 1.03))
])

class SimpleCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(1, 32, 3)
        self.conv2 = nn.Conv2d(32, 64, 3)
        self.pool = nn.MaxPool2d(2)
        self.fc1 = nn.Linear(64 * 5 * 5, 128)
        self.fc2 = nn.Linear(128, 10)

    def forward(self, x):
        x = self.pool(torch.relu(self.conv1(x)))
        x = self.pool(torch.relu(self.conv2(x)))
        return self.fc2(torch.relu(self.fc1(x.flatten(1))))

simple_model, simple_acc, _ = train_model(
    SimpleCNN, "00_simple_cnn", epochs=20, augmentation=original_aug
)
submit("00_simple_cnn", predict(simple_model, x_test), simple_acc)

bigger_model, bigger_acc, _ = train_model(
    BiggerCNN, "01_bigger_baseline", epochs=75, augmentation=original_aug
)
submit("01_bigger_baseline", predict(bigger_model, x_test), bigger_acc)

adamw_model, adamw_acc, _ = train_model(
    BiggerCNN, "02_bigger_adamw_cosine", epochs=75,
    augmentation=original_aug, optimizer="AdamW", cosine=True
)
submit("02_bigger_adamw_cosine", predict(adamw_model, x_test), adamw_acc)

light_model, light_acc, _ = train_model(
    BiggerCNN, "03_bigger_light_aug", epochs=75,
    augmentation=light_aug, optimizer="AdamW", cosine=True
)
submit("03_bigger_light_aug", predict(light_model, x_test), light_acc)

candidates = [
    (bigger_acc, bigger_model, original_aug, "Adam"),
    (adamw_acc, adamw_model, original_aug, "AdamW"),
    (light_acc, light_model, light_aug, "AdamW")
]
best_acc, best_model, best_augmentation, best_optimizer = max(candidates, key=lambda item: item[0])
print("Best BiggerCNN validation accuracy:", round(best_acc, 4))

ensemble = [best_model]
for seed in [13, 37]:
    model, accuracy, _ = train_model(
        BiggerCNN, f"04_bigger_seed_{seed}", epochs=75,
        augmentation=best_augmentation, optimizer=best_optimizer,
        cosine=(best_optimizer == "AdamW"), seed=seed
    )
    ensemble.append(model)

ensemble_val = np.mean([predict(model, x_val) for model in ensemble], axis=0)
ensemble_test = np.mean([predict(model, x_test) for model in ensemble], axis=0)
ensemble_acc = (ensemble_val.argmax(axis=1) == y_val.numpy()).mean()
submit("04_bigger_seed_ensemble", ensemble_test, ensemble_acc)

class ResidualBlock(nn.Module):
    def __init__(self, in_channels, out_channels, stride=1):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Conv2d(in_channels, out_channels, 3, stride, padding=1, bias=False),
            nn.BatchNorm2d(out_channels), nn.ReLU(),
            nn.Conv2d(out_channels, out_channels, 3, padding=1, bias=False),
            nn.BatchNorm2d(out_channels)
        )
        self.skip = (nn.Identity() if in_channels == out_channels and stride == 1
                     else nn.Sequential(
                         nn.Conv2d(in_channels, out_channels, 1, stride, bias=False),
                         nn.BatchNorm2d(out_channels)))

    def forward(self, x):
        return torch.relu(self.layers(x) + self.skip(x))


class SmallResNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Conv2d(1, 32, 3, padding=1, bias=False),
            nn.BatchNorm2d(32), nn.ReLU(),
            ResidualBlock(32, 32),
            ResidualBlock(32, 64, stride=2),
            ResidualBlock(64, 128, stride=2),
            nn.AdaptiveAvgPool2d(1), nn.Flatten(), nn.Linear(128, 10)
        )

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


resnet_model, resnet_acc, _ = train_model(
    SmallResNet, "05_resnet_challenger", epochs=50,
    augmentation=light_aug, optimizer="AdamW", cosine=True
)
submit("05_resnet_challenger", predict(resnet_model, x_test), resnet_acc)