Обучение модели на основе CNN для анализа графиков
Сверточные нейронные сети (CNN) традиционно используются для обработки изображений. В применении к крипто-торговле есть два подхода: анализ графиков как изображений (буквально скриншоты) и применение 1D CNN к числовым временным рядам OHLCV данных. Оба подхода дают интересные результаты.
Подход 1: 1D CNN для временных рядов
1D CNN применяет свёрточные фильтры вдоль временной оси. Каждый фильтр учится распознавать локальные паттерны определённой длины — аналог «ручного» поиска паттернов, но автоматический.
import torch
import torch.nn as nn
class CryptoCNN1D(nn.Module):
def __init__(self, input_channels, seq_len=60):
super().__init__()
# Multi-scale convolutions: разные размеры ядер для разных таймфреймов
self.conv_short = nn.Sequential(
nn.Conv1d(input_channels, 64, kernel_size=3, padding=1),
nn.BatchNorm1d(64),
nn.ReLU()
)
self.conv_medium = nn.Sequential(
nn.Conv1d(input_channels, 64, kernel_size=9, padding=4),
nn.BatchNorm1d(64),
nn.ReLU()
)
self.conv_long = nn.Sequential(
nn.Conv1d(input_channels, 64, kernel_size=21, padding=10),
nn.BatchNorm1d(64),
nn.ReLU()
)
# Объединяем все масштабы
self.residual_blocks = nn.ModuleList([
ResidualBlock1D(192, 128),
ResidualBlock1D(128, 64),
])
self.global_avg_pool = nn.AdaptiveAvgPool1d(1)
self.global_max_pool = nn.AdaptiveMaxPool1d(1)
self.classifier = nn.Sequential(
nn.Linear(128, 64),
nn.ReLU(),
nn.Dropout(0.3),
nn.Linear(64, 3) # buy / hold / sell
)
def forward(self, x):
# x: (batch, channels, seq_len) — нужна транспозиция!
x_t = x.permute(0, 2, 1)
short = self.conv_short(x_t)
medium = self.conv_medium(x_t)
long = self.conv_long(x_t)
combined = torch.cat([short, medium, long], dim=1)
for block in self.residual_blocks:
combined = block(combined)
avg = self.global_avg_pool(combined).squeeze(-1)
max_ = self.global_max_pool(combined).squeeze(-1)
pooled = torch.cat([avg, max_], dim=1)
return self.classifier(pooled)
class ResidualBlock1D(nn.Module):
def __init__(self, in_channels, out_channels):
super().__init__()
self.conv1 = nn.Conv1d(in_channels, out_channels, 3, padding=1)
self.bn1 = nn.BatchNorm1d(out_channels)
self.conv2 = nn.Conv1d(out_channels, out_channels, 3, padding=1)
self.bn2 = nn.BatchNorm1d(out_channels)
self.shortcut = nn.Conv1d(in_channels, out_channels, 1) if in_channels != out_channels else nn.Identity()
def forward(self, x):
residual = self.shortcut(x)
x = torch.relu(self.bn1(self.conv1(x)))
x = self.bn2(self.conv2(x))
return torch.relu(x + residual)
Подход 2: CNN на графиках как изображениях
Рендерим свечной график как изображение, передаём в ResNet/EfficientNet для классификации сигнала.
from PIL import Image, ImageDraw
import numpy as np
import torchvision.models as models
def render_candlestick_image(ohlcv_data, width=224, height=224):
"""Рендерим последние N свечей как PIL Image"""
img = Image.new('RGB', (width, height), color='black')
draw = ImageDraw.Draw(img)
n_candles = len(ohlcv_data)
candle_width = width / n_candles * 0.8
price_min = ohlcv_data['low'].min()
price_max = ohlcv_data['high'].max()
price_range = price_max - price_min
def price_to_y(price):
return height - int((price - price_min) / price_range * height * 0.9) - int(height * 0.05)
for i, (_, row) in enumerate(ohlcv_data.iterrows()):
x = int(i * width / n_candles) + int(candle_width / 2)
# Фитиль
draw.line([(x, price_to_y(row['high'])), (x, price_to_y(row['low']))],
fill='white', width=1)
# Тело свечи
open_y = price_to_y(row['open'])
close_y = price_to_y(row['close'])
color = (0, 200, 0) if row['close'] >= row['open'] else (200, 0, 0)
x1 = x - int(candle_width / 2)
x2 = x + int(candle_width / 2)
draw.rectangle([x1, min(open_y, close_y), x2, max(open_y, close_y)],
fill=color)
return img
class CandlestickCNN(nn.Module):
def __init__(self, n_classes=3, pretrained=True):
super().__init__()
# Используем pretrained EfficientNet как backbone
self.backbone = models.efficientnet_b0(pretrained=pretrained)
n_features = self.backbone.classifier[1].in_features
self.backbone.classifier = nn.Sequential(
nn.Dropout(0.3),
nn.Linear(n_features, n_classes)
)
def forward(self, x):
return self.backbone(x)
Pretraining на синтетических данных
Важная техника: предварительно обучаем CNN на синтетических свечных данных с известными паттернами (программно генерируем «голову и плечи», треугольники и т.д. с метками). Это даёт модели начальное понимание паттернов перед дообучением на реальных данных.
TCN (Temporal Convolutional Network)
Более современная альтернативная 1D CNN архитектура с dilated causal convolutions:
class TCNBlock(nn.Module):
def __init__(self, in_channels, out_channels, kernel_size, dilation):
super().__init__()
padding = (kernel_size - 1) * dilation
self.conv = nn.Conv1d(in_channels, out_channels, kernel_size,
padding=padding, dilation=dilation)
self.chomp = nn.Identity() # обрезаем будущее: x[:, :, :-padding]
self.relu = nn.ReLU()
self.dropout = nn.Dropout(0.1)
def forward(self, x):
out = self.conv(x)
# Causal: оставляем только прошлое
if self.conv.padding[0] > 0:
out = out[:, :, :-self.conv.padding[0]]
return self.dropout(self.relu(out))
Dilated convolutions экспоненциально увеличивают receptive field без роста параметров: с dilation 1, 2, 4, 8 при kernel=3 охватываем 32 timestep'а.
CNN+LSTM гибрид
CNN извлекает локальные паттерны, LSTM захватывает долгосрочные зависимости:
class CNN_LSTM(nn.Module):
def __init__(self, input_size, cnn_channels=64, lstm_hidden=128):
super().__init__()
self.cnn = nn.Sequential(
nn.Conv1d(input_size, cnn_channels, 3, padding=1),
nn.ReLU(),
nn.Conv1d(cnn_channels, cnn_channels, 3, padding=1),
nn.ReLU()
)
self.lstm = nn.LSTM(cnn_channels, lstm_hidden, batch_first=True)
self.fc = nn.Linear(lstm_hidden, 1)
def forward(self, x):
# CNN expects (batch, channels, seq)
cnn_out = self.cnn(x.permute(0, 2, 1)).permute(0, 2, 1)
lstm_out, _ = self.lstm(cnn_out)
return self.fc(lstm_out[:, -1, :])
Разрабатываем CNN-based систему для анализа графических паттернов: 1D TCN на числовых данных, 2D CNN на rendered charts, CNN+LSTM гибрид, ensemble предсказания и interpretability через activation maps.