Zum Inhalt springen
Alle Bibliotheksdokumente

TimeGAN-Auswertung: Prognosenutzen und Realismus

Code Machine Learning for Trading

Zusammenfassung

Dieses Modul stellt Prüfungen nach dem Training von Zeitreihengeneratoren vor, die auf dem TimeGAN-Auswertungsansatz beruhen. Es misst den Nutzen, indem es einen rekurrenten Prädiktor auf synthetischen Sequenzen trainiert und mit echten Daten testet. Ein Modus folgt dem Aufbau der Studie, prognostiziert nur das letzte Merkmal und gibt den mittleren absoluten Fehler aus; ein umfassenderer Modus prognostiziert alle Merkmale und vergleicht die Leistung des Trainings mit synthetischen Daten mit einer auf echten Daten trainierten Basislinie. Die Datenähnlichkeit wird bewertet, indem ein Sequenzklassifikator darauf trainiert wird, echte von generierten Daten zu unterscheiden; eine schwächere Klassifikation bedeutet, dass sich die beiden Quellen schwerer trennen lassen.

Das Modul unterstützt rekurrente Architekturen und bietet eine kombinierte Bewertungsroutine, die Prognose- und Unterscheidungsergebnisse sowie optional einen Score nach Art der Studie ausgibt. Es weist darauf hin, dass die Kennzahlen nach Art der Studie Eingaben voraussetzen, die auf einen bestimmten Wertebereich normalisiert sind, und Ergebnisse bei einer Verletzung dieser Annahme möglicherweise nicht vergleichbar sind. Diese Scores sind Diagnosen und kein Beweis dafür, dass synthetische Daten alle für den Handel relevanten Eigenschaften bewahren; ihre Interpretation hängt von der Merkmalsauswahl, Normalisierung, Datenteilung und den Modelltrainingsentscheidungen ab.

Kernaussagen

  • Der Prognosenutzen lässt sich messen, indem ein Sequenzmodell für den nächsten Zeitschritt mit synthetischen Daten trainiert und mit echten Sequenzen getestet wird.
  • Ein an die Studie angelehnter Prognosescore sagt nur das letzte Merkmal voraus und gibt den mittleren absoluten Fehler aus.
  • Eine umfassende Prognosebewertung vergleicht die Leistung des Trainings mit synthetischen Daten mit einer Trainingsbasislinie aus echten Daten.
  • Ein Klassifikator zur Unterscheidung prüft, ob echte und synthetische Sequenzen auseinanderzuhalten sind.
  • Normalisierung und Bewertungsdesign beeinflussen, ob Ergebnisse vergleichbar und aussagekräftig sind.

Schlagwörter

Volltext
# timegan_metrics.py


