TS-Unity: Time Series Unified Framework

A comprehensive framework for time series analysis including forecasting, anomaly detection, imputation, and classification.

Features

• Multi-Task Support: Forecasting, Anomaly Detection, Imputation, Classification
• Multiple Models: Autoformer, Transformer, Informer, and more
• Type Safety: Comprehensive type hints throughout the codebase
• Error Handling: Robust error handling and validation
• Real-Time Inference: Streaming inference with sliding window support
• Batch Inference: Process entire datasets efficiently
• REST API: HTTP endpoints for easy integration
• Modular Design: Clean, maintainable architecture

Framework Structure

src/
├── config/
│ └── base_config.py # Enhanced configuration management
├── core/
│ ├── base_trainer.py # Enhanced base trainer
│ └── pipeline.py # Multi-task inference pipeline
├── models/ # Model implementations
├── utils/
│ └── anomaly_detection_metrics.py # Refactored metrics
├── api/
│ └── inference_server.py # FastAPI inference server
├── scripts/
│ ├── realtime_inference.py # Real-time inference demo
│ ├── batch_inference.py # Batch inference demo
│ ├── anomaly_detection_inference.py # Anomaly detection demo
│ └── api_client_example.py # API client example
└── main.py # Enhanced main entry point

Installation

git clone https://github.com/your-username/TS-Unity.git
cd TS-Unity
pip install -r requirements.txt

REST API Server

TS-Unity now provides a REST API server for easy integration with other systems and applications.

Starting the API Server

# Start the server
python src/api/inference_server.py --host 0.0.0.0 --port 8000
# Or with auto-reload for development
python src/api/inference_server.py --reload

API Endpoints

Health Check

GET /health

Load Model

POST /load_model
{
"task_type": "anomaly_detection",
"checkpoint_path": "/path/to/checkpoint.pth",
"config_overrides": {
"seq_len": 100,
"enc_in": 7
}
}

Forecasting Inference

POST /forecast
{
"task_type": "forecasting",
"data": [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], ...],
"num_steps": 24,
"window_size": 100,
"stride": 10
}

Anomaly Detection

POST /detect_anomalies
{
"task_type": "anomaly_detection",
"data": [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], ...],
"threshold": 0.8,
"window_size": 100,
"stride": 10
}

Generic Inference

POST /inference
{
"task_type": "imputation",
"data": [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], ...],
"window_size": 100,
"stride": 10
}

File-based Inference

POST /inference/file?file_path=data.csv&task_type=forecasting&num_steps=24

Model Information

GET /model/info

Unload Model

DELETE /model

API Client Example

fromsrc.scripts.api_client_exampleimportTSUnityAPIClient# Initialize clientclient=TSUnityAPIClient("http://localhost:8000")
# Load modelclient.load_model("anomaly_detection", "/path/to/checkpoint.pth")
# Run anomaly detectionresult=client.run_anomaly_detection(
data=[[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], ...],
threshold=0.8
)
print(f"Anomalies detected: {result['anomalies_detected']}")
print(f"Anomaly rate: {result['anomaly_rate']:.2f}%")

Interactive API Documentation

Once the server is running, you can access:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

Real-Time Inference

TS-Unity supports real-time streaming inference for live time series data.

Python API Usage

fromsrc.core.pipelineimportInferencePipeline, RealTimeInferenceManagerfromsrc.config.base_configimportForecastingConfig# Create configurationconfig=ForecastingConfig(
task_name='short_term_forecast',
model='Autoformer',
seq_len=100,
enc_in=7,
dec_in=7,
c_out=7,
pred_len=24
)
# Initialize inference pipelinepipeline=InferencePipeline(config, checkpoint_path='model.pth')
# Add real-time data pointspipeline.add_data_point([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0])
# Get predictionspredictions=pipeline.predict_next(num_steps=12)

Command Line Usage

# Real-time inference demo
python src/scripts/realtime_inference.py --checkpoint_path model.pth --mode interactive
# Simulate streaming data
python src/scripts/realtime_inference.py --checkpoint_path model.pth --mode simulation
# Stream from CSV file
python src/scripts/realtime_inference.py --checkpoint_path model.pth --mode csv --input_file data.csv

Batch Inference

TS-Unity supports efficient batch inference for processing large datasets.

Python API Usage

# Single sequence inferencepredictions=pipeline.predict_batch(sequence_data, num_steps=24)
# Multiple sequencesbatch_predictions=pipeline.predict_batch(batch_data, num_steps=24)
# Sliding window on long seriesresults=pipeline.predict_with_sliding_window(
long_series, window_size=100, stride=10, num_steps=24
)
# File-based inferenceresults=pipeline.predict_from_file(
'data.csv', num_steps=24, output_path='predictions.csv'
)

Command Line Usage

# Single sequence inference
python src/scripts/batch_inference.py --checkpoint_path model.pth --mode single
# Batch inference
python src/scripts/batch_inference.py --checkpoint_path model.pth --mode batch --batch_size 10
# Sliding window inference
python src/scripts/batch_inference.py --checkpoint_path model.pth --mode sliding --data_length 1000
# File-based inference
python src/scripts/batch_inference.py --checkpoint_path model.pth --mode file --input_file data.csv
# Performance benchmark
python src/scripts/batch_inference.py --checkpoint_path model.pth --mode benchmark

Anomaly Detection

