MCP Zero Shot Agentic Forecaster

Executive-Überblick & Geschäftswert

Was ist dieses Repository?

Der MCP Zero Shot Agentic Forecaster ist eine produktionsreife, mikroservicebasierte Zeitreihen-Prognose-Engine, die über das Model Context Protocol (MCP) bereitgestellt wird. Angetrieben von modernsten Foundation-Modellen (Google TimesFM 2.5 XReg und Amazon Chronos-2) ermöglicht er autonomen KI-Agenten (LangGraph-Zustandsmaschinen, CrewAI-Schwärme, OPA/Rego-gesteuerte NeSy-Stacks und standardmäßige ReAct-Schleifen), bedarfsgerecht probabilistische Nachfrageprognosen abzufragen – ohne Offline-Modelltraining, Hyperparameter-Tuning oder Pro-SKU-Datensatzvorbereitung.

Geschäftswert & ROI

• Eliminiert Kaltstart-Latenz: Liefert sofortige Zero-Shot-Wahrscheinlichkeitsprognosen für neue Produkteinführungen, Aktionen und SKUs mit kurzer Historie – ohne Trainingspipelines.

• Quantilbegrenzte Risikokontrolle: Gibt kalibrierte $p_{10}$-, $p_{50}$- und $p_{90}$-Nachfragequantile aus, sodass autonome Einkaufsagenten Sicherheitsbestandspuffer gegen Kapitalbindungskosten abwägen können.

• Niedrigere Gesamtbetriebskosten (TCO): Ersetzt komplexe Fine-Tuning-Pipelines durch eine einheitliche 3-stufige Fallback-Engine und senkt so den GPU-Rechenbedarf und die Infrastrukturdrift drastisch.

• Agentische Resilienz: Gibt bei ungültigen Eingaben strukturierte AgentFriendlyError-Payloads mit Hinweisen zur Behebung zurück, sodass aufrufende Agenten sich in Ausführungsschleifen selbst korrigieren können, ohne stillschweigend zu scheitern oder unbehandelte Ausnahmen auszulösen.

So funktioniert es

1. Agentenaufruf: Aufrufende Agenten rufen forecast_demand oder forecast_batch per stdio/HTTP über die MCP-Tool-Schnittstelle auf.

2. Vertragsdurchsetzung: Pydantic-v2-Schemas führen strenge Prüfungen endlicher numerischer und zeitlicher Grenzen durch ([Type-Safe Input Contract]).

3. Nicht-blockierende Inferenz: FastMCP lagert schwere Tensoroperationen über asyncio.to_thread an Thread-Pools aus, um die Reaktionsfähigkeit des Gateways zu erhalten.

4. Einheitliche 3-stufige Pipeline: Leitet immer über TimesFM 2.5 (Stufe 1), mit Fallback auf Chronos-2 (Stufe 2, behält Kovariaten, falls vorhanden) und ARIMA111 (Stufe 3, verwirft Kovariaten). Bei CUDA-OOM oder Fehlern löst die Engine eine Speicherwiederherstellung aus (gc.collect() + torch.cuda.empty_cache()), bevor sie auf die nächste Stufe degradiert.

5. Mathematische Bereinigung: Wendet isotonische Sortierung an, um die Monotonie der Ausgabequantile zu gewährleisten ($p_{10} \le p_{50} \le p_{90}$), und normalisiert epistemische Konfidenzwerte, bevor strukturierte JSON-Payloads zurückgegeben werden.

Related MCP server: Geneva Forecasting MCP

Systemarchitektur

Der Mikroservice folgt einer strikten Trennung der Zuständigkeiten: Die MCP-Tool-Schicht verwaltet nicht-blockierenden Transport, Speichersicherheit und die Bereinigung der Modellausgaben, während domänenspezifische Geschäftsrichtlinien den nachgelagerten Agenten-Orchestratoren überlassen bleiben.

