Stable Diffusion Project Implementation
What Is Stable Diffusion?
Stable Diffusion Project Implementation
What Is Stable Diffusion?
Stable Diffusion is a text-to-image generative AI model that creates high-quality images from natural-language prompts. It belongs to the Latent Diffusion Model (LDM) family and works by iteratively denoising random noise into a coherent image guided by text embeddings.
Key Characteristics
- Text-to-image generation — Creates images from descriptions
- Image-to-image transformation — Modifies existing images
- Inpainting / Outpainting — Edits or extends images
- Open & extensible — Open-weight, highly customizable
- Efficient architecture — Operates in compressed latent space
- Runs anywhere — Local or cloud deployment
Core principle:
The model learns to reverse a noise-adding process, gradually removing noise until an image matching the prompt emerges.
High-Level Architecture Overview
Text Prompt
↓
[ Text Encoder (CLIP) ]
↓
Text Embeddings ─────────────┐
↓
Random Noise → [ U-Net Denoiser ] → Latent Representation
↓
[ VAE Decoder ]
↓
Final ImageStable Diffusion consists of three neural networks working together.
Text Encoder (CLIP)
Purpose
Converts text prompts into numerical embeddings that guide image generation.
Architecture
- CLIP text encoder
- Transformer-based (similar to BERT)
- Tokenized input (max 77 tokens)
Process Example
“A cat on a skateboard”
→ Tokenization
→ CLIP Transformer
→ 77 × 768 embeddings
→ Conditioning vector
Key Features
- Semantic understanding (meaning, not keywords)
- Cross-attention conditioning
- Negative prompts (remove unwanted concepts)
2. U-Net (Denoising Model)
Purpose
Core engine that iteratively removes noise from the latent image.
Input
- Noisy latent: 64 × 64 × 4
- Text embeddings
- Timestep embedding
Architecture
- Modified U-Net
- ResNet blocks
- Self-attention + cross-attention
- Encoder–decoder with skip connections
Encoder (Downsampling)
- 64×64×320
- 32×32×640
- 16×16×1280
- 8×8×1280
Bottleneck
- ResNet + attention layers
Decoder (Upsampling)
- Uses skip connections
- Gradually reconstructs details
- Outputs predicted noise (64×64×4)
Key Components
- ResNet blocks — Stable training
- Self-attention — Global spatial awareness
- Cross-attention — Injects text conditioning
- Skip connections — Preserve fine details
- Timestep embeddings — Track denoising step
3. Variational Autoencoder (VAE)
Purpose
Compresses and reconstructs images efficiently.
Encoder
512×512×3 image
→ Convolutions
→ 64×64×4 latent
Decoder
64×64×4 latent
→ Transposed convolutions
→ 512×512×3 image
Why VAE Matters
- 64× faster than pixel-space diffusion
- Preserves semantic structure
- High-quality reconstruction
- Pre-trained separately
Training Phase (How the Model Learns)
- Take a real image
- Encode to latent space
- Add noise at random timestep (t = 1–1000)
- U-Net predicts added noise
- Compute loss
- Backpropagate
- Condition on text embeddings
Loss Function
L = || ε − εθ(zt, t, c) ||²
Where:
- ε = true noise
- εθ = predicted noise
- zt = noisy latent
- c = text embedding
Inference Phase (Image Generation)
Step 1: Text Encoding
Prompt → CLIP → text embeddings
Step 2: Noise Initialization
Random Gaussian noise (seeded)
Step 3: Iterative Denoising
- 20–50 steps typical
- U-Net predicts noise
- Scheduler controls update
Step 4: Decode
Latent → VAE decoder → image
Step 5: Post-processing
- Denormalization
- Clipping
- Format conversion
Key Architectural Innovations
1. Latent Diffusion
- 48× dimensionality reduction
- Faster training & inference
- Maintains quality
2. Cross-Attention Conditioning
- Image features attend to text embeddings
- Enables fine-grained control
3. Classifier-Free Guidance (CFG)
Controls prompt adherence.
| CFG Scale | Effect |
| --------- | -------------------------------- |
| 1 | Creative, weak prompt |
| 7–9 | Balanced (default) |
| 15+ | Strong adherence, risk artifacts |4. Sampling Schedulers
- DDPM — Slow, high quality
- DDIM — Fast, deterministic
- DPM-Solver++ — Very fast
- Euler / Euler-A — Balanced
Model Variants
Stable Diffusion 1.5
- ~1B parameters
- 512×512 resolution
Stable Diffusion 2.1
- OpenCLIP text encoder
- 768×768 resolution
SDXL
- 2.6B+ parameters
- Dual text encoders
- Native 1024×1024
- Optional refiner model
Advanced Capabilities
ControlNet
Adds spatial control:
- Pose
- Edges
- Depth
- Layout
LoRA (Low-Rank Adaptation)
- Lightweight fine-tuning
- 2–200 MB
- Stackable styles
DreamBooth
- Learns new concepts
- Fine-tunes full U-Net
- Uses few example images
Why Stable Diffusion Became Popular
- Open-weight & customizable
- Runs on consumer GPUs
- No per-image API cost
- Strong community ecosystem
- High creative control
- Reproducible outputs
Use Cases
Creative
- Concept art
- Illustration
- Game assets
E-Commerce
- Product mockups
- Virtual try-on
- Catalog generation
AI & Research
- Synthetic data
- Augmentation
- Bias testing
Healthcare (Non-Diagnostic)
- Medical illustrations
- Education
Architecture & Design
- Interior visualization
- Mood boards
Marketing
- Campaign creatives
- A/B testing
Education
- Visual learning aids
Strengths vs Limitations
Strengths
- Open & extensible
- Low cost
- Offline capable
- Scalable
- Reproducible
Limitations
- Prompt sensitivity
- Imperfect anatomy
- Dataset bias
- Ethical concerns
- GPU memory needs
Where Stable Diffusion Fits in the GenAI Stack
Role: Visual generation layer
Complements
- LLMs (prompt generation)
- RAG systems (context grounding)
- Agentic workflows
Stable Diffusion is an open-source latent diffusion model that generates and edits images by iteratively denoising compressed latent representations under semantic guidance from CLIP embeddings using a U-Net with cross-attention.
Why Diffusion Models for Market Surveillance?
Traditional surveillance limits
- Rule-based alerts → brittle, high false positives
- Supervised ML → label scarcity (true manipulation is rare)
- Autoencoders → weak at modeling complex joint distributions
Why diffusion models fit
- Learn full data distribution of normal market behavior
- Excellent at rare-event detection
- Can generate counterfactual market paths
- Naturally suited for anomaly scoring
“Market manipulation is a rare, evolving, multivariate time-series problem. Diffusion models are best suited because they learn the full distribution of normal market behavior rather than memorizing past abuse patterns.”
The Ways to Solve this Use Case Of Market Surveillance
| Rank | Model | Why |
| ---- | -------------------- | ---------------------------- |
| 🥇 | **Diffusion Models** | Learn normal market dynamics |
| 🥈 | Transformers | Long-range sequence modeling |
| 🥉 | LSTM Autoencoders | Simple anomaly detection |
| ⚙️ | Graph Models | Collusion & networks |
| ❌ | Rules alone | Legacy only |Project Implementation with Stable Diffusion for Market surveillance
Diffusion Model
"""
Stable Diffusion Model for Market Surveillance
Implements forward and reverse diffusion processes for anomaly detection
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from typing import Tuple, Optional
from config import Config
class SinusoidalPositionEmbedding(nn.Module):
"""Sinusoidal position embedding for timestep encoding"""
def __init__(self, dim: int):
super().__init__()
self.dim = dim
def forward(self, timesteps: torch.Tensor) -> torch.Tensor:
"""
Args:
timesteps: (batch_size,) tensor of timesteps
Returns:
embeddings: (batch_size, dim) tensor
"""
device = timesteps.device
half_dim = self.dim // 2
embeddings = np.log(10000) / (half_dim - 1)
embeddings = torch.exp(torch.arange(half_dim, device=device) * -embeddings)
embeddings = timesteps[:, None] * embeddings[None, :]
embeddings = torch.cat([torch.sin(embeddings), torch.cos(embeddings)], dim=-1)
return embeddings
class ResidualBlock(nn.Module):
"""Residual block with time embedding"""
def __init__(self, in_channels: int, out_channels: int, time_dim: int, dropout: float = 0.1):
super().__init__()
self.conv1 = nn.Conv1d(in_channels, out_channels, kernel_size=3, padding=1)
self.conv2 = nn.Conv1d(out_channels, out_channels, kernel_size=3, padding=1)
self.time_mlp = nn.Linear(time_dim, out_channels)
self.norm1 = nn.GroupNorm(8, out_channels)
self.norm2 = nn.GroupNorm(8, out_channels)
self.dropout = nn.Dropout(dropout)
# Residual connection
if in_channels != out_channels:
self.residual_conv = nn.Conv1d(in_channels, out_channels, kernel_size=1)
else:
self.residual_conv = nn.Identity()
def forward(self, x: torch.Tensor, t: torch.Tensor) -> torch.Tensor:
"""
Args:
x: (batch, channels, length)
t: (batch, time_dim)
"""
residual = self.residual_conv(x)
# First conv
h = self.conv1(x)
h = self.norm1(h)
# Add time embedding
time_emb = self.time_mlp(t)
h = h + time_emb[:, :, None]
h = F.silu(h)
# Second conv
h = self.dropout(h)
h = self.conv2(h)
h = self.norm2(h)
h = F.silu(h)
return h + residual
class UNet1D(nn.Module):
"""1D U-Net for denoising time series data"""
def __init__(
self,
in_channels: int,
hidden_dim: int = 128,
num_layers: int = 4,
dropout: float = 0.1
):
super().__init__()
# Time embedding
time_dim = hidden_dim * 4
self.time_mlp = nn.Sequential(
SinusoidalPositionEmbedding(hidden_dim),
nn.Linear(hidden_dim, time_dim),
nn.SiLU(),
nn.Linear(time_dim, time_dim),
)
# Encoder (downsampling)
self.encoder_blocks = nn.ModuleList()
self.downsample_blocks = nn.ModuleList()
channels = [in_channels] + [hidden_dim * (2 ** i) for i in range(num_layers)]
for i in range(num_layers):
self.encoder_blocks.append(
ResidualBlock(channels[i], channels[i + 1], time_dim, dropout)
)
if i < num_layers - 1:
self.downsample_blocks.append(
nn.Conv1d(channels[i + 1], channels[i + 1], kernel_size=4, stride=2, padding=1)
)
# Bottleneck
bottleneck_dim = channels[-1]
self.bottleneck = ResidualBlock(bottleneck_dim, bottleneck_dim, time_dim, dropout)
# Decoder (upsampling)
self.decoder_blocks = nn.ModuleList()
self.upsample_blocks = nn.ModuleList()
for i in range(num_layers - 1, 0, -1):
self.upsample_blocks.append(
nn.ConvTranspose1d(channels[i + 1], channels[i], kernel_size=4, stride=2, padding=1)
)
self.decoder_blocks.append(
ResidualBlock(channels[i] * 2, channels[i], time_dim, dropout) # *2 for skip connection
)
# Output
self.output_conv = nn.Conv1d(channels[1], in_channels, kernel_size=1)
def forward(self, x: torch.Tensor, t: torch.Tensor) -> torch.Tensor:
"""
Args:
x: (batch, channels, length)
t: (batch,) timesteps
Returns:
noise prediction: (batch, channels, length)
"""
# Time embedding
t_emb = self.time_mlp(t)
# Encoder
skip_connections = []
h = x
for i, encoder in enumerate(self.encoder_blocks):
h = encoder(h, t_emb)
skip_connections.append(h)
if i < len(self.downsample_blocks):
h = self.downsample_blocks[i](h)
# Bottleneck
h = self.bottleneck(h, t_emb)
# Decoder
for i, (upsample, decoder) in enumerate(zip(self.upsample_blocks, self.decoder_blocks)):
h = upsample(h)
# Concatenate skip connection
skip = skip_connections[-(i + 2)]
# Handle size mismatch
if h.shape[-1] != skip.shape[-1]:
h = F.interpolate(h, size=skip.shape[-1], mode='linear', align_corners=False)
h = torch.cat([h, skip], dim=1)
h = decoder(h, t_emb)
# Output
return self.output_conv(h)
class DiffusionModel(nn.Module):
"""Stable Diffusion Model for time series anomaly detection"""
def __init__(self, config: dict = None):
super().__init__()
if config is None:
config = Config.MODEL_CONFIG
self.timesteps = config['timesteps']
self.sequence_length = config['sequence_length']
# Noise schedule (linear)
self.beta = torch.linspace(
config['beta_start'],
config['beta_end'],
self.timesteps
)
self.alpha = 1.0 - self.beta
self.alpha_bar = torch.cumprod(self.alpha, dim=0)
# Determine input channels (will be set during first forward pass)
self.in_channels = None
self.model = None
self.config = config
def _init_model(self, in_channels: int):
"""Initialize U-Net model"""
self.in_channels = in_channels
self.model = UNet1D(
in_channels=in_channels,
hidden_dim=self.config['hidden_dim'],
num_layers=self.config['num_layers'],
dropout=self.config['dropout']
)
def forward_diffusion(
self,
x0: torch.Tensor,
t: torch.Tensor,
noise: Optional[torch.Tensor] = None
) -> Tuple[torch.Tensor, torch.Tensor]:
"""
Forward diffusion: add noise to data
Args:
x0: (batch, channels, length) original data
t: (batch,) timesteps
noise: optional pre-generated noise
Returns:
xt: noisy data at timestep t
noise: the noise that was added
"""
if noise is None:
noise = torch.randn_like(x0)
# Get alpha_bar for timesteps
alpha_bar_t = self.alpha_bar[t].view(-1, 1, 1).to(x0.device)
# q(x_t | x_0) = sqrt(alpha_bar_t) * x_0 + sqrt(1 - alpha_bar_t) * noise
xt = torch.sqrt(alpha_bar_t) * x0 + torch.sqrt(1 - alpha_bar_t) * noise
return xt, noise
def forward(self, x: torch.Tensor, t: torch.Tensor) -> torch.Tensor:
"""
Predict noise in noisy data
Args:
x: (batch, channels, length) noisy data
t: (batch,) timesteps
Returns:
predicted noise
"""
if self.model is None:
self._init_model(x.shape[1])
return self.model(x, t)
@torch.no_grad()
def reverse_diffusion(
self,
xt: torch.Tensor,
t: int,
return_all: bool = False
) -> torch.Tensor:
"""
Reverse diffusion: denoise data step by step
Args:
xt: (batch, channels, length) noisy data
t: starting timestep
return_all: if True, return all intermediate steps
Returns:
x0: denoised data
"""
device = xt.device
self.alpha_bar = self.alpha_bar.to(device)
self.alpha = self.alpha.to(device)
self.beta = self.beta.to(device)
trajectory = [xt] if return_all else None
for i in reversed(range(t)):
# Predict noise
t_tensor = torch.full((xt.shape[0],), i, device=device, dtype=torch.long)
predicted_noise = self(xt, t_tensor)
# Compute x_{t-1}
alpha_t = self.alpha[i]
alpha_bar_t = self.alpha_bar[i]
beta_t = self.beta[i]
# Mean of p(x_{t-1} | x_t)
mean = (1 / torch.sqrt(alpha_t)) * (
xt - (beta_t / torch.sqrt(1 - alpha_bar_t)) * predicted_noise
)
if i > 0:
# Add noise (except for last step)
noise = torch.randn_like(xt)
sigma = torch.sqrt(beta_t)
xt = mean + sigma * noise
else:
xt = mean
if return_all:
trajectory.append(xt)
return trajectory if return_all else xt
@torch.no_grad()
def compute_reconstruction_error(
self,
x0: torch.Tensor,
num_steps: Optional[int] = None
) -> torch.Tensor:
"""
Compute reconstruction error for anomaly detection
Args:
x0: (batch, channels, length) original data
num_steps: number of diffusion steps (default: self.timesteps)
Returns:
reconstruction error per sample
"""
if num_steps is None:
num_steps = self.timesteps
device = x0.device
batch_size = x0.shape[0]
# Forward diffusion to max timestep
t = torch.full((batch_size,), num_steps - 1, device=device, dtype=torch.long)
xt, _ = self.forward_diffusion(x0, t)
# Reverse diffusion
x0_reconstructed = self.reverse_diffusion(xt, num_steps)
# Compute MSE per sample
error = F.mse_loss(x0_reconstructed, x0, reduction='none')
error = error.mean(dim=[1, 2]) # Average over channels and length
return error
def train_diffusion_model(
model: DiffusionModel,
train_data: torch.Tensor,
config: dict = None
) -> list:
"""
Train the diffusion model
Args:
model: DiffusionModel instance
train_data: (n_samples, channels, length) training data
config: training configuration
Returns:
loss history
"""
if config is None:
config = Config.TRAINING_CONFIG
device = config['device']
model = model.to(device)
train_data = train_data.to(device)
# Initialize model with dummy forward pass
if model.model is None:
dummy_batch = train_data[:1]
dummy_t = torch.zeros(1, dtype=torch.long, device=device)
_ = model(dummy_batch, dummy_t)
optimizer = torch.optim.AdamW(
model.parameters(),
lr=config['learning_rate'],
weight_decay=config['weight_decay']
)
batch_size = min(config['batch_size'], train_data.shape[0])
n_samples = train_data.shape[0]
n_batches = max(1, n_samples // batch_size) # At least 1 batch
loss_history = []
print(f"Training on {device}")
print(f"Samples: {n_samples}, Batches: {n_batches}, Epochs: {config['epochs']}")
for epoch in range(config['epochs']):
model.train()
epoch_loss = 0.0
# Shuffle data
perm = torch.randperm(n_samples)
train_data = train_data[perm]
for batch_idx in range(n_batches):
# Get batch
start_idx = batch_idx * batch_size
end_idx = start_idx + batch_size
x0 = train_data[start_idx:end_idx]
# Sample random timesteps
t = torch.randint(0, model.timesteps, (batch_size,), device=device)
# Forward diffusion
xt, noise = model.forward_diffusion(x0, t)
# Predict noise
predicted_noise = model(xt, t)
# Compute loss
loss = F.mse_loss(predicted_noise, noise)
# Backward pass
optimizer.zero_grad()
loss.backward()
# Gradient clipping
torch.nn.utils.clip_grad_norm_(model.parameters(), config['gradient_clip'])
optimizer.step()
epoch_loss += loss.item()
avg_loss = epoch_loss / n_batches
loss_history.append(avg_loss)
if (epoch + 1) % 10 == 0:
print(f"Epoch {epoch + 1}/{config['epochs']}, Loss: {avg_loss:.6f}")
return loss_history
if __name__ == "__main__":
# Test diffusion model
print("Testing Diffusion Model...")
# Create dummy data
batch_size = 8
channels = 10
length = 100
x = torch.randn(batch_size, channels, length)
t = torch.randint(0, 1000, (batch_size,))
# Create model
model = DiffusionModel()
# Forward pass
noise_pred = model(x, t)
print(f"Input shape: {x.shape}")
print(f"Output shape: {noise_pred.shape}")
# Test reconstruction
error = model.compute_reconstruction_error(x[:2], num_steps=100)
print(f"Reconstruction error shape: {error.shape}")
print(f"Reconstruction errors: {error}")Surveillance System
Market Anomalies to Detect
- Pump & Dump: Artificial price inflation followed by sell-off
- Spoofing: Large fake orders to manipulate prices
- Wash Trading: Self-trading to create false volume
- Insider Trading Patterns: Unusual activity before news events
- Volume Anomalies: Unexpected trading volume spikes
Flow Diagram
Raw Time-Series Data
↓
Sliding Window Creation
↓
Diffusion Model (Reconstruction)
↓
Error Computation
↓
Threshold Comparison
↓
Anomaly Scores
↓
Binary Detection
↓
Alert Generation (severity)
↓
Metrics + Report"""
Market Surveillance System using Stable Diffusion
Detects anomalies in market data using trained diffusion model
"""
import torch
import numpy as np
from typing import Dict, List, Tuple
from diffusion_model import DiffusionModel
from config import Config
class MarketSurveillance:
"""Market surveillance system for anomaly detection"""
def __init__(self, model: DiffusionModel, config: dict = None):
"""
Initialize surveillance system
Args:
model: Trained diffusion model
config: Surveillance configuration
"""
self.model = model
self.config = config or Config.SURVEILLANCE_CONFIG
self.device = Config.get_device()
self.model.to(self.device)
self.model.eval()
# Detection parameters
self.threshold = None
self.window_size = self.config['window_size']
self.stride = self.config['stride']
def set_threshold(self, normal_data: torch.Tensor, percentile: float = None):
"""
Set anomaly detection threshold based on normal data
Args:
normal_data: (n_samples, channels, length) normal data
percentile: Percentile for threshold (default from config)
"""
if percentile is None:
percentile = self.config['anomaly_threshold']
print(f"Computing threshold on {normal_data.shape[0]} normal samples...")
# Compute reconstruction errors on normal data
errors = []
batch_size = 32
with torch.no_grad():
for i in range(0, normal_data.shape[0], batch_size):
batch = normal_data[i:i + batch_size].to(self.device)
error = self.model.compute_reconstruction_error(batch)
errors.append(error.cpu())
errors = torch.cat(errors)
self.threshold = torch.quantile(errors, percentile)
print(f"Threshold set to {self.threshold:.6f} ({percentile*100}th percentile)")
print(f"Error range: [{errors.min():.6f}, {errors.max():.6f}]")
return self.threshold
def detect_anomalies(
self,
data: torch.Tensor,
return_scores: bool = True
) -> Dict:
"""
Detect anomalies in data using sliding window
Args:
data: (n_timesteps, n_features) time series data
return_scores: If True, return anomaly scores
Returns:
Dictionary with detection results
"""
n_timesteps = data.shape[0]
n_features = data.shape[1]
# Prepare sliding windows
windows = []
window_indices = []
for i in range(0, n_timesteps - self.window_size + 1, self.stride):
window = data[i:i + self.window_size]
windows.append(window)
window_indices.append((i, i + self.window_size))
windows = torch.stack(windows) # (n_windows, window_size, n_features)
windows = windows.transpose(1, 2) # (n_windows, n_features, window_size)
print(f"Analyzing {len(windows)} windows...")
# Compute reconstruction errors
errors = []
batch_size = 32
with torch.no_grad():
for i in range(0, len(windows), batch_size):
batch = windows[i:i + batch_size].to(self.device)
error = self.model.compute_reconstruction_error(batch)
errors.append(error.cpu())
errors = torch.cat(errors).numpy()
# Map window errors to timesteps
anomaly_scores = np.zeros(n_timesteps)
anomaly_counts = np.zeros(n_timesteps)
for idx, (start, end) in enumerate(window_indices):
anomaly_scores[start:end] += errors[idx]
anomaly_counts[start:end] += 1
# Average scores
anomaly_scores = anomaly_scores / np.maximum(anomaly_counts, 1)
# Detect anomalies
if self.threshold is not None:
anomalies = anomaly_scores > self.threshold.item()
else:
# Use default percentile
threshold = np.percentile(anomaly_scores,
self.config['anomaly_threshold'] * 100)
anomalies = anomaly_scores > threshold
# Generate alerts
alerts = self._generate_alerts(anomaly_scores, anomalies)
results = {
'anomaly_scores': anomaly_scores,
'anomalies': anomalies,
'alerts': alerts,
'threshold': self.threshold.item() if self.threshold is not None else threshold,
'n_anomalies': anomalies.sum(),
'anomaly_ratio': anomalies.sum() / len(anomalies),
}
print(f"Detected {results['n_anomalies']} anomalous timesteps "
f"({results['anomaly_ratio']*100:.1f}%)")
print(f"Generated {len(alerts)} alerts")
return results
def _generate_alerts(
self,
scores: np.ndarray,
anomalies: np.ndarray
) -> List[Dict]:
"""
Generate alerts from anomaly detections
Args:
scores: Anomaly scores
anomalies: Binary anomaly flags
Returns:
List of alert dictionaries
"""
alerts = []
alert_levels = self.config['alert_levels']
# Find contiguous anomalous regions
in_anomaly = False
start_idx = 0
for i in range(len(anomalies)):
if anomalies[i] and not in_anomaly:
# Start of anomaly
in_anomaly = True
start_idx = i
elif not anomalies[i] and in_anomaly:
# End of anomaly
in_anomaly = False
# Create alert
region_scores = scores[start_idx:i]
max_score = region_scores.max()
avg_score = region_scores.mean()
# Determine severity
severity = 'low'
for level, threshold in sorted(alert_levels.items(),
key=lambda x: x[1], reverse=True):
if max_score > threshold:
severity = level
break
alerts.append({
'start': start_idx,
'end': i,
'duration': i - start_idx,
'max_score': float(max_score),
'avg_score': float(avg_score),
'severity': severity,
})
# Handle case where anomaly extends to end
if in_anomaly:
region_scores = scores[start_idx:]
max_score = region_scores.max()
avg_score = region_scores.mean()
severity = 'low'
for level, threshold in sorted(alert_levels.items(),
key=lambda x: x[1], reverse=True):
if max_score > threshold:
severity = level
break
alerts.append({
'start': start_idx,
'end': len(anomalies),
'duration': len(anomalies) - start_idx,
'max_score': float(max_score),
'avg_score': float(avg_score),
'severity': severity,
})
return alerts
def analyze_batch(
self,
data_dict: Dict,
ground_truth_labels: np.ndarray = None
) -> Dict:
"""
Analyze batch of data and compute metrics
Args:
data_dict: Dictionary with 'features' key
ground_truth_labels: Optional ground truth labels
Returns:
Analysis results
"""
features = torch.FloatTensor(data_dict['features'])
# Detect anomalies
results = self.detect_anomalies(features)
# Compute metrics if ground truth available
if ground_truth_labels is not None:
metrics = self._compute_metrics(
results['anomalies'],
ground_truth_labels
)
results['metrics'] = metrics
return results
def _compute_metrics(
self,
predictions: np.ndarray,
ground_truth: np.ndarray
) -> Dict:
"""
Compute detection metrics
Args:
predictions: Binary predictions
ground_truth: Binary ground truth
Returns:
Dictionary of metrics
"""
# Convert to binary
predictions = predictions.astype(bool)
ground_truth = ground_truth.astype(bool)
# Compute confusion matrix
tp = np.sum(predictions & ground_truth)
fp = np.sum(predictions & ~ground_truth)
tn = np.sum(~predictions & ~ground_truth)
fn = np.sum(~predictions & ground_truth)
# Compute metrics
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
accuracy = (tp + tn) / len(predictions)
metrics = {
'true_positives': int(tp),
'false_positives': int(fp),
'true_negatives': int(tn),
'false_negatives': int(fn),
'precision': float(precision),
'recall': float(recall),
'f1_score': float(f1),
'accuracy': float(accuracy),
}
return metrics
def print_report(self, results: Dict):
"""Print surveillance report"""
print("\n" + "=" * 60)
print("MARKET SURVEILLANCE REPORT")
print("=" * 60)
print(f"\nAnomaly Detection:")
print(f" Threshold: {results['threshold']:.6f}")
print(f" Anomalies Detected: {results['n_anomalies']}")
print(f" Anomaly Ratio: {results['anomaly_ratio']*100:.2f}%")
print(f"\nAlerts Generated: {len(results['alerts'])}")
# Group by severity
severity_counts = {}
for alert in results['alerts']:
severity = alert['severity']
severity_counts[severity] = severity_counts.get(severity, 0) + 1
for severity in ['high', 'medium', 'low']:
count = severity_counts.get(severity, 0)
print(f" {severity.upper()}: {count}")
# Show top alerts
if results['alerts']:
print(f"\nTop 5 Alerts:")
sorted_alerts = sorted(results['alerts'],
key=lambda x: x['max_score'],
reverse=True)
for i, alert in enumerate(sorted_alerts[:5], 1):
print(f" {i}. Timesteps {alert['start']}-{alert['end']} "
f"(duration: {alert['duration']})")
print(f" Severity: {alert['severity'].upper()}, "
f"Max Score: {alert['max_score']:.4f}")
# Print metrics if available
if 'metrics' in results:
metrics = results['metrics']
print(f"\nPerformance Metrics:")
print(f" Precision: {metrics['precision']:.4f}")
print(f" Recall: {metrics['recall']:.4f}")
print(f" F1 Score: {metrics['f1_score']:.4f}")
print(f" Accuracy: {metrics['accuracy']:.4f}")
print(f"\nConfusion Matrix:")
print(f" TP: {metrics['true_positives']}, FP: {metrics['false_positives']}")
print(f" FN: {metrics['false_negatives']}, TN: {metrics['true_negatives']}")
print("=" * 60)
if __name__ == "__main__":
print("Testing Market Surveillance System...")
# Create dummy model and data
from diffusion_model import DiffusionModel
model = DiffusionModel()
surveillance = MarketSurveillance(model)
# Generate dummy data
normal_data = torch.randn(100, 10, 100)
test_data = torch.randn(500, 10)
# Set threshold
surveillance.set_threshold(normal_data, percentile=0.9)
# Detect anomalies
results = surveillance.detect_anomalies(test_data)
surveillance.print_report(results)This system uses a diffusion model trained on normal time-series data to compute reconstruction errors. During inference, data is processed using sliding windows, and anomaly scores are derived from reconstruction error. A threshold learned from normal data is used to classify anomalies. Contiguous anomalous regions are grouped into alerts with severity levels, and evaluation metrics are computed if ground truth is available.”
Data Generator
For Testing/Development: YES — Use synthetic data
- For Production: NO — Use real Kaggle/Yahoo Finance data only
- Best Practice: Use BOTH for comprehensive testing
""
Kaggle Dataset Integration for Market Surveillance
Load and preprocess real financial data from Kaggle
"""
import pandas as pd
import numpy as np
import os
from pathlib import Path
from typing import Dict, Tuple
import yfinance as yf
from datetime import datetime, timedelta
class KaggleDataLoader:
"""Load and preprocess financial data from various sources"""
def __init__(self):
self.data_dir = Path("data/kaggle")
self.data_dir.mkdir(parents=True, exist_ok=True)
def download_stock_data(
self,
tickers: list = ['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'TSLA'],
period: str = '2y',
interval: str = '1d'
) -> pd.DataFrame:
"""
Download stock data using yfinance (alternative to Kaggle)
Args:
tickers: List of stock symbols
period: Data period ('1y', '2y', '5y', etc.)
interval: Data interval ('1d', '1h', etc.)
Returns:
DataFrame with stock data
"""
print(f"Downloading data for {len(tickers)} stocks...")
data_frames = []
for ticker in tickers:
try:
print(f" Fetching {ticker}...")
stock = yf.Ticker(ticker)
df = stock.history(period=period, interval=interval)
if not df.empty:
df['Ticker'] = ticker
data_frames.append(df)
except Exception as e:
print(f" Error fetching {ticker}: {e}")
if data_frames:
combined_df = pd.concat(data_frames)
# Save to file
save_path = self.data_dir / f"stock_data_{datetime.now().strftime('%Y%m%d')}.csv"
combined_df.to_csv(save_path)
print(f"\nData saved to {save_path}")
return combined_df
else:
raise ValueError("No data downloaded")
def load_csv_data(self, filepath: str) -> pd.DataFrame:
"""Load data from CSV file"""
df = pd.DataFrame(filepath)
print(f"Loaded {len(df)} rows from {filepath}")
return df
def preprocess_for_surveillance(
self,
df: pd.DataFrame,
price_col: str = 'Close',
volume_col: str = 'Volume',
ticker_col: str = 'Ticker'
) -> Dict:
"""
Preprocess financial data for surveillance system
Args:
df: Raw financial data
price_col: Column name for prices
volume_col: Column name for volumes
ticker_col: Column name for ticker symbols
Returns:
Dictionary compatible with surveillance system
"""
print("Preprocessing data for surveillance system...")
# Get unique tickers
if ticker_col in df.columns:
tickers = df[ticker_col].unique()
n_assets = len(tickers)
# Pivot data to get prices and volumes per asset
prices_list = []
volumes_list = []
for ticker in tickers:
ticker_data = df[df[ticker_col] == ticker].sort_index()
prices_list.append(ticker_data[price_col].values)
volumes_list.append(ticker_data[volume_col].values)
# Find minimum length
min_length = min(len(p) for p in prices_list)
# Truncate to same length
prices = np.array([p[:min_length] for p in prices_list]).T
volumes = np.array([v[:min_length] for v in volumes_list]).T
else:
# Single asset
prices = df[price_col].values.reshape(-1, 1)
volumes = df[volume_col].values.reshape(-1, 1)
n_assets = 1
n_timesteps = prices.shape[0]
# Compute returns
price_returns = np.diff(np.log(prices + 1e-8), axis=0)
price_returns = np.vstack([np.zeros((1, n_assets)), price_returns])
# Normalize volumes
volume_normalized = (volumes - volumes.mean(axis=0)) / (volumes.std(axis=0) + 1e-8)
# Combine features
features = np.concatenate([price_returns, volume_normalized], axis=1)
# Create dataset dictionary
dataset = {
'prices': prices,
'volumes': volumes,
'features': features,
'labels': np.zeros(n_timesteps), # No ground truth labels for real data
'n_assets': n_assets,
'n_timesteps': n_timesteps,
'tickers': tickers.tolist() if ticker_col in df.columns else ['Asset_1'],
}
print(f"Preprocessed: {n_timesteps} timesteps, {n_assets} assets")
return dataset
def inject_synthetic_anomalies(
self,
dataset: Dict,
anomaly_ratio: float = 0.1
) -> Dict:
"""
Inject synthetic anomalies into real data for testing
Args:
dataset: Preprocessed dataset
anomaly_ratio: Fraction of data to make anomalous
Returns:
Dataset with injected anomalies and labels
"""
from data_generator import MarketDataGenerator
print(f"Injecting synthetic anomalies ({anomaly_ratio*100:.0f}%)...")
prices = dataset['prices'].copy()
volumes = dataset['volumes'].copy()
n_timesteps = dataset['n_timesteps']
n_assets = dataset['n_assets']
# Create generator for anomaly injection
generator = MarketDataGenerator()
generator.n_assets = n_assets
generator.n_timesteps = n_timesteps
# Inject anomalies
prices, volumes, labels = generator.inject_anomalies(
prices, volumes, anomaly_ratio=anomaly_ratio
)
# Update features
price_returns = np.diff(np.log(prices + 1e-8), axis=0)
price_returns = np.vstack([np.zeros((1, n_assets)), price_returns])
volume_normalized = (volumes - volumes.mean(axis=0)) / (volumes.std(axis=0) + 1e-8)
features = np.concatenate([price_returns, volume_normalized], axis=1)
dataset['prices'] = prices
dataset['volumes'] = volumes
dataset['features'] = features
dataset['labels'] = labels
dataset['anomaly_details'] = generator.anomaly_details
print(f"Injected {labels.sum()} anomalous timesteps")
return dataset
def create_train_test_split(
self,
dataset: Dict,
train_ratio: float = 0.7
) -> Tuple[Dict, Dict]:
"""
Split dataset into training and testing sets
Args:
dataset: Full dataset
train_ratio: Fraction for training
Returns:
(train_dataset, test_dataset)
"""
n_timesteps = dataset['n_timesteps']
split_idx = int(n_timesteps * train_ratio)
train_dataset = {
'prices': dataset['prices'][:split_idx],
'volumes': dataset['volumes'][:split_idx],
'features': dataset['features'][:split_idx],
'labels': np.zeros(split_idx), # Training on normal data only
'n_assets': dataset['n_assets'],
'n_timesteps': split_idx,
}
test_dataset = {
'prices': dataset['prices'][split_idx:],
'volumes': dataset['volumes'][split_idx:],
'features': dataset['features'][split_idx:],
'labels': dataset['labels'][split_idx:] if 'labels' in dataset else np.zeros(n_timesteps - split_idx),
'n_assets': dataset['n_assets'],
'n_timesteps': n_timesteps - split_idx,
}
if 'anomaly_details' in dataset:
test_dataset['anomaly_details'] = dataset['anomaly_details']
print(f"Split: {split_idx} train, {n_timesteps - split_idx} test")
return train_dataset, test_dataset
def download_sample_data():
"""Download sample stock data for demonstration"""
loader = KaggleDataLoader()
# Download data for major tech stocks
tickers = ['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'TSLA']
df = loader.download_stock_data(tickers=tickers, period='1y', interval='1d')
# Preprocess
dataset = loader.preprocess_for_surveillance(df)
# Inject anomalies for testing
dataset = loader.inject_synthetic_anomalies(dataset, anomaly_ratio=0.15)
# Split into train/test
train_data, test_data = loader.create_train_test_split(dataset)
# Save
np.savez('data/real_train_data.npz', **train_data)
np.savez('data/real_test_data.npz', **test_data)
print("\n✅ Sample data downloaded and preprocessed!")
print(f"Tickers: {', '.join(tickers)}")
print(f"Training: {train_data['n_timesteps']} timesteps")
print(f"Testing: {test_data['n_timesteps']} timesteps")
return train_data, test_data
if __name__ == "__main__":
# Download and prepare sample data
train_data, test_data = download_sample_data()Config File
""
Configuration settings for Market Surveillance System with Stable Diffusion
"""
import torch
class Config:
"""Central configuration for the market surveillance system"""
# Data Generation Parameters
DATA_CONFIG = {
'n_assets': 5, # Number of securities to simulate
'n_timesteps': 1000, # Length of time series
'train_ratio': 0.7, # Train/test split
'anomaly_ratio': 0.15, # Percentage of anomalies in test set
'base_price': 100.0, # Starting price
'volatility': 0.02, # Daily volatility
'trend_strength': 0.0001, # Trend component
'correlation': 0.3, # Asset correlation
}
# Anomaly Types and Parameters
ANOMALY_CONFIG = {
'pump_and_dump': {
'duration': 20, # Duration of pump phase
'magnitude': 0.3, # Price increase magnitude
'dump_speed': 5, # Speed of dump
},
'spoofing': {
'duration': 15,
'fake_volume_mult': 5.0, # Fake volume multiplier
},
'wash_trading': {
'duration': 25,
'volume_mult': 3.0, # Volume increase
},
'volume_spike': {
'duration': 5,
'magnitude': 10.0, # Volume spike magnitude
},
'insider_trading': {
'duration': 30,
'price_drift': 0.05, # Gradual price increase
}
}
# Diffusion Model Parameters
MODEL_CONFIG = {
'timesteps': 1000, # Number of diffusion steps
'beta_start': 0.0001, # Starting noise schedule
'beta_end': 0.02, # Ending noise schedule
'sequence_length': 100, # Length of input sequences
'hidden_dim': 128, # Hidden dimension for U-Net
'num_layers': 4, # Number of U-Net layers
'dropout': 0.1, # Dropout rate
}
# Training Parameters
TRAINING_CONFIG = {
'batch_size': 32,
'learning_rate': 1e-4,
'epochs': 100,
'weight_decay': 1e-5,
'gradient_clip': 1.0,
'save_interval': 10, # Save model every N epochs
'device': 'cuda' if torch.cuda.is_available() else 'cpu',
}
# Surveillance Parameters
SURVEILLANCE_CONFIG = {
'anomaly_threshold': 0.75, # Percentile threshold for anomaly detection
'alert_levels': {
'low': 0.75,
'medium': 0.85,
'high': 0.95,
},
'window_size': 100, # Sliding window for detection
'stride': 10, # Stride for sliding window
}
# Visualization Parameters
VIZ_CONFIG = {
'figure_size': (15, 8),
'dpi': 100,
'style': 'seaborn-v0_8-darkgrid',
'save_format': 'png',
}
# File Paths
PATHS = {
'data_dir': 'data',
'model_dir': 'models',
'results_dir': 'results',
'plots_dir': 'plots',
}
@classmethod
def get_device(cls):
"""Get the computing device"""
return cls.TRAINING_CONFIG['device']
@classmethod
def print_config(cls):
"""Print current configuration"""
print("=" * 60)
print("Market Surveillance System Configuration")
print("=" * 60)
print(f"\nDevice: {cls.get_device()}")
print(f"\nData: {cls.DATA_CONFIG['n_assets']} assets, "
f"{cls.DATA_CONFIG['n_timesteps']} timesteps")
print(f"Anomaly Ratio: {cls.DATA_CONFIG['anomaly_ratio']*100:.1f}%")
print(f"\nModel: {cls.MODEL_CONFIG['timesteps']} diffusion steps, "
f"{cls.MODEL_CONFIG['hidden_dim']} hidden dim")
print(f"Training: {cls.TRAINING_CONFIG['epochs']} epochs, "
f"batch size {cls.TRAINING_CONFIG['batch_size']}")
print("=" * 60)Gradio Web App Interface
""
Gradio Web Interface for Market Surveillance System
Interactive UI for visualizing anomaly detection results and alerts
"""
import gradio as gr
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import torch
import os
from pathlib import Path
from datetime import datetime
from config import Config
from data_generator import MarketDataGenerator
from diffusion_model import DiffusionModel, train_diffusion_model
from surveillance_system import MarketSurveillance
from visualization import SurveillanceVisualizer
class MarketSurveillanceUI:
"""Gradio UI for Market Surveillance System"""
def __init__(self):
self.model = None
self.surveillance = None
self.train_data = None
self.test_data = None
self.results = None
self.viz = SurveillanceVisualizer()
# Create directories
for path in Config.PATHS.values():
Path(path).mkdir(parents=True, exist_ok=True)
def generate_synthetic_data(self, n_assets, n_timesteps, anomaly_ratio, progress=gr.Progress()):
"""Generate synthetic market data"""
progress(0, desc="Generating data...")
# Update config
Config.DATA_CONFIG['n_assets'] = n_assets
Config.DATA_CONFIG['n_timesteps'] = n_timesteps
Config.DATA_CONFIG['anomaly_ratio'] = anomaly_ratio / 100.0
# Generate training data (normal)
progress(0.3, desc="Generating training data...")
train_gen = MarketDataGenerator()
self.train_data = train_gen.generate_dataset(include_anomalies=False)
# Generate test data (with anomalies)
progress(0.6, desc="Generating test data with anomalies...")
test_gen = MarketDataGenerator()
test_gen.config['anomaly_ratio'] = anomaly_ratio / 100.0
self.test_data = test_gen.generate_dataset(include_anomalies=True)
# Create visualizations
progress(0.9, desc="Creating visualizations...")
train_plot_path = os.path.join(Config.PATHS['plots_dir'], 'ui_train_data.png')
test_plot_path = os.path.join(Config.PATHS['plots_dir'], 'ui_test_data.png')
self.viz.plot_market_data(self.train_data, save_path=train_plot_path, show_anomalies=False)
self.viz.plot_market_data(self.test_data, save_path=test_plot_path, show_anomalies=True)
summary = f"""
✅ Data Generation Complete!
Training Data: {n_timesteps} timesteps, {n_assets} assets (normal patterns only)
Test Data: {n_timesteps} timesteps, {n_assets} assets
Anomalies Injected: {int(self.test_data['labels'].sum())} timesteps ({self.test_data['labels'].sum()/n_timesteps*100:.1f}%)
Anomaly Types: {len(self.test_data['anomaly_details'])} instances
"""
return summary, train_plot_path, test_plot_path
def train_model(self, epochs, progress=gr.Progress()):
"""Train the diffusion model"""
if self.train_data is None:
return "❌ Please generate data first!", None
progress(0, desc="Preparing training data...")
# Prepare sequences
features = self.train_data['features']
sequence_length = Config.MODEL_CONFIG['sequence_length']
sequences = []
for i in range(0, features.shape[0] - sequence_length, sequence_length // 2):
seq = features[i:i + sequence_length]
sequences.append(seq)
sequences = np.array(sequences)
sequences = torch.FloatTensor(sequences).transpose(1, 2)
# Create model
progress(0.1, desc="Initializing model...")
self.model = DiffusionModel()
# Update training config
Config.TRAINING_CONFIG['epochs'] = epochs
# Train
progress(0.2, desc=f"Training for {epochs} epochs...")
loss_history = train_diffusion_model(self.model, sequences, Config.TRAINING_CONFIG)
# Save model
progress(0.9, desc="Saving model...")
model_path = os.path.join(Config.PATHS['model_dir'], 'ui_diffusion_model.pt')
torch.save({
'model_state_dict': self.model.state_dict(),
'config': Config.MODEL_CONFIG,
'loss_history': loss_history,
}, model_path)
# Plot training curve
loss_plot_path = os.path.join(Config.PATHS['plots_dir'], 'ui_training_loss.png')
self.viz.plot_training_history(loss_history, save_path=loss_plot_path)
summary = f"""
✅ Training Complete!
Epochs: {epochs}
Final Loss: {loss_history[-1]:.6f}
Initial Loss: {loss_history[0]:.6f}
Improvement: {(1 - loss_history[-1]/loss_history[0])*100:.1f}%
Model Size: {os.path.getsize(model_path) / 1024 / 1024:.1f} MB
"""
return summary, loss_plot_path
def run_surveillance(self, threshold_percentile, progress=gr.Progress()):
"""Run market surveillance"""
if self.model is None:
return "❌ Please train model first!", None, None, None
if self.test_data is None:
return "❌ Please generate data first!", None, None, None
progress(0, desc="Initializing surveillance system...")
# Create surveillance system
self.surveillance = MarketSurveillance(self.model)
# Prepare training sequences for threshold
progress(0.2, desc="Calibrating threshold...")
features = self.train_data['features']
sequence_length = Config.MODEL_CONFIG['sequence_length']
sequences = []
for i in range(0, features.shape[0] - sequence_length, sequence_length // 2):
seq = features[i:i + sequence_length]
sequences.append(seq)
train_sequences = torch.FloatTensor(np.array(sequences)).transpose(1, 2)
# Set threshold
self.surveillance.set_threshold(train_sequences, percentile=threshold_percentile / 100.0)
# Run detection
progress(0.5, desc="Detecting anomalies...")
self.results = self.surveillance.analyze_batch(
self.test_data,
ground_truth_labels=self.test_data['labels']
)
# Create visualizations
progress(0.8, desc="Generating visualizations...")
detection_plot_path = os.path.join(Config.PATHS['plots_dir'], 'ui_anomaly_detection.png')
self.viz.plot_anomaly_detection(self.test_data, self.results, save_path=detection_plot_path)
if self.results['alerts']:
alerts_plot_path = os.path.join(Config.PATHS['plots_dir'], 'ui_alerts_dashboard.png')
self.viz.plot_alerts_dashboard(self.results, save_path=alerts_plot_path)
else:
alerts_plot_path = None
# Create summary
metrics = self.results['metrics']
summary = f"""
✅ Surveillance Complete!
🎯 Detection Results:
• Anomalies Detected: {self.results['n_anomalies']} / {len(self.results['anomalies'])} timesteps
• Detection Rate: {self.results['anomaly_ratio']*100:.1f}%
• Threshold: {self.results['threshold']:.6f}
📊 Performance Metrics:
• Precision: {metrics['precision']*100:.2f}%
• Recall: {metrics['recall']*100:.2f}%
• F1 Score: {metrics['f1_score']*100:.2f}%
• Accuracy: {metrics['accuracy']*100:.2f}%
🚨 Alerts Generated: {len(self.results['alerts'])}
"""
# Create alerts table
if self.results['alerts']:
alerts_df = pd.DataFrame(self.results['alerts'])
alerts_df = alerts_df[['start', 'end', 'duration', 'severity', 'max_score', 'avg_score']]
alerts_df = alerts_df.sort_values('max_score', ascending=False)
else:
alerts_df = pd.DataFrame()
return summary, detection_plot_path, alerts_plot_path, alerts_df
def create_interface(self):
"""Create enhanced Gradio interface"""
# Custom CSS for better styling
custom_css = """
.gradio-container {
font-family: 'Inter', sans-serif;
}
.tab-nav button {
font-size: 16px;
font-weight: 600;
}
.metric-box {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
padding: 20px;
border-radius: 10px;
color: white;
text-align: center;
}
.alert-high {
background-color: #fee;
border-left: 4px solid #f44;
}
.alert-medium {
background-color: #ffeaa7;
border-left: 4px solid #fdcb6e;
}
.alert-low {
background-color: #dfe6e9;
border-left: 4px solid #74b9ff;
}
"""
with gr.Blocks(
title="Market Surveillance System",
theme=gr.themes.Soft(
primary_hue="purple",
secondary_hue="blue",
neutral_hue="slate",
),
css=custom_css
) as demo:
gr.Markdown("""
# 🔍 Market Surveillance System with Stable Diffusion
<div style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); padding: 20px; border-radius: 10px; color: white; margin-bottom: 20px;">
<h3 style="margin: 0; color: white;">AI-Powered Anomaly Detection for Financial Markets</h3>
<p style="margin: 10px 0 0 0; opacity: 0.9;">Detect pump & dump, spoofing, wash trading, and insider trading patterns using deep learning</p>
</div>
""")
with gr.Tabs() as tabs:
# Tab 1: Data Generation
with gr.Tab("📊 Data Generation", id=0):
gr.Markdown("""
### Generate Synthetic Market Data
Create realistic market data with various manipulation patterns for testing
""")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("#### Configuration")
n_assets = gr.Slider(1, 10, value=5, step=1, label="Number of Assets",
info="Number of securities to simulate")
n_timesteps = gr.Slider(500, 2000, value=1000, step=100, label="Number of Timesteps",
info="Length of time series")
anomaly_ratio = gr.Slider(0, 30, value=15, step=5, label="Anomaly Ratio (%)",
info="Percentage of data with anomalies")
gr.Markdown("""
**Anomaly Types Injected:**
- 🚨 Pump & Dump
- 🎭 Spoofing
- 🔄 Wash Trading
- 📊 Volume Spikes
- 💼 Insider Trading
""")
generate_btn = gr.Button("🎲 Generate Data", variant="primary", size="lg")
with gr.Column(scale=2):
data_summary = gr.Textbox(label="📋 Generation Summary", lines=10,
show_label=True, container=True)
gr.Markdown("### 📈 Visualizations")
with gr.Row():
train_plot = gr.Image(label="Training Data (Normal Patterns)", height=400)
test_plot = gr.Image(label="Test Data (With Anomalies)", height=400)
generate_btn.click(
fn=self.generate_synthetic_data,
inputs=[n_assets, n_timesteps, anomaly_ratio],
outputs=[data_summary, train_plot, test_plot]
)
# Tab 2: Model Training
with gr.Tab("🧠 Model Training", id=1):
gr.Markdown("""
### Train Stable Diffusion Model
Train a U-Net based diffusion model to learn normal market patterns
""")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("#### Training Configuration")
epochs = gr.Slider(10, 200, value=50, step=10, label="Training Epochs",
info="More epochs = better performance")
with gr.Accordion("💡 Training Tips", open=False):
gr.Markdown("""
**Recommended Settings:**
- **Quick Test**: 20-30 epochs (~5 min)
- **Good Performance**: 50-75 epochs (~15 min)
- **Best Results**: 100+ epochs (~30 min)
**What to Expect:**
- Loss should decrease steadily
- Final loss < 0.2 is good
- Model size: ~90-100 MB
""")
train_btn = gr.Button("🚀 Train Model", variant="primary", size="lg")
with gr.Column(scale=2):
train_summary = gr.Textbox(label="📊 Training Summary", lines=10,
show_label=True, container=True)
gr.Markdown("### 📉 Training Progress")
loss_plot = gr.Image(label="Training Loss Curve", height=400)
train_btn.click(
fn=self.train_model,
inputs=[epochs],
outputs=[train_summary, loss_plot]
)
# Tab 3: Surveillance
with gr.Tab("🔍 Market Surveillance", id=2):
gr.Markdown("""
### Run Anomaly Detection & Generate Alerts
Analyze market data for suspicious patterns and generate alerts
""")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("#### Detection Settings")
threshold_percentile = gr.Slider(70, 99, value=90, step=1,
label="Detection Threshold (Percentile)",
info="Higher = fewer alerts, more confident")
with gr.Accordion("🎯 Threshold Guide", open=False):
gr.Markdown("""
**Sensitivity Levels:**
- **70-80%**: High sensitivity
- More alerts, catches subtle anomalies
- May have false positives
- **85-90%**: Balanced (Recommended)
- Good precision/recall tradeoff
- Reliable alerts
- **95-99%**: Low sensitivity
- Fewer alerts, very confident
- May miss some anomalies
""")
detect_btn = gr.Button("🎯 Run Detection", variant="primary", size="lg")
with gr.Column(scale=2):
detect_summary = gr.Textbox(label="📊 Detection Summary", lines=14,
show_label=True, container=True)
gr.Markdown("### 📈 Detection Visualizations")
with gr.Tabs():
with gr.Tab("Anomaly Detection"):
detection_plot = gr.Image(label="Anomaly Detection Results", height=500)
with gr.Tab("Alert Dashboard"):
alerts_plot = gr.Image(label="Alerts Dashboard", height=500)
with gr.Tab("Performance Metrics"):
gr.Markdown("""
View detailed performance metrics including precision, recall, F1-score,
and confusion matrix in the Detection Summary above.
""")
gr.Markdown("### 🚨 Alert Details")
with gr.Row():
with gr.Column():
gr.Markdown("#### All Detected Alerts")
alerts_table = gr.Dataframe(
headers=["Start", "End", "Duration", "Severity", "Max Score", "Avg Score"],
label="Alerts (Sorted by Score)",
interactive=False,
wrap=True
)
with gr.Accordion("📋 Alert Breakdown by Type", open=True):
gr.Markdown("""
**Alert Severity Levels:**
🔴 **HIGH** - Critical anomalies requiring immediate attention
- Score > 95th percentile
- Strong deviation from normal patterns
- Likely market manipulation
🟡 **MEDIUM** - Suspicious patterns worth investigating
- Score between 85th-95th percentile
- Moderate deviation from normal
- Potential manipulation or unusual activity
🟢 **LOW** - Minor anomalies for monitoring
- Score between 75th-85th percentile
- Slight deviation from normal
- May be normal volatility
**Detected Anomaly Types:**
- **Pump & Dump**: Rapid price increase followed by crash
- **Spoofing**: Fake high volume with price manipulation
- **Wash Trading**: High volume with minimal price change
- **Volume Spikes**: Sudden unusual volume increases
- **Insider Trading**: Gradual accumulation before events
""")
detect_btn.click(
fn=self.run_surveillance,
inputs=[threshold_percentile],
outputs=[detect_summary, detection_plot, alerts_plot, alerts_table]
)
# Tab 4: About
with gr.Tab("ℹ️ About", id=3):
gr.Markdown("""
## About This System
### 🎯 What It Does
This market surveillance system uses **stable diffusion models** to detect anomalous
trading patterns in financial markets. It learns what "normal" market behavior looks
like and flags deviations as potential market manipulation.
### 🔬 How It Works
<div style="background: #f8f9fa; padding: 20px; border-radius: 10px; margin: 20px 0;">
**1. Data Generation** 📊
- Creates synthetic market data with realistic patterns
- Injects known anomalies for testing
- Simulates multiple correlated assets
**2. Model Training** 🧠
- Trains a U-Net based diffusion model
- Learns the distribution of normal market behavior
- Uses denoising diffusion probabilistic models (DDPM)
**3. Anomaly Detection** 🔍
- Computes reconstruction error for test data
- High error = Anomaly (deviation from learned patterns)
- Generates alerts with severity levels
**4. Alert Generation** 🚨
- Groups contiguous anomalous regions
- Assigns severity based on anomaly score
- Provides detailed metrics and visualizations
</div>
### 🚨 Detected Anomaly Types
| Type | Characteristics | Detection Method |
|------|----------------|------------------|
| **Pump & Dump** | Artificial price inflation → crash | Rapid price spike + volume |
| **Spoofing** | Fake orders to manipulate | High volume, minimal price impact |
| **Wash Trading** | Self-trading for false volume | Circular trading patterns |
| **Volume Spikes** | Unusual trading volume | Sudden volume increases |
| **Insider Trading** | Pre-event accumulation | Gradual price drift + volume |
### 📊 Performance Metrics Explained
- **Precision**: When system flags anomaly, how often is it correct?
- High precision = Few false alarms
- **Recall**: What % of actual anomalies does it detect?
- High recall = Catches most anomalies
- **F1 Score**: Balance between precision and recall
- Higher is better (0-1 scale)
- **Accuracy**: Overall correctness
- Includes both normal and anomalous classifications
### 🛠️ Technical Stack
- **PyTorch**: Deep learning framework
- **Stable Diffusion**: DDPM architecture with U-Net
- **Gradio**: Interactive web interface
- **yfinance**: Real market data integration
- **NumPy/Pandas**: Data processing
- **Matplotlib**: Visualization
### 📚 Model Architecture
Input (Market Data)
↓
Forward Diffusion (Add Noise)
↓
U-Net Denoising Network
- 4 Encoder Layers
- 4 Decoder Layers
- Skip Connections
- Time Embeddings
↓
Reverse Diffusion (Denoise)
↓
Reconstruction Error → Anomaly Score
```
### 🎓 References
- Ho et al. (2020) - Denoising Diffusion Probabilistic Models
- Song et al. (2021) - Score-Based Generative Modeling
- Market manipulation detection literature
---
**Version**: 2.0 (Enhanced UI)
**Last Updated**: 2025-12-23
**License**: MIT
""")
gr.Markdown("""
---
<div style="text-align: center; padding: 20px; background: #f8f9fa; border-radius: 10px;">
<h3>💡 Quick Start Guide</h3>
<p><b>1.</b> Generate Data → <b>2.</b> Train Model (50+ epochs) → <b>3.</b> Run Surveillance → <b>4.</b> Review Alerts</p>
<p style="margin-top: 10px; color: #666;">All visualizations and models are automatically saved to the respective directories</p>
</div>
""")
return demodef launch_ui(share=False): """Launch the Gradio interface""" ui = MarketSurveillanceUI() demo = ui.create_interface() demo.launch(share=share, server_name="0.0.0.0", server_port=7860)
if name == "main": launch_ui(share=False)
## Model Architecture
## Input (Market Data)
↓
Forward Diffusion (Add Noise)
↓
U-Net Denoising Network
- 4 Encoder Layers
- 4 Decoder Layers
- Skip Connections
- Time Embeddings
↓
Reverse Diffusion (Denoise)
↓
Reconstruction Error → Anomaly Score
Skip connections in a diffusion U-Net pass features from each encoder layer to the corresponding decoder layer, helping the model preserve fine-grained details while the bottleneck captures global context during denoising.
In diffusion, the model must remove noise step by step. The bottleneck captures more **global structure**, while skip connections carry more **local, high-frequency details** that help reconstruction during denoising. Recent work analyzing diffusion U-Nets notes that the backbone mainly contributes denoising semantics, while skip connections mainly inject high-frequency features into the decoder.
Input Noisy Market Data ↓ Encoder Block 1 ───────────────┐ ↓ │ Encoder Block 2 ───────────┐ │ ↓ │ │ Encoder Block 3 ───────┐ │ │ ↓ │ │ │ Encoder Block 4 │ │ │ ↓ │ │ │ Bottleneck │ │ │ ↓ │ │ │ Decoder Block 4 ◄──────┘ │ │ ↓ │ │ Decoder Block 3 ◄──────────┘ │ ↓ │ Decoder Block 2 ◄──────────────┘ ↓ Decoder Block 1 ↓ Predicted Noise / Reconstructed Clean Signal ↓ Reconstruction Error → Anomaly Score
## Visualization
""" Visualization utilities for market surveillance system """ import numpy as np import matplotlib.pyplot as plt import matplotlib.patches as mpatches from typing import Dict, List, Optional, Tuple from config import Config import os
class SurveillanceVisualizer: """Visualization tools for market surveillance results"""
def __init__(self, config: dict = None):
"""Initialize visualizer"""
self.config = config or Config.VIZ_CONFIG
plt.style.use('seaborn-v0_8-darkgrid')
def plot_market_data(
self,
data_dict: Dict,
save_path: Optional[str] = None,
show_anomalies: bool = True
):
"""
Plot market data (prices and volumes)
Args:
data_dict: Dictionary with 'prices', 'volumes', 'labels'
save_path: Optional path to save figure
show_anomalies: Whether to highlight anomalies
"""
prices = data_dict['prices']
volumes = data_dict['volumes']
labels = data_dict.get('labels', None)
n_assets = data_dict['n_assets']
fig, axes = plt.subplots(2, 1, figsize=(15, 10))
# Plot prices
ax = axes[0]
for i in range(min(n_assets, 5)): # Plot first 5 assets
ax.plot(prices[:, i], label=f'Asset {i+1}', alpha=0.7)
if show_anomalies and labels is not None:
# Highlight anomalous regions
anomaly_regions = self._get_anomaly_regions(labels)
for start, end in anomaly_regions:
ax.axvspan(start, end, alpha=0.2, color='red')
ax.set_xlabel('Time')
ax.set_ylabel('Price')
ax.set_title('Market Prices')
ax.legend(loc='upper left')
ax.grid(True, alpha=0.3)
# Plot volumes
ax = axes[1]
for i in range(min(n_assets, 5)):
ax.plot(volumes[:, i], label=f'Asset {i+1}', alpha=0.7)
if show_anomalies and labels is not None:
for start, end in anomaly_regions:
ax.axvspan(start, end, alpha=0.2, color='red')
ax.set_xlabel('Time')
ax.set_ylabel('Volume')
ax.set_title('Trading Volumes')
ax.legend(loc='upper left')
ax.grid(True, alpha=0.3)
# Add legend for anomalies
if show_anomalies and labels is not None:
red_patch = mpatches.Patch(color='red', alpha=0.2, label='Anomaly')
axes[0].legend(handles=[*axes[0].get_legend_handles_labels()[0], red_patch],
loc='upper left')
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=self.config['dpi'], bbox_inches='tight')
print(f"Saved plot to {save_path}")
plt.close()
def plot_anomaly_detection(
self,
data_dict: Dict,
results: Dict,
save_path: Optional[str] = None
):
"""
Plot anomaly detection results
Args:
data_dict: Original data dictionary
results: Detection results from surveillance system
save_path: Optional path to save figure
"""
prices = data_dict['prices']
anomaly_scores = results['anomaly_scores']
anomalies = results['anomalies']
threshold = results['threshold']
ground_truth = data_dict.get('labels', None)
fig, axes = plt.subplots(3, 1, figsize=(15, 12))
# Plot first asset price
ax = axes[0]
ax.plot(prices[:, 0], label='Asset 1 Price', color='blue', alpha=0.7)
# Highlight detected anomalies
anomaly_regions = self._get_anomaly_regions(anomalies)
for start, end in anomaly_regions:
ax.axvspan(start, end, alpha=0.3, color='red', label='Detected Anomaly')
# Highlight ground truth if available
if ground_truth is not None:
gt_regions = self._get_anomaly_regions(ground_truth)
for start, end in gt_regions:
ax.axvspan(start, end, alpha=0.2, color='orange', label='True Anomaly')
ax.set_xlabel('Time')
ax.set_ylabel('Price')
ax.set_title('Price with Anomaly Detection')
ax.grid(True, alpha=0.3)
# Remove duplicate labels
handles, labels = ax.get_legend_handles_labels()
by_label = dict(zip(labels, handles))
ax.legend(by_label.values(), by_label.keys(), loc='upper left')
# Plot anomaly scores
ax = axes[1]
ax.plot(anomaly_scores, label='Anomaly Score', color='purple', alpha=0.7)
ax.axhline(y=threshold, color='red', linestyle='--',
label=f'Threshold ({threshold:.4f})')
ax.fill_between(range(len(anomaly_scores)), 0, anomaly_scores,
where=anomalies, alpha=0.3, color='red',
label='Detected Anomaly')
ax.set_xlabel('Time')
ax.set_ylabel('Anomaly Score')
ax.set_title('Anomaly Scores')
ax.legend(loc='upper left')
ax.grid(True, alpha=0.3)
# Plot comparison if ground truth available
if ground_truth is not None:
ax = axes[2]
# Create comparison array
comparison = np.zeros(len(anomalies))
comparison[anomalies & ground_truth.astype(bool)] = 3 # True Positive
comparison[anomalies & ~ground_truth.astype(bool)] = 2 # False Positive
comparison[~anomalies & ground_truth.astype(bool)] = 1 # False Negative
# Plot as colored regions
colors = ['white', 'orange', 'red', 'green']
labels_map = ['True Negative', 'False Negative', 'False Positive', 'True Positive']
for i in range(4):
mask = comparison == i
if mask.any():
ax.fill_between(range(len(mask)), 0, 1, where=mask,
alpha=0.6, color=colors[i], label=labels_map[i])
ax.set_xlabel('Time')
ax.set_ylabel('Classification')
ax.set_title('Detection Performance')
ax.set_ylim([0, 1])
ax.legend(loc='upper left')
ax.grid(True, alpha=0.3)
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=self.config['dpi'], bbox_inches='tight')
print(f"Saved plot to {save_path}")
plt.close()
def plot_training_history(
self,
loss_history: List[float],
save_path: Optional[str] = None
):
"""
Plot training loss history
Args:
loss_history: List of loss values
save_path: Optional path to save figure
"""
fig, ax = plt.subplots(figsize=(10, 6))
ax.plot(loss_history, color='blue', linewidth=2)
ax.set_xlabel('Epoch')
ax.set_ylabel('Loss')
ax.set_title('Training Loss History')
ax.grid(True, alpha=0.3)
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=self.config['dpi'], bbox_inches='tight')
print(f"Saved plot to {save_path}")
plt.close()
def plot_alerts_dashboard(
self,
results: Dict,
save_path: Optional[str] = None
):
"""
Create alerts dashboard
Args:
results: Detection results with alerts
save_path: Optional path to save figure
"""
alerts = results['alerts']
if not alerts:
print("No alerts to visualize")
return
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# Alert severity distribution
ax = axes[0, 0]
severity_counts = {}
for alert in alerts:
severity = alert['severity']
severity_counts[severity] = severity_counts.get(severity, 0) + 1
severities = ['low', 'medium', 'high']
counts = [severity_counts.get(s, 0) for s in severities]
colors = ['green', 'orange', 'red']
ax.bar(severities, counts, color=colors, alpha=0.7)
ax.set_xlabel('Severity')
ax.set_ylabel('Count')
ax.set_title('Alert Severity Distribution')
ax.grid(True, alpha=0.3, axis='y')
# Alert duration distribution
ax = axes[0, 1]
durations = [alert['duration'] for alert in alerts]
ax.hist(durations, bins=20, color='purple', alpha=0.7, edgecolor='black')
ax.set_xlabel('Duration (timesteps)')
ax.set_ylabel('Count')
ax.set_title('Alert Duration Distribution')
ax.grid(True, alpha=0.3, axis='y')
# Alert scores
ax = axes[1, 0]
max_scores = [alert['max_score'] for alert in alerts]
avg_scores = [alert['avg_score'] for alert in alerts]
x = range(len(alerts))
ax.scatter(x, max_scores, label='Max Score', alpha=0.7, s=50)
ax.scatter(x, avg_scores, label='Avg Score', alpha=0.7, s=50)
ax.set_xlabel('Alert Index')
ax.set_ylabel('Score')
ax.set_title('Alert Scores')
ax.legend()
ax.grid(True, alpha=0.3)
# Timeline of alerts
ax = axes[1, 1]
severity_colors = {'low': 'green', 'medium': 'orange', 'high': 'red'}
for alert in alerts:
color = severity_colors[alert['severity']]
ax.barh(0, alert['end'] - alert['start'], left=alert['start'],
height=0.5, color=color, alpha=0.7)
ax.set_xlabel('Time')
ax.set_yticks([])
ax.set_title('Alert Timeline')
ax.grid(True, alpha=0.3, axis='x')
# Add legend
legend_elements = [mpatches.Patch(color=c, alpha=0.7, label=s.upper())
for s, c in severity_colors.items()]
ax.legend(handles=legend_elements, loc='upper right')
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=self.config['dpi'], bbox_inches='tight')
print(f"Saved plot to {save_path}")
plt.close()
def plot_reconstruction_comparison(
self,
original: np.ndarray,
reconstructed: np.ndarray,
save_path: Optional[str] = None
):
"""
Plot original vs reconstructed data
Args:
original: Original data
reconstructed: Reconstructed data
save_path: Optional path to save figure
"""
fig, axes = plt.subplots(2, 1, figsize=(15, 8))
# Plot original
ax = axes[0]
ax.plot(original, label='Original', color='blue', alpha=0.7)
ax.set_xlabel('Time')
ax.set_ylabel('Value')
ax.set_title('Original Data')
ax.legend()
ax.grid(True, alpha=0.3)
# Plot reconstructed
ax = axes[1]
ax.plot(reconstructed, label='Reconstructed', color='red', alpha=0.7)
ax.set_xlabel('Time')
ax.set_ylabel('Value')
ax.set_title('Reconstructed Data')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=self.config['dpi'], bbox_inches='tight')
print(f"Saved plot to {save_path}")
plt.close()
@staticmethod
def _get_anomaly_regions(labels: np.ndarray) -> List[Tuple[int, int]]:
"""
Get contiguous anomaly regions
Args:
labels: Binary labels
Returns:
List of (start, end) tuples
"""
regions = []
in_anomaly = False
start_idx = 0
for i in range(len(labels)):
if labels[i] and not in_anomaly:
in_anomaly = True
start_idx = i
elif not labels[i] and in_anomaly:
in_anomaly = False
regions.append((start_idx, i))
# Handle case where anomaly extends to end
if in_anomaly:
regions.append((start_idx, len(labels)))
return regionsif name == "main": print("Testing Visualization...")
# Create dummy data
n_timesteps = 500
n_assets = 3
data_dict = {
'prices': np.random.randn(n_timesteps, n_assets).cumsum(axis=0) + 100,
'volumes': np.random.lognormal(10, 0.5, (n_timesteps, n_assets)),
'labels': np.random.rand(n_timesteps) > 0.9,
'n_assets': n_assets,
}
results = {
'anomaly_scores': np.random.rand(n_timesteps),
'anomalies': np.random.rand(n_timesteps) > 0.85,
'threshold': 0.7,
'alerts': [
{'start': 50, 'end': 70, 'duration': 20, 'max_score': 0.95,
'avg_score': 0.85, 'severity': 'high'},
{'start': 150, 'end': 160, 'duration': 10, 'max_score': 0.82,
'avg_score': 0.75, 'severity': 'medium'},
]
}
viz = SurveillanceVisualizer()
# Test plots
viz.plot_market_data(data_dict, save_path='test_market.png')
viz.plot_anomaly_detection(data_dict, results, save_path='test_detection.png')
viz.plot_alerts_dashboard(results, save_path='test_alerts.png')
print("Visualization tests complete!")
## Main File
""" End-to-End Market Surveillance Pipeline Main script to run the complete system """ import os import argparse import numpy as np import torch from pathlib import Path
from config import Config from data_generator import MarketDataGenerator from diffusion_model import DiffusionModel, train_diffusion_model from surveillance_system import MarketSurveillance from visualization import SurveillanceVisualizer
def create_directories(): """Create necessary directories""" for path in Config.PATHS.values(): Path(path).mkdir(parents=True, exist_ok=True) print("Created directories")
def generate_data(args): """Generate synthetic market data""" print("\n" + "=" 60) print("STEP 1: GENERATING SYNTHETIC MARKET DATA") print("=" 60)
generator = MarketDataGenerator()
# Generate training data (normal patterns only)
print("\nGenerating training data (normal patterns)...")
Config.DATA_CONFIG['anomaly_ratio'] = 0.0
train_generator = MarketDataGenerator()
train_dataset = train_generator.generate_dataset(include_anomalies=False)
# Generate test data (with anomalies)
print("\nGenerating test data (with anomalies)...")
Config.DATA_CONFIG['anomaly_ratio'] = 0.15
test_generator = MarketDataGenerator()
test_dataset = test_generator.generate_dataset(include_anomalies=True)
# Save datasets
train_path = os.path.join(Config.PATHS['data_dir'], 'train_data.npz')
test_path = os.path.join(Config.PATHS['data_dir'], 'test_data.npz')
train_generator.save_dataset(train_dataset, train_path)
test_generator.save_dataset(test_dataset, test_path)
# Visualize data
if args.visualize:
print("\nGenerating visualizations...")
viz = SurveillanceVisualizer()
viz.plot_market_data(
train_dataset,
save_path=os.path.join(Config.PATHS['plots_dir'], 'train_data.png'),
show_anomalies=False
)
viz.plot_market_data(
test_dataset,
save_path=os.path.join(Config.PATHS['plots_dir'], 'test_data.png'),
show_anomalies=True
)
print("\n[SUCCESS] Data generation complete!")
return train_dataset, test_datasetdef train_model(args, train_dataset): """Train diffusion model""" print("\n" + "=" 60) print("STEP 2: TRAINING DIFFUSION MODEL") print("=" 60)
# Prepare training data
features = train_dataset['features']
n_samples = features.shape[0]
sequence_length = Config.MODEL_CONFIG['sequence_length']
# Create sequences
sequences = []
for i in range(0, n_samples - sequence_length, sequence_length // 2):
seq = features[i:i + sequence_length]
sequences.append(seq)
sequences = np.array(sequences)
sequences = torch.FloatTensor(sequences).transpose(1, 2) # (N, features, length)
print(f"\nTraining sequences: {sequences.shape}")
# Create and train model
model = DiffusionModel()
# Adjust epochs if quick mode
if args.quick:
Config.TRAINING_CONFIG['epochs'] = 20
print("Quick mode: Using 20 epochs")
loss_history = train_diffusion_model(model, sequences, Config.TRAINING_CONFIG)
# Save model
model_path = os.path.join(Config.PATHS['model_dir'], 'diffusion_model.pt')
torch.save({
'model_state_dict': model.state_dict(),
'config': Config.MODEL_CONFIG,
'loss_history': loss_history,
}, model_path)
print(f"\n[SUCCESS] Model saved to {model_path}")
# Visualize training
if args.visualize:
viz = SurveillanceVisualizer()
viz.plot_training_history(
loss_history,
save_path=os.path.join(Config.PATHS['plots_dir'], 'training_loss.png')
)
print("\n[SUCCESS] Training complete!")
return model, sequencesdef run_surveillance(args, model, train_sequences, test_dataset): """Run surveillance system""" print("\n" + "=" 60) print("STEP 3: RUNNING MARKET SURVEILLANCE") print("=" 60)
# Create surveillance system
surveillance = MarketSurveillance(model)
# Set threshold using training data
print("\nCalibrating anomaly detection threshold...")
surveillance.set_threshold(train_sequences, percentile=0.90)
# Analyze test data
print("\nAnalyzing test data...")
results = surveillance.analyze_batch(
test_dataset,
ground_truth_labels=test_dataset['labels']
)
# Print report
surveillance.print_report(results)
# Save results
results_path = os.path.join(Config.PATHS['results_dir'], 'surveillance_results.npz')
np.savez(results_path, **results)
print(f"\n[SUCCESS] Results saved to {results_path}")
# Visualize results
if args.visualize:
print("\nGenerating surveillance visualizations...")
viz = SurveillanceVisualizer()
viz.plot_anomaly_detection(
test_dataset,
results,
save_path=os.path.join(Config.PATHS['plots_dir'], 'anomaly_detection.png')
)
viz.plot_alerts_dashboard(
results,
save_path=os.path.join(Config.PATHS['plots_dir'], 'alerts_dashboard.png')
)
print("\n[SUCCESS] Surveillance complete!")
return resultsdef main(): """Main pipeline""" parser = argparse.ArgumentParser( description='Market Surveillance with Stable Diffusion' ) parser.add_argument( '--mode', type=str, default='full', choices=['generate', 'train', 'detect', 'full'], help='Pipeline mode: generate data, train model, detect anomalies, or full pipeline' ) parser.add_argument( '--visualize', action='store_true', default=True, help='Generate visualizations' ) parser.add_argument( '--quick', action='store_true', help='Quick mode with fewer epochs' )
args = parser.parse_args()
# Print configuration
print("\n")
Config.print_config()
# Create directories
create_directories()
# Run pipeline based on mode
if args.mode in ['generate', 'full']:
train_dataset, test_dataset = generate_data(args)
else:
# Load existing data
train_path = os.path.join(Config.PATHS['data_dir'], 'train_data.npz')
test_path = os.path.join(Config.PATHS['data_dir'], 'test_data.npz')
train_dataset = MarketDataGenerator.load_dataset(train_path)
test_dataset = MarketDataGenerator.load_dataset(test_path)
if args.mode in ['train', 'full']:
model, train_sequences = train_model(args, train_dataset)
else:
# Load existing model
model_path = os.path.join(Config.PATHS['model_dir'], 'diffusion_model.pt')
checkpoint = torch.load(model_path)
model = DiffusionModel()
model.load_state_dict(checkpoint['model_state_dict'])
# Prepare sequences
features = train_dataset['features']
sequence_length = Config.MODEL_CONFIG['sequence_length']
sequences = []
for i in range(0, features.shape[0] - sequence_length, sequence_length // 2):
seq = features[i:i + sequence_length]
sequences.append(seq)
train_sequences = torch.FloatTensor(np.array(sequences)).transpose(1, 2)
if args.mode in ['detect', 'full']:
results = run_surveillance(args, model, train_sequences, test_dataset)
# Final summary
print("\n" + "=" * 60)
print("PIPELINE COMPLETE!")
print("=" * 60)
print(f"\nResults saved in: {Config.PATHS['results_dir']}")
if args.visualize:
print(f"Plots saved in: {Config.PATHS['plots_dir']}")
print("\nNext steps:")
print(" 1. Review visualizations in the plots directory")
print(" 2. Check surveillance results for detected anomalies")
print(" 3. Adjust thresholds in config.py if needed")
print(" 4. Re-run with different parameters")
print("=" * 60 + "\n")if name == "main": main()
# Output
## Training Loss
## Test Data With Anomalies
# Conclusion
Stable Diffusion represents a major shift in how generative image models are designed and deployed. By moving the diffusion process from raw pixel space into a **compressed latent space**, it achieves an exceptional balance between **computational efficiency and image quality** — making high-fidelity image generation accessible beyond large research labs.
Its architecture — combining a **CLIP-based text encoder**, a **U-Net denoising network with cross-attention**, and a **pre-trained VAE** — demonstrates how modular, well-designed components can work together to produce controllable, scalable, and reproducible results. Innovations such as **classifier-free guidance, advanced sampling schedulers, ControlNet, LoRA, and DreamBooth** extend the core model into a flexible creative platform rather than a single-purpose tool.
More importantly, Stable Diffusion’s **open-weight ecosystem** has reshaped the generative AI landscape. It enables local deployment, domain-specific customization, cost-effective experimentation, and community-driven innovation — while also demanding responsible use through proper guardrails, bias awareness, and ethical oversight.
In the broader GenAI stack, Stable Diffusion serves as the **visual generation layer**, complementing LLMs, RAG systems, and agentic workflows to power end-to-end creative and production pipelines. Whether used for art, e-commerce, research, education, or enterprise applications, Stable Diffusion illustrates a core lesson of modern AI systems: **efficient representations, strong conditioning, and openness are what turn powerful models into practical, real-world technology.**
Thanks for reading! I hope this post helped you gain a clearer understanding of the topic. If it added value, a few claps or a follow on Medium would go a long way — they help the content reach a wider audience and keep me motivated to share more.
Grateful for your time and support!메타데이터
- post_id
- b32f8b39eb9c
- slug
- stable-diffusion-project-implementation-b32f8b39eb9c
- url
- https://pub.towardsai.net/stable-diffusion-project-implementation-b32f8b39eb9c
- canonical_url
- https://pub.towardsai.net/stable-diffusion-project-implementation-b32f8b39eb9c
- author_url
- https://medium.com/@rashmi18patel
- status
- ok
- fetched_at
- 2026-08-07 08:37:44
