AI Market Scenario Generation with GAN and LLM
Synthetic market scenarios are used for stress-testing trading strategies, training ML models on rare events (crises, flash crashes), and backtesting without lookahead bias. GAN approach generates statistically plausible price series; LLM approach generates narrative economic scenarios.
TimeGAN for Synthetic Time Series
import torch
import torch.nn as nn
import numpy as np
from dataclasses import dataclass
@dataclass
class TimeGANConfig:
seq_len: int = 24 # sequence length
n_features: int = 5 # OHLCV
hidden_dim: int = 24
num_layers: int = 3
batch_size: int = 128
epochs: int = 1000
learning_rate: float = 1e-3
class EmbeddingNetwork(nn.Module):
"""Encodes real data into latent space"""
def __init__(self, input_dim: int, hidden_dim: int, num_layers: int):
super().__init__()
self.rnn = nn.GRU(input_dim, hidden_dim, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_dim, hidden_dim)
def forward(self, x):
h, _ = self.rnn(x)
return torch.sigmoid(self.fc(h))
class Generator(nn.Module):
"""Generates synthetic data from noise"""
def __init__(self, noise_dim: int, hidden_dim: int, output_dim: int, num_layers: int):
super().__init__()
self.rnn = nn.GRU(noise_dim + hidden_dim, hidden_dim, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_dim, hidden_dim)
def forward(self, z, h):
# z: noise, h: historical context from embedding
combined = torch.cat([z, h], dim=-1)
out, _ = self.rnn(combined)
return torch.sigmoid(self.fc(out))
class TimeGAN:
def __init__(self, config: TimeGANConfig):
self.config = config
self.embedder = EmbeddingNetwork(config.n_features, config.hidden_dim, config.num_layers)
self.generator = Generator(config.hidden_dim, config.hidden_dim, config.n_features, config.num_layers)
self.discriminator = nn.GRU(config.hidden_dim, config.hidden_dim, config.num_layers, batch_first=True)
def train(self, real_data: np.ndarray) -> None:
"""
real_data: (N, seq_len, n_features) normalized OHLCV
4 phases: Embedder, Supervised, Generator, Joint
"""
real_tensor = torch.FloatTensor(real_data)
# ... training across 4 TimeGAN phases
def generate(self, n_samples: int) -> np.ndarray:
with torch.no_grad():
z = torch.randn(n_samples, self.config.seq_len, self.config.hidden_dim)
h_init = torch.zeros(n_samples, self.config.seq_len, self.config.hidden_dim)
synthetic = self.generator(z, h_init)
# Decode through recovery network
return synthetic.numpy()
LLM-Generated Narrative Scenarios
from openai import AsyncOpenAI
import json
client = AsyncOpenAI()
async def generate_market_scenario(
asset: str,
timeframe: str = "3 months",
scenario_type: str = "stress" # stress, bull, bear, sideways, black_swan
) -> dict:
SCENARIO_CONTEXTS = {
"stress": "financial crisis, volatility spike, liquidity decline",
"black_swan": "unexpected event: geopolitics, tech failure, natural disaster",
"bull": "sustained growth, positive macro data",
"bear": "recession, inflation spike, monetary tightening"
}
response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": f"""You are a qualified financial analyst.
Generate a detailed market scenario.
Scenario type: {scenario_type} — {SCENARIO_CONTEXTS.get(scenario_type, '')}.
Return JSON with fields:
- narrative: text description of scenario
- macro_drivers: macroeconomic triggers (list)
- price_trajectory: expected price dynamics [{{"month": N, "expected_change_pct": X}}]
- volatility_profile: expected volatility by periods
- key_risk_factors: key risks
- correlation_shifts: how correlations with other assets change
Horizon: {timeframe}.
IMPORTANT: this is hypothetical scenario for strategy testing, not investment advice."""
}, {
"role": "user",
"content": f"Asset: {asset}"
}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
Synthetic Scenarios for Backtesting
import pandas as pd
from scipy.stats import norm
class MarketScenarioGenerator:
def generate_gbm_scenario(
self,
initial_price: float,
mu: float, # annual drift
sigma: float, # annual volatility
T: float = 1.0, # years
n_steps: int = 252,
n_scenarios: int = 1000
) -> np.ndarray:
"""Geometric Brownian Motion — baseline scenario"""
dt = T / n_steps
prices = np.zeros((n_scenarios, n_steps + 1))
prices[:, 0] = initial_price
for t in range(1, n_steps + 1):
z = np.random.standard_normal(n_scenarios)
prices[:, t] = prices[:, t-1] * np.exp(
(mu - 0.5 * sigma**2) * dt + sigma * np.sqrt(dt) * z
)
return prices
def generate_jump_diffusion(
self,
initial_price: float,
mu: float,
sigma: float,
lambda_jump: float = 0.1, # jump frequency (per year)
mu_jump: float = -0.1, # mean jump size
sigma_jump: float = 0.05,
T: float = 1.0,
n_steps: int = 252
) -> np.ndarray:
"""Merton Jump-Diffusion — models flash crash scenarios"""
dt = T / n_steps
prices = [initial_price]
for _ in range(n_steps):
# Diffusion component
diffusion = (mu - 0.5 * sigma**2) * dt + sigma * np.sqrt(dt) * norm.rvs()
# Jump component
n_jumps = np.random.poisson(lambda_jump * dt)
jump = sum(norm.rvs(mu_jump, sigma_jump) for _ in range(n_jumps))
prices.append(prices[-1] * np.exp(diffusion + jump))
return np.array(prices)
Validation of Synthetic Data
def validate_synthetic_data(real: np.ndarray, synthetic: np.ndarray) -> dict:
"""Check statistical plausibility of synthetic data"""
from scipy.stats import ks_2samp
real_returns = np.diff(np.log(real.flatten()))
synth_returns = np.diff(np.log(synthetic.flatten()))
ks_stat, ks_p = ks_2samp(real_returns, synth_returns)
return {
"ks_statistic": float(ks_stat), # < 0.1 good
"ks_p_value": float(ks_p), # > 0.05 accept H0
"real_mean": float(real_returns.mean()),
"synth_mean": float(synth_returns.mean()),
"real_std": float(real_returns.std()),
"synth_std": float(synth_returns.std()),
"real_skewness": float(pd.Series(real_returns).skew()),
"synth_skewness": float(pd.Series(synth_returns).skew()),
"real_kurtosis": float(pd.Series(real_returns).kurtosis()),
"synth_kurtosis": float(pd.Series(synth_returns).kurtosis()),
}
Timeline: GBM/Jump-Diffusion scenarios + LLM narratives — 1–2 weeks. Full TimeGAN with training on historical data — 3–4 weeks.