Real-time market data agents with AutoGen

Building a real-time market data agent autogen system means wiring together live data feeds, technical analysis, and multi-agent reasoning without the latency that kills trading strategies. This tutorial walks through a production-ready implementation: a primary agent that pulls streaming quotes, a specialist agent that computes indicators, and a coordinator that synthesizes signals — all running locally with zero external API dependencies beyond market data.

Prerequisites

You need Python 3.10+, an OpenAI-compatible endpoint (we use n4n.ai for model routing in the examples, but any OpenAI-compatible endpoint works), and the following packages:

pip install autogen-agentchat==0.2.0 autogen-ext==0.2.0 yfinance pandas numpy python-dotenv

Create a .env file with your endpoint configuration:

# .env
OPENAI_API_KEY=your_key_here
OPENAI_BASE_URL=https://api.n4n.ai/v1  # or your preferred endpoint
MODEL_NAME=gpt-4o-mini

The code below assumes this environment. Adjust MODEL_NAME to whatever your endpoint serves.

Architecture overview

We’ll build three agents:

1. Data Agent — Fetches real-time and historical data via yfinance, caches recent bars, exposes a clean tool interface.
2. Analysis Agent — Receives data frames, computes technical indicators (RSI, MACD, Bollinger Bands), returns structured JSON.
3. Coordinator Agent — Takes a user query (e.g., “Should I buy AAPL right now?”), delegates to the other agents, synthesizes a recommendation with citations.

AutoGen’s SelectorGroupChat routes messages based on agent descriptions. We’ll define each agent with a focused system prompt and register tools using the FunctionTool pattern.

Step 1: Data agent with caching and retries

Market data is flaky. Providers throttle, connections drop, timestamps drift. The data agent handles this once so downstream agents don’t.

# agents/data_agent.py
import os
import time
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional
from autogen_core.tools import FunctionTool
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_core.model_context import BufferedChatCompletionContext
from autogen_ext.models.openai import OpenAIChatCompletionClient

# ---- Configuration ----
CACHE_TTL_SECONDS = 30
MAX_RETRIES = 3
RETRY_BACKOFF = 1.5

# ---- In-memory cache ----
_price_cache: dict[str, tuple[pd.DataFrame, float]] = {}

def _fetch_with_retry(ticker: str, period: str, interval: str) -> pd.DataFrame:
    """Fetch with exponential backoff. Raises on final failure."""
    last_exc = None
    for attempt in range(MAX_RETRIES):
        try:
            df = yf.download(
                tickers=ticker,
                period=period,
                interval=interval,
                progress=False,
                auto_adjust=True,
                prepost=False,
                threads=False,
            )
            if df.empty:
                raise ValueError(f"No data returned for {ticker}")
            # Flatten MultiIndex columns if present
            if isinstance(df.columns, pd.MultiIndex):
                df.columns = df.columns.get_level_values(0)
            return df
        except Exception as e:
            last_exc = e
            wait = RETRY_BACKOFF ** attempt
            time.sleep(wait)
    raise RuntimeError(f"Failed to fetch {ticker} after {MAX_RETRIES} attempts: {last_exc}")

def get_market_data(ticker: str, period: str = "1d", interval: str = "1m") -> str:
    """
    Fetch OHLCV data for a ticker. Returns JSON string for LLM consumption.
    Caches for CACHE_TTL_SECONDS to avoid hammering the provider.
    """
    now = time.time()
    cache_key = f"{ticker}:{period}:{interval}"
    
    if cache_key in _price_cache:
        df, cached_at = _price_cache[cache_key]
        if now - cached_at < CACHE_TTL_SECONDS:
            return df.tail(50).to_json(orient="records", date_format="iso")
    
    df = _fetch_with_retry(ticker, period, interval)
    _price_cache[cache_key] = (df, now)
    return df.tail(50).to_json(orient="records", date_format="iso")

def get_current_price(ticker: str) -> str:
    """Convenience tool: latest close price only."""
    df_json = get_market_data(ticker, period="1d", interval="1m")
    df = pd.read_json(df_json, orient="records")
    if df.empty:
        return f"ERROR: No data for {ticker}"
    latest = df.iloc[-1]
    return f"{ticker}: ${latest['Close']:.2f} as of {latest.name.isoformat()}"

