Chapter 336: Continuous Normalizing Flows — Modeling Market Dynamics with Neural ODEs
Chapter 336: Continuous Normalizing Flows — Modeling Market Dynamics with Neural ODEs
Overview
Continuous Normalizing Flows (CNFs) represent a paradigm shift in generative modeling by replacing discrete transformation steps with continuous dynamics governed by neural ordinary differential equations (Neural ODEs). In trading, CNFs enable modeling the continuous evolution of market states, learning complex return distributions, and generating realistic market scenarios for risk management and strategy development.
This chapter explores how to leverage CNFs for cryptocurrency trading, using the power of continuous transformations to capture market dynamics that discrete models may miss.
Core Concepts
What are Continuous Normalizing Flows?
Unlike discrete normalizing flows that apply a sequence of fixed transformations, CNFs define a continuous-time transformation from a simple base distribution to a complex target distribution:
Discrete Flow: z₀ → f₁ → z₁ → f₂ → z₂ → ... → zₙContinuous Flow: z(0) → ODE dynamics → z(T)
Key insight: The transformation is defined by an ODE:dz/dt = f(z(t), t; θ)
Where f is a neural network parameterizing the velocity field.Why Continuous Normalizing Flows for Trading?
- Flexible Distributions: Model arbitrary return distributions without architectural constraints
- Continuous Dynamics: Capture smooth market transitions rather than discrete jumps
- Efficient Sampling: Generate market scenarios by solving ODEs forward
- Exact Likelihood: Compute exact log-probabilities via instantaneous change of variables
- Memory Efficient: Constant memory cost regardless of transformation depth (adjoint method)
From Discrete to Continuous Flows
Discrete Normalizing Flow:├── Fixed number of layers├── Change of variables: log p(x) = log p(z) - Σ log|det(∂fᵢ/∂zᵢ₋₁)|├── Jacobian determinant at each layer└── Memory scales with depth
Continuous Normalizing Flow (FFJORD):├── Continuous transformation via ODE├── Change of variables: log p(x) = log p(z(0)) - ∫₀ᵀ tr(∂f/∂z(t)) dt├── Trace of Jacobian (not full determinant!)└── O(1) memory via adjoint methodTrading Strategy
Strategy Overview: Use CNFs to learn the joint distribution of market features and future returns. Trading signals are generated by:
- Computing the likelihood of current market states
- Sampling conditional return distributions
- Identifying regime changes through distribution dynamics
Signal Generation
1. Feature Extraction: - Compute market features: returns, volatility, orderbook imbalance - Normalize features to match training distribution
2. Likelihood Computation: - Transform current state through learned flow - Compute log-likelihood via trace integral - High likelihood → familiar pattern
3. Conditional Sampling: - Given current features, sample future return distribution - Compute expected return and confidence intervals - Mean > 0 with high confidence → Long signal
4. Regime Detection: - Track likelihood trajectory over time - Sudden drops indicate regime change - Reduce exposure during transitionsEntry Signals
- Long Signal: Conditional return distribution centered above zero with narrow variance
- Short Signal: Conditional return distribution centered below zero with narrow variance
- No Trade: Wide variance (uncertain) or likelihood below threshold (novel state)
Risk Management
- Likelihood Filtering: Only trade when current state has high likelihood under learned distribution
- Variance-Based Sizing: Position size inversely proportional to conditional variance
- Regime Detection: Reduce exposure when likelihood drops significantly
- ODE Divergence: Monitor numerical stability of flow transformations
Technical Specification
Mathematical Foundation
Neural ODE Definition
The core of CNF is a Neural ODE that defines the transformation:
State dynamics:dz/dt = f_θ(z(t), t)
Where:├── z(t) ∈ ℝᵈ is the state at time t├── f_θ: ℝᵈ × ℝ → ℝᵈ is a neural network├── t ∈ [0, T] is the integration time└── θ are learnable parameters
Solution via numerical integration:z(T) = z(0) + ∫₀ᵀ f_θ(z(t), t) dtInstantaneous Change of Variables
The log-probability evolves according to:
d log p(z(t))/dt = -tr(∂f_θ/∂z(t))
This gives us:log p(z(T)) = log p(z(0)) - ∫₀ᵀ tr(∂f_θ/∂z(t)) dt
Key properties:├── Only need trace of Jacobian, not full determinant!├── Trace is O(d) while determinant is O(d³)├── Enables high-dimensional modeling└── Hutchinson's trace estimator: O(d) → O(1)Hutchinson’s Trace Estimator
For efficient trace computation:
tr(A) = E_v[v^T A v]
Where v is a random vector with E[vv^T] = I
For Jacobian:tr(∂f/∂z) ≈ E_ε[ε^T (∂f/∂z) ε] = E_ε[ε^T ∂(f^T ε)/∂z] (via VJP)
This requires only one vector-Jacobian product!FFJORD Training Objective
Loss = -E_{x~p_data}[log p_θ(x)]
Where:log p_θ(x) = log p(z(0)) - ∫₀ᵀ tr(∂f_θ/∂z(t)) dtz(0) = ODESolve(z(T)=x, f_θ, T→0) # Reverse ODE
Training procedure:1. Sample x from data2. Solve ODE backward to get z(0)3. Compute log p(z(0)) under base distribution4. Estimate trace integral during backward pass5. Minimize negative log-likelihoodArchitecture Diagram
Market Data Stream │ ▼ ┌─────────────────────────────┐ │ Feature Engineering │ │ ├── Multi-scale returns │ │ ├── Volatility measures │ │ ├── Volume patterns │ │ └── Technical indicators │ └──────────────┬──────────────┘ │ ▼ x = Market State ┌─────────────────────────────┐ │ Continuous Normalizing │ │ Flow (CNF) │ │ │ │ ┌───────────────────────┐ │ │ │ Inverse Transform │ │ │ │ dz/dt = f_θ(z, t) │ │ │ │ ODE: x → z(0) │ │ │ └───────────┬───────────┘ │ │ │ │ │ ┌───────────▼───────────┐ │ │ │ Base Distribution │ │ │ │ p(z) = N(0, I) │ │ │ │ log p(z(0)) │ │ │ └───────────┬───────────┘ │ │ │ │ │ ┌───────────▼───────────┐ │ │ │ Trace Integral │ │ │ │ ∫ tr(∂f/∂z) dt │ │ │ │ Hutchinson est. │ │ │ └───────────────────────┘ │ └──────────────┬──────────────┘ │ ┌──────────────┴──────────────┐ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Log- │ │ Conditional│ │ Regime │ │ Likelihood │ │ Samples │ │ Detection │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ └───────────────┼───────────────┘ ▼ ┌─────────────────────────────┐ │ Trading Decision │ │ ├── Signal Direction │ │ ├── Position Size │ │ ├── Confidence Interval │ │ └── Risk Parameters │ └─────────────────────────────┘Velocity Field Network
import torchimport torch.nn as nnimport numpy as np
class VelocityField(nn.Module): """ Neural network that defines the ODE dynamics.
dz/dt = f(z, t; θ)
The network takes (z, t) as input and outputs dz/dt. """
def __init__(self, dim: int, hidden_dim: int = 128, num_layers: int = 3, time_embed_dim: int = 16): super().__init__()
self.dim = dim self.time_embed_dim = time_embed_dim
# Time embedding (sinusoidal) self.time_embed = nn.Sequential( SinusoidalEmbedding(time_embed_dim), nn.Linear(time_embed_dim, hidden_dim), nn.GELU() )
# Input projection self.input_proj = nn.Linear(dim, hidden_dim)
# Main network (residual MLP) layers = [] for _ in range(num_layers): layers.append(ConcatResBlock(hidden_dim)) self.layers = nn.ModuleList(layers)
# Output projection self.output_proj = nn.Sequential( nn.LayerNorm(hidden_dim), nn.Linear(hidden_dim, hidden_dim), nn.GELU(), nn.Linear(hidden_dim, dim) )
# Zero initialization for stable training nn.init.zeros_(self.output_proj[-1].weight) nn.init.zeros_(self.output_proj[-1].bias)
def forward(self, z: torch.Tensor, t: torch.Tensor) -> torch.Tensor: """ Compute velocity field at state z and time t.
Args: z: (batch, dim) current state t: (batch,) or scalar current time
Returns: dz_dt: (batch, dim) velocity """ # Handle scalar time if t.dim() == 0: t = t.expand(z.shape[0])
# Embed time t_emb = self.time_embed(t)
# Project input h = self.input_proj(z)
# Apply residual blocks with time conditioning for layer in self.layers: h = layer(h, t_emb)
# Output dz_dt = self.output_proj(h)
return dz_dt
class SinusoidalEmbedding(nn.Module): """Sinusoidal time embedding (from Transformer/Diffusion models)"""
def __init__(self, dim: int, max_period: float = 10000.0): super().__init__() self.dim = dim self.max_period = max_period
def forward(self, t: torch.Tensor) -> torch.Tensor: half = self.dim // 2 freqs = torch.exp( -np.log(self.max_period) * torch.arange(half, device=t.device) / half ) args = t.unsqueeze(-1) * freqs return torch.cat([torch.cos(args), torch.sin(args)], dim=-1)
class ConcatResBlock(nn.Module): """Residual block with time conditioning via concatenation"""
def __init__(self, dim: int, dropout: float = 0.1): super().__init__()
self.norm1 = nn.LayerNorm(dim) self.linear1 = nn.Linear(dim * 2, dim * 4) self.norm2 = nn.LayerNorm(dim * 4) self.dropout = nn.Dropout(dropout) self.linear2 = nn.Linear(dim * 4, dim)
def forward(self, x: torch.Tensor, t_emb: torch.Tensor) -> torch.Tensor: h = self.norm1(x) h = torch.cat([h, t_emb], dim=-1) h = self.linear1(h) h = nn.functional.gelu(h) h = self.norm2(h) h = self.dropout(h) h = self.linear2(h) return x + hODE Solver Implementation
class ODESolver: """ Numerical ODE solver for continuous normalizing flows.
Supports multiple integration methods: - Euler (fast, less accurate) - RK4 (balanced) - Dopri5 (adaptive, most accurate) """
def __init__(self, method: str = 'rk4', atol: float = 1e-5, rtol: float = 1e-5): self.method = method self.atol = atol self.rtol = rtol
def solve(self, func, z0: torch.Tensor, t_span: tuple, num_steps: int = 100) -> torch.Tensor: """ Solve ODE from t_span[0] to t_span[1].
Args: func: velocity field function f(z, t) z0: (batch, dim) initial state t_span: (t0, t1) integration interval num_steps: number of integration steps
Returns: z1: (batch, dim) final state """ if self.method == 'euler': return self._euler(func, z0, t_span, num_steps) elif self.method == 'rk4': return self._rk4(func, z0, t_span, num_steps) elif self.method == 'dopri5': return self._dopri5(func, z0, t_span) else: raise ValueError(f"Unknown method: {self.method}")
def solve_with_trace(self, func, z0: torch.Tensor, t_span: tuple, num_steps: int = 100) -> tuple: """ Solve ODE and compute trace integral for log-det Jacobian.
Returns: z1: final state trace_integral: ∫ tr(∂f/∂z) dt """ if self.method == 'euler': return self._euler_with_trace(func, z0, t_span, num_steps) elif self.method == 'rk4': return self._rk4_with_trace(func, z0, t_span, num_steps) else: raise ValueError(f"Trace not implemented for {self.method}")
def _euler(self, func, z, t_span, num_steps): """Euler method""" t0, t1 = t_span dt = (t1 - t0) / num_steps t = t0
for _ in range(num_steps): dz = func(z, torch.tensor(t, device=z.device)) z = z + dt * dz t = t + dt
return z
def _rk4(self, func, z, t_span, num_steps): """4th-order Runge-Kutta method""" t0, t1 = t_span dt = (t1 - t0) / num_steps t = t0 device = z.device
for _ in range(num_steps): t_tensor = torch.tensor(t, device=device) k1 = func(z, t_tensor) k2 = func(z + 0.5 * dt * k1, t_tensor + 0.5 * dt) k3 = func(z + 0.5 * dt * k2, t_tensor + 0.5 * dt) k4 = func(z + dt * k3, t_tensor + dt)
z = z + (dt / 6) * (k1 + 2*k2 + 2*k3 + k4) t = t + dt
return z
def _euler_with_trace(self, func, z, t_span, num_steps): """Euler with Hutchinson trace estimation""" t0, t1 = t_span dt = (t1 - t0) / num_steps t = t0 trace_integral = 0.0 device = z.device
for _ in range(num_steps): t_tensor = torch.tensor(t, device=device)
# Compute trace using Hutchinson estimator epsilon = torch.randn_like(z) with torch.enable_grad(): z_in = z.detach().requires_grad_(True) dz = func(z_in, t_tensor)
# Vector-Jacobian product vjp = torch.autograd.grad( dz, z_in, epsilon, create_graph=True )[0] trace_est = (epsilon * vjp).sum(dim=-1)
z = z + dt * dz.detach() trace_integral = trace_integral + dt * trace_est t = t + dt
return z, trace_integral
def _rk4_with_trace(self, func, z, t_span, num_steps): """RK4 with trace estimation at each step""" t0, t1 = t_span dt = (t1 - t0) / num_steps t = t0 trace_integral = torch.zeros(z.shape[0], device=z.device) device = z.device
for _ in range(num_steps): t_tensor = torch.tensor(t, device=device)
# RK4 with trace at midpoint epsilon = torch.randn_like(z)
with torch.enable_grad(): z_mid = z.detach().requires_grad_(True) dz = func(z_mid, t_tensor + 0.5 * dt)
vjp = torch.autograd.grad( dz, z_mid, epsilon, create_graph=True )[0] trace_est = (epsilon * vjp).sum(dim=-1)
# Standard RK4 step k1 = func(z, t_tensor) k2 = func(z + 0.5 * dt * k1, t_tensor + 0.5 * dt) k3 = func(z + 0.5 * dt * k2, t_tensor + 0.5 * dt) k4 = func(z + dt * k3, t_tensor + dt)
z = z + (dt / 6) * (k1 + 2*k2 + 2*k3 + k4) trace_integral = trace_integral + dt * trace_est t = t + dt
return z, trace_integralContinuous Normalizing Flow Model
class ContinuousNormalizingFlow(nn.Module): """ FFJORD-style Continuous Normalizing Flow for market data.
Features: - Flexible neural ODE dynamics - Hutchinson trace estimator for O(1) log-det computation - Regularization for stable training - Bidirectional sampling and density evaluation """
def __init__(self, dim: int, hidden_dim: int = 128, num_layers: int = 3, t_span: tuple = (0.0, 1.0)): super().__init__()
self.dim = dim self.t_span = t_span
# Velocity field network self.velocity_field = VelocityField( dim=dim, hidden_dim=hidden_dim, num_layers=num_layers )
# ODE solver self.solver = ODESolver(method='rk4')
# Base distribution self.register_buffer('base_mean', torch.zeros(dim)) self.register_buffer('base_std', torch.ones(dim))
def forward(self, x: torch.Tensor, reverse: bool = False) -> tuple: """ Transform data through the flow.
Args: x: (batch, dim) input data reverse: if True, sample (z→x); if False, encode (x→z)
Returns: z_or_x: transformed data log_det_jacobian: log |det(dx/dz)| """ if reverse: # Sampling: z → x (forward ODE) return self._sample(x) else: # Encoding: x → z (backward ODE) return self._encode(x)
def _encode(self, x: torch.Tensor) -> tuple: """Encode data to latent space (x → z)""" # Solve ODE backward in time t_span = (self.t_span[1], self.t_span[0])
z, neg_trace = self.solver.solve_with_trace( self.velocity_field, x, t_span, num_steps=50 )
# log_det = -∫ tr(∂f/∂z) dt (negative because we go backward) log_det = -neg_trace
return z, log_det
def _sample(self, z: torch.Tensor) -> tuple: """Sample from latent space (z → x)""" x, trace = self.solver.solve_with_trace( self.velocity_field, z, self.t_span, num_steps=50 )
log_det = -trace
return x, log_det
def log_prob(self, x: torch.Tensor) -> torch.Tensor: """ Compute log probability of data under the flow.
log p(x) = log p(z) + log |det(dz/dx)| = log p(z) - ∫ tr(∂f/∂z) dt """ z, log_det = self._encode(x)
# Log probability under base distribution log_p_z = self._log_prob_base(z)
return log_p_z + log_det
def _log_prob_base(self, z: torch.Tensor) -> torch.Tensor: """Log probability under standard normal""" return -0.5 * ( z.shape[-1] * np.log(2 * np.pi) + (z ** 2).sum(dim=-1) )
def sample(self, num_samples: int, device: str = 'cpu') -> torch.Tensor: """Generate samples from the learned distribution""" z = torch.randn(num_samples, self.dim, device=device) x, _ = self._sample(z) return x
def sample_conditional(self, condition: torch.Tensor, cond_dims: list, num_samples: int = 100) -> torch.Tensor: """ Sample from conditional distribution p(x_other | x_cond)
Uses rejection sampling based on conditioning. """ samples = [] attempts = 0 max_attempts = num_samples * 100
while len(samples) < num_samples and attempts < max_attempts: # Generate proposals z = torch.randn(num_samples, self.dim, device=condition.device) x, _ = self._sample(z)
# Check condition match (within tolerance) cond_match = torch.abs(x[:, cond_dims] - condition).max(dim=-1)[0] valid = cond_match < 0.1 # Tolerance
samples.extend(x[valid].unbind(0)) attempts += num_samples
if len(samples) < num_samples: # Fall back to noisy conditioning z = torch.randn(num_samples, self.dim, device=condition.device) x, _ = self._sample(z) x[:, cond_dims] = condition return x
return torch.stack(samples[:num_samples])
class CNFLoss(nn.Module): """ Training loss for Continuous Normalizing Flows.
Includes: - Negative log-likelihood - Kinetic regularization (for smoother dynamics) - Jacobian Frobenius regularization """
def __init__(self, cnf: ContinuousNormalizingFlow, kinetic_weight: float = 0.01, jacobian_weight: float = 0.0): super().__init__() self.cnf = cnf self.kinetic_weight = kinetic_weight self.jacobian_weight = jacobian_weight
def forward(self, x: torch.Tensor) -> dict: """ Compute training loss.
Returns dict with individual loss components. """ # Main loss: negative log-likelihood log_prob = self.cnf.log_prob(x) nll_loss = -log_prob.mean()
# Kinetic regularization: penalize large velocities kinetic_loss = torch.tensor(0.0, device=x.device) if self.kinetic_weight > 0: kinetic_loss = self._kinetic_regularization(x)
# Total loss total_loss = nll_loss + self.kinetic_weight * kinetic_loss
return { 'total': total_loss, 'nll': nll_loss, 'kinetic': kinetic_loss }
def _kinetic_regularization(self, x: torch.Tensor) -> torch.Tensor: """Penalize kinetic energy of the flow""" t_samples = torch.rand(10, device=x.device) kinetic = 0.0
for t in t_samples: v = self.cnf.velocity_field(x, t.expand(x.shape[0])) kinetic = kinetic + (v ** 2).sum(dim=-1).mean()
return kinetic / len(t_samples)Feature Engineering for CNF
def compute_market_features(candles: list, lookback: int = 20) -> np.ndarray: """ Compute feature vector for CNF input.
Features designed to capture market state: - Returns at multiple scales - Volatility patterns - Volume dynamics - Price position """ import numpy as np
closes = np.array([c.close for c in candles]) highs = np.array([c.high for c in candles]) lows = np.array([c.low for c in candles]) volumes = np.array([c.volume for c in candles])
# Returns at multiple scales returns_1 = closes[-1] / closes[-2] - 1 returns_5 = closes[-1] / closes[-5] - 1 returns_10 = closes[-1] / closes[-10] - 1 returns_20 = closes[-1] / closes[-lookback] - 1
# Volatility log_returns = np.log(closes[1:] / closes[:-1]) volatility = log_returns[-lookback:].std() vol_ratio = log_returns[-5:].std() / (log_returns[-lookback:].std() + 1e-8)
# Volume volume_ma = volumes[-lookback:].mean() volume_ratio = volumes[-1] / (volume_ma + 1e-8)
# Price position high_20 = highs[-lookback:].max() low_20 = lows[-lookback:].min() price_position = (closes[-1] - low_20) / (high_20 - low_20 + 1e-8)
# Trend ema_fast = _ema(closes, 5)[-1] ema_slow = _ema(closes, 20)[-1] trend = (ema_fast - ema_slow) / closes[-1]
features = np.array([ returns_1, returns_5, returns_10, returns_20, volatility, vol_ratio, volume_ratio, price_position, trend ])
return features
def _ema(data: np.ndarray, period: int) -> np.ndarray: """Exponential moving average""" alpha = 2 / (period + 1) ema = np.zeros_like(data) ema[0] = data[0] for i in range(1, len(data)): ema[i] = alpha * data[i] + (1 - alpha) * ema[i-1] return emaCNF Trading System
class CNFTrader: """ Trading system based on Continuous Normalizing Flows.
Uses the learned distribution for: - Likelihood-based anomaly detection - Conditional return prediction - Confidence-weighted position sizing """
def __init__(self, cnf: ContinuousNormalizingFlow, return_idx: int = 0, likelihood_threshold: float = -10.0, confidence_threshold: float = 0.6): self.cnf = cnf self.return_idx = return_idx self.likelihood_threshold = likelihood_threshold self.confidence_threshold = confidence_threshold
# For tracking regime self.likelihood_history = [] self.likelihood_ma = None
def generate_signal(self, features: np.ndarray) -> dict: """ Generate trading signal from market features.
Returns dict with: - signal: trading direction (-1, 0, 1) - confidence: signal strength [0, 1] - log_likelihood: novelty measure - expected_return: predicted return """ self.cnf.eval()
x = torch.tensor(features, dtype=torch.float32).unsqueeze(0)
with torch.no_grad(): # Compute likelihood log_prob = self.cnf.log_prob(x).item()
# Update regime detection self._update_likelihood_tracking(log_prob)
# Check if in distribution if log_prob < self.likelihood_threshold: return { 'signal': 0, 'confidence': 0.0, 'log_likelihood': log_prob, 'expected_return': 0.0, 'regime_change': self._detect_regime_change() }
# Sample conditional returns expected_return, return_std = self._estimate_conditional_return(x)
# Confidence based on return distribution # High confidence if expected return is many stds from zero z_score = abs(expected_return) / (return_std + 1e-8) confidence = min(z_score / 3.0, 1.0) # Normalize to [0, 1]
# Signal direction if confidence < self.confidence_threshold: signal = 0 else: signal = 1 if expected_return > 0 else -1
return { 'signal': signal, 'confidence': confidence, 'log_likelihood': log_prob, 'expected_return': expected_return, 'return_std': return_std, 'regime_change': self._detect_regime_change() }
def _estimate_conditional_return(self, x: torch.Tensor, num_samples: int = 100) -> tuple: """Estimate expected return and std via sampling""" # Generate samples near current state z, _ = self.cnf._encode(x)
# Perturb in latent space and decode noise = torch.randn(num_samples, z.shape[-1]) * 0.1 z_perturbed = z + noise
samples, _ = self.cnf._sample(z_perturbed)
# Extract return predictions returns = samples[:, self.return_idx].numpy()
return returns.mean(), returns.std()
def _update_likelihood_tracking(self, log_prob: float): """Track likelihood for regime detection""" self.likelihood_history.append(log_prob)
# Keep last 50 values if len(self.likelihood_history) > 50: self.likelihood_history = self.likelihood_history[-50:]
# Update moving average if len(self.likelihood_history) >= 10: self.likelihood_ma = np.mean(self.likelihood_history[-10:])
def _detect_regime_change(self) -> bool: """Detect regime change via likelihood drop""" if self.likelihood_ma is None or len(self.likelihood_history) < 20: return False
recent = self.likelihood_history[-1] baseline = np.mean(self.likelihood_history[-20:-10])
# Regime change if recent likelihood is significantly below baseline return recent < baseline - 2.0 # 2 units belowTraining Loop
def train_cnf( model: ContinuousNormalizingFlow, train_data: torch.Tensor, val_data: torch.Tensor, epochs: int = 100, batch_size: int = 128, lr: float = 1e-3, weight_decay: float = 1e-5) -> ContinuousNormalizingFlow: """ Train Continuous Normalizing Flow model. """ optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=weight_decay) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, epochs) loss_fn = CNFLoss(model, kinetic_weight=0.01)
best_val_loss = float('inf') best_state = None patience = 10 patience_counter = 0
for epoch in range(epochs): model.train() total_loss = 0 n_batches = 0
# Shuffle data perm = torch.randperm(len(train_data))
for i in range(0, len(train_data), batch_size): batch = train_data[perm[i:i+batch_size]]
optimizer.zero_grad() losses = loss_fn(batch) losses['total'].backward()
# Gradient clipping for stability torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step()
total_loss += losses['nll'].item() n_batches += 1
scheduler.step()
# Validation model.eval() with torch.no_grad(): val_log_prob = model.log_prob(val_data) val_loss = -val_log_prob.mean().item()
# Early stopping if val_loss < best_val_loss: best_val_loss = val_loss best_state = {k: v.cpu().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
if (epoch + 1) % 10 == 0: print(f"Epoch {epoch+1}/{epochs}: " f"Train NLL={total_loss/n_batches:.4f}, " f"Val NLL={val_loss:.4f}")
# Load best model model.load_state_dict(best_state) return modelBacktesting Framework
class CNFBacktester: """ Backtest CNF-based trading strategy. """
def __init__(self, trader: CNFTrader, lookback: int = 20): self.trader = trader self.lookback = lookback
def run(self, candles: list, warmup: int = 100) -> pd.DataFrame: """ Run backtest on historical candle data. """ results = { 'timestamp': [], 'close': [], 'signal': [], 'confidence': [], 'log_likelihood': [], 'expected_return': [], 'position': [], 'pnl': [], 'cumulative_pnl': [] }
position = 0.0 cumulative_pnl = 0.0
for i in range(warmup, len(candles)): window = candles[i-self.lookback:i] features = compute_market_features(window, self.lookback)
# Generate signal signal_info = self.trader.generate_signal(features)
# Calculate PnL if i > warmup: daily_return = candles[i].close / candles[i-1].close - 1 pnl = position * daily_return cumulative_pnl += pnl else: pnl = 0.0
# Update position new_position = signal_info['signal'] * signal_info['confidence']
# Reduce position on regime change if signal_info.get('regime_change', False): new_position *= 0.5
position = new_position
# Record results results['timestamp'].append(candles[i].timestamp) results['close'].append(candles[i].close) results['signal'].append(signal_info['signal']) results['confidence'].append(signal_info['confidence']) results['log_likelihood'].append(signal_info['log_likelihood']) results['expected_return'].append(signal_info['expected_return']) results['position'].append(position) results['pnl'].append(pnl) results['cumulative_pnl'].append(cumulative_pnl)
return pd.DataFrame(results)
def calculate_metrics(self, results: pd.DataFrame) -> dict: """Calculate performance metrics""" returns = results['pnl']
# Basic metrics total_return = results['cumulative_pnl'].iloc[-1]
# Risk-adjusted metrics sharpe = returns.mean() / (returns.std() + 1e-8) * np.sqrt(252)
downside = returns[returns < 0] sortino = returns.mean() / (downside.std() + 1e-8) * np.sqrt(252)
# Drawdown cumulative = results['cumulative_pnl'] rolling_max = cumulative.expanding().max() drawdown = cumulative - rolling_max max_drawdown = drawdown.min()
# Win rate trades = returns[returns != 0] win_rate = (trades > 0).mean() if len(trades) > 0 else 0.0
# CNF-specific metrics avg_likelihood = results['log_likelihood'].mean() avg_confidence = results['confidence'].mean()
return { 'total_return': total_return, 'sharpe_ratio': sharpe, 'sortino_ratio': sortino, 'max_drawdown': max_drawdown, 'win_rate': win_rate, 'avg_likelihood': avg_likelihood, 'avg_confidence': avg_confidence, 'n_trades': (results['signal'] != 0).sum() }Data Requirements
Historical OHLCV Data:├── Minimum: 6 months of hourly data├── Recommended: 1+ years for robust distribution learning├── Frequency: 1-hour to daily└── Source: Bybit, Binance, or other exchanges
Required Fields:├── timestamp├── open, high, low, close├── volume└── Optional: funding rate, open interest
Preprocessing:├── Log-returns for stationarity├── Z-score normalization per feature├── Outlier clipping to ±4 std├── Train/Val/Test split: 70/15/15└── Temporal ordering preservedKey Metrics
- Negative Log-Likelihood (NLL): Training objective, lower is better
- Bits Per Dimension (BPD): NLL / (dim * log(2)), comparable across dims
- Sample Quality: Visual inspection of generated market scenarios
- ODE NFE: Number of function evaluations (computational cost)
- Sharpe Ratio: Risk-adjusted returns from trading
- Maximum Drawdown: Worst peak-to-trough decline
Dependencies
# Corenumpy>=1.23.0pandas>=1.5.0scipy>=1.10.0
# Deep Learningtorch>=2.0.0
# ODE Solvers (optional, for advanced methods)torchdiffeq>=0.2.3
# Market Dataccxt>=4.0.0
# Visualizationmatplotlib>=3.6.0seaborn>=0.12.0
# Utilitiesscikit-learn>=1.2.0tqdm>=4.65.0Expected Outcomes
- Distribution Learning: CNF captures complex, multi-modal return distributions
- Anomaly Detection: Likelihood-based regime change detection
- Scenario Generation: Realistic market scenarios for risk management
- Trading Signals: Expected Sharpe Ratio 0.7-1.3 with proper tuning
- Computational Efficiency: O(1) memory via adjoint method
References
-
FFJORD: Free-form Continuous Dynamics for Scalable Reversible Generative Models (Grathwohl et al., 2018)
-
Neural Ordinary Differential Equations (Chen et al., 2018)
-
Normalizing Flows for Probabilistic Modeling and Inference (Papamakarios et al., 2021)
-
Flow Matching for Generative Modeling (Lipman et al., 2023)
-
Scalable Reversible Generative Models with Free-form Continuous Dynamics (Grathwohl et al., 2019)
Rust Implementation
This chapter includes a complete Rust implementation for high-performance CNF trading on cryptocurrency data from Bybit. See rust/ directory.
Features:
- Real-time data fetching from Bybit API
- Custom neural ODE solver implementation
- Velocity field network with time conditioning
- Efficient trace estimation for log-det computation
- Backtesting framework with comprehensive metrics
- Modular and extensible design
Difficulty Level
⭐⭐⭐⭐⭐ (Expert)
Requires understanding of: Ordinary Differential Equations, Numerical Methods, Probability Theory, Neural Networks, Change of Variables Formula, Generative Modeling