Build an AI trading bot with live market data and guardrails

python3 -m venv .venv && source .venv/bin/activate
pip install "anthropic>=1.8" "requests>=2.31"
export TICKERLAYER_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"

"""Paper-trading agent: Claude reads market data through the TickerLayer MCP
server, and plain Python decides whether a proposed order may touch the paper
book. Educational only. Nothing here sends an order to a broker."""

import json
import os
import sys
import time
from pathlib import Path

import anthropic
import requests

TL_KEY = os.environ["TICKERLAYER_API_KEY"]
REST = "https://api.tickerlayer.com"
MCP_URL = "https://mcp.tickerlayer.com/mcp"
MODEL = os.environ.get("CLAUDE_MODEL", "claude-opus-5")
LEDGER = Path("paper_ledger.json")

# Your risk settings. Example values to tune, not recommendations.
MAX_QUOTE_AGE_S = {"stocks": 120, "crypto": 30}
MAX_SPREAD_BPS = 30.0
MAX_ORDER_USD = 1_000.0
MAX_TURNS = 10

http = requests.Session()
http.headers["x-api-key"] = TL_KEY
claude = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY

SYSTEM = (
    "You are a cautious paper-trading assistant. Use the TickerLayer tools to check "
    "market status, the latest quote and recent daily bars for each symbol on the "
    "watchlist. Then either propose one paper order with propose_paper_order or "
    "propose nothing. Quote prices only from tool results and give their timestamps. "
    "If a market is closed or data looks stale, say so and do not trade."
)

PROPOSE_ORDER = {
    "name": "propose_paper_order",
    "description": (
        "Propose one paper order. Deterministic guardrails re-check market status, "
        "quote freshness, spread and size, then fill it on a paper ledger or reject "
        "it with a reason. Nothing is sent to a broker."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "asset_class": {"type": "string", "enum": ["stocks", "crypto"]},
            "symbol": {"type": "string"},
            "side": {"type": "string", "enum": ["buy", "sell"]},
            "notional_usd": {"type": "number"},
            "reason": {"type": "string"},
        },
        "required": ["asset_class", "symbol", "side", "notional_usd", "reason"],
        "additionalProperties": False,
    },
    "strict": True,
}

# Read-only allowlist: the model sees only the tools this bot needs.
TOOLSET = {
    "type": "mcp_toolset",
    "mcp_server_name": "tickerlayer",
    "default_config": {"enabled": False},
    "configs": {name: {"enabled": True} for name in (
        "get_market_status", "get_quote", "get_snapshot", "get_history", "list_symbols"
    )},
}


def tl_get(path, params=None):
    """GET a TickerLayer REST path, honouring Retry-After on a per-second 429."""
    for _ in range(3):
        resp = http.get(REST + path, params=params, timeout=10)
        if resp.status_code == 429 and "Retry-After" in resp.headers:
            time.sleep(float(resp.headers["Retry-After"]))
            continue
        resp.raise_for_status()
        return resp.json()
    resp.raise_for_status()
    return resp.json()


def load_book():
    if LEDGER.exists():
        return json.loads(LEDGER.read_text())
    return {"cash": 10_000.0, "positions": {}, "fills": []}


def check_order(order):
    """Return (ok, detail, price). Every check re-reads the market itself."""
    asset, symbol = order["asset_class"], order["symbol"].strip().upper()

    # 1. Regular session only. is_open is also true in pre- and post-market.
    params = {"asset": asset}
    if asset == "stocks":
        params["market"] = symbol.split(":")[0]
    status = tl_get("/markets/status", params)
    if status.get("status") != "open":
        return False, f"market {status.get('status')}, next open {status.get('next_open')}", None

    # 2. The quote must be fresh, uncrossed and tight.
    quote = tl_get(f"/{asset}/quote/{symbol}")
    bid, ask = float(quote["bid"]), float(quote["ask"])
    age_s = time.time() - quote["timestamp"] / 1000
    if age_s > MAX_QUOTE_AGE_S[asset]:
        return False, f"stale quote, {age_s:.0f}s old", None
    if bid <= 0 or ask <= bid:
        return False, f"unusable quote, bid {bid} ask {ask}", None
    spread_bps = (ask - bid) / ((ask + bid) / 2) * 10_000
    if spread_bps > MAX_SPREAD_BPS:
        return False, f"spread {spread_bps:.1f} bps is over the limit", None

    # 3. Size and inventory. No shorting, even on paper.
    notional = float(order["notional_usd"])
    if not 0 < notional <= MAX_ORDER_USD:
        return False, f"notional {notional} outside (0, {MAX_ORDER_USD}]", None
    price = ask if order["side"] == "buy" else bid
    held = load_book()["positions"].get(symbol, 0.0)
    if order["side"] == "sell" and notional / price > held + 1e-9:
        return False, f"sell of {notional / price:.6f} exceeds position {held:.6f}", None
    return True, f"quote inside the {MAX_QUOTE_AGE_S[asset]}s budget, spread {spread_bps:.2f} bps", price


