TensorFlow for Trading: Neural Network Price Prediction
Deep learning models offer the ability to learn complex temporal patterns from raw financial data. LSTMs capture long-range dependencies in price sequences, convolutional networks detect local patterns similar to chart formations, and attention mechanisms weigh the relevance of different time steps dynamically. However, deep learning for trading is significantly harder than for computer vision or NLP. Financial time series have low signal-to-noise ratios, non-stationarity, and regime changes that violate the i.i.d. assumptions underlying most neural network training. This guide covers architectures that address these challenges with production-tested patterns.Key Takeaways
- LSTMs remain the workhorse for financial time series due to their ability to model sequential dependencies with forget gates.
- Attention mechanisms improve LSTM performance by focusing on relevant historical time steps.
- Data preprocessing (stationarity, normalization, sequence construction) matters more than architecture choice.
- Regularization is essential: dropout, early stopping, and small model capacity prevent overfitting to noise.
Data Preparation for Neural Networks
Neural networks require careful preprocessing that differs from traditional ML pipelines. ``python
import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
from sklearn.preprocessing import StandardScaler
def prepare_sequences(
features: np.ndarray,
target: np.ndarray,
sequence_length: int = 60,
forecast_horizon: int = 5,
) -> tuple[np.ndarray, np.ndarray]:
"""
Create sliding window sequences for time series models.
Args:
features: (T, F) array of feature values
target: (T,) array of target values
sequence_length: number of past time steps per sample
forecast_horizon: gap between end of input and target
Returns:
X: (N, sequence_length, F) input sequences
y: (N,) targets
"""
X, y = [], []
for i in range(sequence_length, len(features) - forecast_horizon + 1):
X.append(features[i - sequence_length:i])
y.append(target[i + forecast_horizon - 1])
return np.array(X), np.array(y)
def create_dataset(
df: pd.DataFrame,
target_col: str = "target",
feature_cols: list[str] | None = None,
sequence_length: int = 60,
forecast_horizon: int = 5,
test_ratio: float = 0.2,
val_ratio: float = 0.1,
) -> dict:
"""
Full preprocessing pipeline: scale, sequence, and split temporally.
"""
if feature_cols is None:
feature_cols = [c for c in df.columns if c != target_col]
features = df[feature_cols].values
target = df[target_col].values
# Temporal split points
n = len(features)
train_end = int(n * (1 - test_ratio - val_ratio))
val_end = int(n * (1 - test_ratio))
# Scale on training data only
scaler = StandardScaler()
features[:train_end] = scaler.fit_transform(features[:train_end])
features[train_end:val_end] = scaler.transform(features[train_end:val_end])
features[val_end:] = scaler.transform(features[val_end:])
# Create sequences
X, y = prepare_sequences(features, target, sequence_length, forecast_horizon)
# Adjust split indices for sequence creation offset
offset = sequence_length + forecast_horizon - 1
train_samples = train_end - offset
val_samples = val_end - offset
return {
"X_train": X[:train_samples],
"y_train": y[:train_samples],
"X_val": X[train_samples:val_samples],
"y_val": y[train_samples:val_samples],
"X_test": X[val_samples:],
"y_test": y[val_samples:],
"scaler": scaler,
"feature_names": feature_cols,
}
`
LSTM Model for Price Direction
The LSTM architecture with dropout regularization is the most reliable starting point for financial time series.
`python
def build_lstm_model(
sequence_length: int,
n_features: int,
lstm_units: list[int] = [64, 32],
dropout_rate: float = 0.3,
learning_rate: float = 0.001,
task: str = "classification",
) -> keras.Model:
"""
LSTM model with stacked layers, dropout, and batch normalization.
Args:
sequence_length: length of input sequences
n_features: number of input features
lstm_units: units per LSTM layer
dropout_rate: dropout probability
task: 'classification' for direction, 'regression' for returns
"""
inputs = keras.Input(shape=(sequence_length, n_features))
x = inputs
# Stacked LSTM layers
for i, units in enumerate(lstm_units):
return_sequences = i < len(lstm_units) - 1
x = layers.LSTM(
units,
return_sequences=return_sequences,
kernel_regularizer=keras.regularizers.l2(1e-4),
)(x)
x = layers.BatchNormalization()(x)
x = layers.Dropout(dropout_rate)(x)
# Dense head
x = layers.Dense(16, activation="relu")(x)
x = layers.Dropout(dropout_rate / 2)(x)
if task == "classification":
outputs = layers.Dense(1, activation="sigmoid")(x)
loss = "binary_crossentropy"
metrics = ["accuracy"]
else:
outputs = layers.Dense(1)(x)
loss = "huber"
metrics = ["mae"]
model = keras.Model(inputs, outputs)
model.compile(
optimizer=keras.optimizers.Adam(learning_rate=learning_rate),
loss=loss,
metrics=metrics,
)
return model
Build and inspect
model = build_lstm_model(
sequence_length=60,
n_features=25,
lstm_units=[64, 32],
)
model.summary()
`
CNN-LSTM Hybrid Architecture
Combine 1D convolutions (for local pattern detection) with LSTMs (for sequential modeling).
`python
def build_cnn_lstm_model(
sequence_length: int,
n_features: int,
conv_filters: list[int] = [64, 32],
kernel_size: int = 3,
lstm_units: int = 32,
dropout_rate: float = 0.3,
) -> keras.Model:
"""
CNN extracts local patterns, LSTM models temporal dependencies.
Particularly effective for detecting chart-pattern-like structures.
"""
inputs = keras.Input(shape=(sequence_length, n_features))
x = inputs
# Convolutional feature extraction
for filters in conv_filters:
x = layers.Conv1D(
filters, kernel_size, padding="causal", activation="relu"
)(x)
x = layers.BatchNormalization()(x)
x = layers.MaxPooling1D(pool_size=2)(x)
x = layers.Dropout(dropout_rate)(x)
# LSTM temporal modeling
x = layers.LSTM(lstm_units)(x)
x = layers.Dropout(dropout_rate)(x)
# Output
x = layers.Dense(16, activation="relu")(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)
model.compile(
optimizer=keras.optimizers.Adam(0.001),
loss="binary_crossentropy",
metrics=["accuracy"],
)
return model
`
Attention-Enhanced LSTM
Adding an attention mechanism lets the model focus on the most relevant time steps rather than relying solely on the final hidden state.
`python
class TemporalAttention(layers.Layer):
"""Attention mechanism for time series sequences."""
def __init__(self, units: int = 32, **kwargs):
super().__init__(**kwargs)
self.units = units
self.W = layers.Dense(units, activation="tanh")
self.V = layers.Dense(1, use_bias=False)
def call(self, hidden_states):
# hidden_states: (batch, timesteps, features)
score = self.V(self.W(hidden_states)) # (batch, timesteps, 1)
attention_weights = tf.nn.softmax(score, axis=1)
context = tf.reduce_sum(attention_weights * hidden_states, axis=1)
return context, attention_weights
def build_attention_lstm(
sequence_length: int,
n_features: int,
lstm_units: int = 64,
attention_units: int = 32,
dropout_rate: float = 0.3,
) -> keras.Model:
"""LSTM with temporal attention for interpretable predictions."""
inputs = keras.Input(shape=(sequence_length, n_features))
# LSTM returns all hidden states
lstm_out = layers.LSTM(
lstm_units, return_sequences=True,
kernel_regularizer=keras.regularizers.l2(1e-4),
)(inputs)
lstm_out = layers.Dropout(dropout_rate)(lstm_out)
# Attention layer
context, attention_weights = TemporalAttention(attention_units)(lstm_out)
# Dense head
x = layers.Dense(32, activation="relu")(context)
x = layers.Dropout(dropout_rate)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)
model.compile(
optimizer=keras.optimizers.Adam(0.001),
loss="binary_crossentropy",
metrics=["accuracy"],
)
return model
`
Training with Proper Callbacks
Use callbacks to prevent overfitting and save the best model during training.
`python
def train_model(
model: keras.Model,
data: dict,
epochs: int = 100,
batch_size: int = 64,
patience: int = 15,
) -> keras.callbacks.History:
"""
Train with early stopping, learning rate reduction,
and model checkpointing.
"""
callbacks = [
keras.callbacks.EarlyStopping(
monitor="val_loss",
patience=patience,
restore_best_weights=True,
),
keras.callbacks.ReduceLROnPlateau(
monitor="val_loss",
factor=0.5,
patience=5,
min_lr=1e-6,
),
keras.callbacks.ModelCheckpoint(
"best_model.keras",
monitor="val_loss",
save_best_only=True,
),
]
history = model.fit(
data["X_train"], data["y_train"],
validation_data=(data["X_val"], data["y_val"]),
epochs=epochs,
batch_size=batch_size,
callbacks=callbacks,
verbose=1,
)
return history
`
Model Evaluation and Backtesting
Evaluate neural network predictions with both ML metrics and financial performance. If you want a quick baseline to compare against, our Strategy Builder lets you backtest a simple rules-based strategy on the same data without writing this evaluation pipeline yourself.
`python
def evaluate_nn_predictions(
model: keras.Model,
X_test: np.ndarray,
y_test: np.ndarray,
actual_returns: np.ndarray,
threshold: float = 0.5,
) -> dict:
"""
Comprehensive evaluation combining ML and financial metrics.
"""
# Raw predictions
y_prob = model.predict(X_test, verbose=0).flatten()
y_pred = (y_prob >= threshold).astype(int)
# ML metrics
from sklearn.metrics import (
accuracy_score, precision_score, recall_score, roc_auc_score
)
ml_metrics = {
"accuracy": accuracy_score(y_test, y_pred),
"precision": precision_score(y_test, y_pred, zero_division=0),
"recall": recall_score(y_test, y_pred, zero_division=0),
"auc_roc": roc_auc_score(y_test, y_prob),
}
# Financial metrics
strategy_returns = np.where(y_pred == 1, actual_returns, 0)
cumulative = np.cumprod(1 + strategy_returns)
peak = np.maximum.accumulate(cumulative)
drawdown = (cumulative - peak) / peak
financial_metrics = {
"total_return": cumulative[-1] - 1,
"sharpe_ratio": (
np.mean(strategy_returns) / np.std(strategy_returns)
- np.sqrt(252) if np.std(strategy_returns) > 0 else 0
),
"max_drawdown": drawdown.min(),
"win_rate": (
np.sum((y_pred == 1) & (actual_returns > 0))
/ np.sum(y_pred == 1) if np.sum(y_pred == 1) > 0 else 0
),
"trade_count": np.sum(y_pred == 1),
}
return {ml_metrics, financial_metrics}
`
Production Deployment Considerations
Deploying neural network models for live trading requires additional infrastructure.
`python
Model serialization for deployment
model.save("models/lstm_trading_v1.keras")
loaded_model = keras.models.load_model(
"models/lstm_trading_v1.keras",
custom_objects={"TemporalAttention": TemporalAttention},
)
TensorFlow Lite for low-latency inference
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()
with open("models/lstm_trading_v1.tflite", "wb") as f:
f.write(tflite_model)
Inference benchmarking
import time
test_input = np.random.randn(1, 60, 25).astype(np.float32)
times = []
for _ in range(100):
start = time.perf_counter()
_ = model.predict(test_input, verbose=0)
times.append(time.perf_counter() - start)
print(f"Avg inference: {np.mean(times)*1000:.1f}ms")
``