CNNs — Shopify Product Image Classification

Imagine inspecting a large fabric roll for defects. You would not look at the entire roll simultaneously — you would use a small magnifying glass and slide it across, looking for the same pattern (a tear, a stain) at every position. You use the same visual skill (the same filter weights) everywhere.

A CNN's convolutional layer is that magnifying glass — a small filter sliding across the image, applying the same weights at every position. Early layers detect edges and textures. Middle layers detect shapes. Deep layers detect objects. The hierarchy of features is learned automatically from labelled images.

import numpy as np
import torch
import torch.nn as nn

# ── Manual 2D convolution — see every step ────────────────────────────
def conv2d_manual(input_2d, kernel, stride=1, padding=0):
    """
    input_2d: (H, W)
    kernel:   (kH, kW)
    Returns:  feature map (H_out, W_out)
    """
    H, W   = input_2d.shape
    kH, kW = kernel.shape

    if padding > 0:
        input_2d = np.pad(input_2d, padding, mode='constant')
        H, W     = input_2d.shape

    H_out = (H - kH) // stride + 1
    W_out = (W - kW) // stride + 1
    output = np.zeros((H_out, W_out))

    for i in range(H_out):
        for j in range(W_out):
            patch         = input_2d[i*stride:i*stride+kH, j*stride:j*stride+kW]
            output[i, j]  = (patch * kernel).sum()
    return output

# ── Test: vertical edge detector ──────────────────────────────────────
image = np.array([
    [0, 0, 0, 255, 255],
    [0, 0, 0, 255, 255],
    [0, 0, 0, 255, 255],
    [0, 0, 0, 255, 255],
    [0, 0, 0, 255, 255],
], dtype=float)

vertical_edge = np.array([[-1, 0, 1], [-1, 0, 1], [-1, 0, 1]], dtype=float)
horizontal_edge = np.array([[-1,-1,-1], [0,0,0], [1,1,1]], dtype=float)

print("Vertical edge filter response:")
print(conv2d_manual(image, vertical_edge))
print("
Horizontal edge filter response:")
print(conv2d_manual(image, horizontal_edge))

# ── Output size formula ────────────────────────────────────────────────
def output_size(H, kernel, stride, padding):
    return (H + 2*padding - kernel) // stride + 1

print("
Output size examples (input=224):")
for k, s, p in [(3,1,0),(3,1,1),(3,2,1),(5,1,2),(7,2,3)]:
    out = output_size(224, k, s, p)
    print(f"  kernel={k} stride={s} pad={p}: output={out}×{out}")

import torch
import torch.nn as nn

# ── CNN from scratch for Shopify product classification ─────────────────
# 6 categories: kurta, saree, jeans, sneakers, watch, handbag

class ShopifyCNN(nn.Module):
    def __init__(self, n_classes=6):
        super().__init__()

        # Feature extractor — conv layers learn visual features
        self.features = nn.Sequential(
            # Block 1: 3 → 32 channels
            nn.Conv2d(3, 32, kernel_size=3, padding=1),   # (B,3,128,128) → (B,32,128,128)
            nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),                            # → (B,32,64,64)

            # Block 2: 32 → 64 channels
            nn.Conv2d(32, 64, kernel_size=3, padding=1),  # → (B,64,64,64)
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),                            # → (B,64,32,32)

            # Block 3: 64 → 128 channels
            nn.Conv2d(64, 128, kernel_size=3, padding=1), # → (B,128,32,32)
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2),                            # → (B,128,16,16)

            # Block 4: 128 → 256 channels
            nn.Conv2d(128, 256, kernel_size=3, padding=1),# → (B,256,16,16)
            nn.BatchNorm2d(256),
            nn.ReLU(inplace=True),
            nn.AdaptiveAvgPool2d((4, 4)),                  # → (B,256,4,4) fixed
        )

        # Classifier head — fully connected layers make final decision
        self.classifier = nn.Sequential(
            nn.Flatten(),                                  # → (B, 4096)
            nn.Linear(256 * 4 * 4, 512),
            nn.ReLU(inplace=True),
            nn.Dropout(0.4),
            nn.Linear(512, n_classes),                    # → (B, 6) raw logits
        )

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

model = ShopifyCNN(n_classes=6)

# ── Parameter count ───────────────────────────────────────────────────
total  = sum(p.numel() for p in model.parameters())
trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f"Total parameters:     {total:,}")
print(f"Trainable parameters: {trainable:,}")

