FA690 Mini Project¶

predict 5-day binary stock price movement using CNN, based on the paper "(Re-)Imag(in)ing Price Trends" (Jiang, Kelly, Xiu 2023).

the idea is to convert price/volume data into chart images and feed them into a CNN to classify whether the stock goes up or down in the next 5 trading days.

In [1]:
from google.colab import drive
drive.mount('/content/drive')
Mounted at /content/drive
In [2]:
import pandas as pd
import numpy as np
import os
import os.path as op
import gc
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, confusion_matrix, classification_report
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')
In [3]:
IMAGE_WIDTH = {5: 15, 20: 60, 60: 180}
IMAGE_HEIGHT = {5: 32, 20: 64, 60: 96}

# data is in the same Google Drive folder as this notebook
img_path = "/content/drive/My Drive/690/mini project"

# full dataset
TRAIN_YEARS = list(range(2000, 2016))   # 2000-2015 for training
VAL_YEARS = [2016, 2017]                # 2016-2017 for validation
TEST_YEARS = [2018, 2019]               # 2018-2019 for test

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f'Using device: {device}')
Using device: cuda

Data Loading and Exploration¶

load image and label data from .dat and .feather files. each year has its own file pair. images are stored as memmap with shape (N, 64, 60), dtype uint8.

I use memmap to avoid loading everything into RAM at once, then copy only the subset I need.

In [4]:
def load_year_data(year):
    """load images (as memmap) and labels for a single year"""
    img_file = op.join(img_path, f"20d_month_has_vb_[20]_ma_{year}_images.dat")
    label_file = op.join(img_path, f"20d_month_has_vb_[20]_ma_{year}_labels_w_delay.feather")
    images = np.memmap(img_file, dtype=np.uint8, mode='r').reshape(
        (-1, IMAGE_HEIGHT[20], IMAGE_WIDTH[20]))
    labels = pd.read_feather(label_file)
    return images, labels

# quick check with one year
test_imgs, test_labels = load_year_data(2017)
print(f'2017: {test_imgs.shape[0]} samples, image shape = {test_imgs.shape[1:]}')
del test_imgs, test_labels; gc.collect()
2017: 67858 samples, image shape = (64, 60)
Out[4]:
0
In [5]:
# load all years 2000-2019 and check distribution
all_years = list(range(2000, 2020))
year_stats = []

for yr in all_years:
    try:
        _, lbl = load_year_data(yr)
        lbl['binary'] = (lbl['Ret_5d'] > 0).astype(int)
        n = len(lbl)
        pos_ratio = lbl['binary'].mean()
        year_stats.append({'year': yr, 'n_samples': n, 'pos_ratio': round(pos_ratio, 4)})
    except Exception as e:
        print(f'year {yr} failed: {e}')

stats_df = pd.DataFrame(year_stats)
print(stats_df.to_string(index=False))
 year  n_samples  pos_ratio
 2000      98148     0.5224
 2001      91985     0.5078
 2002      85729     0.4214
 2003      80894     0.6007
 2004      79512     0.5017
 2005      79883     0.5922
 2006      80163     0.4618
 2007      81140     0.4774
 2008      78186     0.4052
 2009      70065     0.5798
 2010      68111     0.5724
 2011      67325     0.4582
 2012      66250     0.5300
 2013      65486     0.5155
 2014      67302     0.4595
 2015      68827     0.4679
 2016      68141     0.5523
 2017      67858     0.4987
 2018      68481     0.5297
 2019      68637     0.5051

looking at the class balance, most years have a positive ratio around 0.48-0.52, so its pretty balanced. the baseline accuracy for random guessing is roughly 50%