graph TD
    subgraph External Agent Orchestrator
        Agent[LLM Agent / Swarm / State Machine<br/>LangGraph / OPA Sidecar / ReAct Loop]
    end

    subgraph MCP Microservice Boundary
        Gateway[FastMCP Async Gateway Server<br/>mcp_server.py]
        Sanitizer[Pydantic v2 Input Contract<br/>TimeSeriesInputPayload]
        ErrorFormatter[AgentFriendlyError Formatter]
        
        subgraph Engine Memory & Concurrency Boundary
            ExecThread[Thread Executor<br/>asyncio.to_thread]
            Engine[ZeroShotForecastingEngine<br/>src/models/forecaster.py<br/>Lazy-Load Lock Protected]
            
            subgraph 3-Tier Fallback Model Chain
                T1[Tier 1: TimesFM 2.5<br/>XReg / Univariate]
                T2[Tier 2: Chronos-2<br/>Multivariate / Univariate]
                T3[Tier 3: ARIMA111<br/>CPU Baseline Fallback]
            end
            
            IsoSanitizer[Isotonic Quantile Sanitizer<br/>Enforces p10 ≤ p50 ≤ p90]
        end
    end

    Agent -->|FastMCP Tool Call<br/>forecast_demand / forecast_batch| Gateway
    Gateway -->|1. Validate Schema| Sanitizer
    Sanitizer -->|Validation Error| ErrorFormatter
    ErrorFormatter -.->|Structured Error + Remediation| Agent
    
    Sanitizer -->|2. Valid Payload| ExecThread
    ExecThread -->|3. Route Request| Engine
    Engine --> T1
    T1 -.->|CUDA OOM / Fail| T2
    T2 -.->|Fail| T3
    
    T1 -->|Raw Quantiles| IsoSanitizer
    T2 -->|Raw Quantiles| IsoSanitizer
    T3 -->|Raw Quantiles| IsoSanitizer
    
    IsoSanitizer -->|4. Validated ForecastResponse| Gateway
    Gateway -->|5. Return JSON Payload| Agent

Aufschlüsselung der Komponentenarchitektur

FastMCP-Gateway (mcp_server.py): Stellt asynchronen JSON-RPC-Transport bereit und setzt Grenzen für die Batch-Nebenläufigkeit durch (asyncio.Semaphore(4)).

Typsichere Vertragsgrenze (src/schemas/payloads.py): Setzt zeitliche Ausrichtung, Garantien endlicher Zahlen sowie Kontext-/Horizontgrenzen durch.

Thread-sicherer Forecaster-Kern (src/models/forecaster.py): Verwendet Double-Checked Locking (threading.Lock()) für das Lazy Model Loading und übernimmt die automatische CUDA-OOM-Wiederherstellung (gc.collect() + torch.cuda.empty_cache()).

Isotoner Ausgabebereiniger: Verarbeitet rohe Foundation-Modell-Quantile mithilfe monotoner Sortierung nach, um statistische Anomalien zu eliminieren ($p_{10} > p_{50}$), bevor Prognosen an Agenten zurückgegeben werden.

Systemausführungsablauf (Sequenzdiagramm)

sequenceDiagram
    autonumber
    actor Agent as LLM Agent / Orchestrator
    participant Gateway as FastMCP Gateway (Async)
    participant Sanitizer as Pydantic Input Contract
    participant Executor as Thread Executor (asyncio.to_thread)
    participant Pipeline as 3-Tier Fallback Pipeline
    participant Std as Isotonic Quantile Sanitizer

    Agent->>Gateway: forecast_demand / forecast_batch (JSON)
    Gateway->>Sanitizer: Validate TimeSeriesInputPayload
    alt Validation Failure
        Sanitizer-->>Gateway: AgentFriendlyError {error_code, expected, received, remediation}
        Gateway-->>Agent: Structured Error Response
    else Validation Success
        Sanitizer->>Executor: Offload sync inference
        Executor->>Pipeline: Execute prediction
        Note right of Pipeline: Tier 1: TimesFM 2.5 → Tier 2: Chronos-2 → Tier 3: ARIMA111
        alt CUDA OOM / Transient Failure
            Pipeline->>Pipeline: gc.collect() + torch.cuda.empty_cache()
            Pipeline->>Pipeline: Degrade to next tier (retain covariates where possible)
        end
        Pipeline->>Std: Apply _enforce_quantile_monotonicity()
        Std-->>Executor: ForecastResponse {model_used, exogenous_dropped, warnings}
        Executor-->>Gateway: Return validated response
        Gateway-->>Agent: 200 OK with ForecastResponse
    end

Zentrale Architekturprinzipien

Nicht-blockierender asynchroner Transport

Alle Tensor-Forward-Pässe werden über asyncio.to_thread in einem Thread-Pool ausgeführt, sodass die FastMCP-Ereignisschleife auch unter Last für gleichzeitige Health Checks und Tool-Aufrufe reaktionsfähig bleibt.

