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.
from google.colab import drive
drive.mount('/content/drive')
Mounted at /content/drive
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')
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.
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)
0
# 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%
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()
# 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()
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.
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
# 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]
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:
- BaselineCNN: simple 3-layer CNN with ReLU, no BatchNorm, high dropout (naive approach)
- 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
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)
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
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)
# 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.
# 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)
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
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()
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()
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()
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.
# 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.
# 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()
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
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¶
- Time-series split is essential: using chronological train/val/test split avoids look-ahead bias, which is critical for financial data
- 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
- 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
- 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
- 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