Create IBM Granite Time Series Models MCP Server

[crivetimihai]

Overview

Create a comprehensive MCP Server for IBM Granite Time Series Models supporting forecasting, anomaly detection, trend analysis, and temporal data processing with enterprise-grade capabilities.

Server Specifications

Server Details

• Name: granite-timeseries-server
• Language: Python 3.11+
• Location: mcp-servers/python/granite_timeseries_server/
• Purpose: Provide access to IBM Granite time series models via MCP

Supported Models

From IBM Granite Time Series on Hugging Face:

• Forecasting: granite-timeseries-forecast-v1, granite-multivariate-forecast
• Anomaly Detection: granite-timeseries-anomaly-v1, granite-outlier-detection
• Classification: granite-timeseries-classify-v1, granite-pattern-recognition
• Foundation: granite-timeseries-foundation, granite-temporal-embeddings

Provider Support

• Ollama: Local inference with time series models
• watsonx.ai: IBM's enterprise time series platform
• Hugging Face: Direct access to time series transformers
• Custom Endpoints: Flexible time series API integration

Tools Provided

1. forecast_timeseries

Time series forecasting with multiple horizons

@dataclass
class ForecastRequest:
    data: List[Dict[str, Any]]  # Time series data
    model: str = "granite-timeseries-forecast-v1"
    provider: str = "ollama"
    forecast_horizon: int = 12
    confidence_intervals: bool = True
    confidence_level: float = 0.95
    seasonal_periods: Optional[int] = None
    external_features: Optional[List[str]] = None
    frequency: str = "auto"  # auto, daily, weekly, monthly

2. detect_anomalies

Anomaly detection in time series data

@dataclass
class AnomalyDetectionRequest:
    data: List[Dict[str, Any]]
    model: str = "granite-timeseries-anomaly-v1"
    provider: str = "watsonx"
    sensitivity: float = 0.95
    method: str = "statistical"  # statistical, ml_based, hybrid
    window_size: Optional[int] = None
    return_scores: bool = True
    contextual_anomalies: bool = True

3. classify_patterns

Time series pattern recognition and classification

@dataclass
class PatternClassificationRequest:
    data: List[Dict[str, Any]]
    model: str = "granite-timeseries-classify-v1"
    provider: str = "huggingface"
    pattern_types: List[str]  # trend, seasonal, cyclical, irregular
    classification_level: str = "series"  # series, segment, point
    confidence_threshold: float = 0.8
    window_analysis: bool = True

4. analyze_trends

Trend analysis and decomposition

@dataclass
class TrendAnalysisRequest:
    data: List[Dict[str, Any]]
    model: str = "granite-timeseries-foundation"
    provider: str = "ollama"
    decomposition_method: str = "additive"  # additive, multiplicative, auto
    trend_components: List[str] = ["trend", "seasonal", "residual"]
    change_point_detection: bool = True
    seasonal_adjustment: bool = True

5. multivariate_analysis

Multi-dimensional time series analysis

@dataclass
class MultivariateAnalysisRequest:
    data: List[Dict[str, Any]]  # Multiple time series
    model: str = "granite-multivariate-forecast"
    provider: str = "watsonx"
    analysis_type: str = "correlation"  # correlation, causality, clustering
    variables: List[str]
    lag_analysis: bool = True
    cross_correlation: bool = True
    dimensionality_reduction: Optional[str] = None  # pca, tsne, umap

6. generate_embeddings

Time series embeddings for similarity and clustering

@dataclass
class EmbeddingRequest:
    data: List[Dict[str, Any]]
    model: str = "granite-temporal-embeddings"
    provider: str = "huggingface"
    embedding_dimension: int = 128
    window_size: int = 100
    stride: int = 1
    normalize_embeddings: bool = True
    similarity_metric: str = "cosine"  # cosine, euclidean, manhattan

7. real_time_monitoring

Real-time time series monitoring and alerting

@dataclass
class RealTimeMonitoringRequest:
    stream_source: str  # data stream endpoint
    model: str = "granite-timeseries-anomaly-v1"
    provider: str = "ollama"
    monitoring_metrics: List[str] = ["anomalies", "drift", "quality"]
    alert_thresholds: Dict[str, float]
    buffer_size: int = 1000
    update_frequency: str = "1min"  # 1min, 5min, 1hour

8. batch_timeseries_processing

Efficient batch processing of multiple time series