# mcp_server.py
result = await asyncio.to_thread(engine.predict, validated_payload)

Lazy Loading des VRAM

Modellgewichte werden nur bei erster Verwendung über Zugriffsmethoden materialisiert – beim Start wird kein VRAM verbraucht. Thread-sicheres Double-Checked Locking verhindert doppelte Instanziierung bei gleichzeitigen Kaltstarts.

# src/models/forecaster.py
def _get_timesfm(self):
    if self._timesfm is None:
        with self._timesfm_lock:
            if self._timesfm is None:
                import timesfm
                logger.info(f"Lazily loading TimesFM-2.5 ({self.timesfm_repo_id}) onto {self._device}")
                self._timesfm = timesfm.TimesFM_2p5_200M_torch.from_pretrained(self.timesfm_repo_id)
    return self._timesfm

def _get_chronos(self):
    if self._chronos is None:
        with self._chronos_lock:
            if self._chronos is None:
                from chronos import BaseChronosPipeline
                logger.info(f"Lazily loading Chronos-2 ({self.chronos_repo_id}) onto {self._device}")
                self._chronos = BaseChronosPipeline.from_pretrained(
                    self.chronos_repo_id, device_map=self._device, dtype=torch.float32
                )
    return self._chronos

3-stufige Fallback-Engine

Die Engine unterhält unabhängig vom Payload-Inhalt eine einzige einheitliche deterministische Degradationskette.

Bei TimesFM-Fehlern (einschließlich torch.cuda.OutOfMemoryError):

1. gc.collect() + torch.cuda.empty_cache()

2. Exogene Signale werden für Chronos-2 über _build_chronos_covariates() beibehalten (vergangene/zukünftige Kovariaten)

3. exogenous_dropped = true nur wenn der Fallback auf Stufe 3 (ARIMA111) degradiert

4. Ausführung wird an Chronos-2 weitergeleitet

5. Wenn Chronos fehlschlägt → ARIMA111-Baseline

Durchsetzung zeitlicher Regelmäßigkeit

Pydantic-v2-Validatoren weisen ungültige Telemetrie an der Grenze zurück:

Isotonische Quantilbereinigung

Alle Backends (TimesFM, Chronos-2, AutoARIMA) geben rohe Quantile aus, die bei extremen OOD-Eingaben gelegentlich kreuzen können ($p_{10} > p_{50}$ oder $p_{50} > p_{90}$). Die Engine wendet einen leichtgewichtigen Nachbearbeitungsschritt _enforce_quantile_monotonicity() an, der pro Zeitschritt eine isotonische Sortierung durchführt – stapelt $(p_{10}, p_{50}, p_{90})$, sortiert entlang der Quantilachse und gibt die geordneten Tripel zurück. Dies garantiert mathematisch gültige $p_{10} \le p_{50} \le p_{90}$ für jeden Prognosehorizont-Schritt, ohne die Verteilungsform zu verzerren.

Begrenzte Batch-Nebenläufigkeit

Das MCP-Tool forecast_batch führt Multi-SKU-Inferenz gleichzeitig mit asyncio.gather aus, begrenzt durch ein asyncio.Semaphore(4). Dies bietet parallelen Durchsatz und schützt gleichzeitig den GPU/CPU-Speicher vor unbegrenzten gleichzeitigen Tensorzuordnungen. Jedes Element erwirbt das Semaphor, validiert seinen Payload, lagert engine.predict über asyncio.to_thread an einen Thread-Pool aus und gibt eine strukturierte ForecastResponse mit Fallback-Metadaten pro Element zurück. Der Zusammenfassungsblock meldet die Gesamtzahl der Elemente, die Fehleranzahl und die Nutzung pro Backend (model_usage).

Circuit Breaker pro Backend

Jedes Foundation-Modell-Backend unterhält einen unabhängigen Circuit Breaker (CircuitBreakerState), um Kaskadenfehler zu verhindern, wenn ein Modell-Hub nicht erreichbar ist oder dauerhaft Fehler liefert. Nach 5 aufeinanderfolgenden Fehlern öffnet sich der Schalter und leitet den Datenverkehr sofort für 60 Sekunden an die nächste Fallback-Stufe weiter, bevor ein Testaufruf zugelassen wird.

Dadurch wird sichergestellt, dass eine vorübergehende HuggingFace-Hub-Störung oder ein fehlerhafter Gewichtsdownload den Agenten nicht auf unbestimmte Zeit blockiert.