In [6]:
fig, axes = plt.subplots(1, 2, figsize=(14, 4))
axes[0].bar(stats_df['year'], stats_df['n_samples'], color='steelblue')
axes[0].set_xlabel('Year'); axes[0].set_ylabel('Number of Samples')
axes[0].set_title('Samples per Year'); axes[0].tick_params(axis='x', rotation=45)
axes[1].plot(stats_df['year'], stats_df['pos_ratio'], 'o-', color='coral')
axes[1].axhline(y=0.5, color='gray', linestyle='--', alpha=0.5)
axes[1].set_xlabel('Year'); axes[1].set_ylabel('Positive Ratio')
axes[1].set_title('Positive Class Ratio per Year'); axes[1].tick_params(axis='x', rotation=45)
plt.tight_layout(); plt.show()
No description has been provided for this image
In [7]:
# show some sample images
sample_imgs, sample_labels = load_year_data(2015)
sample_labels['binary'] = (sample_labels['Ret_5d'] > 0).astype(int)
fig, axes = plt.subplots(2, 5, figsize=(15, 6))
for i in range(10):
    ax = axes[i // 5][i % 5]
    ax.imshow(sample_imgs[i], cmap='gray')
    lbl = 'Up' if sample_labels.iloc[i]['binary'] == 1 else 'Down'
    ax.set_title(f'{lbl}', fontsize=10); ax.axis('off')
plt.suptitle('Sample Chart Images (2015)', fontsize=13)
plt.tight_layout(); plt.show()
del sample_imgs, sample_labels; gc.collect()
No description has been provided for this image
Out[7]:
7700

Data Preprocessing and Split¶

time-series split is the key here. I can't use random split because that causes look-ahead bias, the model would "see" future data during training. so I split by year:

  • train: 2000-2015
  • validation: 2016-2017
  • test: 2018-2019

images also need to be normalized to [0,1], the raw pixel values are 0-255 uint8. I load subsets per year to manage memory, but on a machine with more RAM you should use full years and ideally train on 2000-2015.

In [8]:
def load_year_subset(year, max_samples=None):
    imgs, lbls = load_year_data(year)
    binary = (lbls['Ret_5d'] > 0).astype(int).values
    n = min(len(imgs), max_samples) if max_samples else len(imgs)
    imgs_np = np.array(imgs[:n])
    binary = binary[:n]
    del imgs, lbls; gc.collect()
    return imgs_np, binary

def load_multi_year(years, max_per_year=None):
    """load and concatenate data from multiple years"""
    imgs_list, lbls_list = [], []
    for yr in years:
        im, lb = load_year_subset(yr, max_per_year)
        imgs_list.append(im); lbls_list.append(lb)
        print(f'  {yr}: {len(im)} samples')
    return np.concatenate(imgs_list), np.concatenate(lbls_list)

# load training data
print('Loading training data...')
X_train_raw, y_train_raw = load_multi_year(TRAIN_YEARS)

print('Loading validation data...')
X_val_raw, y_val_raw = load_multi_year(VAL_YEARS)

print('Loading test data...')
X_test_raw, y_test_raw = load_multi_year(TEST_YEARS)

print(f'\nTrain: {len(X_train_raw)}, Val: {len(X_val_raw)}, Test: {len(X_test_raw)}')
Loading training data...
  2000: 98148 samples
  2001: 91985 samples
  2002: 85729 samples
  2003: 80894 samples
  2004: 79512 samples
  2005: 79883 samples
  2006: 80163 samples
  2007: 81140 samples
  2008: 78186 samples
  2009: 70065 samples
  2010: 68111 samples
  2011: 67325 samples
  2012: 66250 samples
  2013: 65486 samples
  2014: 67302 samples
  2015: 68827 samples
Loading validation data...
  2016: 68141 samples
  2017: 67858 samples
Loading test data...
  2018: 68481 samples
  2019: 68637 samples

Train: 1229006, Val: 135999, Test: 137118
In [9]:
# normalize to [0,1] and convert to tensors
X_train_t = torch.tensor(X_train_raw, dtype=torch.float32).unsqueeze(1) / 255.0
y_train_t = torch.tensor(y_train_raw, dtype=torch.float32)
del X_train_raw, y_train_raw; gc.collect()

X_val_t = torch.tensor(X_val_raw, dtype=torch.float32).unsqueeze(1) / 255.0
y_val_t = torch.tensor(y_val_raw, dtype=torch.float32)
del X_val_raw, y_val_raw; gc.collect()

X_test_t = torch.tensor(X_test_raw, dtype=torch.float32).unsqueeze(1) / 255.0
y_test_t = torch.tensor(y_test_raw, dtype=torch.float32)
del X_test_raw, y_test_raw; gc.collect()

print(f'X_train tensor shape: {X_train_t.shape}')
print(f'Pixel value range: [{X_train_t.min():.1f}, {X_train_t.max():.1f}]')
X_train tensor shape: torch.Size([1229006, 1, 64, 60])
Pixel value range: [0.0, 1.0]
In [10]:
BATCH_SIZE = 64
train_loader = DataLoader(TensorDataset(X_train_t, y_train_t), batch_size=BATCH_SIZE, shuffle=True)
val_loader = DataLoader(TensorDataset(X_val_t, y_val_t), batch_size=BATCH_SIZE, shuffle=False)
test_loader = DataLoader(TensorDataset(X_test_t, y_test_t), batch_size=BATCH_SIZE, shuffle=False)
print(f'Train batches: {len(train_loader)}, Val batches: {len(val_loader)}, Test batches: {len(test_loader)}')
Train batches: 19204, Val batches: 2125, Test batches: 2143

CNN Model Architecture¶

I define two models for comparison:

  1. BaselineCNN: simple 3-layer CNN with ReLU, no BatchNorm, high dropout (naive approach)
  2. ImprovedStockCNN: enhanced architecture with:
    • 4 conv blocks with BatchNorm after each layer for stable training
    • LeakyReLU instead of ReLU to avoid dying neuron problem
    • configurable dropout (default 0.3, lower than baseline's 0.5)
    • more conv layers to capture complex chart patterns
In [11]:
class BaselineCNN(nn.Module):
    """Naive baseline: 3 conv layers, ReLU, no BatchNorm"""
    def __init__(self):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, 16, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2),
            nn.Conv2d(16, 32, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2),
            nn.Conv2d(32, 64, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2),
        )
        # after 3x pool: 64->32->16->8, 60->30->15->7
        self.classifier = nn.Sequential(
            nn.Linear(64 * 8 * 7, 128), nn.ReLU(), nn.Dropout(0.5),
            nn.Linear(128, 1), nn.Sigmoid()
        )
    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        return self.classifier(x)