@dataclass
class BatchTimeSeriesRequest:
    datasets: List[Dict[str, Any]]  # Multiple time series datasets
    model: str = "granite-timeseries-forecast-v1"
    provider: str = "watsonx"
    processing_type: str = "forecast"  # forecast, anomaly, classify, analyze
    parallel_processing: bool = True
    max_concurrent: int = 4
    aggregate_results: bool = True

Implementation Requirements

Directory Structure

mcp-servers/python/granite_timeseries_server/
├── src/
│   └── granite_timeseries_server/
│       ├── __init__.py
│       ├── server.py
│       ├── providers/
│       │   ├── __init__.py
│       │   ├── ollama_timeseries.py
│       │   ├── watsonx_timeseries.py
│       │   ├── huggingface_timeseries.py
│       │   └── streaming_provider.py
│       ├── models/
│       │   ├── __init__.py
│       │   ├── granite_ts_models.py
│       │   └── model_registry.py
│       ├── processing/
│       │   ├── __init__.py
│       │   ├── data_preprocessor.py
│       │   ├── forecasting.py
│       │   ├── anomaly_detection.py
│       │   ├── pattern_analysis.py
│       │   └── trend_analysis.py
│       ├── tools/
│       │   ├── __init__.py
│       │   ├── forecasting_tools.py
│       │   ├── anomaly_tools.py
│       │   ├── analysis_tools.py
│       │   └── monitoring_tools.py
│       └── utils/
│           ├── __init__.py
│           ├── time_utils.py
│           ├── data_validation.py
│           └── visualization.py
├── tests/
├── requirements.txt
├── README.md
└── examples/
    ├── forecasting_example.py
    ├── anomaly_detection.py
    └── real_time_monitoring.py

Dependencies

# requirements.txt
mcp>=1.0.0
transformers>=4.35.0
torch>=2.1.0
pandas>=2.1.0
numpy>=1.24.0
scipy>=1.11.0
scikit-learn>=1.3.0
statsmodels>=0.14.0
prophet>=1.1.4
plotly>=5.17.0
matplotlib>=3.7.0
seaborn>=0.13.0
requests>=2.31.0
pydantic>=2.5.0
ollama>=0.1.7
ibm-watson-machine-learning>=1.0.325

Configuration

# config.yaml
providers:
  ollama:
    base_url: "http://localhost:11434"
    timeseries_models_enabled: true
    timeout: 300
    
  watsonx:
    url: "https://us-south.ml.cloud.ibm.com"
    apikey: "${WATSONX_API_KEY}"
    project_id: "${WATSONX_PROJECT_ID}"
    timeseries_endpoint: "/ml/v1/timeseries"
    
  huggingface:
    api_key: "${HF_API_KEY}"
    cache_dir: "./hf_timeseries_cache"
    device: "auto"

models:
  default_forecast: "granite-timeseries-forecast-v1"
  default_anomaly: "granite-timeseries-anomaly-v1"
  default_classify: "granite-timeseries-classify-v1"
  default_foundation: "granite-timeseries-foundation"

processing:
  max_data_points: 1000000
  max_series_length: 10000
  supported_frequencies: ["1min", "5min", "1hour", "1day", "1week", "1month"]
  auto_preprocessing: true
  missing_value_handling: "interpolate"  # drop, interpolate, forward_fill

forecasting:
  default_horizon: 12
  max_horizon: 365
  confidence_levels: [0.8, 0.9, 0.95]
  seasonal_detection: true
  
anomaly_detection:
  default_sensitivity: 0.95
  methods: ["statistical", "ml_based", "hybrid"]
  real_time_capable: true

monitoring:
  buffer_size: 10000
  alert_cooldown: 300  # seconds
  max_concurrent_streams: 10

Usage Examples

Time Series Forecasting

# Forecast sales data for next 12 months
result = await mcp_client.call_tool("forecast_timeseries", {
    "data": [
        {"timestamp": "2024-01-01", "value": 1000, "category": "electronics"},
        {"timestamp": "2024-01-02", "value": 1150, "category": "electronics"},
        # ... more data points
    ],
    "model": "granite-timeseries-forecast-v1",
    "provider": "ollama",
    "forecast_horizon": 12,
    "confidence_intervals": True,
    "seasonal_periods": 12,
    "frequency": "monthly"
})

Anomaly Detection