Normalisierte Konfidenzmetrik

Der Konfidenzwert verwendet ein begrenztes relatives Unsicherheitsverhältnis anstelle einer linearen Untergrenze, die breite Varianz auf 0,0 komprimiert:

$$\text{Confidence} = \frac{1}{1 + \frac{p_{90} - p_{10}}{\vert p_{50}\vert + \epsilon}}$$

wobei $\epsilon = 10^{-5}$. Eigenschaften:

• Ausgabebereich $(0, 1]$ — nie negativ, nie auf 0 komprimiert

• Wenn die Spannweite $(p_{90} - p_{10}) \to 0$, Konfidenz $\to 1$ (enge Grenzen)

• Wenn die Spannweite $\to \infty$, Konfidenz $\to 0$ asymptotisch (extreme Unsicherheit)

• Skaleninvariant durch Division durch die Median-Größenordnung $|p_{50}|$

Agentik-Architektur-agnostisch

Als zustandsloser, schema-gebundener MCP-Tool-Mikroservice integriert sich diese Engine nahtlos in jeden Agenten-Orchestrator – einschließlich neuro-symbolischer Stacks, die durch OPA/Rego-Richtlinien gesteuert werden, LangGraph-Zustandsmaschinen, CrewAI-Schwärme oder standardmäßige ReAct-Schleifen.

MCP-Toolspezifikationen & API-Verträge

forecast_demand

Einzelne Zeitreihenprognose.

Anfrage (TimeSeriesInputPayload)

{
  "target_series": [120.5, 115.0, 130.2, 125.8, 140.1],
  "forecast_horizon": 30,
  "price_index": [19.99, 19.99, 24.99, 24.99, 24.99, 24.99, ...],
  "promo_flag": [0, 0, 1, 0, 1, 0, ...]
}

Antwort (ForecastResponse)

{
  "model_used": "Chronos-2-Fallback",
  "mean_prediction": [142.3, 145.1, 140.8, 148.2, 150.0],
  "p10_quantile": [120.1, 122.4, 118.7, 125.3, 127.9],
  "p50_quantile": [142.3, 145.1, 140.8, 148.2, 150.0],
  "p90_quantile": [165.2, 168.5, 162.1, 170.4, 172.8],
  "confidence_score": 0.87,
  "horizon_length": 30,
  "exogenous_dropped": false,
  "warnings": ["CUDA unavailable; running on CPU. Expect degraded inference performance."]
}

forecast_batch

Array-basierte Multi-SKU-Prognose mit Fallback-Zusammenfassung pro Element.

Anfrage

{
  "payloads": [
    {"target_series": [10.0]*20, "forecast_horizon": 5},
    {"target_series": [11.0]*30, "forecast_horizon": 3, "price_index": [20.0]*33}
  ]
}

Antwort

{
  "results": [
    {"model_used": "Chronos-2-Fallback", "mean_prediction": [...], ...},
    {"model_used": "TimesFM-2.5", "mean_prediction": [...], ...}
  ],
  "summary": {
    "total": 2,
    "errors": 0,
    "model_usage": {"Chronos-2-Fallback": 1, "TimesFM-2.5": 1}
  }
}

AgentFriendlyError

Selbstkorrigierendes Fehlerschema, das bei Validierungs- oder Ausführungsfehlern zurückgegeben wird.

{
  "error_code": "VALIDATION_ERROR",
  "message": "Input validation failed at 'price_index': Price array misalignment. Expected 25 elements (Context: 20 + Horizon: 5), got 2.",
  "expected": "Payload matching TimeSeriesInputPayload schema (context 16-16000 finite values, aligned exogenous signals).",
  "received": "{\"location\": \"price_index\", \"message\": \"Price array misalignment. Expected 25 elements (Context: 20 + Horizon: 5), got 2.\", \"context\": {\"expected\": \"25\", \"got\": \"2\"}}",
  "remediation_suggestion": "Correct field 'price_index' (Price array misalignment. Expected 25 elements (Context: 20 + Horizon: 5), got 2.) and resubmit. Ensure context length is between 16 and 16000, values are finite (no NaN/Inf), and exogenous arrays align to len(target_series) + forecast_horizon."
}

Fehlercodes: VALIDATION_ERROR, MODEL_UNAVAILABLE, TRANSIENT_FAILURE, MODEL_FAILURE, INTERNAL_ERROR

