Разработка AI-модели на базе GAN для генерации рыночных сценариев

AI-генерация рыночных сценариев с GAN и LLM

Синтетические рыночные сценарии используются для стресс-тестирования торговых стратегий, обучения ML-моделей на редких событиях (кризисы, flash crash), бэктестинга без lookahead bias. GAN-подход генерирует статистически правдоподобные ценовые ряды, LLM-подход — нарративные экономические сценарии.

TimeGAN для синтетических временных рядов

import torch
import torch.nn as nn
import numpy as np
from dataclasses import dataclass

@dataclass
class TimeGANConfig:
    seq_len: int = 24          # длина последовательности
    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):
    """Кодирует реальные данные в латентное пространство"""
    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):
    """Генерирует синтетические данные из шума"""
    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: шум, h: исторический контекст из эмбеддинга
        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) нормализованный OHLCV
        4 фазы: Embedder, Supervised, Generator, Joint
        """
        real_tensor = torch.FloatTensor(real_data)
        # ... обучение по 4 фазам TimeGAN

    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)
            # Декодируем через recovery network
        return synthetic.numpy()

LLM-генерация нарративных сценариев

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": "финансовый кризис, рост волатильности, снижение ликвидности",
        "black_swan": "неожиданное событие: геополитика, технологический сбой, природная катастрофа",
        "bull": "устойчивый рост, позитивные макроэкономические данные",
        "bear": "рецессия, рост инфляции, ужесточение монетарной политики"
    }

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": f"""Ты — квалифицированный финансовый аналитик.
            Сгенерируй детальный рыночный сценарий.
            Тип сценария: {scenario_type} — {SCENARIO_CONTEXTS.get(scenario_type, '')}.
            Верни JSON с полями:
            - narrative: текстовое описание сценария
            - macro_drivers: макроэкономические триггеры (список)
            - price_trajectory: ожидаемая динамика цены [{{"month": N, "expected_change_pct": X}}]
            - volatility_profile: ожидаемая волатильность по периодам
            - key_risk_factors: ключевые риски
            - correlation_shifts: как меняются корреляции с другими активами
            Горизонт: {timeframe}.
            ВАЖНО: это гипотетический сценарий для тестирования стратегий, не инвестиционная рекомендация."""
        }, {
            "role": "user",
            "content": f"Актив: {asset}"
        }],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

Синтетические сценарии для бэктестинга

import pandas as pd
from scipy.stats import norm

class MarketScenarioGenerator:
    def generate_gbm_scenario(
        self,
        initial_price: float,
        mu: float,       # годовой дрейф
        sigma: float,    # годовая волатильность
        T: float = 1.0,  # лет
        n_steps: int = 252,
        n_scenarios: int = 1000
    ) -> np.ndarray:
        """Geometric Brownian Motion — baseline сценарий"""
        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,  # частота прыжков (в год)
        mu_jump: float = -0.1,     # средний размер прыжка
        sigma_jump: float = 0.05,
        T: float = 1.0,
        n_steps: int = 252
    ) -> np.ndarray:
        """Merton Jump-Diffusion — моделирует flash crash сценарии"""
        dt = T / n_steps
        prices = [initial_price]

        for _ in range(n_steps):
            # Диффузионная компонента
            diffusion = (mu - 0.5 * sigma**2) * dt + sigma * np.sqrt(dt) * norm.rvs()

            # Прыжковая компонента
            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)

Валидация синтетических данных

def validate_synthetic_data(real: np.ndarray, synthetic: np.ndarray) -> dict:
    """Проверяем статистическое правдоподобие синтетических данных"""
    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 хорошо
        "ks_p_value": float(ks_p),                # > 0.05 принимаем 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()),
    }

Сроки: GBM/Jump-Diffusion сценарии + LLM нарративы — 1–2 недели. Полноценный TimeGAN с обучением на исторических данных — 3–4 недели.