TS-Unity now supports anomaly detection inference with dedicated methods and analysis.

Anomaly Detection Approaches

TS-Unity supports two different approaches to anomaly detection:

1. Reconstruction-based Anomaly Detection

• Models: AnomalyTransformer, OmniAnomaly, USAD, DAGMM, AutoEncoder, VAE
• Method: Compares input with reconstructed output
• Anomaly Score: Reconstruction error (higher = more anomalous)
• Best for: Detecting structural anomalies, pattern changes, distribution shifts
• How it works: The model learns to reconstruct normal patterns, and anomalies have higher reconstruction errors

2. Prediction-based Anomaly Detection

• Models: Autoformer, Transformer, TimesNet, and other forecasting models
• Method: Analyzes prediction patterns and variance
• Anomaly Score: Prediction variance or error (higher = more anomalous)
• Best for: Detecting temporal anomalies, trend changes, unexpected patterns
• How it works: Uses forecasting models to predict future values and detects anomalies based on prediction confidence

Automatic Method Selection

The system automatically selects the appropriate detection method based on the model type:

# Reconstruction-based detection (e.g., AnomalyTransformer)config=AnomalyDetectionConfig(
task_name='anomaly_detection',
model='AnomalyTransformer', # Reconstruction modelseq_len=100,
enc_in=7
)
# Prediction-based detection (e.g., Autoformer)config=ForecastingConfig(
task_name='anomaly_detection',
model='Autoformer', # Forecasting modelseq_len=100,
enc_in=7
)

Enhanced Experiment Classes

The experiment classes now support both approaches:

Exp_Anomaly_Detection

• Reconstruction-based: Uses _reconstruction_based_scoring() for models like AnomalyTransformer
• Prediction-based: Uses _prediction_based_scoring() for forecasting models
• Automatic selection: Detects model type and applies appropriate method
• Method information: Saves detection method details in results

Exp_Long_Term_Forecast & Exp_Short_Term_Forecast

• Forecasting: Primary functionality for time series prediction
• Anomaly detection: Secondary functionality using prediction-based approach
• Dual purpose: Can be used for both forecasting and anomaly detection
• Flexible input: Handles various model architectures (Linear, TST, Transformer-based)

Usage Examples

# Using forecasting model for anomaly detectionfromexp.exp_long_term_forecastingimportExp_Long_Term_Forecastexp=Exp_Long_Term_Forecast(args)
exp.model.load_state_dict(torch.load('forecasting_model.pth'))
# Anomaly detection using prediction-based methodanomaly_scores=exp.detect_anomaly(input_data)
# Regular forecastingpredictions=exp.predict_single(input_data, num_steps=24)

Python API Usage

fromsrc.config.base_configimportAnomalyDetectionConfig# Create anomaly detection configurationconfig=AnomalyDetectionConfig(
task_name='anomaly_detection',
model='AnomalyTransformer',
seq_len=100,
enc_in=7,
dec_in=7,
c_out=7,
anomaly_ratio=0.1,
win_size=100
)
# Initialize inference pipelinepipeline=InferencePipeline(config, checkpoint_path='anomaly_model.pth')
# Run anomaly detectionanomaly_scores=pipeline.predict_batch(sequence_data, num_steps=1)
# Analyze resultsprint(f"Anomaly scores shape: {anomaly_scores.shape}")

Command Line Usage

# Single sequence anomaly detection
python src/scripts/anomaly_detection_inference.py --checkpoint_path anomaly_model.pth --mode single
# Batch anomaly detection
python src/scripts/anomaly_detection_inference.py --checkpoint_path anomaly_model.pth --mode batch --batch_size 10
# Sliding window anomaly detection
python src/scripts/anomaly_detection_inference.py --checkpoint_path anomaly_model.pth --mode sliding --data_length 1000
# File-based anomaly detection
python src/scripts/anomaly_detection_inference.py --checkpoint_path anomaly_model.pth --mode file --input_file data.csv
# Performance benchmark
python src/scripts/anomaly_detection_inference.py --checkpoint_path anomaly_model.pth --mode benchmark

Data Format Requirements

Input Data

• Real-time: Single data points as 1D arrays
• Batch: 2D arrays (time_steps, features) or 3D arrays (batch_size, time_steps, features)
• File: CSV, NPY, or NPZ files

Output Data

• Forecasting: Predictions for specified number of steps ahead
• Anomaly Detection: Anomaly scores for each time step
• Imputation: Imputed values
• Classification: Class probabilities

Performance Tips

Real-Time Inference

• Use appropriate buffer size for your use case
• Consider batch processing for multiple data points
• Monitor memory usage with large buffers

Batch Inference

• Use sliding window for very long time series
• Adjust stride based on your requirements
• Consider parallel processing for large batches

API Usage

• Keep models loaded for multiple requests
• Use appropriate batch sizes for your data
• Monitor API response times

Examples

See the src/scripts/ directory for comprehensive examples:

• realtime_inference.py: Real-time streaming examples
• batch_inference.py: Batch processing examples
• anomaly_detection_inference.py: Anomaly detection examples
• api_client_example.py: API usage examples

Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests if applicable
5. Submit a pull request

License

This project is licensed under the MIT License - see the LICENSE file for details.

---

TS-Unity v2.0 - Enhanced with type safety, error handling, real-time inference, batch processing, and REST API support.