# ---- Tool registration ----
market_data_tool = FunctionTool(get_market_data, description="Fetch OHLCV bars for a ticker. Args: ticker (str), period (str, default '1d'), interval (str, default '1m'). Returns JSON array of bars.")
current_price_tool = FunctionTool(get_current_price, description="Get the latest price for a ticker. Arg: ticker (str). Returns formatted string.")

# ---- Agent definition ----
def create_data_agent(model_client: OpenAIChatCompletionClient) -> AssistantAgent:
    return AssistantAgent(
        name="DataAgent",
        model_client=model_client,
        tools=[market_data_tool, current_price_tool],
        system_message=(
            "You are a market data specialist. Your ONLY job is to fetch and return raw market data. "
            "Never compute indicators, never give opinions. When asked for data, call the appropriate tool "
            "and return the raw JSON or formatted string exactly as the tool provides it. "
            "If a tool errors, return the error message verbatim."
        ),
        model_context=BufferedChatCompletionContext(max_tokens=8000),
    )

Checkpoint — Run a quick smoke test:

# test_data_agent.py
import asyncio
from dotenv import load_dotenv
from autogen_ext.models.openai import OpenAIChatCompletionClient
from agents.data_agent import create_data_agent

load_dotenv()

async def main():
    client = OpenAIChatCompletionClient(
        model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
        base_url=os.getenv("OPENAI_BASE_URL"),
        api_key=os.getenv("OPENAI_API_KEY"),
    )
    agent = create_data_agent(client)
    
    # Direct tool call bypasses LLM for speed
    from agents.data_agent import get_current_price
    print(get_current_price("AAPL"))
    print(get_current_price("MSFT"))
    
    # Via agent (tests tool routing)
    result = await agent.on_messages([TextMessage(content="Get current price for NVDA", source="user")], cancellation_token=None)
    print(result.chat_message.content)

if __name__ == "__main__":
    asyncio.run(main())

Expected output (prices will differ):

AAPL: $189.42 as of 2025-01-15T19:59:00
MSFT: $421.33 as of 2025-01-15T19:59:00
NVDA: $134.56 as of 2025-01-15T19:59:00

Step 2: Analysis agent — pure computation, no I/O

The analysis agent receives DataFrames as JSON, computes indicators, returns structured results. No network calls, no opinions — just math.

# agents/analysis_agent.py
import json
import pandas as pd
import numpy as np
from autogen_core.tools import FunctionTool
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.model_context import BufferedChatCompletionContext

def compute_indicators(data_json: str) -> str:
    """
    Compute RSI(14), MACD(12,26,9), Bollinger Bands(20,2) from OHLCV JSON.
    Returns JSON with latest values and signals.
    """
    try:
        df = pd.read_json(data_json, orient="records")
        if df.empty or len(df) < 26:
            return json.dumps({"error": "Insufficient data for indicators (need 26+ bars)"})
        
        # Ensure datetime index
        if "Datetime" in df.columns:
            df["Datetime"] = pd.to_datetime(df["Datetime"])
            df.set_index("Datetime", inplace=True)
        elif not isinstance(df.index, pd.DatetimeIndex):
            df.index = pd.to_datetime(df.index)
        
        close = df["Close"].astype(float)
        
        # RSI(14)
        delta = close.diff()
        gain = delta.where(delta > 0, 0).rolling(14).mean()
        loss = -delta.where(delta < 0, 0).rolling(14).mean()
        rs = gain / loss.replace(0, np.nan)
        rsi = 100 - (100 / (1 + rs))
        
        # MACD
        ema12 = close.ewm(span=12, adjust=False).mean()
        ema26 = close.ewm(span=26, adjust=False).mean()
        macd_line = ema12 - ema26
        signal_line = macd_line.ewm(span=9, adjust=False).mean()
        histogram = macd_line - signal_line
        
        # Bollinger Bands(20,2)
        sma20 = close.rolling(20).mean()
        std20 = close.rolling(20).std()
        upper = sma20 + 2 * std20
        lower = sma20 - 2 * std20
        
        latest = {
            "timestamp": df.index[-1].isoformat(),
            "close": float(close.iloc[-1]),
            "rsi": float(rsi.iloc[-1]) if not np.isnan(rsi.iloc[-1]) else None,
            "macd": {
                "macd": float(macd_line.iloc[-1]),
                "signal": float(signal_line.iloc[-1]),
                "histogram": float(histogram.iloc[-1]),
            },
            "bollinger": {
                "upper": float(upper.iloc[-1]),
                "middle": float(sma20.iloc[-1]),
                "lower": float(lower.iloc[-1]),
            },
            "signals": {
                "rsi_overbought": bool(rsi.iloc[-1] > 70) if not np.isnan(rsi.iloc[-1]) else False,
                "rsi_oversold": bool(rsi.iloc[-1] < 30) if not np.isnan(rsi.iloc[-1]) else False,
                "macd_bullish_cross": bool(macd_line.iloc[-1] > signal_line.iloc[-1] and macd_line.iloc[-2] <= signal_line.iloc[-2]),
                "macd_bearish_cross": bool(macd_line.iloc[-1] < signal_line.iloc[-1] and macd_line.iloc[-2] >= signal_line.iloc[-2]),
                "price_above_upper_bb": bool(close.iloc[-1] > upper.iloc[-1]),
                "price_below_lower_bb": bool(close.iloc[-1] < lower.iloc[-1]),
            }
        }
        return json.dumps(latest, indent=2)
    except Exception as e:
        return json.dumps({"error": f"Indicator computation failed: {str(e)}"})

