Build an AI Hedge Fund Research Pipeline

pip install requests python-dotenv
export SWARMS_API_KEY="your-api-key-here"

import json
import os

import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}

PORTFOLIO_MANAGER_PROMPT = (
    "You are a Portfolio Manager at a long/short equity hedge fund. "
    "Review the briefs from your Fundamentals, Technicals, and Macro analysts. "
    "Issue a single decision in this exact format:\n\n"
    "TICKER: <symbol>\n"
    "CALL: <BUY | SELL | HOLD>\n"
    "CONVICTION: <LOW | MEDIUM | HIGH>\n"
    "KEY SIGNAL: <one sentence on the single strongest driver of the call>\n"
    "RISK: <one sentence on the primary risk to the thesis>\n\n"
    "Be decisive. Do not hedge across all three analysts — pick the dominant signal."
)

FUNDAMENTALS_PROMPT = (
    "You are a Fundamental Equity Analyst. Given a ticker, write a brief covering: "
    "revenue growth trajectory, margin profile, FCF generation, balance sheet health, "
    "and the single most important catalyst over the next two quarters. "
    "Keep it under 200 words. Be specific."
)

TECHNICALS_PROMPT = (
    "You are a Technical Analyst. Given a ticker, write a brief covering: "
    "current trend regime, key support and resistance levels, momentum indicators "
    "(RSI, MACD), volume profile, and a near-term target with stop. "
    "Keep it under 200 words. Be specific."
)

MACRO_PROMPT = (
    "You are a Macro Analyst. Given a ticker, write a brief covering: "
    "sector positioning vs. the current rate regime, FX and commodity sensitivities, "
    "policy and regulatory tailwinds or headwinds, and how the name behaves in a "
    "risk-off rotation. Keep it under 200 words. Be specific."
)


def build_swarm_for_ticker(ticker: str) -> dict:
    return {
        "name": f"AI Hedge Fund Research — {ticker}",
        "description": "Portfolio Manager directing Fundamentals, Technicals, and Macro analysts.",
        "swarm_type": "HierarchicalSwarm",
        "max_loops": 1,
        "task": (
            f"Produce a same-day research note for {ticker}. "
            f"Each analyst writes their brief, then the Portfolio Manager issues a final call."
        ),
        "agents": [
            {
                "agent_name": "Portfolio Manager",
                "description": "Director — synthesizes analyst briefs into a single call.",
                "system_prompt": PORTFOLIO_MANAGER_PROMPT,
                "model_name": "gpt-4.1",
                "role": "coordinator",
                "max_loops": 1,
                "max_tokens": 4096,
                "temperature": 0.2,
            },
            {
                "agent_name": "Fundamentals Analyst",
                "description": "Earnings, margins, balance sheet, catalysts.",
                "system_prompt": FUNDAMENTALS_PROMPT,
                "model_name": "gpt-4.1",
                "role": "worker",
                "max_loops": 1,
                "max_tokens": 2048,
                "temperature": 0.4,
            },
            {
                "agent_name": "Technicals Analyst",
                "description": "Trend, levels, momentum, volume.",
                "system_prompt": TECHNICALS_PROMPT,
                "model_name": "gpt-4.1",
                "role": "worker",
                "max_loops": 1,
                "max_tokens": 2048,
                "temperature": 0.4,
            },
            {
                "agent_name": "Macro Analyst",
                "description": "Rates, FX, commodities, policy.",
                "system_prompt": MACRO_PROMPT,
                "model_name": "gpt-4.1",
                "role": "worker",
                "max_loops": 1,
                "max_tokens": 2048,
                "temperature": 0.4,
            },
        ],
    }

def run_single_ticker(ticker: str) -> dict:
    payload = build_swarm_for_ticker(ticker)
    response = requests.post(
        f"{BASE_URL}/v1/swarm/completions",
        headers=headers,
        json=payload,
        timeout=300,
    )
    response.raise_for_status()
    return response.json()


result = run_single_ticker("NVDA")

for output in result.get("output", []):
    print("=" * 60)
    print(output["role"])
    print("=" * 60)
    content = output["content"]
    if isinstance(content, list):
        content = " ".join(str(c) for c in content)
    print(str(content)[:600])

print(f"\nTotal cost: ${result['usage']['billing_info']['total_cost']:.4f}")
print(f"Execution time: {result['execution_time']:.1f}s")

WATCHLIST = [
    "NVDA", "AAPL", "MSFT", "GOOGL", "META",
    "AMZN", "TSLA", "AMD", "AVGO", "CRM",
    "ORCL", "ADBE", "NFLX", "PYPL", "SHOP",
    "UBER", "COIN", "PLTR", "SNOW", "DDOG",
]


def run_watchlist_batch(tickers: list[str]) -> list[dict]:
    payload = [build_swarm_for_ticker(t) for t in tickers]
    response = requests.post(
        f"{BASE_URL}/v1/swarm/batch/completions",
        headers=headers,
        json=payload,
        timeout=900,
    )
    response.raise_for_status()
    return response.json()


results = run_watchlist_batch(WATCHLIST)

with open("morning_notes.jsonl", "w") as f:
    for ticker, result in zip(WATCHLIST, results):
        # Extract the Portfolio Manager's final call
        pm_call = next(
            (o["content"] for o in result.get("output", []) if "Portfolio Manager" in o.get("role", "")),
            "",
        )
        if isinstance(pm_call, list):
            pm_call = " ".join(str(c) for c in pm_call)
        f.write(json.dumps({"ticker": ticker, "call": pm_call}) + "\n")

total_cost = sum(r.get("usage", {}).get("billing_info", {}).get("total_cost", 0) for r in results)
print(f"Generated {len(results)} morning notes for ${total_cost:.2f}")