class ImprovedStockCNN(nn.Module):
    """Improved: 4 conv blocks + BatchNorm + LeakyReLU"""
    def __init__(self, dropout=0.3):
        super().__init__()
        self.conv1 = nn.Sequential(nn.Conv2d(1, 32, kernel_size=3, padding=1), nn.BatchNorm2d(32), nn.LeakyReLU(0.1), nn.MaxPool2d(2, 2))
        self.conv2 = nn.Sequential(nn.Conv2d(32, 64, kernel_size=3, padding=1), nn.BatchNorm2d(64), nn.LeakyReLU(0.1), nn.MaxPool2d(2, 2))
        self.conv3 = nn.Sequential(nn.Conv2d(64, 128, kernel_size=3, padding=1), nn.BatchNorm2d(128), nn.LeakyReLU(0.1), nn.MaxPool2d(2, 2))
        self.conv4 = nn.Sequential(nn.Conv2d(128, 128, kernel_size=3, padding=1), nn.BatchNorm2d(128), nn.LeakyReLU(0.1), nn.MaxPool2d(2, 2))
        # after 4x pool: 64->32->16->8->4, 60->30->15->7->3
        self.classifier = nn.Sequential(
            nn.Linear(128 * 4 * 3, 256), nn.BatchNorm1d(256), nn.LeakyReLU(0.1), nn.Dropout(dropout),
            nn.Linear(256, 64), nn.BatchNorm1d(64), nn.LeakyReLU(0.1), nn.Dropout(dropout),
            nn.Linear(64, 1), nn.Sigmoid()
        )
    def forward(self, x):
        x = self.conv1(x)
        x = self.conv2(x)
        x = self.conv3(x)
        x = self.conv4(x)
        x = x.view(x.size(0), -1)
        return self.classifier(x)
In [12]:
baseline_model = BaselineCNN().to(device)
improved_model = ImprovedStockCNN(dropout=0.3).to(device)

print("Baseline CNN")
print(f'  Parameters: {sum(p.numel() for p in baseline_model.parameters()):,}\n')
print("Improved CNN")
print(improved_model)
print(f'\n  Parameters: {sum(p.numel() for p in improved_model.parameters()):,}')
Baseline CNN
  Parameters: 482,305