indicators_tool = FunctionTool(compute_indicators, description="Compute technical indicators from OHLCV JSON. Arg: data_json (str). Returns JSON with RSI, MACD, Bollinger Bands, and boolean signals.")

def create_analysis_agent(model_client: OpenAIChatCompletionClient) -> AssistantAgent:
    return AssistantAgent(
        name="AnalysisAgent",
        model_client=model_client,
        tools=[indicators_tool],
        system_message=(
            "You are a quantitative analysis engine. You receive raw OHLCV data as JSON and MUST call "
            "the compute_indicators tool. Return the tool's JSON output exactly. Do not interpret, "
            "summarize, or add commentary. If the input is not valid JSON or lacks required columns, "
            "return the tool's error message."
        ),
        model_context=BufferedChatCompletionContext(max_tokens=8000),
    )

Checkpoint — Test the analysis agent in isolation:

# test_analysis_agent.py
import asyncio
import json
from dotenv import load_dotenv
from autogen_ext.models.openai import OpenAIChatCompletionClient
from agents.analysis_agent import create_analysis_agent, compute_indicators

load_dotenv()

async def main():
    # Direct function test
    sample_bars = [
        {"Datetime": "2025-01-15T19:00:00", "Open": 189.0, "High": 189.5, "Low": 188.8, "Close": 189.2, "Volume": 100000},
        {"Datetime": "2025-01-15T19:01:00", "Open": 189.2, "High": 189.7, "Low": 189.0, "Close": 189.5, "Volume": 120000},
        # ... add 24 more bars for minimum 26
    ]
    # Pad with synthetic data for demo
    import pandas as pd
    import numpy as np
    base = 189.0
    bars = []
    for i in range(30):
        base += np.random.normal(0, 0.15)
        bars.append({
            "Datetime": f"2025-01-15T{19:02d}:{i:02d}:00".replace("19:02d", f"{19+i//60:02d}"),
            "Open": base, "High": base+0.2, "Low": base-0.2, "Close": base, "Volume": 100000
        })
    print(compute_indicators(json.dumps(bars)))
    
    # Via agent
    client = OpenAIChatCompletionClient(
        model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
        base_url=os.getenv("OPENAI_BASE_URL"),
        api_key=os.getenv("OPENAI_API_KEY"),
    )
    agent = create_analysis_agent(client)
    result = await agent.on_messages([
        TextMessage(content=json.dumps(bars), source="user")
    ], cancellation_token=None)
    print(result.chat_message.content)

if __name__ == "__main__":
    asyncio.run(main())

Expected output (values will vary):

{
  "timestamp": "2025-01-15T19:29:00",
  "close": 189.42,
  "rsi": 52.3,
  "macd": {"macd": 0.15, "signal": 0.12, "histogram": 0.03},
  "bollinger": {"upper": 190.1, "middle": 189.3, "lower": 188.5},
  "signals": {
    "rsi_overbought": false,
    "rsi_oversold": false,
    "macd_bullish_cross": true,
    "macd_bearish_cross": false,
    "price_above_upper_bb": false,
    "price_below_lower_bb": false
  }
}

Step 3: Coordinator agent — synthesis and routing

The coordinator owns the user conversation. It decides which specialist to call, merges results, and produces a final answer with citations.

