Building Autonomous Trading Agents with LLMs

import time
import json
from datetime import datetime

class TradingAgent:
    def __init__(self, api_key: str, llm_client, symbols: list[str]):
        self.api_key = api_key
        self.llm = llm_client
        self.symbols = symbols
        self.memory = []  # Track past observations and decisions
        self.check_interval = 300  # 5 minutes

    def run(self):
        """Main agent loop — runs indefinitely."""
        print(f"[{datetime.utcnow()}] Agent started. Watching {self.symbols}")

        while True:
            try:
                observations = self.observe()
                analysis = self.think(observations)
                self.act(analysis)
            except Exception as e:
                print(f"Agent error: {e}")

            time.sleep(self.check_interval)

import requests

CLAW_BASE = "https://tickatlas.com/v1"

def observe(self) -> list[dict]:
    """Fetch indicators for all watched symbols."""
    observations = []
    headers = {"X-API-Key": self.api_key}

    for symbol in self.symbols:
        resp = requests.get(f"{CLAW_BASE}/indicators", params={
            "symbol": symbol,
            "timeframe": "H1",
            "indicators": "RSI_14,MACD,BBANDS_20,ADX_14,ATR_14"
        }, headers=headers)

        if resp.status_code == 200:
            data = resp.json()["data"]
            observations.append({
                "symbol": symbol,
                "indicators": data["indicators"],
                "price": data.get("price", {}),
                "timestamp": data["timestamp"]
            })

    return observations

def think(self, observations: list[dict]) -> dict:
    """Use LLM to analyze observations and decide on actions."""
    # Build context from memory
    recent_memory = self.memory[-10:]  # Last 10 observations

    prompt = f"""You are an autonomous trading analyst agent.

Current observations:
{json.dumps(observations, indent=2)}

Recent memory (previous analyses):
{json.dumps(recent_memory, indent=2)}

Analyze each symbol and return a JSON object with this structure:
{{
  "signals": [
    {{
      "symbol": "EURUSD",
      "action": "ALERT_BUY" | "ALERT_SELL" | "WATCH" | "NO_ACTION",
      "confidence": 0.0-1.0,
      "reasoning": "brief explanation",
      "key_levels": {{"support": 1.0820, "resistance": 1.0890}}
    }}
  ],
  "market_summary": "One paragraph overview of current conditions"
}}

Rules:
- Only signal ALERT_BUY/ALERT_SELL when confidence > 0.7
- Consider what has changed since the last observation
- Flag any divergences between indicators
- Note when ADX < 20 (no trend) — avoid trend signals in ranging markets"""

    response = self.llm.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )

    # Parse the structured response
    text = response.content[0].text
    analysis = json.loads(text)

    # Save to memory
    self.memory.append({
        "timestamp": datetime.utcnow().isoformat(),
        "observations": observations,
        "analysis": analysis
    })

    return analysis

def act(self, analysis: dict):
    """Execute the agent's decisions."""
    for signal in analysis.get("signals", []):
        if signal["action"] in ("ALERT_BUY", "ALERT_SELL"):
            self.send_alert(signal)

        elif signal["action"] == "WATCH":
            print(f"[WATCH] {signal['symbol']}: {signal['reasoning']}")

def send_alert(self, signal: dict):
    """Send alert via Telegram, email, or webhook."""
    msg = (
        f"{'BUY' if 'BUY' in signal['action'] else 'SELL'} Alert: "
        f"{signal['symbol']}\n"
        f"Confidence: {signal['confidence']:.0%}\n"
        f"Reason: {signal['reasoning']}\n"
        f"Support: {signal['key_levels']['support']}\n"
        f"Resistance: {signal['key_levels']['resistance']}"
    )
    print(f"[ALERT] {msg}")
    # Send to Telegram, Slack, email, etc.