# ── Shape check — forward pass with dummy batch ───────────────────────
dummy = torch.randn(4, 3, 128, 128)   # batch of 4 RGB images at 128×128
out   = model(dummy)
print(f"
Input shape:  {dummy.shape}")
print(f"Output shape: {out.shape}  ← (4 samples, 6 class logits)")

# ── Inspect intermediate feature map shapes ───────────────────────────
print("
Feature map shapes through each block:")
x = dummy
for i, layer in enumerate(model.features):
    x = layer(x)
    if isinstance(layer, (nn.MaxPool2d, nn.AdaptiveAvgPool2d)):
        print(f"  After {layer.__class__.__name__}: {x.shape}")

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision.transforms as T
import numpy as np
from torch.utils.data import Dataset, DataLoader
import copy, warnings
warnings.filterwarnings('ignore')

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

# ── Simulate Shopify product image dataset ─────────────────────────────
# In production: torchvision.datasets.ImageFolder pointing to your image directory
# Here: synthetic RGB images with class-specific colour patterns

CATEGORIES = ['kurta', 'saree', 'jeans', 'sneakers', 'watch', 'handbag']
N_CLASSES  = len(CATEGORIES)

class SyntheticShopifyDataset(Dataset):
    def __init__(self, n_samples=2000, img_size=64, transform=None):
        self.n  = n_samples
        self.sz = img_size
        self.transform = transform
        np.random.seed(42)
        self.labels = np.random.randint(0, N_CLASSES, n_samples)
        # Each class has a distinct colour bias — simulates real category difference
        self.color_bias = np.array([
            [0.8, 0.3, 0.2],   # kurta    — warm red
            [0.2, 0.7, 0.5],   # saree    — green/teal
            [0.2, 0.3, 0.8],   # jeans    — blue
            [0.7, 0.7, 0.2],   # sneakers — yellow/white
            [0.5, 0.5, 0.5],   # watch    — grey/silver
            [0.6, 0.2, 0.6],   # handbag  — purple
        ])

    def __len__(self): return self.n

    def __getitem__(self, idx):
        label = self.labels[idx]
        bias  = self.color_bias[label]
        # Generate image with class-specific colour + noise
        img = np.random.randn(3, self.sz, self.sz) * 0.15
        for c in range(3):
            img[c] += bias[c]
        img = np.clip(img, 0, 1).astype(np.float32)
        img = torch.FloatTensor(img)
        if self.transform:
            img = self.transform(img)
        return img, label

# ── Data augmentation for training ────────────────────────────────────
train_transform = T.Compose([
    T.RandomHorizontalFlip(p=0.5),
    T.RandomVerticalFlip(p=0.1),
    T.RandomRotation(degrees=15),
    T.ColorJitter(brightness=0.3, contrast=0.3, saturation=0.2),
    T.RandomErasing(p=0.2),       # randomly mask patches — prevents overfit
])
val_transform = None   # no augmentation at validation time

train_ds = SyntheticShopifyDataset(1600, transform=train_transform)
val_ds   = SyntheticShopifyDataset(400,  transform=val_transform)

train_loader = DataLoader(train_ds, batch_size=64, shuffle=True,  num_workers=0)
val_loader   = DataLoader(val_ds,   batch_size=64, shuffle=False, num_workers=0)

# ── Model, loss, optimiser ─────────────────────────────────────────────
class ShopifyCNN(nn.Module):
    def __init__(self, n=6):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3,32,3,padding=1), nn.BatchNorm2d(32), nn.ReLU(True), nn.MaxPool2d(2),
            nn.Conv2d(32,64,3,padding=1), nn.BatchNorm2d(64), nn.ReLU(True), nn.MaxPool2d(2),
            nn.Conv2d(64,128,3,padding=1), nn.BatchNorm2d(128), nn.ReLU(True),
            nn.AdaptiveAvgPool2d((4,4)),
        )
        self.classifier = nn.Sequential(
            nn.Flatten(), nn.Linear(128*16,256),
            nn.ReLU(True), nn.Dropout(0.4), nn.Linear(256,n),
        )
    def forward(self, x): return self.classifier(self.features(x))

model     = ShopifyCNN()
criterion = nn.CrossEntropyLoss()
optimizer = optim.AdamW(model.parameters(), lr=1e-3, weight_decay=0.01)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=30)

# ── Training loop with early stopping ─────────────────────────────────
best_acc, best_wts, patience_count = 0.0, None, 0
PATIENCE = 8

print(f"Training ShopifyCNN from scratch:")
print(f"{'Epoch':>6} {'Train loss':>12} {'Val acc':>10} {'LR':>12}")
print("─" * 44)

