Розробка 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 — базовий сценарій"""
        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 тижні.