DEV Community

Ama
Ama

Posted on

Building a Robust Market Research Assistant: Clean Architecture and LLM Tool Routing in Python

#programming #ai #python #llm

When designing AI-powered financial or analytics pipelines, developers frequently run into two major failure modes:

  • Tight Coupling: LLM orchestration logic is directly bound to external market APIs. Any breaking change from a data vendor breaks the entire agent pipeline.
  • Fragile Outputs: Relying on raw text generation for deterministic indicators creates hallucinated figures and pipeline crashes downstream.

To solve this in Trading-research-assistant, the system applies Hexagonal Architecture (Ports and Adapters), strict schema validation with Pydantic, and decoupled inference routing.

High-Level Architecture (Ports & Adapters)

The core domain layer remains completely isolated from external HTTP clients, third-party market APIs, and specific inference engines.

       +---------------------------------------------+
       |               User / CLI / API              |
       +---------------------------------------------+
                              |
                              v
       +---------------------------------------------+
       |               Application Layer             |
       |  (ResearchCoordinator, AnalysisOrchestrator) |
       +---------------------------------------------+
            |                                   |
            v                                   v
   [ MarketDataPort ]                  [ LLMInferencePort ]
            ^                                   ^
            | (implements)                      | (implements)
+------------------------+          +------------------------+
| Adapters:              |          | Adapters:              |
| - OandaAdapter         |          | - OllamaAdapter        |
| - TwelveDataAdapter    |          | - OpenRouterAdapter    |
| - MockDataAdapter      |          | - ClaudeAdapter        |
+------------------------+          +------------------------+
Enter fullscreen mode Exit fullscreen mode

Key Architectural Benefits

  • Zero-Cost Unit Testing: Fast mock adapters allow full integration tests without consuming rate limits or paid API credits.
  • Resilient Failovers: If a primary provider hits rate limits (HTTP 429) or service outages, the orchestrator switches to a fallback adapter implementing the identical port contract.

Strict Interface Contracts

Data boundaries between adapters and application services are enforced using typing.Protocol and immutable Pydantic schemas.

from datetime import datetime
from typing import Protocol, Sequence
from pydantic import BaseModel, Field


class MarketCandle(BaseModel):
    timestamp: datetime
    open: float
    high: float
    low: float
    close: float
    volume: float


class IndicatorSnapshot(BaseModel):
    rsi: float = Field(ge=0.0, le=100.0)
    macd_signal: str
    summary: str


class MarketDataProviderPort(Protocol):
    async def fetch_historical_candles(
        self, symbol: str, timeframe: str, limit: int
    ) -> Sequence[MarketCandle]:
        """Fetch historical OHLCV data."""
        ...

    async def get_indicator_snapshot(self, symbol: str) -> IndicatorSnapshot:
        """Fetch pre-calculated technical indicators."""
        ...
Enter fullscreen mode Exit fullscreen mode

Deterministic Routing & Local Inference

The execution pipeline isolates computation from reasoning:

  1. Deterministic Calculation: Technical indicators (RSI, Moving Averages, MACD) are computed directly in Python adapters.
  2. Context Synthesis: Pre-validated metrics are passed to the inference layer solely for semantic summarization and risk profiling. 
import httpx
from pydantic import BaseModel


class SynthesisRequest(BaseModel):
    symbol: str
    indicators: IndicatorSnapshot
    sentiment_score: float


class OllamaResearchAdapter:
    def __init__(self, base_url: str = "http://localhost:11434", model: str = "llama3") -> None:
        self.base_url = base_url
        self.model = model

    async def generate_executive_summary(self, payload: SynthesisRequest) -> str:
        prompt = (
            f"Analyze the verified market metrics for {payload.symbol}:\n"
            f"- RSI: {payload.indicators.rsi}\n"
            f"- MACD Signal: {payload.indicators.macd_signal}\n"
            f"- Sentiment Index: {payload.sentiment_score}\n"
            "Synthesize risk exposure. Do not generate or modify numerical metrics."
        )

        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{self.base_url}/api/generate",
                json={
                    "model": self.model,
                    "prompt": prompt,
                    "stream": False,
                    "format": "json",
                },
            )
            response.raise_for_status()
            return response.json().get("response", "")
Enter fullscreen mode Exit fullscreen mode

Production Takeaways

  • Avoid LLM Arithmetic: Compute technical indicators natively; feed only validated snapshots into model context.
  • Enforce Structured JSON: Utilize JSON schemas and Pydantic validation on all model responses to prevent parsing errors.
  • Decouple Infrastructure: Isolating data ingestion and inference behind ports enables zero-downtime provider migrations.

I build custom AI pipelines, robust trading infrastructure, and backend automation systems for production environments. If you are designing complex backend architectures or need resilient custom implementations:

Connect on Fiverr for Custom Architecture & Engineering

Top comments (0)