```py
"""
TimeGAN Evaluation Metrics

Implements the evaluation protocol from:
Yoon, J., Jarrett, D., & van der Schaar, M. (2019).
"Time-series Generative Adversarial Networks." NeurIPS 2019.

Also used for GT-GAN evaluation (same metrics apply to GRU-ODE generators).

Paper Section 5.2 specifies:
1. Predictive Score: Train post-hoc RNN to predict next step
   - Train on synthetic, test on real (TSTR with MAE)
   - Lower MAE = better utility

2. Discriminative Score: Train post-hoc RNN classifier
   - Binary: distinguish real vs synthetic
   - Lower accuracy = better fidelity (harder to tell apart)

Two evaluation modes:

1. **Yoon Mode** (compute_predictive_score_yoon):
   - Matches official TimeGAN repo
   - Predicts ONLY the last feature using other features as input
   - GRU with sigmoid output (expects [0,1] normalized data)
   - Reports raw MAE

2. **Comprehensive Mode** (compute_predictive_score):
   - Predicts ALL features
   - Compares TSTR vs TRTR baseline
   - Reports MAE ratio

Reference: [ref:NCTZ79FF] TimeGAN
"""

from __future__ import annotations

from typing import Any

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from sklearn.metrics import accuracy_score, roc_auc_score
from torch.utils.data import DataLoader, TensorDataset
from tqdm import tqdm

# Progress bars write to stderr, and papermill captures every redraw as a separate
# output: one training run left several thousand of them in the committed notebook.
# `verbose` still controls the informative stdout prints, which are worth keeping.
# Set this True in an interactive session to watch a long run.
PROGRESS_BARS = False


class PostHocLSTMPredictor(nn.Module):
    """
    2-layer LSTM for next-step prediction per TimeGAN paper.

    Architecture (Section 5.2):
    - Input: sequences of shape (batch, seq_len, n_features)
    - 2 LSTM layers with hidden_dim units
    - Dense output layer to n_features
    - Predicts next timestep given previous timesteps

    Used for computing predictive score (TSTR with MAE).
    """

    def __init__(self, input_dim: int, hidden_dim: int = 128, num_layers: int = 2):
        super().__init__()
        self.lstm = nn.LSTM(
            input_dim,
            hidden_dim,
            num_layers=num_layers,
            batch_first=True,
            bidirectional=False,
        )
        self.fc = nn.Linear(hidden_dim, input_dim)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Forward pass.

        Args:
            x: Input sequences (batch, seq_len, n_features)

        Returns:
            Predictions for next timestep at each position (batch, seq_len, n_features)
        """
        h, _ = self.lstm(x)
        return self.fc(h)


class PostHocLSTMDiscriminator(nn.Module):
    """
    2-layer LSTM classifier for discriminative score per TimeGAN paper.

    Architecture (Section 5.2):
    - Input: sequences of shape (batch, seq_len, n_features)
    - 2 LSTM layers with hidden_dim units
    - Dense output layer to 1 (sigmoid for binary classification)
    - Classifies sequences as real (1) or synthetic (0)

    Used for computing discriminative score (lower accuracy = better fidelity).
    """

    def __init__(self, input_dim: int, hidden_dim: int = 128, num_layers: int = 2):
        super().__init__()
        self.lstm = nn.LSTM(
            input_dim,
            hidden_dim,
            num_layers=num_layers,
            batch_first=True,
            bidirectional=False,
        )
        self.fc = nn.Linear(hidden_dim, 1)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Forward pass.

        Args:
            x: Input sequences (batch, seq_len, n_features)

        Returns:
            Logits for real/fake classification (batch, 1)
        """
        h, _ = self.lstm(x)
        # Use last hidden state for classification
        return self.fc(h[:, -1, :])


class GRUPredictor(nn.Module):
    """
    GRU predictor matching official TimeGAN repo.

    Architecture:
    - GRU (not LSTM)
    - hidden_dim = dim // 2
    - Sigmoid output (expects [0, 1] normalized data)
    - Predicts last feature using other features
    """

    def __init__(self, input_dim: int, hidden_dim: int | None = None):
        super().__init__()
        # Paper: hidden_dim = int(dim/2) where dim is total features
        if hidden_dim is None:
            hidden_dim = max(input_dim // 2, 4)
        self.gru = nn.GRU(
            input_dim,
            hidden_dim,
            num_layers=1,  # Paper uses single layer
            batch_first=True,
            bidirectional=False,
        )
        self.fc = nn.Linear(hidden_dim, 1)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Forward pass.

        Args:
            x: Input sequences (batch, seq_len, n_features-1)
               Note: Last feature already excluded by caller

        Returns:
            Predictions for last feature (batch, seq_len, 1) with sigmoid
        """
        h, _ = self.gru(x)
        return torch.sigmoid(self.fc(h))


class GRUDiscriminator(nn.Module):
    """
    GRU discriminator for TimeGAN evaluation.

    Architecture: GRU with hidden_dim = input_dim // 2, single layer.
    Reports |accuracy - 0.5| where 0.0 = indistinguishable, 0.5 = trivially separable.
    """

    def __init__(self, input_dim: int, hidden_dim: int | None = None):
        super().__init__()
        if hidden_dim is None:
            hidden_dim = max(input_dim // 2, 4)
        self.hidden_dim = hidden_dim
        self.gru = nn.GRU(input_dim, hidden_dim, num_layers=1, batch_first=True)
        self.fc = nn.Linear(hidden_dim, 1)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        h, _ = self.gru(x)
        return self.fc(h[:, -1, :])


def compute_predictive_score_yoon(
    synthetic: np.ndarray,
    real: np.ndarray,
    iterations: int = 5000,
    batch_size: int = 128,
    device: str | None = None,
    quick_test: bool = False,
    verbose: bool = True,
) -> dict[str, Any]:
    """
    Compute predictive score per Yoon et al. (2019).

    Replicates metrics/predictive_metrics.py from official repo.

    Methodology:
    - Predicts ONLY the last feature using other features
    - GRU with sigmoid output (expects [0,1] normalized data)
    - Reports raw MAE (no baseline comparison)

    Args:
        synthetic: Synthetic sequences (n_samples, seq_len, n_features)
        real: Real sequences for evaluation (same shape)
        iterations: Training iterations (default: 5000)
        batch_size: Batch size (default: 128)
        device: 'cuda' or 'cpu' (auto-detected if None)
        quick_test: If True, reduce iterations for testing
        verbose: Print progress

    Returns:
        dict with mae, feature_idx, model
    """
    if quick_test:
        iterations = min(100, iterations)

    if device is None:
        device = "cuda" if torch.cuda.is_available() else "cpu"

    device = torch.device(device)
    n_features = synthetic.shape[-1]

    # Check if data appears to be normalized to [0, 1]
    data_min, data_max = synthetic.min(), synthetic.max()
    if data_min < -0.1 or data_max > 1.1:
        if verbose:
            print(f"\nWARNING: Data range [{data_min:.2f}, {data_max:.2f}]")
            print("   Yoon metric expects [0, 1] normalized data.")
            print("   Results may not be comparable to paper.\n")

    if verbose:
        print("Predictive Score (Yoon et al.)")
        print(f"  Features: {n_features - 1} input → 1 output (last feature)")
        print(f"  Iterations: {iterations}, Batch size: {batch_size}")

    hidden_dim = max(n_features // 2, 4)
    model = GRUPredictor(n_features - 1, hidden_dim).to(device)
    optimizer = optim.Adam(model.parameters())
    criterion = nn.L1Loss()  # MAE loss

    # Prepare training data from synthetic
    # X: all features except last, for all timesteps except last
    # Y: last feature only, shifted by 1 (next-step prediction)
    def prepare_paper_data(data: np.ndarray) -> tuple[torch.Tensor, torch.Tensor]:
        """Prepare X and Y per paper's methodology."""
        # X: [:-1, :(dim-1)] - all but last timestep, all but last feature
        X = torch.tensor(data[:, :-1, :-1], dtype=torch.float32)
        # Y: [1:, (dim-1)] - shifted by 1, only last feature
        Y = torch.tensor(data[:, 1:, -1:], dtype=torch.float32)
        return X, Y

    X_train, Y_train = prepare_paper_data(synthetic)
    n_train = len(X_train)

    # Training loop (iteration-based, not epoch-based)
    model.train()
    iterator = tqdm(range(iterations), desc="Yoon Predictive", disable=not PROGRESS_BARS)
    for _ in iterator:
        # Random mini-batch (paper's approach)
        idx = np.random.permutation(n_train)[:batch_size]
        X_batch = X_train[idx].to(device)
        Y_batch = Y_train[idx].to(device)

        optimizer.zero_grad()
        Y_pred = model(X_batch)
        loss = criterion(Y_pred, Y_batch)
        loss.backward()
        optimizer.step()

        if verbose:
            iterator.set_postfix(loss=loss.item())

    # Evaluate on real data
    model.eval()
    X_real, Y_real = prepare_paper_data(real)
    X_real = X_real.to(device)
    Y_real = Y_real.to(device)

    with torch.no_grad():
        Y_pred = model(X_real)
        mae = torch.abs(Y_pred - Y_real).mean().item()

    if verbose:
        print("\nPredictive Score (Yoon et al.):")
        print(f"  MAE on real data: {mae:.6f}")
        print("  (Paper reports ~0.04 for GOOG stock)")

    return {
        "mae": mae,
        "feature_idx": n_features - 1,
        "model": model,
    }


def compute_discriminative_score_yoon(
    real: np.ndarray,
    synthetic: np.ndarray,
    iterations: int = 2000,
    batch_size: int = 128,
    device: str | None = None,
    quick_test: bool = False,
    verbose: bool = True,
) -> dict[str, Any]:
    """
    Compute discriminative score per Yoon et al. (2019).

    Metric: |accuracy - 0.5| where 0.0 = indistinguishable, 0.5 = trivially separable.
    Data should be MinMax normalized to [0, 1].

    Args:
        real: Real sequences (n_samples, seq_len, n_features)
        synthetic: Synthetic sequences (same shape)
        iterations: Training iterations (default: 2000)
        batch_size: Batch size (default: 128)
        device: 'cuda' or 'cpu' (auto-detected if None)
        quick_test: If True, reduce iterations for testing
        verbose: Print progress

    Returns:
        dict with score, accuracy, model
    """
    if quick_test:
        iterations = min(100, iterations)

    if device is None:
        device = "cuda" if torch.cuda.is_available() else "cpu"

    device = torch.device(device)
    n_features = real.shape[-1]

    # Warn if data not normalized
    data_min, data_max = real.min(), real.max()
    if (data_min < -0.1 or data_max > 1.1) and verbose:
        print(f"WARNING: Data range [{data_min:.2f}, {data_max:.2f}] - expected [0, 1]")

    hidden_dim = max(n_features // 2, 4)
    model = GRUDiscriminator(n_features, hidden_dim).to(device)
    optimizer = optim.Adam(model.parameters())
    criterion = nn.BCEWithLogitsLoss()

    # Balance and combine datasets
    n_samples = min(len(real), len(synthetic))
    X = np.concatenate([real[:n_samples], synthetic[:n_samples]], axis=0)
    y = np.array([1] * n_samples + [0] * n_samples)

    # Shuffle and split 70/30
    indices = np.random.permutation(len(X))
    X, y = X[indices], y[indices]
    n_train = int(0.7 * len(X))
    X_train, X_test = X[:n_train], X[n_train:]
    y_train, y_test = y[:n_train], y[n_train:]

    X_train_t = torch.tensor(X_train, dtype=torch.float32)
    y_train_t = torch.tensor(y_train, dtype=torch.float32).unsqueeze(1)
    X_test_t = torch.tensor(X_test, dtype=torch.float32)

    # Training (iteration-based)
    model.train()
    iterator = tqdm(range(iterations), desc="Discriminator", disable=not PROGRESS_BARS)
    for _ in iterator:
        idx = np.random.permutation(len(X_train_t))[:batch_size]
        X_batch = X_train_t[idx].to(device)
        y_batch = y_train_t[idx].to(device)

        optimizer.zero_grad()
        loss = criterion(model(X_batch), y_batch)
        loss.backward()
        optimizer.step()

    # Evaluation
    model.eval()
    with torch.no_grad():
        probs = torch.sigmoid(model(X_test_t.to(device))).cpu().numpy().flatten()
        preds = (probs > 0.5).astype(int)

    accuracy = accuracy_score(y_test, preds)
    score = abs(accuracy - 0.5)

    if verbose:
        print(f"Discriminative: accuracy={accuracy:.1%}, score={score:.4f}")

    return {"score": score, "accuracy": accuracy, "model": model}


def compute_predictive_score(
    synthetic: np.ndarray,
    real_holdout: np.ndarray,
    hidden_dim: int = 128,
    num_layers: int = 2,
    epochs: int = 100,
    batch_size: int = 64,
    learning_rate: float = 1e-3,
    device: str | None = None,
    quick_test: bool = False,
    verbose: bool = True,
) -> dict[str, Any]:
    """
    Compute predictive score per TimeGAN paper (Section 5.2).

    Protocol:
    1. Train LSTM predictor on SYNTHETIC data to predict next timestep
    2. Evaluate on REAL holdout data using MAE
    3. Compare to TRTR baseline (train on real, test on real)

    The ratio TSTR_MAE / TRTR_MAE indicates utility:
    - Ratio ~1.0: Synthetic data preserves predictive patterns
    - Ratio >> 1.0: Synthetic data lost temporal structure

    IMPORTANT: Data is normalized per-feature before evaluation to prevent
    high-variance features (e.g., volume) from dominating the MAE calculation.

    Args:
        synthetic: Synthetic sequences (n_samples, seq_len, n_features)
        real_holdout: Real holdout sequences (never seen by generator)
        hidden_dim: LSTM hidden dimension (paper: 128)
        num_layers: Number of LSTM layers (paper: 2)
        epochs: Training epochs
        batch_size: Batch size
        learning_rate: Learning rate
        device: 'cuda' or 'cpu' (auto-detected if None)
        quick_test: If True, reduce epochs for faster testing
        verbose: Print progress

    Returns:
        dict with:
        - mae_tstr: MAE when trained on synthetic, tested on real (normalized)
        - mae_trtr: MAE when trained on real, tested on real (baseline, normalized)
        - ratio: mae_tstr / mae_trtr (lower is better, 1.0 = perfect)
        - model_tstr: Trained TSTR model
        - model_trtr: Trained TRTR model
    """
    if quick_test:
        epochs = min(10, epochs)

    if device is None:
        device = "cuda" if torch.cuda.is_available() else "cpu"

    device = torch.device(device)
    n_features = synthetic.shape[-1]

    # Normalize data per-feature to ensure equal contribution to MAE
    # Compute mean/std from real holdout (the evaluation target)
    holdout_flat = real_holdout.reshape(-1, n_features)
    feat_mean = holdout_flat.mean(axis=0, keepdims=True)
    feat_std = holdout_flat.std(axis=0, keepdims=True) + 1e-8

    if verbose:
        print("\nNormalizing data for fair evaluation...")
        print(f"  Feature means: {feat_mean.flatten()[:3]}... (showing first 3)")
        print(f"  Feature stds:  {feat_std.flatten()[:3]}... (showing first 3)")

    # Normalize all data using holdout statistics
    synthetic_norm = (synthetic - feat_mean) / feat_std
    real_holdout_norm = (real_holdout - feat_mean) / feat_std

    # Split real holdout into train (for TRTR) and test
    n_holdout = len(real_holdout_norm)
    n_train_real = int(0.7 * n_holdout)
    real_train = real_holdout_norm[:n_train_real]
    real_test = real_holdout_norm[n_train_real:]

    def prepare_prediction_data(sequences: np.ndarray) -> tuple[torch.Tensor, torch.Tensor]:
        """Prepare X (all but last timestep) and Y (shifted by 1)."""
        X = torch.tensor(sequences[:, :-1, :], dtype=torch.float32)
        Y = torch.tensor(sequences[:, 1:, :], dtype=torch.float32)
        return X, Y

    def train_predictor(
        train_data: np.ndarray,
        n_epochs: int,
        desc: str,
    ) -> PostHocLSTMPredictor:
        """Train LSTM predictor on given data."""
        model = PostHocLSTMPredictor(n_features, hidden_dim, num_layers).to(device)
        optimizer = optim.Adam(model.parameters(), lr=learning_rate)
        criterion = nn.L1Loss()  # MAE loss

        X_train, Y_train = prepare_prediction_data(train_data)
        dataset = TensorDataset(X_train, Y_train)
        loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

        model.train()
        iterator = tqdm(range(n_epochs), desc=desc, disable=not PROGRESS_BARS)
        for _ in iterator:
            epoch_loss = 0.0
            for X_batch, Y_batch in loader:
                X_batch, Y_batch = X_batch.to(device), Y_batch.to(device)

                optimizer.zero_grad()
                Y_pred = model(X_batch)
                loss = criterion(Y_pred, Y_batch)
                loss.backward()
                optimizer.step()

                epoch_loss += loss.item()

            if verbose:
                iterator.set_postfix(loss=epoch_loss / len(loader))

        return model

    def evaluate_predictor(model: PostHocLSTMPredictor, test_data: np.ndarray) -> float:
        """Evaluate MAE on test data."""
        model.eval()
        X_test, Y_test = prepare_prediction_data(test_data)
        X_test, Y_test = X_test.to(device), Y_test.to(device)

        with torch.no_grad():
            Y_pred = model(X_test)
            mae = torch.abs(Y_pred - Y_test).mean().item()

        return mae

    # TSTR: Train on Synthetic (normalized), Test on Real (normalized)
    model_tstr = train_predictor(synthetic_norm, epochs, "TSTR Predictor")
    mae_tstr = evaluate_predictor(model_tstr, real_test)

    # TRTR: Train on Real, Test on Real (baseline)
    model_trtr = train_predictor(real_train, epochs, "TRTR Predictor")
    mae_trtr = evaluate_predictor(model_trtr, real_test)

    # Compute ratio
    ratio = mae_tstr / mae_trtr if mae_trtr > 0 else float("inf")

    if verbose:
        print("\nPredictive Score (MAE, lower is better):")
        print(f"  TRTR (baseline): {mae_trtr:.6f}")
        print(f"  TSTR (synthetic): {mae_tstr:.6f}")
        print(f"  Ratio (TSTR/TRTR): {ratio:.3f}")
        if ratio <= 1.1:
            print("  Assessment: [OK] Good utility")
        elif ratio <= 1.5:
            print("  Assessment: WARNING: Moderate utility")
        else:
            print("  Assessment: [FAIL] Poor utility")

    return {
        "mae_tstr": mae_tstr,
        "mae_trtr": mae_trtr,
        "ratio": ratio,
        "model_tstr": model_tstr,
        "model_trtr": model_trtr,
    }


def compute_discriminative_score(
    real: np.ndarray,
    synthetic: np.ndarray,
    hidden_dim: int = 128,
    num_layers: int = 2,
    epochs: int = 50,
    batch_size: int = 64,
    learning_rate: float = 1e-3,
    device: str | None = None,
    quick_test: bool = False,
    verbose: bool = True,
) -> dict[str, Any]:
    """
    Compute discriminative score per TimeGAN paper (Section 5.2).

    Protocol:
    1. Train LSTM classifier to distinguish real vs synthetic
    2. Evaluate accuracy on held-out test set
    3. Lower accuracy = better fidelity (harder to tell apart)

    Target: Accuracy ~50% = indistinguishable
    Good: Accuracy < 70%
    Poor: Accuracy > 85%

    Args:
        real: Real sequences (n_samples, seq_len, n_features)
        synthetic: Synthetic sequences (same shape)
        hidden_dim: LSTM hidden dimension (paper: 128)
        num_layers: Number of LSTM layers (paper: 2)
        epochs: Training epochs
        batch_size: Batch size
        learning_rate: Learning rate
        device: 'cuda' or 'cpu' (auto-detected if None)
        quick_test: If True, reduce epochs for faster testing
        verbose: Print progress

    Returns:
        dict with:
        - accuracy: Classification accuracy (lower = better fidelity)
        - auc: ROC AUC score
        - model: Trained discriminator
    """
    if quick_test:
        epochs = min(10, epochs)

    if device is None:
        device = "cuda" if torch.cuda.is_available() else "cpu"

    device = torch.device(device)
    n_features = real.shape[-1]

    # Normalize data per-feature (using real data statistics)
    real_flat = real.reshape(-1, n_features)
    feat_mean = real_flat.mean(axis=0, keepdims=True)
    feat_std = real_flat.std(axis=0, keepdims=True) + 1e-8

    real_norm = (real - feat_mean) / feat_std
    synth_norm = (synthetic - feat_mean) / feat_std

    # Balance dataset sizes
    n_samples = min(len(real_norm), len(synth_norm))
    real_subset = real_norm[:n_samples]
    synth_subset = synth_norm[:n_samples]

    # Fixed sequential split (80/20) - matches benchmark methodology
    # This ensures temporal consistency: test set has most recent data from both
    n_train = int(0.8 * n_samples)
    train_idx = np.arange(n_train)
    test_idx = np.arange(n_train, n_samples)

    # Stack real and synthetic for train/test separately
    X_train = np.concatenate([real_subset[train_idx], synth_subset[train_idx]], axis=0)
    X_test = np.concatenate([real_subset[test_idx], synth_subset[test_idx]], axis=0)

    n_train_samples = len(train_idx)
    n_test_samples = len(test_idx)
    y_train = np.concatenate([np.ones(n_train_samples), np.zeros(n_train_samples)])
    y_test = np.concatenate([np.ones(n_test_samples), np.zeros(n_test_samples)])

    # Convert to tensors
    X_train_t = torch.tensor(X_train, dtype=torch.float32)
    y_train_t = torch.tensor(y_train, dtype=torch.float32).unsqueeze(1)
    X_test_t = torch.tensor(X_test, dtype=torch.float32)
    y_test_t = torch.tensor(y_test, dtype=torch.float32).unsqueeze(1)

    # Create data loader
    train_dataset = TensorDataset(X_train_t, y_train_t)
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

    # Initialize model
    model = PostHocLSTMDiscriminator(n_features, hidden_dim, num_layers).to(device)
    optimizer = optim.Adam(model.parameters(), lr=learning_rate)
    criterion = nn.BCEWithLogitsLoss()

    # Training loop
    model.train()
    iterator = tqdm(range(epochs), desc="Discriminator", disable=not PROGRESS_BARS)
    for _ in iterator:
        epoch_loss = 0.0
        for X_batch, y_batch in train_loader:
            X_batch, y_batch = X_batch.to(device), y_batch.to(device)

            optimizer.zero_grad()
            logits = model(X_batch)
            loss = criterion(logits, y_batch)
            loss.backward()
            optimizer.step()

            epoch_loss += loss.item()

        if verbose:
            iterator.set_postfix(loss=epoch_loss / len(train_loader))

    # Evaluation
    model.eval()
    X_test_t = X_test_t.to(device)

    with torch.no_grad():
        logits = model(X_test_t)
        probs = torch.sigmoid(logits).cpu().numpy().flatten()
        preds = (probs > 0.5).astype(int)

    accuracy = accuracy_score(y_test, preds)
    try:
        auc = roc_auc_score(y_test, probs)
    except ValueError:
        auc = 0.5

    if verbose:
        print("\nDiscriminative Score (lower accuracy = better):")
        print(f"  Accuracy: {accuracy:.1%}")
        print(f"  ROC AUC: {auc:.4f}")
        if accuracy <= 0.55:
            print("  Assessment: [OK] Excellent fidelity (indistinguishable)")
        elif accuracy <= 0.70:
            print("  Assessment: [OK] Good fidelity")
        elif accuracy <= 0.85:
            print("  Assessment: WARNING: Moderate fidelity")
        else:
            print("  Assessment: [FAIL] Poor fidelity (easily distinguished)")

    return {
        "accuracy": accuracy,
        "auc": auc,
        "model": model,
    }


def run_timegan_evaluation(
    synthetic: np.ndarray,
    real_train: np.ndarray,
    real_holdout: np.ndarray,
    hidden_dim: int = 128,
    num_layers: int = 2,
    epochs_pred: int = 100,
    epochs_disc: int = 50,
    iterations_yoon: int = 5000,
    device: str | None = None,
    quick_test: bool = False,
    verbose: bool = True,
    include_yoon: bool = True,
) -> dict[str, Any]:
    """
    Run complete TimeGAN evaluation suite.

    Combines:
    1. Yoon predictive score (optional): Last-feature prediction
    2. Comprehensive predictive score: All-feature TSTR with baseline
    3. Discriminative score

    Args:
        synthetic: Synthetic sequences from generator
        real_train: Real sequences for discriminative training
        real_holdout: Real sequences for predictive evaluation
        hidden_dim: LSTM hidden dimension
        num_layers: Number of LSTM layers
        epochs_pred: Epochs for comprehensive predictive model
        epochs_disc: Epochs for discriminator
        iterations_yoon: Iterations for Yoon metric (default: 5000)
        device: Compute device
        quick_test: Reduce epochs/iterations for testing
        verbose: Print results
        include_yoon: If True, run Yoon metric for comparison

    Returns:
        dict with yoon, predictive, discriminative, summary
    """
    if verbose:
        print("=" * 60)
        print("TimeGAN Evaluation Suite")
        print("=" * 60)
        print("\nReference: Yoon et al., NeurIPS 2019, Section 5.2")

    # Align shapes
    min_samples = min(len(synthetic), len(real_train), len(real_holdout))
    min_seq_len = min(synthetic.shape[1], real_train.shape[1], real_holdout.shape[1])
    min_features = min(synthetic.shape[2], real_train.shape[2], real_holdout.shape[2])

    synthetic = synthetic[:min_samples, :min_seq_len, :min_features]
    real_train = real_train[:min_samples, :min_seq_len, :min_features]
    real_holdout = real_holdout[:min_samples, :min_seq_len, :min_features]

    results = {}

    # 1. Yoon predictive score (optional)
    if include_yoon:
        if verbose:
            print("\n1. YOON PREDICTIVE SCORE")
            print("-" * 40)
            print("(Last-feature prediction per Yoon et al.)")
        yoon_results = compute_predictive_score_yoon(
            synthetic=synthetic,
            real=real_holdout,
            iterations=iterations_yoon,
            device=device,
            quick_test=quick_test,
            verbose=verbose,
        )
        results["yoon"] = yoon_results

    # 2. Comprehensive predictive score
    if verbose:
        section_num = "2" if include_yoon else "1"
        print(f"\n{section_num}. COMPREHENSIVE PREDICTIVE SCORE (Utility)")
        print("-" * 40)
        print("(All-feature TSTR with TRTR baseline)")
    pred_results = compute_predictive_score(
        synthetic=synthetic,
        real_holdout=real_holdout,
        hidden_dim=hidden_dim,
        num_layers=num_layers,
        epochs=epochs_pred,
        device=device,
        quick_test=quick_test,
        verbose=verbose,
    )
    results["predictive"] = pred_results

    # 3. Discriminative score
    if verbose:
        section_num = "3" if include_yoon else "2"
        print(f"\n{section_num}. DISCRIMINATIVE SCORE (Fidelity)")
        print("-" * 40)
    disc_results = compute_discriminative_score(
        real=real_train,
        synthetic=synthetic,
        hidden_dim=hidden_dim,
        num_layers=num_layers,
        epochs=epochs_disc,
        device=device,
        quick_test=quick_test,
        verbose=verbose,
    )
    results["discriminative"] = disc_results

    # Combined summary
    if verbose:
        print("\n" + "=" * 60)
        print("SUMMARY")
        print("=" * 60)

        if include_yoon:
            print("\nYoon Metric (last-feature prediction):")
            print(f"  MAE on last feature: {yoon_results['mae']:.6f}")
            print("  (Paper reports ~0.04 for GOOG stock)")

        print("\nComprehensive Assessment:")
        print(f"  Predictive Score (MAE ratio): {pred_results['ratio']:.3f}")
        print(f"  Discriminative Score (accuracy): {disc_results['accuracy']:.1%}")

        # Overall assessment
        pred_good = pred_results["ratio"] <= 1.2
        disc_good = disc_results["accuracy"] <= 0.70

        if pred_good and disc_good:
            print("\n  Overall: [OK] Generator passes both tests")
        elif pred_good or disc_good:
            print("\n  Overall: WARNING: Generator passes one test")
        else:
            print("\n  Overall: [FAIL] Generator fails both tests")

    # Build summary
    summary = {
        "mae_ratio": pred_results["ratio"],
        "disc_accuracy": disc_results["accuracy"],
        "disc_auc": disc_results["auc"],
    }
    if include_yoon:
        summary["yoon_mae"] = yoon_results["mae"]

    results["summary"] = summary
    return results


if __name__ == "__main__":
    # Quick test with random data
    print("TimeGAN Metrics Module - Quick Test")
    print("=" * 40)

    # Generate random test data
    np.random.seed(42)
    n_samples = 100
    seq_len = 24
    n_features = 6

    real_train = np.random.randn(n_samples, seq_len, n_features).astype(np.float32)
    real_holdout = np.random.randn(n_samples, seq_len, n_features).astype(np.float32)
    synthetic = np.random.randn(n_samples, seq_len, n_features).astype(np.float32)

    # Run quick evaluation
    results = run_timegan_evaluation(
        synthetic=synthetic,
        real_train=real_train,
        real_holdout=real_holdout,
        quick_test=True,
        verbose=True,
    )

    print("\nTest completed successfully!")

```

Vollständig mit Quellenangabe unter der Lizenz der Quelle angezeigt. Lizenz: MIT

Diese Zusammenfassung wurde vom Research-Agenten von Stratmill anhand des Originals verfasst; sie ist keine Kopie der Quelle.