Генеративний AI для синтетичних даних: коли реальних даних недостатньо

Хочеш натренувати модель для розпізнавання рідкісних захворювань. Проблема: таких випадків — сотні. Потрібно — тисячі.

Хочеш autonomous driving для snow conditions. Проблема: твоя компанія в Каліфорнії, снігу немає.

Хочеш fraud detection. Проблема: шахрайських транзакцій — 0.1%, legitimate — 99.9%.

Рішення? Згенерувати дані, яких не вистачає. Synthetic data — не хак, а технологія, якою користуються Tesla, Waymo, pharma giants. Gartner прогнозує, що до 2030 року синтетичні дані перевищать реальні у тренуванні AI-систем. Це не майбутнє — це вже сьогодення.

Проблема даних у ML: чому реальних завжди недостатньо

Data-hungry reality сучасного ML:

• GPT-4: трильйони токенів (весь доступний інтернет вичерпаний)
• ImageNet: 14 мільйонів images (роки збору та розмітки)
• Autonomous driving: мільярди miles (десятиліття збору)
• AlphaFold: мільйони protein structures (століття експериментів)

Фундаментальні bottlenecks:

1. Рідкісні події (Class Imbalance)

Cancer cases:        0.5% населення
Fraud transactions:  0.1% операцій
Critical failures:   0.001% систем
Black swan events:   Практично ніколи

Модель бачить тисячі нормальних випадків і одиниці аномалій. Результат — bias до majority class.

2. Privacy та регуляції

• Medical records: HIPAA в США, GDPR в Європі
• Financial data: PCI DSS, банківська таємниця
• Personal information: право на забуття, consent
• Children's data: COPPA, посилені вимоги

3. Annotation cost

• Simple labels: $0.10-0.50 per item
• Medical expert: $5-50 per diagnosis
• Legal review: $100+ per document
• Мільйони labels = мільйони доларів
• Expert disagreement: потрібно кілька експертів

4. Edge cases та corner scenarios

• Rare weather conditions (снігопад у пустелі)
• Unusual user behavior (нетипові патерни)
• Hardware failures (рідкісні комбінації)
• Things that should never happen (але трапляються)

Математичний фреймворк синтетичних даних

Формальне визначення:

Нехай $P_{real}(x)$ — розподіл реальних даних.

Генеративна модель $G_\theta$ навчається апроксимувати цей розподіл:

$P{synth}(x) = G\theta(z), \quad z \sim P(z)$

Мета: мінімізувати дивергенцію між розподілами:

$\min\theta D(P{real} || P_{synth})$

Різні методи використовують різні дивергенції:

• GAN: Jensen-Shannon Divergence
• VAE: KL Divergence (ELBO bound)
• Diffusion: Score matching
• Flow: Exact likelihood

import torch
import torch.nn as nn
import numpy as np
from typing import Tuple, List, Optional
from dataclasses import dataclass

@dataclass
class SyntheticDataConfig:
    """Конфігурація для генерації синтетичних даних."""
    generator_type: str  # 'gan', 'vae', 'diffusion', 'simulator'
    latent_dim: int = 128
    num_samples: int = 10000
    quality_threshold: float = 0.8
    privacy_budget: Optional[float] = None  # epsilon для DP

class SyntheticDataGenerator:
    """Базовий клас для генерації синтетичних даних."""

    def __init__(self, config: SyntheticDataConfig):
        self.config = config
        self.quality_metrics = {}

    def fit(self, real_data: np.ndarray) -> 'SyntheticDataGenerator':
        """Навчання на реальних даних."""
        raise NotImplementedError

    def generate(self, n_samples: int) -> np.ndarray:
        """Генерація синтетичних samples."""
        raise NotImplementedError

    def evaluate_quality(
        self,
        synthetic: np.ndarray,
        real: np.ndarray
    ) -> dict:
        """Оцінка якості синтетичних даних."""
        return {
            'fidelity': self._compute_fidelity(synthetic, real),
            'diversity': self._compute_diversity(synthetic),
            'utility': self._compute_utility(synthetic, real),
            'privacy': self._compute_privacy_risk(synthetic, real)
        }

Типи синтетичних даних та їх характеристики

1. Fully Synthetic Data

• Згенеровані повністю з нуля
• Немає прямого зв'язку з реальними записами
• Privacy-safe by design
• Потребує якісну генеративну модель

2. Augmented/Transformed Real Data

• Реальні дані + детерміновані трансформації
• Flips, rotations, color changes, noise
• Зберігає структуру оригіналу
• Обмежена новизна

3. Simulation-based Synthetic Data