def fill(order, price):
    book = load_book()
    symbol = order["symbol"].strip().upper()
    qty = float(order["notional_usd"]) / price
    signed = qty if order["side"] == "buy" else -qty
    book["cash"] -= signed * price
    book["positions"][symbol] = book["positions"].get(symbol, 0.0) + signed
    book["fills"].append({"ts": int(time.time() * 1000), "symbol": symbol,
                          "side": order["side"], "qty": qty, "price": price,
                          "reason": order["reason"]})
    LEDGER.write_text(json.dumps(book, indent=2))
    return qty


def handle_proposal(order, already_filled):
    if already_filled:
        return "rejected: one order per run", True
    try:
        ok, detail, price = check_order(order)
    except requests.HTTPError as exc:
        return f"rejected: data check failed ({exc.response.status_code})", True
    if not ok:
        print(f"  guard  REJECT {order['side']} {order['symbol']}: {detail}")
        return f"rejected: {detail}", False
    qty = fill(order, price)
    print(f"  guard  FILL {order['side']} {qty:.6f} {order['symbol']} @ {price} ({detail})")
    return f"filled on paper: {qty:.6f} @ {price}", False


def run(task):
    messages = [{"role": "user", "content": task}]
    filled = False
    for _ in range(MAX_TURNS):
        resp = claude.beta.messages.create(
            model=MODEL,
            max_tokens=16000,
            system=SYSTEM,
            betas=["mcp-client-2025-11-20"],
            mcp_servers=[{"type": "url", "url": MCP_URL, "name": "tickerlayer",
                          "authorization_token": TL_KEY}],
            tools=[TOOLSET, PROPOSE_ORDER],
            messages=messages,
        )
        messages.append({"role": "assistant", "content": resp.content})
        for block in resp.content:
            if block.type == "mcp_tool_use":
                print(f"  data   {block.name} {json.dumps(block.input)}")

        if resp.stop_reason == "refusal":
            print("model declined the request; holding")
            return
        if resp.stop_reason == "pause_turn":
            continue  # long server-side tool turn: send it back to resume
        if resp.stop_reason != "tool_use":
            print("\n" + "".join(b.text for b in resp.content if b.type == "text"))
            return

        results = []
        for block in resp.content:
            if block.type == "tool_use" and block.name == "propose_paper_order":
                text, is_error = handle_proposal(block.input, filled)
                filled = filled or text.startswith("filled")
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": text, "is_error": is_error})
        messages.append({"role": "user", "content": results})
    print(f"stopped after {MAX_TURNS} turns without a final answer")


if __name__ == "__main__":
    watchlist = sys.argv[1:] or ["US:KO", "US:JPM", "BTCUSD"]
    try:
        run("Watchlist: " + ", ".join(watchlist) + ". Decide on at most one paper order.")
    except anthropic.APIStatusError as exc:
        sys.exit(f"Claude API error {exc.status_code}: {exc.message}")
    except requests.RequestException as exc:
        sys.exit(f"TickerLayer request failed: {exc}")

python paper_agent.py US:KO US:JPM BTCUSD

  data   get_market_status {"asset_class": "stocks", "market": "US"}
  data   get_snapshot {"asset_class": "stocks", "symbol": "US:KO"}
  data   get_snapshot {"asset_class": "stocks", "symbol": "US:JPM"}
  data   get_market_status {"asset_class": "crypto"}
  data   get_snapshot {"asset_class": "crypto", "symbol": "BTCUSD"}
  guard  REJECT buy US:KO: market pre_market, next open 2026-09-28T13:30:00.000Z
  guard  FILL buy 0.006025 BTCUSD @ 82993.1 (quote inside the 30s budget, spread 0.00 bps)

US equities are in pre-market until 13:30 UTC, so the US:KO idea was rejected.
Bought 0.006025 BTCUSD on paper at 82,993.10 (quote at 10:54 UTC).

{
  "cash": 9500.0,
  "positions": { "BTCUSD": 0.006024597225552485 },
  "fills": [
    { "ts": 1790592856369, "symbol": "BTCUSD", "side": "buy",
      "qty": 0.006024597225552485, "price": 82993.1, "reason": "..." }
  ]
}