Building a Real-Time Algorithmic Trading Bot Using LLMs and the Alpaca Markets API
Scope and a necessary disclaimer
This tutorial is educational systems-engineering content, not investment advice. It builds a paper-trading bot end to end — real-time bar streaming, a rules-based technical signal, an LLM-based news-sentiment gate, and order execution through Alpaca's paper trading endpoint. Live trading with real capital introduces regulatory, risk-management, and capital-loss considerations far beyond this tutorial's scope; treat everything here as a foundation to rigorously backtest and risk-review before ever pointing it at a live account.
1. Technical architecture
┌───────────────────┐ WebSocket (bars) ┌────────────────────────┐
│ Alpaca Market Data │ ─────────────────────▶│ Bar aggregator │
│ Stream (IEX/SIP) │ │ (in-memory OHLCV buffer) │
└───────────────────┘ └───────────┬────────────┘
▼
┌────────────────────────┐
│ Technical signal engine │
│ (SMA crossover + RSI) │
└───────────┬────────────┘
│ candidate signal
▼
┌───────────────────┐ REST (poll, cached) ┌────────────────────────┐
│ News API (headlines) │ ───────────────────▶ │ LLM sentiment gate │
│ │ │ (Claude via API, JSON out) │
└───────────────────┘ └───────────┬────────────┘
│ confirmed signal
▼
┌────────────────────────┐
│ Risk manager │
│ (position size, daily loss │
│ cap, max concurrent trades)│
└───────────┬────────────┘
▼
┌────────────────────────┐
│ Alpaca Trading API │
│ (paper endpoint, bracket │
│ orders w/ stop-loss) │
└────────────────────────┘
2. Complete code implementation
2.1 Real-time bar stream and technical signal
# trading/signal_engine.py
from collections import deque
from dataclasses import dataclass, field
@dataclass
class SymbolState:
closes: deque = field(default_factory=lambda: deque(maxlen=50))
def sma(self, window: int) -> float | None:
if len(self.closes) < window:
return None
recent = list(self.closes)[-window:]
return sum(recent) / window
def rsi(self, window: int = 14) -> float | None:
if len(self.closes) < window + 1:
return None
prices = list(self.closes)[-(window + 1):]
gains, losses = [], []
for i in range(1, len(prices)):
delta = prices[i] - prices[i - 1]
gains.append(max(delta, 0))
losses.append(max(-delta, 0))
avg_gain = sum(gains) / window
avg_loss = sum(losses) / window
if avg_loss == 0:
return 100.0
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
class SignalEngine:
"""SMA(10)/SMA(30) crossover confirmed by RSI not already overbought/oversold."""
def __init__(self):
self.state: dict[str, SymbolState] = {}
def on_bar(self, symbol: str, close_price: float) -> str | None:
s = self.state.setdefault(symbol, SymbolState())
s.closes.append(close_price)
fast, slow, rsi = s.sma(10), s.sma(30), s.rsi(14)
if fast is None or slow is None or rsi is None:
return None # not enough history yet
if fast > slow and rsi < 70:
return "buy"
if fast < slow and rsi > 30:
return "sell"
return None
2.2 LLM sentiment gate
# trading/sentiment_gate.py
import json
import time
from anthropic import Anthropic, APIStatusError, APITimeoutError
client = Anthropic() # reads ANTHROPIC_API_KEY from environment
SYSTEM_PROMPT = """You are a financial news sentiment classifier. Given recent
headlines about a stock ticker, respond with strict JSON only:
{"sentiment": "positive"|"neutral"|"negative", "confidence": 0.0-1.0,
"reason": "one short sentence"}. Base your answer only on the headlines given."""
def classify_sentiment(symbol: str, headlines: list[str], max_retries: int = 3) -> dict:
if not headlines:
return {"sentiment": "neutral", "confidence": 0.0, "reason": "no recent headlines"}
user_content = f"Ticker: {symbol}\nHeadlines:\n" + "\n".join(f"- {h}" for h in headlines[:10])
for attempt in range(max_retries):
try:
resp = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=200,
system=SYSTEM_PROMPT,
messages=[{"role": "user", "content": user_content}],
)
text = resp.content[0].text
return json.loads(text)
except (APIStatusError, APITimeoutError) as exc:
if attempt == max_retries - 1:
# Fail safe: on repeated API failure, treat as neutral rather
# than blocking trading entirely or defaulting to "positive".
return {"sentiment": "neutral", "confidence": 0.0, "reason": f"sentiment API error: {exc}"}
time.sleep(2 ** attempt)
except (json.JSONDecodeError, IndexError):
# Model didn't return valid JSON — do not guess, fail safe to neutral.
return {"sentiment": "neutral", "confidence": 0.0, "reason": "unparseable model response"}
2.3 Risk manager and order execution
# trading/risk_manager.py
from dataclasses import dataclass
@dataclass
class RiskLimits:
max_position_pct: float = 0.05 # max 5% of equity per position
max_concurrent_positions: int = 8
daily_loss_limit_pct: float = 0.03 # halt trading after 3% daily drawdown
class RiskManager:
def __init__(self, limits: RiskLimits):
self.limits = limits
self.starting_equity: float | None = None
self.trading_halted = False
def check_daily_loss(self, current_equity: float) -> bool:
if self.starting_equity is None:
self.starting_equity = current_equity
return True
drawdown = (self.starting_equity - current_equity) / self.starting_equity
if drawdown >= self.limits.daily_loss_limit_pct:
self.trading_halted = True
return not self.trading_halted
def position_size(self, equity: float, price: float) -> int:
max_dollar_amount = equity * self.limits.max_position_pct
qty = int(max_dollar_amount // price)
return max(qty, 0)
# trading/executor.py
import logging
from alpaca.trading.client import TradingClient
from alpaca.trading.requests import MarketOrderRequest, StopLossRequest, TakeProfitRequest
from alpaca.trading.enums import OrderSide, TimeInForce
from alpaca.common.exceptions import APIError
logger = logging.getLogger("executor")
class OrderExecutor:
def __init__(self, api_key: str, secret_key: str, paper: bool = True):
self.client = TradingClient(api_key, secret_key, paper=paper)
def place_bracket_order(self, symbol: str, qty: int, side: str,
stop_loss_pct: float = 0.02, take_profit_pct: float = 0.04) -> dict | None:
if qty <= 0:
logger.info("Skipping order for %s: computed quantity is 0", symbol)
return None
try:
quote = self.client.get_latest_quote(symbol) # last traded price reference
ref_price = float(quote.ask_price or quote.bid_price)
except APIError as exc:
logger.error("Failed to fetch quote for %s: %s", symbol, exc)
return None
order_side = OrderSide.BUY if side == "buy" else OrderSide.SELL
stop_price = round(ref_price * (1 - stop_loss_pct), 2) if side == "buy" else round(ref_price * (1 + stop_loss_pct), 2)
take_price = round(ref_price * (1 + take_profit_pct), 2) if side == "buy" else round(ref_price * (1 - take_profit_pct), 2)
request = MarketOrderRequest(
symbol=symbol,
qty=qty,
side=order_side,
time_in_force=TimeInForce.DAY,
order_class="bracket",
stop_loss=StopLossRequest(stop_price=stop_price),
take_profit=TakeProfitRequest(limit_price=take_price),
)
try:
order = self.client.submit_order(request)
except APIError as exc:
status = getattr(exc, "status_code", None)
if status == 403:
logger.error("Order rejected — insufficient buying power or PDT restriction for %s", symbol)
elif status == 429:
logger.error("Alpaca rate limit hit submitting order for %s", symbol)
else:
logger.error("Order submission failed for %s: %s", symbol, exc)
return None
logger.info("Submitted %s bracket order for %d shares of %s (order id=%s)", side, qty, symbol, order.id)
return {"id": str(order.id), "symbol": symbol, "qty": qty, "side": side}
2.4 Main event loop
# trading/main.py
import os
import logging
from alpaca.data.live import StockDataStream
from alpaca.trading.client import TradingClient
from signal_engine import SignalEngine
from sentiment_gate import classify_sentiment
from risk_manager import RiskManager, RiskLimits
from executor import OrderExecutor
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("main")
API_KEY = os.environ["ALPACA_API_KEY"]
SECRET_KEY = os.environ["ALPACA_SECRET_KEY"]
WATCHLIST = ["AAPL", "MSFT", "NVDA"]
signal_engine = SignalEngine()
risk_manager = RiskManager(RiskLimits())
executor = OrderExecutor(API_KEY, SECRET_KEY, paper=True)
trading_client = TradingClient(API_KEY, SECRET_KEY, paper=True)
async def on_bar(bar):
account = trading_client.get_account()
if not risk_manager.check_daily_loss(float(account.equity)):
logger.warning("Daily loss limit hit — trading halted for the session")
return
action = signal_engine.on_bar(bar.symbol, bar.close)
if action is None:
return
headlines = fetch_recent_headlines(bar.symbol) # implement via your news provider
sentiment = classify_sentiment(bar.symbol, headlines)
if action == "buy" and sentiment["sentiment"] == "negative" and sentiment["confidence"] > 0.6:
logger.info("Technical buy signal for %s overridden by negative sentiment", bar.symbol)
return
if action == "sell" and sentiment["sentiment"] == "positive" and sentiment["confidence"] > 0.6:
logger.info("Technical sell signal for %s overridden by positive sentiment", bar.symbol)
return
qty = risk_manager.position_size(float(account.equity), bar.close)
executor.place_bracket_order(bar.symbol, qty, action)
def fetch_recent_headlines(symbol: str) -> list[str]:
# Plug in a news provider (e.g. Alpaca News API, Polygon, NewsAPI) here.
return []
def run():
stream = StockDataStream(API_KEY, SECRET_KEY)
stream.subscribe_bars(on_bar, *WATCHLIST)
stream.run()
if __name__ == "__main__":
run()
3. Step-by-step configuration guide
- Create an Alpaca paper trading account at alpaca.markets and generate an API key/secret from the dashboard (Paper Trading tab).
- Install dependencies:
pip install alpaca-py anthropic --break-system-packages - Set environment variables:
export ALPACA_API_KEY="your_paper_key" export ALPACA_SECRET_KEY="your_paper_secret" export ANTHROPIC_API_KEY="your_anthropic_key" - Wire in a news provider in
fetch_recent_headlines— Alpaca's News API, Polygon.io, or NewsAPI.org all work; cache results per symbol for a few minutes to avoid rate-limit and cost blowup. - Run the bot against paper trading:
python trading/main.py. - Monitor fills in the Alpaca dashboard's Paper Trading → Orders tab, and tail your process logs for
RiskManagerhalt events. - Backtest before any live consideration — replay historical bars through
SignalEngineandclassify_sentimentoffline, and evaluate Sharpe ratio, max drawdown, and win rate over multiple market regimes before trusting the strategy with capital.
4. Error handling and edge cases
- Alpaca API rate limits (429) are caught explicitly in
place_bracket_orderand logged distinctly from other failures, so you can add a backoff/queue layer without conflating rate limiting with genuine rejections. - Insufficient buying power or PDT (pattern day trader) restrictions (403) are surfaced with a specific log message — these are account-state failures, not bugs, and should not trigger automatic retries.
- WebSocket disconnects —
alpaca-py'sStockDataStreamreconnects automatically on transient network drops, but wrapstream.run()in a supervising process (e.g., systemd withRestart=on-failure, or a Kubernetes liveness probe) so a hard crash doesn't leave the bot silently offline. - LLM sentiment API failures never block or crash the trading loop —
classify_sentimentfails safe to a neutral, zero-confidence result on both API errors and malformed JSON responses, so a sentiment-provider outage degrades gracefully to technical-signal-only trading rather than halting everything. - Daily loss limit breach is checked before every order via
risk_manager.check_daily_loss, and oncetrading_haltedis set it stays set for the session — the bot does not silently resume after a brief equity recovery. - Zero-quantity orders (position size rounds down to 0 shares, common for high-priced stocks on smaller accounts) are explicitly skipped in
place_bracket_orderrather than submitted as invalid orders. - Stale or missing quotes — if
get_latest_quotefails or returnsNonefor both bid/ask, the order is aborted rather than computing stop-loss/take-profit off a garbage reference price.