Improved CNN
ImprovedStockCNN(
  (conv1): Sequential(
    (0): Conv2d(1, 32, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
    (1): BatchNorm2d(32, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    (2): LeakyReLU(negative_slope=0.1)
    (3): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  )
  (conv2): Sequential(
    (0): Conv2d(32, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
    (1): BatchNorm2d(64, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    (2): LeakyReLU(negative_slope=0.1)
    (3): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  )
  (conv3): Sequential(
    (0): Conv2d(64, 128, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
    (1): BatchNorm2d(128, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    (2): LeakyReLU(negative_slope=0.1)
    (3): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  )
  (conv4): Sequential(
    (0): Conv2d(128, 128, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
    (1): BatchNorm2d(128, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    (2): LeakyReLU(negative_slope=0.1)
    (3): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
  )
  (classifier): Sequential(
    (0): Linear(in_features=1536, out_features=256, bias=True)
    (1): BatchNorm1d(256, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    (2): LeakyReLU(negative_slope=0.1)
    (3): Dropout(p=0.3, inplace=False)
    (4): Linear(in_features=256, out_features=64, bias=True)
    (5): BatchNorm1d(64, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    (6): LeakyReLU(negative_slope=0.1)
    (7): Dropout(p=0.3, inplace=False)
    (8): Linear(in_features=64, out_features=1, bias=True)
    (9): Sigmoid()
  )
)

  Parameters: 651,585

Training¶

training setup:

  • optimizer: Adam with weight_decay for L2 regularization
  • lr scheduler: ReduceLROnPlateau, reduce lr by 0.5 if val loss doesnt improve for 3 epochs
  • early stopping to prevent overfitting
  • baseline uses higher lr (0.001) since it's a simpler model
  • improved uses lower lr (0.0005) for more careful optimization
In [13]:
def train_model(model, train_loader, val_loader, num_epochs=30, lr=0.0005,
                weight_decay=1e-4, patience=5, model_name="Model"):
    """Reusable training function with ReduceLROnPlateau and early stopping"""
    criterion = nn.BCELoss()
    optimizer = optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=3)

    history = {'train_loss': [], 'val_loss': [], 'train_acc': [], 'val_acc': []}
    best_val_loss = float('inf')
    patience_counter = 0
    best_state = None

    print(f'\n{"="*65}')
    print(f'Training {model_name} | epochs={num_epochs}, lr={lr}, wd={weight_decay}')
    print(f'{"="*65}')

    for epoch in range(num_epochs):
        model.train()
        running_loss = 0.0; correct = 0; total = 0
        for inputs, labels in train_loader:
            inputs, labels = inputs.to(device), labels.to(device)
            optimizer.zero_grad()
            outputs = model(inputs).squeeze()
            loss = criterion(outputs, labels)
            loss.backward(); optimizer.step()
            running_loss += loss.item() * inputs.size(0)
            predicted = (outputs > 0.5).float()
            total += labels.size(0)
            correct += (predicted == labels).sum().item()

        train_loss = running_loss / total
        train_acc = correct / total
        history['train_loss'].append(train_loss)
        history['train_acc'].append(train_acc)

        model.eval()
        vl = 0.0; vc = 0; vt = 0
        with torch.no_grad():
            for inputs, labels in val_loader:
                inputs, labels = inputs.to(device), labels.to(device)
                outputs = model(inputs).squeeze()
                loss = criterion(outputs, labels)
                vl += loss.item() * inputs.size(0)
                predicted = (outputs > 0.5).float()
                vt += labels.size(0)
                vc += (predicted == labels).sum().item()

        val_loss = vl / vt; val_acc = vc / vt
        history['val_loss'].append(val_loss)
        history['val_acc'].append(val_acc)
        scheduler.step(val_loss)

        lr_now = optimizer.param_groups[0]['lr']
        print(f'  Epoch {epoch+1:2d}/{num_epochs}  '
              f'Train: {train_loss:.4f}/{train_acc:.4f}  '
              f'Val: {val_loss:.4f}/{val_acc:.4f}  LR: {lr_now:.6f}')

        if val_loss < best_val_loss:
            best_val_loss = val_loss
            best_state = {k: v.clone() for k, v in model.state_dict().items()}
            patience_counter = 0
        else:
            patience_counter += 1
            if patience_counter >= patience:
                print(f'  Early stopping at epoch {epoch+1}')
                break
        gc.collect()

    model.load_state_dict(best_state)
    print(f'  Best val loss: {best_val_loss:.4f}\n')
    return model, history


def evaluate_model(model, loader):
    """Get predictions from a trained model"""
    model.eval()
    all_preds, all_labels, all_probs = [], [], []
    with torch.no_grad():
        for inputs, labels in loader:
            inputs = inputs.to(device)
            outputs = model(inputs).squeeze()
            all_probs.extend(outputs.cpu().numpy())
            all_preds.extend((outputs > 0.5).float().cpu().numpy())
            all_labels.extend(labels.numpy())
    return np.array(all_preds), np.array(all_labels), np.array(all_probs)
In [14]:
# train baseline (simpler model, higher lr)
baseline_model, baseline_hist = train_model(
    baseline_model, train_loader, val_loader,
    num_epochs=20, lr=0.001, weight_decay=1e-4, patience=5,
    model_name="Baseline CNN"
)

# train improved (same config as original v1 that achieved ~52%)
improved_model, improved_hist = train_model(
    improved_model, train_loader, val_loader,
    num_epochs=30, lr=0.0005, weight_decay=1e-4, patience=5,
    model_name="Improved CNN"
)
=================================================================
Training Baseline CNN | epochs=20, lr=0.001, wd=0.0001
=================================================================
  Epoch  1/20  Train: 0.6901/0.5312  Val: 0.6920/0.5214  LR: 0.001000
  Epoch  2/20  Train: 0.6874/0.5443  Val: 0.6919/0.5252  LR: 0.001000
  Epoch  3/20  Train: 0.6861/0.5485  Val: 0.6916/0.5324  LR: 0.001000
  Epoch  4/20  Train: 0.6856/0.5496  Val: 0.6910/0.5345  LR: 0.001000
  Epoch  5/20  Train: 0.6852/0.5508  Val: 0.6924/0.5310  LR: 0.001000
  Epoch  6/20  Train: 0.6848/0.5513  Val: 0.6904/0.5330  LR: 0.001000
  Epoch  7/20  Train: 0.6846/0.5521  Val: 0.6915/0.5323  LR: 0.001000
  Epoch  8/20  Train: 0.6846/0.5518  Val: 0.6912/0.5329  LR: 0.001000
  Epoch  9/20  Train: 0.6843/0.5523  Val: 0.6915/0.5336  LR: 0.001000
  Epoch 10/20  Train: 0.6840/0.5524  Val: 0.6923/0.5350  LR: 0.000500
  Epoch 11/20  Train: 0.6825/0.5560  Val: 0.6914/0.5315  LR: 0.000500
  Early stopping at epoch 11
  Best val loss: 0.6904


=================================================================
Training Improved CNN | epochs=30, lr=0.0005, wd=0.0001
=================================================================
  Epoch  1/30  Train: 0.6866/0.5458  Val: 0.6955/0.5203  LR: 0.000500
  Epoch  2/30  Train: 0.6792/0.5643  Val: 0.6973/0.5236  LR: 0.000500
  Epoch  3/30  Train: 0.6756/0.5704  Val: 0.6960/0.5273  LR: 0.000500
  Epoch  4/30  Train: 0.6730/0.5748  Val: 0.6984/0.5208  LR: 0.000500
  Epoch  5/30  Train: 0.6708/0.5782  Val: 0.6998/0.5219  LR: 0.000250
  Epoch  6/30  Train: 0.6650/0.5868  Val: 0.7012/0.5257  LR: 0.000250
  Early stopping at epoch 6
  Best val loss: 0.6955

Evaluation¶

evaluate both models on the held-out test set (2018-2019). compare accuracy, precision, recall, and F1-score side by side.

In [15]:
# evaluate both models on test set
base_preds, base_labels, base_probs = evaluate_model(baseline_model, test_loader)
imp_preds, imp_labels, imp_probs = evaluate_model(improved_model, test_loader)
In [16]:
def calc_metrics(labels, preds):
    return {
        'Accuracy': accuracy_score(labels, preds),
        'Precision': precision_score(labels, preds, zero_division=0),
        'Recall': recall_score(labels, preds, zero_division=0),
        'F1': f1_score(labels, preds, zero_division=0),
    }

base_m = calc_metrics(base_labels, base_preds)
imp_m = calc_metrics(imp_labels, imp_preds)

results_df = pd.DataFrame({'Baseline CNN': base_m, 'Improved CNN': imp_m}).T
results_df = results_df.round(4)
print("=" * 60)
print("MODEL COMPARISON ON TEST SET (2018-2019)")
print("=" * 60)
print(results_df.to_string())
print()
print("Detailed Classification Report (Improved CNN):")
print(classification_report(imp_labels, imp_preds, target_names=['Down', 'Up'], zero_division=0))
============================================================
MODEL COMPARISON ON TEST SET (2018-2019)
============================================================
              Accuracy  Precision  Recall      F1
Baseline CNN    0.5191     0.5254  0.7283  0.6105
Improved CNN    0.5168     0.5340  0.5182  0.5260

Detailed Classification Report (Improved CNN):
              precision    recall  f1-score   support

        Down       0.50      0.52      0.51     66173
          Up       0.53      0.52      0.53     70945

    accuracy                           0.52    137118
   macro avg       0.52      0.52      0.52    137118
weighted avg       0.52      0.52      0.52    137118

In [17]:
fig, axes = plt.subplots(2, 2, figsize=(14, 10))

axes[0,0].plot(baseline_hist['train_loss'], label='Train'); axes[0,0].plot(baseline_hist['val_loss'], label='Val')
axes[0,0].set_title('Baseline CNN - Loss'); axes[0,0].legend(); axes[0,0].grid(True, alpha=0.3)
axes[0,1].plot(baseline_hist['train_acc'], label='Train'); axes[0,1].plot(baseline_hist['val_acc'], label='Val')
axes[0,1].set_title('Baseline CNN - Accuracy'); axes[0,1].legend(); axes[0,1].grid(True, alpha=0.3)

axes[1,0].plot(improved_hist['train_loss'], label='Train'); axes[1,0].plot(improved_hist['val_loss'], label='Val')
axes[1,0].set_title('Improved CNN - Loss'); axes[1,0].legend(); axes[1,0].grid(True, alpha=0.3)
axes[1,1].plot(improved_hist['train_acc'], label='Train'); axes[1,1].plot(improved_hist['val_acc'], label='Val')
axes[1,1].set_title('Improved CNN - Accuracy'); axes[1,1].legend(); axes[1,1].grid(True, alpha=0.3)

for ax in axes.flat:
    ax.set_xlabel('Epoch')
plt.suptitle('Training Curves Comparison', fontsize=14)
plt.tight_layout(); plt.show()
No description has been provided for this image
In [18]:
fig, axes = plt.subplots(1, 2, figsize=(12, 5))
classes = ['Down', 'Up']
for ax, preds, labels, title in [(axes[0], base_preds, base_labels, 'Baseline CNN'),
                                  (axes[1], imp_preds, imp_labels, 'Improved CNN')]:
    cm = confusion_matrix(labels, preds)
    im = ax.imshow(cm, interpolation='nearest', cmap='Blues')
    ax.figure.colorbar(im, ax=ax)
    ax.set(xticks=[0,1], yticks=[0,1], xticklabels=classes, yticklabels=classes,
           ylabel='True Label', xlabel='Predicted Label', title=title)
    for i in range(2):
        for j in range(2):
            ax.text(j, i, format(cm[i,j], 'd'), ha='center', va='center',
                    color='white' if cm[i,j] > cm.max()/2 else 'black')
plt.suptitle('Confusion Matrices', fontsize=13)
plt.tight_layout(); plt.show()
No description has been provided for this image
In [19]:
fig, axes = plt.subplots(1, 2, figsize=(14, 4))
for ax, probs, labels, title in [(axes[0], base_probs, base_labels, 'Baseline CNN'),
                                  (axes[1], imp_probs, imp_labels, 'Improved CNN')]:
    ax.hist(probs[labels == 0], bins=50, alpha=0.5, label='Actual Down', color='red')
    ax.hist(probs[labels == 1], bins=50, alpha=0.5, label='Actual Up', color='green')
    ax.axvline(x=0.5, color='black', linestyle='--', label='Threshold')
    ax.set_xlabel('Predicted Probability'); ax.set_ylabel('Count')
    ax.set_title(title); ax.legend()
plt.suptitle('Distribution of Predicted Probabilities', fontsize=13)
plt.tight_layout(); plt.show()
No description has been provided for this image

Hyperparameter Experiments¶

I test three configurations of the improved model to see how learning rate and dropout affect performance. each config trains for 15 epochs to save compute time.

In [20]:
# hyperparameter experiment: test different configurations (15 epochs each)
configs = [
    {'name': 'A: lr=1e-3, dropout=0.5',  'lr': 0.001,  'dropout': 0.5},
    {'name': 'B: lr=5e-4, dropout=0.3',  'lr': 0.0005, 'dropout': 0.3},
    {'name': 'C: lr=3e-4, dropout=0.2',  'lr': 0.0003, 'dropout': 0.2},
]

exp_results = []
for cfg in configs:
    m = ImprovedStockCNN(dropout=cfg['dropout']).to(device)
    m, hist = train_model(m, train_loader, val_loader,
                          num_epochs=15, lr=cfg['lr'], weight_decay=1e-4,
                          patience=5, model_name=cfg['name'])
    preds, labels, _ = evaluate_model(m, test_loader)
    exp_results.append({
        'Config': cfg['name'],
        'Best Val Loss': f"{min(hist['val_loss']):.4f}",
        'Best Val Acc': f"{max(hist['val_acc']):.4f}",
        'Test Acc': f"{accuracy_score(labels, preds):.4f}",
        'Test F1': f"{f1_score(labels, preds, zero_division=0):.4f}",
    })
    del m; gc.collect(); torch.cuda.empty_cache()

exp_df = pd.DataFrame(exp_results)
print('\n' + '='*70)
print('HYPERPARAMETER EXPERIMENT RESULTS')
print('='*70)
print(exp_df.to_string(index=False))
=================================================================
Training A: lr=1e-3, dropout=0.5 | epochs=15, lr=0.001, wd=0.0001
=================================================================
  Epoch  1/15  Train: 0.6886/0.5395  Val: 0.6903/0.5312  LR: 0.001000
  Epoch  2/15  Train: 0.6847/0.5532  Val: 0.6913/0.5310  LR: 0.001000
  Epoch  3/15  Train: 0.6831/0.5568  Val: 0.6908/0.5363  LR: 0.001000
  Epoch  4/15  Train: 0.6824/0.5587  Val: 0.6899/0.5362  LR: 0.001000
  Epoch  5/15  Train: 0.6816/0.5604  Val: 0.6919/0.5324  LR: 0.001000
  Epoch  6/15  Train: 0.6815/0.5600  Val: 0.6937/0.5302  LR: 0.001000
  Epoch  7/15  Train: 0.6811/0.5606  Val: 0.6936/0.5287  LR: 0.001000
  Epoch  8/15  Train: 0.6808/0.5612  Val: 0.6947/0.5235  LR: 0.000500
  Epoch  9/15  Train: 0.6779/0.5672  Val: 0.6980/0.5230  LR: 0.000500
  Early stopping at epoch 9
  Best val loss: 0.6899


=================================================================
Training B: lr=5e-4, dropout=0.3 | epochs=15, lr=0.0005, wd=0.0001
=================================================================
  Epoch  1/15  Train: 0.6868/0.5442  Val: 0.6949/0.5278  LR: 0.000500
  Epoch  2/15  Train: 0.6793/0.5640  Val: 0.6996/0.5210  LR: 0.000500
  Epoch  3/15  Train: 0.6759/0.5700  Val: 0.7013/0.5183  LR: 0.000500
  Epoch  4/15  Train: 0.6733/0.5746  Val: 0.6996/0.5292  LR: 0.000500
  Epoch  5/15  Train: 0.6711/0.5767  Val: 0.6971/0.5228  LR: 0.000250
  Epoch  6/15  Train: 0.6653/0.5861  Val: 0.7011/0.5242  LR: 0.000250
  Early stopping at epoch 6
  Best val loss: 0.6949


=================================================================
Training C: lr=3e-4, dropout=0.2 | epochs=15, lr=0.0003, wd=0.0001
=================================================================
  Epoch  1/15  Train: 0.6855/0.5473  Val: 0.6972/0.5278  LR: 0.000300
  Epoch  2/15  Train: 0.6754/0.5711  Val: 0.6985/0.5252  LR: 0.000300
  Epoch  3/15  Train: 0.6696/0.5801  Val: 0.6970/0.5280  LR: 0.000300
  Epoch  4/15  Train: 0.6652/0.5859  Val: 0.7055/0.5285  LR: 0.000300
  Epoch  5/15  Train: 0.6616/0.5907  Val: 0.7077/0.5239  LR: 0.000300
  Epoch  6/15  Train: 0.6591/0.5938  Val: 0.7052/0.5246  LR: 0.000300
  Epoch  7/15  Train: 0.6572/0.5967  Val: 0.7129/0.5184  LR: 0.000150
  Epoch  8/15  Train: 0.6494/0.6069  Val: 0.7098/0.5241  LR: 0.000150
  Early stopping at epoch 8
  Best val loss: 0.6970


======================================================================
HYPERPARAMETER EXPERIMENT RESULTS
======================================================================
                 Config Best Val Loss Best Val Acc Test Acc Test F1
A: lr=1e-3, dropout=0.5        0.6899       0.5363   0.5150  0.5928
B: lr=5e-4, dropout=0.3        0.6949       0.5292   0.5163  0.5965
C: lr=3e-4, dropout=0.2        0.6970       0.5285   0.5191  0.5539

Discussion and Conclusion¶

Misclassification Analysis¶

examine examples where the improved model made wrong predictions with high confidence. comparing confidently wrong vs confidently correct predictions helps us understand what the model has learned and where it struggles.

In [21]:
# load 2018 images for visualization
mis_imgs, mis_lbls = load_year_data(2018)
mis_binary = (mis_lbls['Ret_5d'] > 0).astype(int).values

# get improved model predictions for 2018
mis_tensor = torch.tensor(np.array(mis_imgs[:len(mis_binary)]), dtype=torch.float32).unsqueeze(1) / 255.0
mis_loader = DataLoader(TensorDataset(mis_tensor, torch.tensor(mis_binary, dtype=torch.float32)),
                        batch_size=256, shuffle=False)
mis_preds, mis_labels_arr, mis_probs = evaluate_model(improved_model, mis_loader)

# find misclassified samples, sort by confidence (most confidently wrong first)
wrong_mask = mis_preds != mis_labels_arr
wrong_idx = np.where(wrong_mask)[0]
wrong_conf = np.abs(mis_probs[wrong_idx] - 0.5)
sorted_wrong = wrong_idx[np.argsort(-wrong_conf)]

# show top 10 most confidently wrong predictions
fig, axes = plt.subplots(2, 5, figsize=(16, 7))
for i, idx in enumerate(sorted_wrong[:10]):
    ax = axes[i // 5][i % 5]
    ax.imshow(mis_imgs[idx], cmap='gray')
    true_lbl = 'Up' if mis_labels_arr[idx] == 1 else 'Down'
    pred_lbl = 'Up' if mis_preds[idx] == 1 else 'Down'
    prob = mis_probs[idx]
    ax.set_title(f'True: {true_lbl}\nPred: {pred_lbl} ({prob:.2f})', fontsize=9, color='red')
    ax.axis('off')
plt.suptitle('Most Confidently Misclassified Examples (2018)', fontsize=13)
plt.tight_layout(); plt.show()

# show correctly classified for contrast
correct_idx = np.where(~wrong_mask)[0]
correct_conf = np.abs(mis_probs[correct_idx] - 0.5)
sorted_correct = correct_idx[np.argsort(-correct_conf)]

fig, axes = plt.subplots(2, 5, figsize=(16, 7))
for i, idx in enumerate(sorted_correct[:10]):
    ax = axes[i // 5][i % 5]
    ax.imshow(mis_imgs[idx], cmap='gray')
    true_lbl = 'Up' if mis_labels_arr[idx] == 1 else 'Down'
    prob = mis_probs[idx]
    ax.set_title(f'True: {true_lbl} ({prob:.2f})', fontsize=9, color='green')
    ax.axis('off')
plt.suptitle('Most Confidently Correct Examples (2018)', fontsize=13)
plt.tight_layout(); plt.show()

# summary stats
print(f'Total test samples (2018): {len(mis_labels_arr)}')
print(f'Misclassified: {wrong_mask.sum()} ({wrong_mask.mean()*100:.1f}%)')
print(f'Correct: {(~wrong_mask).sum()} ({(~wrong_mask).mean()*100:.1f}%)')
print(f'\nMisclassification breakdown:')
print(f'  True Up,   predicted Down: {((mis_labels_arr==1) & (mis_preds==0)).sum()}')
print(f'  True Down, predicted Up:   {((mis_labels_arr==0) & (mis_preds==1)).sum()}')

del mis_imgs, mis_lbls, mis_tensor, mis_loader; gc.collect()
No description has been provided for this image
No description has been provided for this image
Total test samples (2018): 68481
Misclassified: 33256 (48.6%)
Correct: 35225 (51.4%)

Misclassification breakdown:
  True Up,   predicted Down: 18329
  True Down, predicted Up:   14927
Out[21]:
39117

Model Comparison¶

the baseline and improved CNN achieve similar overall accuracy (~51.7-51.9%), but behave very differently:

  • Baseline CNN has high recall (72.8%) but low precision (52.5%) — it predicts "Up" most of the time. this happens because the simple model without BatchNorm cannot learn nuanced patterns, so it defaults to the majority-ish class. its high F1 (0.61) is misleading: it's essentially a biased predictor.
  • Improved CNN is more balanced: precision (53.4%) and recall (51.8%) are nearly equal, meaning it doesn't blindly favor one class. this is a more useful model in practice — a model that always says "Up" is useless for trading decisions.

the improved architecture's advantages:

  • BatchNorm stabilizes training and enables the model to learn more discriminative features
  • LeakyReLU avoids dead neurons that standard ReLU can cause
  • 4 conv blocks (vs baseline's 3) capture more complex spatial patterns
  • Lower dropout (0.3 vs 0.5) retains more learned features

Key Findings¶

  1. Time-series split is essential: using chronological train/val/test split avoids look-ahead bias, which is critical for financial data
  2. Stock prediction is inherently noisy: both models achieve ~52% accuracy on out-of-sample data, consistent with the efficient market hypothesis and the low signal-to-noise ratio in financial markets
  3. Balanced prediction matters more than accuracy: a model biased toward one class (like the baseline) can have inflated F1 but provides no real predictive value
  4. Misclassification patterns: the model misses more actual Up cases (18,329) than actual Down cases (14,927), suggesting upward trends in 2018 had patterns not seen during 2000-2015 training period
  5. Confidently wrong predictions: as shown in the misclassification analysis, visually similar chart patterns can lead to opposite outcomes, highlighting the limits of pure image-based prediction

Hyperparameter Sensitivity¶

the experiment results show all three configs achieve similar test accuracy (~51.5-51.9%), confirming that the signal ceiling is low for this task. key observations:

  • Config A (higher lr=1e-3, higher dropout=0.5): best validation loss (0.6899) but early-stopped quickly due to overfitting
  • Config B (moderate lr=5e-4, dropout=0.3): matches our main improved model configuration
  • Config C (lower lr=3e-4, lower dropout=0.2): largest train-val gap, showing that lower dropout leads to more overfitting despite slower learning

Limitations and Future Work¶

  • only used 20-day chart images with volume bars; incorporating additional features (market cap, sector, macro indicators) could improve performance
  • CNN architecture could be enhanced with residual connections (ResNet-style) or attention mechanisms
  • ensemble methods (combining predictions from multiple models) could reduce variance
  • the 0.5 classification threshold could be optimized via ROC analysis
  • rolling-window retraining could help the model adapt to changing market regimes (e.g., 2018 patterns not present in 2000-2015 training data)
  • could explore transfer learning from pre-trained image models