# agents/coordinator_agent.py
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.model_context import BufferedChatCompletionContext
from agents.data_agent import create_data_agent
from agents.analysis_agent import create_analysis_agent

COORDINATOR_SYSTEM = """
You are the lead analyst coordinating a market research team. You have two specialists:

1. DataAgent — Fetches raw OHLCV data. Call it with: "Fetch data for [TICKER] [PERIOD] [INTERVAL]"
2. AnalysisAgent — Computes technical indicators from DataAgent's JSON output. Call it with the raw JSON.

Workflow for a user query like "Should I buy AAPL?":
1. Ask DataAgent for 1d/1m data (intraday) AND 3mo/1d data (context).
2. Pass BOTH JSON responses to AnalysisAgent separately.
3. Synthesize: intraday signals for timing, daily signals for trend.
4. Respond with: Recommendation (Buy/Hold/Sell), Confidence (0-100), Key Evidence (bulleted, citing specific indicator values), Risk Factors.

Rules:
- Never invent data. If a specialist returns an error, report it.
- Cite specific numbers: "RSI(14)=52.3", "MACD histogram turned positive at 19:15".
- Keep responses under 300 words. No fluff.
"""

def create_coordinator_agent(model_client: OpenAIChatCompletionClient) -> AssistantAgent:
    return AssistantAgent(
        name="Coordinator",
        model_client=model_client,
        system_message=COORDINATOR_SYSTEM,
        model_context=BufferedChatCompletionContext(max_tokens=16000),
    )

def create_team(model_client: OpenAIChatCompletionClient) -> SelectorGroupChat:
    data_agent = create_data_agent(model_client)
    analysis_agent = create_analysis_agent(model_client)
    coordinator = create_coordinator_agent(model_client)
    
    # SelectorGroupChat uses the coordinator's system prompt to route
    # We add explicit termination conditions
    termination = MaxMessageTermination(max_messages=15) | TextMentionTermination("FINAL ANSWER")
    
    return SelectorGroupChat(
        participants=[data_agent, analysis_agent, coordinator],
        model_client=model_client,
        termination_condition=termination,
        selector_prompt=(
            "Select the next agent. Options: DataAgent, AnalysisAgent, Coordinator. "
            "Coordinator speaks first and last. DataAgent fetches data. AnalysisAgent computes indicators. "
            "Return ONLY the agent name."
        ),
    )

Step 4: Wiring it together — the entry point

# main.py
import asyncio
import os
from dotenv import load_dotenv
from autogen_ext.models.openai import OpenAIChatCompletionClient
from agents.coordinator_agent import create_team

load_dotenv()

async def run_analysis(query: str):
    client = OpenAIChatCompletionClient(
        model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
        base_url=os.getenv("OPENAI_BASE_URL"),
        api_key=os.getenv("OPENAI_API_KEY"),
    )
    
    team = create_team(client)
    
    # Stream the conversation for visibility
    async for message in team.run_stream(task=query):
        print(f"[{message.source}] {message.content[:200]}...")
        print("---")
    
    # Get final result
    result = await team.run(task=query)
    return result

if __name__ == "__main__":
    # Example queries
    queries = [
        "Should I buy AAPL right now? Intraday perspective.",
        "Give me a technical read on NVDA for the next few days.",
        "Compare MSFT and GOOGL intraday momentum.",
    ]
    
    for q in queries:
        print(f"\n{'='*60}")
        print(f"QUERY: {q}")
        print(f"{'='*60}\n")
        asyncio.run(run_analysis(q))

Checkpoint — Run python main.py. Expected flow (abbreviated):

[Coordinator] Fetching data for AAPL...
---
[DataAgent] [{"Datetime":"2025-01-15T19:00:00","Open":189.0,"High":189.5,"Low":188.8,"Close":189.2,"Volume":100000},...]
---
[Coordinator] Computing indicators on intraday data...
---
[AnalysisAgent] {"timestamp":"2025-01-15T19:59:00","close":189.42,"rsi":52.3,"macd":{"macd":0.15,"signal":0.12,"histogram":0.03},...}
---
[Coordinator] Fetching daily context for AAPL...
---
[DataAgent] [{"Datetime":"2024-10-15","Open":175.0,...},...]
---
[AnalysisAgent] {"timestamp":"2025-01-15","close":189.42,"rsi":58.7,"macd":{"macd":2.1,"signal":1.8,"histogram":0.3},...}
---
[Coordinator] FINAL ANSWER
Recommendation: Buy (Intraday) / Hold (Swing)
Confidence: 65
Key Evidence:
- Intraday RSI(14)=52.3 (neutral), MACD histogram turned positive at 19:15 EST
- Daily RSI(14)=58.7, MACD bullish crossover 3 sessions ago, price above 20-day SMA
- Volume 1.2x 20-day average on up-moves
Risk Factors:
- Earnings in 12 days — gamma risk increases
- Nasdaq breadth weakening; check QQQ correlation

