class TemporalVAE(nn.Module):
"""
Temporal Variational Autoencoder with LSTM encoder/decoder.
Captures temporal dependencies while providing a probabilistic
latent space for anomaly detection.
"""
def __init__(
self,
n_features: int,
seq_len: int,
hidden_size: int = 32,
latent_dim: int = 8,
n_layers: int = 1,
dropout: float = 0.1
):
super().__init__()
self.n_features = n_features
self.seq_len = seq_len
self.hidden_size = hidden_size
self.latent_dim = latent_dim
# Encoder LSTM
self.encoder_lstm = nn.LSTM(
input_size=n_features,
hidden_size=hidden_size,
num_layers=n_layers,
batch_first=True,
dropout=dropout if n_layers > 1 else 0
)
# Latent space projections
self.fc_mu = nn.Linear(hidden_size, latent_dim)
self.fc_logvar = nn.Linear(hidden_size, latent_dim)
# Decoder
self.fc_decoder_input = nn.Linear(latent_dim, hidden_size)
self.decoder_lstm = nn.LSTM(
input_size=hidden_size,
hidden_size=hidden_size,
num_layers=n_layers,
batch_first=True,
dropout=dropout if n_layers > 1 else 0
)
self.fc_output = nn.Linear(hidden_size, n_features)
def encode(self, x: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
"""
Encode input sequence to latent distribution parameters.
Parameters:
-----------
x : tensor
Input of shape (batch, seq_len, n_features)
Returns:
--------
mu, log_var : tensors of shape (batch, latent_dim)
"""
# LSTM encoding
_, (h_n, _) = self.encoder_lstm(x) # h_n: (n_layers, batch, hidden)
h = h_n[-1] # Last layer hidden state: (batch, hidden)
# Project to latent parameters
mu = self.fc_mu(h)
log_var = self.fc_logvar(h)
return mu, log_var
def reparameterize(self, mu: torch.Tensor, log_var: torch.Tensor) -> torch.Tensor:
"""
Reparameterization trick: z = mu + sigma * epsilon
"""
std = torch.exp(0.5 * log_var)
eps = torch.randn_like(std)
return mu + eps * std
def decode(self, z: torch.Tensor) -> torch.Tensor:
"""
Decode latent vector to sequence.
Parameters:
-----------
z : tensor
Latent vector of shape (batch, latent_dim)
Returns:
--------
tensor : Reconstructed sequence (batch, seq_len, n_features)
"""
batch_size = z.shape[0]
# Project latent to hidden size
h = self.fc_decoder_input(z) # (batch, hidden)
# Repeat for each timestep
h_repeated = h.unsqueeze(1).repeat(1, self.seq_len, 1) # (batch, seq_len, hidden)
# LSTM decoding
decoded, _ = self.decoder_lstm(h_repeated)
# Project to output
output = self.fc_output(decoded) # (batch, seq_len, n_features)
return output
def forward(self, x: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""
Forward pass.
Returns:
--------
recon_x, mu, log_var
"""
mu, log_var = self.encode(x)
z = self.reparameterize(mu, log_var)
recon_x = self.decode(z)
return recon_x, mu, log_var
def reconstruction_error(self, x: torch.Tensor) -> torch.Tensor:
"""
Compute reconstruction error (MSE) per sample.
Used for anomaly detection.
"""
recon_x, _, _ = self.forward(x)
mse = torch.mean((x - recon_x) ** 2, dim=(1, 2)) # (batch,)
return mse
def vae_loss(recon_x, x, mu, log_var, beta=0.1):
"""
VAE loss = Reconstruction loss + beta * KL divergence
Parameters:
-----------
beta : float
Weight of KL term (beta-VAE)
"""
# Reconstruction loss (MSE)
recon_loss = F.mse_loss(recon_x, x, reduction='mean')
# KL divergence: KL(q(z|x) || p(z)) where p(z) = N(0, I)
kl_loss = -0.5 * torch.mean(1 + log_var - mu.pow(2) - log_var.exp())
return recon_loss + beta * kl_loss, recon_loss, kl_loss
# Test Temporal VAE
print("Test de TemporalVAE:")
vae = TemporalVAE(n_features=12, seq_len=20, hidden_size=32, latent_dim=8)
test_input = torch.randn(4, 20, 12)
recon, mu, logvar = vae(test_input)
print(f" Input shape: {test_input.shape}")
print(f" Recon shape: {recon.shape}")
print(f" Mu shape: {mu.shape}")
print(f" LogVar shape: {logvar.shape}")
print(f" Parameters: {sum(p.numel() for p in vae.parameters()):,}")
Comment lire ce notebook hors Cloud
Le notebook est commite avec ses outputs reels : chaque tableau et chaque chiffre cite dans les interpretations provient d’une execution QuantConnect effective (meme jeu de donnees simulees, memes seeds). Vous pouvez donc suivre le raisonnement complet sans executer — mais pour reproduire, trois prerequis distinguent le Cloud du local :
QuantBook: l’objet de recherche QC qui expose l’historical data feed (qb.history,qb.add_equity). Les cellules de generation de donnees simulent ce feed pour rester deterministes ; les cellules finales (code de reference QC) montrent la variante Cloud avec de vraies barres SPY.ObjectStore: la persistance QC (cle/valeur, similaire a un dict sauvegarde entre recherches). C’est elle qui permet d’enregistrer le VAE (147.9 KB) et le HMM (1.9 KB) en fin de notebook, puis de les recharger dans l’algorithme de trading sans re-entrainement.pip install torch hmmlearnreproduit le meme comportement pour toutes les cellules hors QuantBook.Plan de lecture : Partie 1 (approches generatives, Temporal VAE puis Transformer VAE), Partie 2 (HMM et regimes), Partie 3 (strategie adaptative QuantConnect et backtest). Les trois exercices jalonnent le parcours : reconstruction (Ex. 1), interpretation des etats caches (Ex. 2), strategie (Ex. 3).