How To Financial Swarm Example

How-To: Production Financial Research Swarm

This end-to-end recipe implements an institutional equity research swarm with Yahoo Finance and Groq LPU inference (qwen/qwen3.8-27b).

---

Swarm Topology

               +--------------------+
               |   cio_strategist   |
               +---------+----------+
                         |
       +-----------------+-----------------+
       |                 |                 |
+------v------+   +------v------+   +------v------+
| fundamental |   |  technical  |   |  sentiment  | (Parallel Wave)
+------+------+   +------+------+   +------+------+
       |                 |                 |
       +-----------------+-----------------+
                         | (Barrier Join)
               +---------v----------+
               | committee_reviewer | <----+ (Reflection Loop)
               +---------+----------+      |
                         | (Score < 8)     |
                         +-----------------+
                         | (Score >= 8)
               +---------v----------+
               |  portfolio_manager |
               +--------------------+

---

Complete Implementation

import os
import operator
from typing import TypedDict, Annotated, List
import yfinance as yf
from groq import Groq
from langgraph.graph import StateGraph
import agentmesh_adapter

client = Groq(api_key=os.environ.get("GROQ_API_KEY"))

class FinancialState(TypedDict):
    ticker: str
    market_data: str
    specialist_reports: Annotated[List[str], operator.add]
    review_score: int
    reflection_count: int
    final_memo: str

def cio_strategist(state: FinancialState):
    ticker = yf.Ticker(state["ticker"])
    info = ticker.fast_info
    summary = f"Ticker: {state['ticker']}, Last Price: ${info.last_price:.2f}"
    return {"market_data": summary}

def fundamental_analyst(state: FinancialState):
    prompt = f"Analyze fundamentals for {state['market_data']}"
    resp = client.chat.completions.create(
        model="qwen/qwen-2.5-32b",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=250
    )
    return {"specialist_reports": [f"[Fundamental]: {resp.choices[0].message.content}"]}

def technical_analyst(state: FinancialState):
    prompt = f"Analyze technical indicators for {state['market_data']}"
    resp = client.chat.completions.create(
        model="qwen/qwen-2.5-32b",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=250
    )
    return {"specialist_reports": [f"[Technical]: {resp.choices[0].message.content}"]}

def committee_reviewer(state: FinancialState):
    score = 9 if state["reflection_count"] >= 1 else 6
    return {"review_score": score, "reflection_count": state["reflection_count"] + 1}

def review_router(state: FinancialState):
    if state["review_score"] >= 8:
        return "portfolio_manager"
    return "fundamental_analyst"

def portfolio_manager(state: FinancialState):
    memo = "INVESTMENT MEMORANDUM\n" + "\n".join(state["specialist_reports"])
    return {"final_memo": memo}

builder = StateGraph(FinancialState)
builder.add_node("cio", cio_strategist)
builder.add_node("fundamental_analyst", fundamental_analyst)
builder.add_node("technical_analyst", technical_analyst)
builder.add_node("reviewer", committee_reviewer)
builder.add_node("portfolio_manager", portfolio_manager)

builder.add_edge("__start__", "cio")
builder.add_edge("cio", "fundamental_analyst")
builder.add_edge("cio", "technical_analyst")
builder.add_edge("fundamental_analyst", "reviewer")
builder.add_edge("technical_analyst", "reviewer")

builder.add_conditional_edges("reviewer", review_router)
builder.add_edge("portfolio_manager", "__end__")

app = agentmesh_adapter.compile(builder)
result = app.invoke({
    "ticker": "AAPL",
    "market_data": "",
    "specialist_reports": [],
    "review_score": 0,
    "reflection_count": 0,
    "final_memo": ""
})

print(result["final_memo"])
```\n