Step 5: Production hardening

Three things separate a demo from a system you trust with capital.

Structured output enforcement

LLMs drift. Force the coordinator to emit valid JSON using a response format schema:

# Add to coordinator creation
from autogen_core.models import ChatCompletionClient
from pydantic import BaseModel
from typing import Literal

class Recommendation(BaseModel):
    action: Literal["Buy", "Sell", "Hold"]
    horizon: Literal["intraday", "swing", "position"]
    confidence: int  # 0-100
    evidence: list[str]
    risks: list[str]
    timestamp: str

# In create_coordinator_agent, pass:
# response_format=Recommendation
# Then parse: result = Recommendation.model_validate_json(message.content)

Observability hooks

Log every tool call, latency, and token count. AutoGen emits events — subscribe to them:

from autogen_core import Event, event_logger
from autogen_agentchat.events import ToolCallEvent, ToolCallResultEvent

@event_logger
async def log_tools(event: Event):
    if isinstance(event, ToolCallEvent):
        print(f"TOOL CALL: {event.tool_name} args={event.arguments}")
    elif isinstance(event, ToolCallResultEvent):
        print(f"TOOL RESULT: {event.tool_name} latency_ms={event.latency_ms} success={event.success}")

Fallback data provider

yfinance is unofficial and rate-limits aggressively. Swap the data agent’s backend without changing its interface:

# In data_agent.py, replace _fetch_with_retry:
def _fetch_with_retry(ticker: str, period: str, interval: str) -> pd.DataFrame:
    # Try primary
    try:
        return _fetch_yfinance(ticker, period, interval)
    except Exception:
        pass
    # Fallback to Polygon, Alpha Vantage, or your n4n.ai-routed provider
    return _fetch_polygon(ticker, period, interval)

The coordinator and analysis agents don’t care — they only see JSON.

Running continuously

For a real-time loop, wrap the team in a scheduler:

# scheduler.py
import asyncio
import signal
from main import run_analysis

WATCHLIST = ["AAPL", "NVDA", "MSFT", "GOOGL", "META"]
INTERVAL_SECONDS = 60

async def monitor():
    while True:
        for symbol in WATCHLIST:
            query = f"Intraday scalp signal for {symbol} — Buy/Sell/Hold with confidence."
            try:
                await run_analysis(query)
            except Exception as e:
                print(f"Error on {symbol}: {e}")
        await asyncio.sleep(INTERVAL_SECONDS)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, loop.stop)
    loop.run_until_complete(monitor())

Deploy this as a systemd service or Kubernetes cronjob. Point Slack/PagerDuty alerts at the FINAL ANSWER events where confidence > 80 and action != Hold.

What we didn’t cover (and why)

• Order execution — This is analysis only. Execution requires broker APIs, idempotency keys, and risk checks that belong in a separate service.
• Fundamental data — Earnings, filings, news need different pipelines. Add a FundamentalAgent with the same pattern.
• Backtesting — Replay historical bars through the same agent graph. The tool interface makes this trivial: swap get_market_data for a CSV reader.
• Multi-timeframe convergence — The coordinator already pulls 1m and 1d. Add 1w/1M for position trades.

Summary

You now have a real-time market data agent autogen system with:

• DataAgent: Cached, retried, rate-limit-aware market data fetch
• AnalysisAgent: Pure-pandas indicator computation, deterministic JSON output
• Coordinator: Structured synthesis with citations, confidence scoring, risk flags
• SelectorGroupChat: Explicit routing, bounded turns, clean termination
• Production hooks: Structured output, event logging, fallback providers

The entire graph runs locally. Swap the model client to any OpenAI-compatible endpoint — including n4n.ai for automatic fallback across 240+ models — without changing agent code. The tools are the contract; the LLM is just the router.