for epoch in range(1, 31):
    model.train()
    total_loss = 0
    for Xb, yb in train_loader:
        optimizer.zero_grad()
        loss = criterion(model(Xb), yb)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    scheduler.step()

    model.eval()
    correct = 0
    with torch.no_grad():
        for Xb, yb in val_loader:
            correct += (model(Xb).argmax(1) == yb).sum().item()
    val_acc = correct / len(val_ds)
    lr_now  = optimizer.param_groups[0]['lr']

    if epoch % 5 == 0:
        print(f"  {epoch:>4}  {total_loss/len(train_loader):>12.4f}  {val_acc:>10.4f}  {lr_now:>12.6f}")

    if val_acc > best_acc:
        best_acc, best_wts = val_acc, copy.deepcopy(model.state_dict())
        patience_count = 0
    else:
        patience_count += 1
        if patience_count >= PATIENCE:
            print(f"  Early stop at epoch {epoch}")
            break

model.load_state_dict(best_wts)
print(f"
Best val accuracy: {best_acc:.4f}")

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision.models as models
import numpy as np
from torch.utils.data import Dataset, DataLoader
import warnings
warnings.filterwarnings('ignore')

torch.manual_seed(42)

CATEGORIES = ['kurta', 'saree', 'jeans', 'sneakers', 'watch', 'handbag']
N_CLASSES  = 6

class SyntheticShopifyDataset(torch.utils.data.Dataset):
    def __init__(self, n=500):
        np.random.seed(42)
        self.labels = np.random.randint(0, N_CLASSES, n)
        self.bias   = np.array([[.8,.3,.2],[.2,.7,.5],[.2,.3,.8],
                                 [.7,.7,.2],[.5,.5,.5],[.6,.2,.6]])
    def __len__(self): return len(self.labels)
    def __getitem__(self, i):
        b   = self.bias[self.labels[i]]
        img = np.random.randn(3,64,64)*.15
        for c in range(3): img[c] += b[c]
        return torch.FloatTensor(np.clip(img,0,1).astype(np.float32)), self.labels[i]

train_ds = SyntheticShopifyDataset(400)
val_ds   = SyntheticShopifyDataset(100)
train_ld = DataLoader(train_ds, 32, shuffle=True)
val_ld   = DataLoader(val_ds,   32)

# ── Strategy 1: Feature extraction — freeze backbone ──────────────────
backbone = models.resnet18(weights=None)   # use pretrained=True in production
# Freeze all layers
for param in backbone.parameters():
    param.requires_grad = False
# Replace the final FC layer — only this trains
backbone.fc = nn.Linear(backbone.fc.in_features, N_CLASSES)
# Only fc parameters have requires_grad=True
trainable = sum(p.numel() for p in backbone.parameters() if p.requires_grad)
total     = sum(p.numel() for p in backbone.parameters())
print(f"Feature extraction: {trainable:,} / {total:,} parameters trainable")

# ── Strategy 2: Partial fine-tuning — unfreeze layer4 + fc ───────────
backbone2 = models.resnet18(weights=None)
for param in backbone2.parameters():
    param.requires_grad = False
# Unfreeze only the last residual block and fc
for param in backbone2.layer4.parameters():
    param.requires_grad = True
backbone2.fc = nn.Linear(backbone2.fc.in_features, N_CLASSES)
for param in backbone2.fc.parameters():
    param.requires_grad = True
trainable2 = sum(p.numel() for p in backbone2.parameters() if p.requires_grad)
print(f"Partial fine-tuning: {trainable2:,} / {total:,} parameters trainable")

# ── Train feature extraction version ──────────────────────────────────
# Resize input to 224×224 for ResNet (designed for ImageNet size)
# Here: use adaptive pool to handle our 64×64 images
model = backbone
criterion = nn.CrossEntropyLoss()
# Only pass trainable parameters to optimiser — good practice
optimizer = optim.AdamW(
    filter(lambda p: p.requires_grad, model.parameters()),
    lr=1e-3, weight_decay=0.01,
)

print("
Fine-tuning classification head only:")
for epoch in range(1, 11):
    model.train()
    for Xb, yb in train_ld:
        optimizer.zero_grad()
        criterion(model(Xb), yb).backward()
        optimizer.step()

    model.eval()
    correct = 0
    with torch.no_grad():
        for Xb, yb in val_ld:
            correct += (model(Xb).argmax(1) == yb).sum().item()
    if epoch % 2 == 0:
        print(f"  Epoch {epoch:2d}: val acc = {correct/len(val_ds):.4f}")

# ── Differential learning rates — best practice for fine-tuning ───────
# Backbone layers: very small lr (they are already good)
# Classification head: normal lr (it is randomly initialised)
backbone3 = models.resnet18(weights=None)
backbone3.fc = nn.Linear(backbone3.fc.in_features, N_CLASSES)

optimizer_diff = optim.AdamW([
    {'params': backbone3.layer1.parameters(), 'lr': 1e-5},
    {'params': backbone3.layer2.parameters(), 'lr': 1e-5},
    {'params': backbone3.layer3.parameters(), 'lr': 1e-4},
    {'params': backbone3.layer4.parameters(), 'lr': 1e-4},
    {'params': backbone3.fc.parameters(),     'lr': 1e-3},   # head: normal lr
], weight_decay=0.01)

print("
Differential lr param groups:")
for group in optimizer_diff.param_groups:
    n = sum(p.numel() for p in group['params'])
    print(f"  lr={group['lr']:.0e}  params={n:,}")

A retail team built a model to detect out-of-stock shelves from in-store camera frames. The first version, a fine-tuned ResNet50, hit 94 percent accuracy in validation. On the actual in-store hardware — a low-power ARM-based camera unit, not a GPU — it took 340 milliseconds per frame and the 98MB checkpoint barely fit in available memory alongside the camera's own firmware. It could not ship as built.

The team switched the backbone to MobileNetV3-Small, applied post-training int8 quantisation, and pruned the classification head. The final model was 4.2MB and ran in 12 milliseconds per frame on the same hardware. Accuracy dropped from 94 percent to 91 percent — a trade the team accepted without much debate, since the alternative was a model the device could not run at all.

import torch
import torch.nn as nn
import torchvision.models as models
import time, os

# ── The production question: does this model fit the deployment target? ─
# Accuracy is necessary but not sufficient — an edge camera app needs a
# model under roughly 5MB and 15ms per frame, not just a high top-1 score.

def model_size_mb(m: nn.Module) -> float:
    torch.save(m.state_dict(), '/tmp/_tmp_model.pt')
    size = os.path.getsize('/tmp/_tmp_model.pt') / 1e6
    os.remove('/tmp/_tmp_model.pt')
    return size

def benchmark_latency(m: nn.Module, input_size=(1, 3, 224, 224), n=50) -> float:
    m.eval()
    dummy = torch.randn(*input_size)
    with torch.no_grad():
        for _ in range(5): _ = m(dummy)          # warmup
        start = time.time()
        for _ in range(n): _ = m(dummy)
        return (time.time() - start) / n * 1000  # ms

candidates = {
    'resnet50':           models.resnet50(weights=None),
    'mobilenet_v3_small': models.mobilenet_v3_small(weights=None),
    'mobilenet_v3_large': models.mobilenet_v3_large(weights=None),
    'efficientnet_b0':    models.efficientnet_b0(weights=None),
}

print(f"{'Backbone':<20} {'Size (MB)':>10} {'Latency (ms)':>13}")
print("─" * 46)
for name, model in candidates.items():
    size_mb = model_size_mb(model)
    latency = benchmark_latency(model)
    fit     = '  fits edge budget' if size_mb < 10 and latency < 20 else ''
    print(f"  {name:<18} {size_mb:>10.1f} {latency:>13.2f}{fit}")

# ── Quantisation — the standard final step before edge deployment ──────
# Post-training dynamic quantisation shrinks Linear-layer weights from
# float32 to int8, roughly a 4x size reduction with minor accuracy loss.
small_model = models.mobilenet_v3_small(weights=None)
quantized   = torch.quantization.quantize_dynamic(
    small_model, {nn.Linear}, dtype=torch.qint8,
)
print(f"
mobilenet_v3_small — float32 classifier head:     "
      f"{model_size_mb(small_model):.1f} MB")
print(f"mobilenet_v3_small — int8-quantised (Linear layers): "
      f"{model_size_mb(quantized):.1f} MB")

Convolution itself is translation equivariant, not invariant — shift the input image and the resulting feature map shifts by the same amount rather than staying fixed. What gives a CNN its practical robustness to small shifts is pooling and downsampling collapsing spatial resolution layer by layer, which only approximates invariance, and imperfectly: research on shift invariance (Zhang, 2019) showed standard strided pooling can alias and actually break invariance in modern architectures, which is why anti-aliased pooling variants exist. True invariance is something you approximate through architecture choices, not a guarantee convolution hands you for free.

Stacking two 3×3 convolutional layers gives the exact same receptive field as one 5×5 layer, and three 3×3 layers match one 7×7 layer — but with fewer total parameters (3×(3×3)=27 vs 7×7=49 weights per channel pair) and, critically, two or three ReLU non-linearities inserted between them instead of one. This was the specific insight behind VGGNet: depth with small filters is strictly more expressive per parameter than shallow networks with large filters, which is why virtually every modern CNN backbone uses stacks of 3×3 (or even 1×1) convolutions instead of large kernels.

CNN filters start as small random matrices — there is no hand-engineering involved at all. What's genuinely striking is that after training on real images, the very first convolutional layer's filters frequently converge to something that looks like edge detectors and colour-opponent blobs — visually similar to, but not copied from, classical filters like Sobel — purely because gradient descent discovers these are the most useful low-level features for the task. Deeper layers converge to increasingly abstract, task-specific patterns with no classical analogue at all: textures, object parts, and eventually whole-object detectors a human would never hand-design.

The features learned by early and middle convolutional layers — edges, corners, textures, colour gradients, simple shapes — are generic visual primitives, not specific to any ImageNet category, and they transfer surprisingly well even to domains that look nothing like natural photos: medical imaging, satellite imagery, and industrial defect detection all see real gains from ImageNet-pretrained backbones. What actually needs to match your domain is not the visual similarity of the images but your fine-tuning strategy — the more your domain differs from natural photos, the more of the later, task-specific layers you typically need to unfreeze and retrain rather than just replacing the final classification head.

MaxPool is one common way to reduce spatial dimensions, but it is not a structural requirement — many strong modern architectures downsample using a strided convolution instead, letting the same layer that extracts features also handle the resolution reduction, or dispense with pooling until a single global-average-pool at the very end. The "Striving for Simplicity" line of research demonstrated that replacing all pooling with strided convolutions can match or beat traditional pooling-based architectures, because pooling inherently discards spatial information (only the max or average survives) that a learned strided convolution can partially preserve.

Two properties, both baked into the convolution operation itself. First, local connectivity — each output value only depends on a small local patch of the input (e.g. 3×3), not the entire image, unlike a fully connected layer where every input connects to every output. Second, and more importantly, weight sharing — the exact same filter weights are reused at every spatial position across the image rather than learning a separate set of weights per position. A single 3×3 filter over a 3-channel image has only 27 weights plus bias regardless of whether the image is 32×32 or 1024×1024, because that same filter simply slides across however many positions the image contains. This is what makes CNNs both parameter-efficient and translation-equivariant.

Output size = floor((H + 2×padding − kernel_size) / stride) + 1. With "valid" padding (padding=0), the output shrinks every layer — a 3×3 filter with stride 1 on a 32×32 input gives 30×30 — which compounds across many layers and eventually leaves too little spatial resolution to keep stacking. With "same" padding (padding=(kernel_size−1)/2 for stride 1, e.g. padding=1 for a 3×3 filter), the output stays the same spatial size as the input, which is what lets you build very deep feature extractors without spatial dimensions collapsing prematurely. In practice: use "same" padding through the feature-extraction backbone so depth is a free architectural choice, and use pooling or strided convolutions as the deliberate, controlled points where you reduce spatial resolution.

With only 500 images I would start with feature extraction — freeze the entire pretrained backbone and train only a new classification head on top. With so little data, unfreezing and updating millions of backbone parameters risks catastrophic forgetting of useful pretrained features and overfitting almost immediately, since the backbone has far more capacity than the dataset can constrain. If feature extraction plateaus below the accuracy I need, I'd move to partial fine-tuning — unfreeze only the last block or two (which hold more task-specific, less universal features) with a very small learning rate. Full fine-tuning is usually reserved for datasets in the tens of thousands of images or more, or when the target domain is visually very different from natural photos.

CNNs are data-hungry and prone to overfitting on the exact pixel patterns of a small training set — augmentation artificially expands the effective dataset and teaches the network that these transformations shouldn't change the predicted label, directly reinforcing translation and rotation robustness the raw architecture doesn't guarantee for free. Applying augmentation at validation or test time is a real bug: it makes your evaluation numbers noisy and non-reproducible, since accuracy varies run to run based on random transforms, and more subtly it means you're evaluating on a shifted data distribution rather than genuinely representative held-out data — you can end up over- or under-estimating true generalisation performance.

MaxPool2d(kernel_size, stride) reduces spatial size by a fixed ratio determined by its kernel and stride — 2×2 with stride 2 always halves H and W — keeping the strongest activation in each local window. AdaptiveAvgPool2d((h, w)) instead is told the exact output size you want and figures out the pooling window itself to hit that target regardless of the input's spatial size — this is what lets the same architecture accept variable input image sizes and still produce a fixed-size feature vector for the classifier head, since a plain Flatten would produce a different-length vector for every different input resolution and break the fixed-size Linear layer that follows it. This is exactly why architectures like ResNet end their feature extractor with AdaptiveAvgPool2d((1,1)) rather than relying on a fixed input size.