Schnellstart & MCP-Konfiguration

Installation

# Requires Python 3.11+
uv sync --extra gpu   # or: pip install -r requirements.txt

Umgebung

# Optional: force CPU if GPU memory constrained
export MODEL_CONFIG_PATH=configs/model_config.yaml
export DATA_STORAGE_ROOT=data/

Die Engine erkennt CUDA automatisch. Wenn nicht verfügbar, fällt sie auf CPU zurück und gibt eine Warnung im Feld warnings aus.

MCP-Client-Konfiguration (Claude Desktop / OpenCode / LangGraph / CrewAI)

Fügen Sie Folgendes zu Ihrer MCP-Client-Konfiguration hinzu (claude_desktop_config.json, opencode.json oder Äquivalent):

{
  "mcpServers": {
    "zero-shot-forecaster": {
      "command": "python",
      "args": ["mcp_server.py"],
      "cwd": "/absolute/path/to/zero-shot-demand-foundation",
      "env": {
        "MODEL_CONFIG_PATH": "configs/model_config.yaml"
      }
    }
  }
}

Starten Sie Ihren MCP-Client neu. Die Tools forecast_demand und forecast_batch registrieren sich automatisch mit ihren vollständigen JSON-Schemas.

Aufrufbeispiel (Claude / LLM-Agent)

{
  "tool": "forecast_demand",
  "arguments": {
    "target_series": [120, 115, 130, 125, 140, 135, 150, 145, 155, 160, 155, 165, 170, 168, 172, 175, 180, 178, 185, 190],
    "forecast_horizon": 7,
    "price_index": [19.99, 19.99, 19.99, 19.99, 19.99, 19.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99],
    "promo_flag": [0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0]
  }
}

Verifizierung & Tests

Vollständige Testsuite ausführen (62 Tests, AC-1 bis AC-5)

uv run pytest tests/ -q
# or
python -m pytest tests/ -q

Erwartete Ausgabe:

.............................................................. [100%]
62 passed in ~3s

Testabdeckungsmatrix

Legacy-Tests (Abwärtskompatibilität)

Alle 48 ursprünglichen Tests bestehen weiterhin:

pytest tests/test_forecasting_engine.py tests/test_forecaster_router.py tests/test_mcp_server.py tests/test_schemas.py tests/test_metrics.py -q

Projektstruktur (nach dem Refactoring)

zero-shot-demand-foundation/
├── configs/
│   └── model_config.yaml            # Model IDs, device_map, num_samples
├── data/                            # Git-ignored (CSV, ZIP)
├── scripts/
│   ├── download_m5.py               # M5 dataset fetcher
│   └── download_favorita.py         # Favorita dataset fetcher
├── src/
│   ├── models/
│   │   └── forecaster.py            # ZeroShotForecastingEngine (refactored)
│   ├── schemas/
│   │   └── payloads.py              # TimeSeriesInputPayload, ForecastResponse, AgentFriendlyError
│   └── utils/
│       ├── data_loader.py           # DemandDataEngine, FavoritaDataLoader
│       └── metrics.py               # WAPE, RMSSE, Pinball Loss, CRPS
├── tests/
│   ├── test_forecasting_engine.py   # Updated for lazy loading
│   ├── test_forecaster_router.py    # Updated fixtures
│   ├── test_mcp_server.py           # Async + AgentFriendlyError
│   ├── test_metrics.py              # Unchanged
│   ├── test_schemas.py              # Unchanged
│   └── test_refactored_forecaster.py # NEW: AC-1..5 coverage
├── main.py                          # CLI evaluation entry point
├── mcp_server.py                    # FastMCP server (async, batch, errors)
├── requirements.txt
├── .gitignore                       # Ignores *.md, data/, __pycache__/
└── README.md                        # This file

Lizenz

MIT-Lizenz. Siehe LICENSE für Details.

Referenzen

• Chronos-2: Ansari et al., Chronos: Learning the Language of Time Series, arXiv:2403.07815

• TimesFM: Das et al., TimesFM: A Decoder-Only Foundation Model for Time-Series Forecasting, arXiv:2402.02592

• M5 Competition: Makridakis et al., M5 Accuracy Competition, IJF 2022

• Corporación Favorita: Kaggle Favorita Grocery Sales Forecasting

• Model Context Protocol: Anthropic MCP Specification