# Detect anomalies in sensor data
result = await mcp_client.call_tool("detect_anomalies", {
    "data": [
        {"timestamp": "2024-01-01T00:00:00", "sensor_value": 23.5, "location": "factory_a"},
        {"timestamp": "2024-01-01T01:00:00", "sensor_value": 24.1, "location": "factory_a"},
        # ... more sensor readings
    ],
    "model": "granite-timeseries-anomaly-v1",
    "provider": "watsonx",
    "sensitivity": 0.95,
    "method": "hybrid",
    "return_scores": True
})

Pattern Classification

# Classify patterns in financial data
result = await mcp_client.call_tool("classify_patterns", {
    "data": [
        {"date": "2024-01-01", "price": 150.25, "volume": 1000000},
        {"date": "2024-01-02", "price": 152.10, "volume": 1200000},
        # ... more financial data
    ],
    "model": "granite-timeseries-classify-v1",
    "provider": "huggingface",
    "pattern_types": ["trend", "seasonal", "cyclical"],
    "classification_level": "series",
    "confidence_threshold": 0.8
})

Multivariate Analysis

# Analyze relationships between multiple time series
result = await mcp_client.call_tool("multivariate_analysis", {
    "data": [
        {"timestamp": "2024-01-01", "temperature": 22.5, "humidity": 65, "sales": 1000},
        {"timestamp": "2024-01-02", "temperature": 23.1, "humidity": 68, "sales": 1100},
        # ... more multivariate data
    ],
    "model": "granite-multivariate-forecast",
    "provider": "watsonx",
    "analysis_type": "correlation",
    "variables": ["temperature", "humidity", "sales"],
    "lag_analysis": True,
    "cross_correlation": True
})

Real-time Monitoring

# Set up real-time anomaly monitoring
result = await mcp_client.call_tool("real_time_monitoring", {
    "stream_source": "kafka://localhost:9092/sensor_data",
    "model": "granite-timeseries-anomaly-v1",
    "provider": "ollama",
    "monitoring_metrics": ["anomalies", "drift"],
    "alert_thresholds": {
        "anomaly_score": 0.9,
        "drift_score": 0.8
    },
    "buffer_size": 1000,
    "update_frequency": "1min"
})

Time Series Embeddings

# Generate embeddings for similarity analysis
result = await mcp_client.call_tool("generate_embeddings", {
    "data": [
        {"timestamp": "2024-01-01", "value": 100},
        {"timestamp": "2024-01-02", "value": 105},
        # ... time series data
    ],
    "model": "granite-temporal-embeddings",
    "provider": "huggingface",
    "embedding_dimension": 128,
    "window_size": 100,
    "normalize_embeddings": True
})

Advanced Features

• Multi-horizon Forecasting: Short, medium, and long-term predictions
• Hierarchical Time Series: Support for grouped and nested series
• External Regressors: Incorporate external variables in models
• Model Ensembling: Combine multiple models for better accuracy
• Probabilistic Forecasting: Full uncertainty quantification
• Online Learning: Models that adapt to new data continuously

Enterprise Features

• High-Frequency Data: Handle minute-level and sub-minute data
• Scalable Processing: Distributed processing for large datasets
• Model Governance: Version control and model lineage tracking
• A/B Testing: Compare different forecasting approaches
• API Integration: REST endpoints for enterprise systems
• Compliance: Audit trails and regulatory compliance

Specialized Applications

• Financial Markets: Stock price prediction and risk analysis
• Supply Chain: Demand forecasting and inventory optimization
• Energy: Load forecasting and renewable energy prediction
• IoT/Sensors: Equipment monitoring and predictive maintenance
• Healthcare: Patient monitoring and outbreak prediction
• Retail: Sales forecasting and seasonal planning

Acceptance Criteria

• Python MCP server with 8+ Granite time series model tools
• Support for all major Granite time series models
• Multi-provider integration (Ollama, watsonx.ai, Hugging Face)
• Time series forecasting with confidence intervals
• Anomaly detection with multiple methods
• Pattern classification and trend analysis
• Multivariate time series analysis
• Real-time monitoring and streaming capabilities
• Time series embeddings and similarity analysis
• Batch processing for efficiency
• Comprehensive test suite with sample time series data (>90% coverage)
• Complete documentation with forecasting examples

Priority

High - Demonstrates IBM's advanced time series AI capabilities via MCP

Use Cases

• Financial forecasting and risk management
• Supply chain demand planning and optimization
• Energy load forecasting and grid management
• IoT sensor monitoring and predictive maintenance
• Healthcare patient monitoring and prediction
• Retail sales forecasting and inventory management
• Manufacturing quality control and process optimization
• Environmental monitoring and climate analysis