• Фізичні симулятори (CARLA, Unity, Unreal)
• Точний контроль над параметрами
• Ground truth автоматично
• Sim-to-real gap проблема

4. Generative Model-based Data

• GANs, Diffusion Models, VAEs, Flows
• Вчаться на реальних, генерують нові
• Найгнучкіші та найпотужніші
• Потребують великих обчислювальних ресурсів

class DataAugmentationPipeline:
    """Pipeline для різних типів аугментації та генерації."""

    def __init__(self):
        self.transforms = []
        self.generators = []

    def add_geometric_transforms(self):
        """Геометричні трансформації для images."""
        self.transforms.extend([
            ('horizontal_flip', lambda x: np.fliplr(x)),
            ('vertical_flip', lambda x: np.flipud(x)),
            ('rotation_90', lambda x: np.rot90(x)),
            ('rotation_180', lambda x: np.rot90(x, 2)),
            ('random_crop', self._random_crop),
            ('scale', self._random_scale),
        ])
        return self

    def add_color_transforms(self):
        """Колірні трансформації."""
        self.transforms.extend([
            ('brightness', lambda x: np.clip(x * np.random.uniform(0.8, 1.2), 0, 255)),
            ('contrast', self._adjust_contrast),
            ('saturation', self._adjust_saturation),
            ('hue_shift', self._hue_shift),
            ('gaussian_noise', lambda x: x + np.random.normal(0, 10, x.shape)),
        ])
        return self

    def add_mixup(self, alpha: float = 0.2):
        """MixUp аугментація для регуляризації."""
        def mixup(x1, x2, y1, y2):
            lam = np.random.beta(alpha, alpha)
            x_mixed = lam * x1 + (1 - lam) * x2
            y_mixed = lam * y1 + (1 - lam) * y2
            return x_mixed, y_mixed
        self.transforms.append(('mixup', mixup))
        return self

    def add_cutmix(self):
        """CutMix аугментація."""
        def cutmix(x1, x2, y1, y2):
            lam = np.random.uniform(0.3, 0.7)
            h, w = x1.shape[:2]
            cut_h = int(h * np.sqrt(1 - lam))
            cut_w = int(w * np.sqrt(1 - lam))
            cx, cy = np.random.randint(0, w), np.random.randint(0, h)

            x1_copy = x1.copy()
            x1_copy[max(0, cy-cut_h//2):min(h, cy+cut_h//2),
                   max(0, cx-cut_w//2):min(w, cx+cut_w//2)] = \
            x2[max(0, cy-cut_h//2):min(h, cy+cut_h//2),
               max(0, cx-cut_w//2):min(w, cx+cut_w//2)]

            y_mixed = lam * y1 + (1 - lam) * y2
            return x1_copy, y_mixed
        self.transforms.append(('cutmix', cutmix))
        return self

Simulation-Based Generation: Autonomous Driving

Як Waymo/Tesla тренують моделі без мільйонів реальних аварій:

1. Base world — 3D reconstruction реальних міст з LiDAR та photogrammetry
2. Scenario generation — рандомізовані ситуації з controlled parameters
3. Sensor simulation — точна модель LiDAR, cameras, radar з noise models
4. Physics engine — реалістична динаміка автомобілів, пішоходів, погоди

CARLA Simulator — повноцінний приклад:

import carla
import numpy as np
import queue
from dataclasses import dataclass
from typing import List, Dict, Any

@dataclass
class SensorConfig:
    """Конфігурація сенсора для симуляції."""
    sensor_type: str
    transform: carla.Transform
    attributes: Dict[str, Any]

class AutonomousDrivingSimulator:
    """Симулятор для генерації training data для autonomous driving."""

    def __init__(self, host: str = 'localhost', port: int = 2000):
        self.client = carla.Client(host, port)
        self.client.set_timeout(10.0)
        self.world = self.client.get_world()
        self.blueprint_library = self.world.get_blueprint_library()

        self.sensors = {}
        self.sensor_queues = {}
        self.vehicle = None

    def setup_vehicle(self, vehicle_type: str = 'vehicle.tesla.model3') -> carla.Actor:
        """Створення ego vehicle."""
        blueprint = self.blueprint_library.find(vehicle_type)
        spawn_points = self.world.get_map().get_spawn_points()
        spawn_point = np.random.choice(spawn_points)

        self.vehicle = self.world.spawn_actor(blueprint, spawn_point)
        return self.vehicle

    def add_camera(
        self,
        name: str,
        image_size: tuple = (1920, 1080),
        fov: float = 90.0,
        position: tuple = (1.5, 0, 2.4)  # x, y, z відносно vehicle
    ) -> carla.Actor:
        """Додавання RGB камери."""
        camera_bp = self.blueprint_library.find('sensor.camera.rgb')
        camera_bp.set_attribute('image_size_x', str(image_size[0]))
        camera_bp.set_attribute('image_size_y', str(image_size[1]))
        camera_bp.set_attribute('fov', str(fov))

        camera_transform = carla.Transform(
            carla.Location(x=position[0], y=position[1], z=position[2])
        )

        camera = self.world.spawn_actor(
            camera_bp,
            camera_transform,
            attach_to=self.vehicle
        )

        # Setup queue for data collection
        self.sensor_queues[name] = queue.Queue()
        camera.listen(lambda data: self.sensor_queues[name].put(data))
        self.sensors[name] = camera

        return camera

    def add_lidar(
        self,
        name: str,
        channels: int = 64,
        range_m: float = 100.0,
        points_per_second: int = 1200000,
        rotation_frequency: float = 20.0
    ) -> carla.Actor:
        """Додавання LiDAR сенсора."""
        lidar_bp = self.blueprint_library.find('sensor.lidar.ray_cast')
        lidar_bp.set_attribute('channels', str(channels))
        lidar_bp.set_attribute('range', str(range_m))
        lidar_bp.set_attribute('points_per_second', str(points_per_second))
        lidar_bp.set_attribute('rotation_frequency', str(rotation_frequency))

        lidar_transform = carla.Transform(carla.Location(z=2.5))
        lidar = self.world.spawn_actor(
            lidar_bp,
            lidar_transform,
            attach_to=self.vehicle
        )

        self.sensor_queues[name] = queue.Queue()
        lidar.listen(lambda data: self.sensor_queues[name].put(data))
        self.sensors[name] = lidar

        return lidar

    def add_semantic_segmentation(self, name: str) -> carla.Actor:
        """Камера з автоматичною семантичною сегментацією — ground truth безкоштовно!"""
        seg_bp = self.blueprint_library.find('sensor.camera.semantic_segmentation')
        seg_bp.set_attribute('image_size_x', '1920')
        seg_bp.set_attribute('image_size_y', '1080')

        seg_transform = carla.Transform(carla.Location(x=1.5, z=2.4))
        seg_camera = self.world.spawn_actor(
            seg_bp,
            seg_transform,
            attach_to=self.vehicle
        )

        self.sensor_queues[name] = queue.Queue()
        seg_camera.listen(lambda data: self.sensor_queues[name].put(data))
        self.sensors[name] = seg_camera

        return seg_camera

    def set_weather(
        self,
        cloudiness: float = 0.0,
        precipitation: float = 0.0,
        fog_density: float = 0.0,
        sun_altitude: float = 70.0
    ):
        """Контроль погодних умов."""
        weather = carla.WeatherParameters(
            cloudiness=cloudiness,
            precipitation=precipitation,
            precipitation_deposits=precipitation * 0.5,
            fog_density=fog_density,
            fog_distance=max(0, 100 - fog_density * 10),
            sun_altitude_angle=sun_altitude
        )
        self.world.set_weather(weather)

    def spawn_traffic(
        self,
        num_vehicles: int = 50,
        num_pedestrians: int = 30
    ):
        """Створення реалістичного трафіку."""
        # Spawn vehicles
        vehicle_bps = self.blueprint_library.filter('vehicle.*')
        spawn_points = self.world.get_map().get_spawn_points()

        for _ in range(min(num_vehicles, len(spawn_points) - 1)):
            bp = np.random.choice(list(vehicle_bps))
            sp = np.random.choice(spawn_points)
            try:
                vehicle = self.world.spawn_actor(bp, sp)
                vehicle.set_autopilot(True)
            except:
                pass

    def collect_scenario(
        self,
        duration_seconds: float = 10.0,
        fps: int = 20
    ) -> Dict[str, List]:
        """Збір даних з усіх сенсорів протягом сценарію."""
        collected_data = {name: [] for name in self.sensors}

        # Enable synchronous mode
        settings = self.world.get_settings()
        settings.synchronous_mode = True
        settings.fixed_delta_seconds = 1.0 / fps
        self.world.apply_settings(settings)

        num_frames = int(duration_seconds * fps)

        for _ in range(num_frames):
            self.world.tick()

            for name, q in self.sensor_queues.items():
                try:
                    data = q.get(timeout=1.0)
                    collected_data[name].append(self._process_sensor_data(name, data))
                except queue.Empty:
                    pass

        return collected_data

Ключові переваги симуляції:

• Необмежена кількість даних (generate on demand)
• Автоматичні labels (ground truth із симулятора)
• Безпечне тестування dangerous scenarios
• Повний контроль над умовами

GAN-Based Synthetic Data Generation

Conditional GAN для контрольованої генерації:

class ConditionalGenerator(nn.Module):
    """Generator з умовним входом для контрольованої генерації."""

    def __init__(
        self,
        latent_dim: int = 128,
        num_classes: int = 10,
        image_size: int = 256
    ):
        super().__init__()
        self.latent_dim = latent_dim
        self.num_classes = num_classes

        # Class embedding
        self.class_embedding = nn.Embedding(num_classes, latent_dim)

        # Generator network
        self.fc = nn.Linear(latent_dim * 2, 512 * 4 * 4)

        self.blocks = nn.ModuleList([
            self._make_block(512, 256),  # 4 -> 8
            self._make_block(256, 128),  # 8 -> 16
            self._make_block(128, 64),   # 16 -> 32
            self._make_block(64, 32),    # 32 -> 64
            self._make_block(32, 16),    # 64 -> 128
            self._make_block(16, 8),     # 128 -> 256
        ])

        self.final = nn.Sequential(
            nn.Conv2d(8, 3, 3, 1, 1),
            nn.Tanh()
        )

    def _make_block(self, in_ch: int, out_ch: int) -> nn.Module:
        return nn.Sequential(
            nn.Upsample(scale_factor=2, mode='nearest'),
            nn.Conv2d(in_ch, out_ch, 3, 1, 1),
            nn.BatchNorm2d(out_ch),
            nn.LeakyReLU(0.2)
        )

    def forward(self, z: torch.Tensor, labels: torch.Tensor) -> torch.Tensor:
        # Combine noise with class embedding
        class_emb = self.class_embedding(labels)
        combined = torch.cat([z, class_emb], dim=1)

        x = self.fc(combined)
        x = x.view(-1, 512, 4, 4)

        for block in self.blocks:
            x = block(x)

        return self.final(x)


class MedicalImageGAN:
    """GAN для генерації медичних зображень з рідкісними патологіями."""

    def __init__(
        self,
        image_size: int = 256,
        num_pathologies: int = 5,
        latent_dim: int = 128
    ):
        self.generator = ConditionalGenerator(
            latent_dim=latent_dim,
            num_classes=num_pathologies,
            image_size=image_size
        )
        self.discriminator = self._build_discriminator(image_size, num_pathologies)

        self.g_optimizer = torch.optim.Adam(
            self.generator.parameters(),
            lr=0.0002,
            betas=(0.5, 0.999)
        )
        self.d_optimizer = torch.optim.Adam(
            self.discriminator.parameters(),
            lr=0.0002,
            betas=(0.5, 0.999)
        )

    def _build_discriminator(self, image_size: int, num_classes: int) -> nn.Module:
        """Projection Discriminator для cGAN."""
        return ProjectionDiscriminator(image_size, num_classes)

    def train_step(
        self,
        real_images: torch.Tensor,
        real_labels: torch.Tensor
    ) -> dict:
        """Один крок тренування GAN."""
        batch_size = real_images.size(0)
        device = real_images.device

        # Train Discriminator
        self.d_optimizer.zero_grad()

        # Real images
        real_output = self.discriminator(real_images, real_labels)
        d_loss_real = torch.relu(1.0 - real_output).mean()

        # Fake images
        z = torch.randn(batch_size, self.generator.latent_dim, device=device)
        fake_images = self.generator(z, real_labels)
        fake_output = self.discriminator(fake_images.detach(), real_labels)
        d_loss_fake = torch.relu(1.0 + fake_output).mean()

        d_loss = d_loss_real + d_loss_fake
        d_loss.backward()
        self.d_optimizer.step()

        # Train Generator
        self.g_optimizer.zero_grad()

        fake_output = self.discriminator(fake_images, real_labels)
        g_loss = -fake_output.mean()

        g_loss.backward()
        self.g_optimizer.step()

        return {
            'd_loss': d_loss.item(),
            'g_loss': g_loss.item(),
            'd_real': real_output.mean().item(),
            'd_fake': fake_output.mean().item()
        }

    def generate_pathology_samples(
        self,
        pathology_id: int,
        num_samples: int = 100
    ) -> torch.Tensor:
        """Генерація samples для конкретної патології."""
        self.generator.eval()
        with torch.no_grad():
            z = torch.randn(num_samples, self.generator.latent_dim)
            labels = torch.full((num_samples,), pathology_id, dtype=torch.long)
            samples = self.generator(z, labels)
        return samples

Diffusion Models для синтетичних даних

Сучасний state-of-the-art для high-quality generation:

from diffusers import StableDiffusionPipeline, ControlNetModel
from diffusers import StableDiffusionControlNetPipeline
import torch

class DiffusionSyntheticGenerator:
    """Генерація синтетичних даних за допомогою Diffusion Models."""

    def __init__(
        self,
        model_id: str = "stabilityai/stable-diffusion-xl-base-1.0",
        device: str = "cuda"
    ):
        self.device = device
        self.pipe = StableDiffusionPipeline.from_pretrained(
            model_id,
            torch_dtype=torch.float16,
            use_safetensors=True
        ).to(device)

        # Enable memory optimizations
        self.pipe.enable_attention_slicing()
        self.pipe.enable_vae_slicing()

    def generate_rare_scenarios(
        self,
        scenario_prompts: List[str],
        num_per_scenario: int = 100,
        guidance_scale: float = 7.5,
        num_inference_steps: int = 30
    ) -> Dict[str, List]:
        """Генерація рідкісних сценаріїв за текстовими описами."""

        generated = {}

        for prompt in scenario_prompts:
            generated[prompt] = []

            for i in range(num_per_scenario):
                # Vary the seed for diversity
                generator = torch.Generator(device=self.device).manual_seed(i)

                image = self.pipe(
                    prompt=prompt,
                    negative_prompt="blurry, low quality, distorted",
                    guidance_scale=guidance_scale,
                    num_inference_steps=num_inference_steps,
                    generator=generator
                ).images[0]

                generated[prompt].append(image)

        return generated

    def generate_with_structure_control(
        self,
        prompt: str,
        control_image: np.ndarray,
        control_type: str = "canny"
    ):
        """Генерація з контролем структури через ControlNet."""

        controlnet = ControlNetModel.from_pretrained(
            f"lllyasviel/control_v11p_sd15_{control_type}",
            torch_dtype=torch.float16
        )

        control_pipe = StableDiffusionControlNetPipeline.from_pretrained(
            "runwayml/stable-diffusion-v1-5",
            controlnet=controlnet,
            torch_dtype=torch.float16
        ).to(self.device)

        return control_pipe(
            prompt=prompt,
            image=control_image,
            controlnet_conditioning_scale=0.8,
            num_inference_steps=30
        ).images[0]


# Приклад використання для autonomous driving
driving_prompts = [
    "traffic accident on snowy highway, dashcam view, realistic photo",
    "pedestrian crossing in heavy fog, autonomous car perspective",
    "construction zone with unusual signage, front camera view",
    "deer crossing road at night, headlights illumination",
    "flooded road after heavy rain, car approaching puddle",
    "fallen tree blocking highway, emergency situation"
]

generator = DiffusionSyntheticGenerator()
rare_scenarios = generator.generate_rare_scenarios(
    driving_prompts,
    num_per_scenario=50
)

Tabular Synthetic Data: CTGAN та SDV

Для фінансових транзакцій, healthcare records, customer data:

from sdv.tabular import GaussianCopula, CTGAN, TVAE
from sdv.evaluation import evaluate
from sdv.metadata import SingleTableMetadata
import pandas as pd
import numpy as np

class TabularSyntheticPipeline:
    """Pipeline для генерації синтетичних табличних даних."""

    def __init__(self, method: str = 'ctgan'):
        self.method = method
        self.model = None
        self.metadata = None
        self.quality_report = None

    def prepare_metadata(self, data: pd.DataFrame) -> SingleTableMetadata:
        """Автоматичне визначення метаданих."""
        metadata = SingleTableMetadata()
        metadata.detect_from_dataframe(data)

        # Manual adjustments for specific columns
        for col in data.columns:
            if 'date' in col.lower():
                metadata.update_column(col, sdtype='datetime')
            elif 'id' in col.lower():
                metadata.update_column(col, sdtype='id')
            elif 'amount' in col.lower() or 'price' in col.lower():
                metadata.update_column(col, sdtype='numerical')

        self.metadata = metadata
        return metadata

    def fit(self, data: pd.DataFrame, epochs: int = 300):
        """Навчання генеративної моделі."""
        if self.metadata is None:
            self.prepare_metadata(data)

        if self.method == 'ctgan':
            self.model = CTGAN(
                metadata=self.metadata,
                epochs=epochs,
                batch_size=500,
                generator_dim=(256, 256, 256),
                discriminator_dim=(256, 256, 256),
                generator_lr=2e-4,
                discriminator_lr=2e-4,
                discriminator_steps=1,
                pac=10  # Number of samples to group together
            )
        elif self.method == 'tvae':
            self.model = TVAE(
                metadata=self.metadata,
                epochs=epochs,
                embedding_dim=128,
                compress_dims=(128, 128),
                decompress_dims=(128, 128)
            )
        elif self.method == 'gaussian_copula':
            self.model = GaussianCopula(metadata=self.metadata)

        self.model.fit(data)
        return self

    def generate(
        self,
        num_rows: int,
        conditions: dict = None
    ) -> pd.DataFrame:
        """Генерація синтетичних записів."""
        if conditions:
            from sdv.sampling import Condition
            condition_obj = Condition(conditions, num_rows=num_rows)
            return self.model.sample_from_conditions([condition_obj])
        else:
            return self.model.sample(num_rows)

    def evaluate_quality(
        self,
        synthetic_data: pd.DataFrame,
        real_data: pd.DataFrame
    ) -> dict:
        """Комплексна оцінка якості синтетичних даних."""
        from sdv.evaluation import evaluate_quality

        report = evaluate_quality(
            real_data=real_data,
            synthetic_data=synthetic_data,
            metadata=self.metadata
        )

        self.quality_report = {
            'overall_score': report.get_score(),
            'column_shapes': report.get_details('Column Shapes'),
            'column_pair_trends': report.get_details('Column Pair Trends')
        }

        return self.quality_report


class FraudDataBalancer:
    """Балансування fraud detection dataset через синтетичні дані."""

    def __init__(self, target_fraud_ratio: float = 0.3):
        self.target_ratio = target_fraud_ratio
        self.fraud_generator = None

    def balance_dataset(
        self,
        data: pd.DataFrame,
        fraud_column: str = 'is_fraud'
    ) -> pd.DataFrame:
        """Балансування через генерацію синтетичних fraud cases."""

        # Розділення на fraud і non-fraud
        fraud_data = data[data[fraud_column] == 1]
        normal_data = data[data[fraud_column] == 0]

        current_ratio = len(fraud_data) / len(data)
        print(f"Current fraud ratio: {current_ratio:.2%}")

        # Скільки fraud samples потрібно згенерувати
        target_fraud_count = int(
            len(normal_data) * self.target_ratio / (1 - self.target_ratio)
        )
        samples_to_generate = max(0, target_fraud_count - len(fraud_data))

        print(f"Generating {samples_to_generate} synthetic fraud samples...")

        # Тренуємо генератор тільки на fraud data
        self.fraud_generator = TabularSyntheticPipeline(method='ctgan')
        fraud_features = fraud_data.drop(columns=[fraud_column])
        self.fraud_generator.fit(fraud_features, epochs=500)

        # Генеруємо синтетичні fraud
        synthetic_fraud = self.fraud_generator.generate(samples_to_generate)
        synthetic_fraud[fraud_column] = 1

        # Об'єднуємо
        balanced_data = pd.concat([
            data,
            synthetic_fraud
        ], ignore_index=True)

        final_ratio = balanced_data[fraud_column].mean()
        print(f"Final fraud ratio: {final_ratio:.2%}")

        return balanced_data

Privacy-Preserving Synthetic Data

Диференціальна приватність для гарантованого захисту:

from opacus import PrivacyEngine
from opacus.validators import ModuleValidator
import torch
import torch.nn as nn

class DifferentiallyPrivateSynthesizer:
    """Генерація синтетичних даних з математичними гарантіями приватності."""

    def __init__(
        self,
        epsilon: float = 1.0,
        delta: float = 1e-5,
        max_grad_norm: float = 1.0
    ):
        self.epsilon = epsilon
        self.delta = delta
        self.max_grad_norm = max_grad_norm
        self.privacy_engine = None

    def _make_private(
        self,
        model: nn.Module,
        optimizer: torch.optim.Optimizer,
        data_loader: torch.utils.data.DataLoader
    ):
        """Обгортка моделі для DP-SGD."""

        # Validate model compatibility
        model = ModuleValidator.fix(model)

        self.privacy_engine = PrivacyEngine()

        model, optimizer, data_loader = self.privacy_engine.make_private_with_epsilon(
            module=model,
            optimizer=optimizer,
            data_loader=data_loader,
            epochs=100,
            target_epsilon=self.epsilon,
            target_delta=self.delta,
            max_grad_norm=self.max_grad_norm
        )

        return model, optimizer, data_loader

    def get_privacy_spent(self) -> tuple:
        """Отримання витраченого privacy budget."""
        if self.privacy_engine:
            return self.privacy_engine.get_epsilon(self.delta), self.delta
        return None, None


class PrivateCTGAN:
    """CTGAN з диференціальною приватністю."""

    def __init__(
        self,
        epsilon: float = 1.0,
        epochs: int = 300,
        batch_size: int = 500
    ):
        self.epsilon = epsilon
        self.epochs = epochs
        self.batch_size = batch_size

    def fit(self, data: pd.DataFrame):
        """Тренування з DP гарантіями."""
        # Розподіляємо epsilon між компонентами
        eps_encoder = self.epsilon * 0.1  # 10% на encoder
        eps_generator = self.epsilon * 0.9  # 90% на generator

        # Ініціалізація з noise для приватності
        self._fit_with_dp(data, eps_generator)

    def _add_noise_to_gradients(
        self,
        gradients: torch.Tensor,
        sensitivity: float,
        epsilon: float
    ) -> torch.Tensor:
        """Додавання Gaussian noise для DP."""
        sigma = sensitivity * np.sqrt(2 * np.log(1.25 / self.delta)) / epsilon
        noise = torch.normal(0, sigma, size=gradients.shape)
        return gradients + noise


# Приклад: генерація приватних медичних даних
def generate_private_medical_data():
    """Генерація синтетичних медичних записів з DP гарантіями."""

    # Завантаження реальних даних (обмежений доступ)
    real_data = pd.DataFrame({
        'age': np.random.randint(18, 90, 1000),
        'blood_pressure': np.random.randint(90, 180, 1000),
        'cholesterol': np.random.randint(150, 300, 1000),
        'has_diabetes': np.random.choice([0, 1], 1000, p=[0.9, 0.1]),
        'diagnosis_code': np.random.choice(['A00', 'B00', 'C00', 'D00'], 1000)
    })

    # Генерація з epsilon=1 (сильний захист)
    synthesizer = PrivateCTGAN(epsilon=1.0, epochs=300)
    synthesizer.fit(real_data)

    # Синтетичні дані можна публікувати
    synthetic_data = synthesizer.sample(10000)

    print("Privacy guarantee: No individual record can be identified with")
    print(f"probability > e^{synthesizer.epsilon} ≈ {np.exp(synthesizer.epsilon):.2f}")

    return synthetic_data

Оцінка якості синтетичних даних

Комплексний фреймворк для валідації:

from scipy import stats
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score
import numpy as np

class SyntheticDataQualityAssessment:
    """Комплексна оцінка якості синтетичних даних."""

    def __init__(self):
        self.metrics = {}

    def assess_fidelity(
        self,
        real_data: np.ndarray,
        synthetic_data: np.ndarray
    ) -> dict:
        """Оцінка статистичної схожості."""

        fidelity_metrics = {}

        for i in range(real_data.shape[1]):
            real_col = real_data[:, i]
            synth_col = synthetic_data[:, i]

            # Kolmogorov-Smirnov test
            ks_stat, ks_pvalue = stats.ks_2samp(real_col, synth_col)

            # Wasserstein distance
            wasserstein = stats.wasserstein_distance(real_col, synth_col)

            # Jensen-Shannon divergence (for discrete)
            js_div = self._jensen_shannon(real_col, synth_col)

            fidelity_metrics[f'column_{i}'] = {
                'ks_statistic': ks_stat,
                'ks_pvalue': ks_pvalue,
                'wasserstein': wasserstein,
                'js_divergence': js_div
            }

        self.metrics['fidelity'] = fidelity_metrics
        return fidelity_metrics

    def assess_utility(
        self,
        real_data: pd.DataFrame,
        synthetic_data: pd.DataFrame,
        target_column: str
    ) -> dict:
        """Train on Synthetic, Test on Real (TSTR) evaluation."""

        X_real = real_data.drop(columns=[target_column])
        y_real = real_data[target_column]
        X_synth = synthetic_data.drop(columns=[target_column])
        y_synth = synthetic_data[target_column]

        # Model trained on real data
        model_real = RandomForestClassifier(n_estimators=100)
        scores_real = cross_val_score(model_real, X_real, y_real, cv=5)

        # Model trained on synthetic, tested on real
        model_synth = RandomForestClassifier(n_estimators=100)
        model_synth.fit(X_synth, y_synth)
        score_tstr = model_synth.score(X_real, y_real)

        utility_metrics = {
            'real_cv_score': scores_real.mean(),
            'real_cv_std': scores_real.std(),
            'tstr_score': score_tstr,
            'utility_ratio': score_tstr / scores_real.mean()  # Ідеально ≈ 1.0
        }

        self.metrics['utility'] = utility_metrics
        return utility_metrics

    def assess_privacy(
        self,
        real_data: np.ndarray,
        synthetic_data: np.ndarray,
        num_attacks: int = 100
    ) -> dict:
        """Membership Inference Attack для оцінки privacy leakage."""

        # Найближчий сусід для кожного синтетичного запису
        from sklearn.neighbors import NearestNeighbors

        nn = NearestNeighbors(n_neighbors=1)
        nn.fit(real_data)

        distances, _ = nn.kneighbors(synthetic_data)
        min_distances = distances.flatten()

        # Порогове значення для "занадто схожих"
        threshold = np.percentile(min_distances, 5)
        potentially_copied = (min_distances < threshold).sum()

        privacy_metrics = {
            'mean_nearest_distance': min_distances.mean(),
            'min_nearest_distance': min_distances.min(),
            'potentially_memorized_ratio': potentially_copied / len(synthetic_data),
            'privacy_safe': potentially_copied / len(synthetic_data) < 0.01
        }

        self.metrics['privacy'] = privacy_metrics
        return privacy_metrics

    def assess_diversity(
        self,
        synthetic_data: np.ndarray
    ) -> dict:
        """Оцінка різноманітності згенерованих даних."""

        # Покриття простору через кластеризацію
        from sklearn.cluster import KMeans

        n_clusters = min(50, len(synthetic_data) // 10)
        kmeans = KMeans(n_clusters=n_clusters)
        clusters = kmeans.fit_predict(synthetic_data)

        cluster_sizes = np.bincount(clusters)

        diversity_metrics = {
            'num_clusters_used': (cluster_sizes > 0).sum(),
            'cluster_entropy': stats.entropy(cluster_sizes / cluster_sizes.sum()),
            'mode_collapse_score': 1 - (cluster_sizes.max() / cluster_sizes.sum()),
            'coverage': (cluster_sizes > 0).sum() / n_clusters
        }

        self.metrics['diversity'] = diversity_metrics
        return diversity_metrics

Практичні Use Cases

1. Autonomous Vehicles (Tesla, Waymo)

• Мільярди synthetic miles щороку
• Simulation-first approach для edge cases
• Zero real accidents для тренування accident avoidance

2. Healthcare & Pharma

• Rare disease training data generation
• Drug discovery compound datasets
• Privacy-compliant data sharing між інституціями

3. Financial Services

• Fraud detection з balanced datasets
• Stress testing під регуляторні вимоги
• Synthetic portfolios для risk modeling

4. Robotics & Manufacturing

• Sim-to-real transfer learning
• Domain randomization для robust perception
• Digital twins для predictive maintenance

Ідеї для наукових досліджень

Для бакалаврської роботи:

• Порівняння GAN vs CTGAN vs VAE для табличних даних
• Synthetic data augmentation для image classification
• Вплив обсягу синтетичних даних на якість моделі

Для магістерської дисертації:

• Domain-specific synthetic generation (medical imaging, satellite)
• Privacy-utility trade-offs у синтетичних даних
• Методи зменшення sim-to-real gap

Для PhD досліджень:

• Theoretical bounds on synthetic data utility
• Foundation models для універсальної генерації даних
• Causally-aware synthetic data generation

Висновок

Синтетичні дані — це не просто спосіб "надути" dataset. Це фундаментальна технологія, яка дозволяє тренувати моделі для сценаріїв, які неможливо зібрати в реальності: рідкісних захворювань, небезпечних дорожніх ситуацій, фінансових шахрайств.

Real data collection: $1-100 per sample.

Synthetic data generation: $0.001 per sample.

Але справа не тільки в економіці. Деякі дані неможливо зібрати (rare events), деякі заборонено ділитись (privacy), деякі небезпечно збирати (edge cases). Synthetic data не замінює реальні — але дозволяє створювати системи, які без неї були б просто неможливі.

Якщо ви плануєте дослідження в галузі генеративних моделей, приватності даних чи симуляційного навчання — фахівці SKP-Degree готові допомогти з формулюванням задачі, реалізацією pipeline та оформленням роботи. Звертайтесь на skp-degree.com.ua або пишіть у Telegram: @kursovi_diplomy — від ідеї до захищеної роботи.

Ключові слова: синтетичні дані, генеративні моделі, GAN, CTGAN, diffusion models, диференціальна приватність, симуляція, CARLA, аугментація даних, privacy-preserving ML, autonomous driving, medical imaging, synthetic data generation, дипломна робота, магістерська, AI-дослідження.