Building a Multi-Agent Trading System with Claude and TickAtlas

import requests
import json
from dataclasses import dataclass

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://tickatlas.com/v1"

@dataclass
class MarketSnapshot:
    symbol: str
    timeframe: str
    close: float
    rsi: float
    macd_hist: float
    ema_50: float
    ema_200: float
    atr: float
    stochastic_k: float

def fetch_snapshot(symbol: str, timeframe: str) -> MarketSnapshot:
    """Fetch all indicators needed for analysis."""
    indicators = ["RSI_14", "MACD", "EMA_50", "EMA_200", "ATR_14", "STOCHASTIC"]
    values = {}
    ohlcv = {}

    for ind in indicators:
        resp = requests.get(
            f"{BASE_URL}/indicator",
            headers={"X-API-Key": API_KEY},
            params={"symbol": symbol, "indicator": ind, "timeframe": timeframe},
        )
        data = resp.json()["data"]
        values[ind] = data["values"]
        if not ohlcv:
            ohlcv = data["ohlcv"]

    return MarketSnapshot(
        symbol=symbol,
        timeframe=timeframe,
        close=ohlcv["close"],
        rsi=values["RSI_14"]["rsi"],
        macd_hist=values["MACD"]["histogram"],
        ema_50=values["EMA_50"]["value"],
        ema_200=values["EMA_200"]["value"],
        atr=values["ATR_14"]["atr"],
        stochastic_k=values["STOCHASTIC"]["k"],
    )

import anthropic

claude = anthropic.Anthropic()

ANALYST_PROMPT = """You are a technical analyst. Based on the data below,
identify any trade setup. Respond in JSON.

Symbol: {symbol} ({timeframe})
Close: {close}
RSI(14): {rsi}
MACD Histogram: {macd_hist}
EMA 50: {ema_50}
EMA 200: {ema_200}
Stochastic %K: {stochastic_k}

Respond with:
{{"setup_found": true/false, "direction": "long"/"short"/null,
  "reasoning": "...", "confidence": "low"/"medium"/"high",
  "entry_zone": [low, high]}}"""

def analyst_agent(snapshot: MarketSnapshot) -> dict:
    response = claude.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        messages=[{"role": "user", "content": ANALYST_PROMPT.format(
            symbol=snapshot.symbol, timeframe=snapshot.timeframe,
            close=snapshot.close, rsi=snapshot.rsi,
            macd_hist=snapshot.macd_hist, ema_50=snapshot.ema_50,
            ema_200=snapshot.ema_200, stochastic_k=snapshot.stochastic_k,
        )}],
    )
    return json.loads(response.content[0].text)

RISK_PROMPT = """You are a risk manager. Given this trade setup and account
info, determine position sizing and stop placement.

Trade Setup: {setup}
ATR(14): {atr}
Account Balance: \${balance}
Max Risk Per Trade: {risk_pct}%

Respond with:
{{"approved": true/false, "position_size_lots": X.XX,
  "stop_loss": X.XXXXX, "take_profit": X.XXXXX,
  "risk_amount": X.XX, "risk_reward_ratio": X.X,
  "rejection_reason": null/"reason"}}"""

def risk_manager_agent(
    setup: dict, snapshot: MarketSnapshot,
    balance: float = 10000, risk_pct: float = 1.0
) -> dict:
    response = claude.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        messages=[{"role": "user", "content": RISK_PROMPT.format(
            setup=json.dumps(setup), atr=snapshot.atr,
            balance=balance, risk_pct=risk_pct,
        )}],
    )
    return json.loads(response.content[0].text)

def run_trading_pipeline(symbol: str, timeframe: str) -> dict:
    """Run the full multi-agent trading pipeline."""
    # Step 1: Fetch market data
    snapshot = fetch_snapshot(symbol, timeframe)

    # Step 2: Analyst identifies setup
    setup = analyst_agent(snapshot)
    if not setup["setup_found"]:
        return {"action": "NO_TRADE", "reason": "No setup found"}

    # Step 3: Risk manager evaluates
    risk = risk_manager_agent(setup, snapshot)
    if not risk["approved"]:
        return {"action": "REJECTED", "reason": risk["rejection_reason"]}

    # Step 4: Return for human review (never auto-execute)
    return {
        "action": "PENDING_REVIEW",
        "symbol": symbol,
        "direction": setup["direction"],
        "confidence": setup["confidence"],
        "entry_zone": setup["entry_zone"],
        "stop_loss": risk["stop_loss"],
        "take_profit": risk["take_profit"],
        "position_size": risk["position_size_lots"],
        "risk_amount": risk["risk_amount"],
        "reasoning": setup["reasoning"],
    }

# Run on multiple symbols
for sym in ["EURUSD", "GBPUSD", "XAUUSD"]:
    result = run_trading_pipeline(sym, "H4")
    print(f"\n{sym}: {result['action']}")
    if result["action"] == "PENDING_REVIEW":
        print(f"  Direction: {result['direction']}")
        print(f"  Size: {result['position_size']} lots")
        print(f"  SL: {result['stop_loss']} / TP: {result['take_profit']}")