AWS Lambda for Trading Bots: Serverless Deployment
Category: Algo Trading Date: 2026-03-16Introduction
AWS Lambda enables deploying trading bots without managing servers, scaling automatically with demand, and paying only for compute time used. Serverless architecture is ideal for event-driven trading strategies that don't require constant running. This guide covers deploying trading systems on AWS Lambda with integration to market data, order execution, and monitoring.Lambda Trading Bot Architecture
``python
import json
import boto3
import logging
from datetime import datetime
from typing import Dict, List, Any
import os
from decimal import Decimal
import asyncio
import httpx
logger = logging.getLogger()
logger.setLevel(logging.INFO)
AWS clients
dynamodb = boto3.resource('dynamodb')
s3_client = boto3.client('s3')
cloudwatch = boto3.client('cloudwatch')
sns_client = boto3.client('sns')
class TradingBotLambda:
"""AWS Lambda trading bot handler."""
def __init__(self):
self.trades_table = dynamodb.Table(os.getenv('TRADES_TABLE'))
self.positions_table = dynamodb.Table(os.getenv('POSITIONS_TABLE'))
self.config_bucket = os.getenv('CONFIG_BUCKET')
async def fetch_market_data(self, symbol: str) -> Dict[str, Any]:
"""Fetch current market data."""
# Integration with data provider (e.g., Alpaca, IB)
async with httpx.AsyncClient() as client:
response = await client.get(
f'https://data.alpaca.markets/v1beta1/crypto/latest/quotes',
params={'symbols': symbol},
headers={'Authorization': f'Bearer {os.getenv("ALPACA_KEY")}'}
)
return response.json()
def load_strategy_config(self, strategy_id: str) -> Dict:
"""Load strategy configuration from S3."""
try:
response = s3_client.get_object(
Bucket=self.config_bucket,
Key=f'strategies/{strategy_id}/config.json'
)
return json.loads(response['Body'].read())
except Exception as e:
logger.error(f"Failed to load config: {e}")
raise
def evaluate_strategy(self, symbol: str, market_data: Dict,
config: Dict) -> Dict:
"""Evaluate trading strategy."""
signal = None
confidence = 0.0
# Simple example: RSI-based signal
current_price = market_data['quote']['ap'] # ask price
volume = market_data['quote']['as'] # ask size
# In production, would calculate sophisticated indicators
if current_price < 100:
signal = 'buy'
confidence = 0.7
elif current_price > 110:
signal = 'sell'
confidence = 0.6
return {
'symbol': symbol,
'signal': signal,
'confidence': confidence,
'price': float(current_price),
'timestamp': datetime.utcnow().isoformat()
}
def submit_order(self, symbol: str, side: str, quantity: int,
price: float) -> Dict:
"""Submit order to broker."""
# Use broker API (Alpaca, Interactive Brokers, etc.)
order_data = {
'symbol': symbol,
'qty': quantity,
'side': side,
'type': 'limit',
'limit_price': price,
'time_in_force': 'day'
}
# Make API call to broker
try:
# order_response = submit_to_broker(order_data)
order_response = {
'order_id': 'test_order_123',
'status': 'pending',
'symbol': symbol,
'side': side,
'quantity': quantity
}
# Store in DynamoDB
self.trades_table.put_item(
Item={
'order_id': order_response['order_id'],
'symbol': symbol,
'side': side,
'quantity': quantity,
'price': Decimal(str(price)),
'timestamp': datetime.utcnow().isoformat(),
'status': 'submitted'
}
)
return order_response
except Exception as e:
logger.error(f"Order submission failed: {e}")
self._send_alert(f"Order failed for {symbol}: {e}")
raise
def update_positions(self, symbol: str, quantity: int, price: float):
"""Update position tracking."""
self.positions_table.update_item(
Key={'symbol': symbol},
UpdateExpression='SET quantity = :q, last_price = :p, updated_at = :t',
ExpressionAttributeValues={
':q': quantity,
':p': Decimal(str(price)),
':t': datetime.utcnow().isoformat()
}
)
def publish_metrics(self, metrics: Dict):
"""Publish metrics to CloudWatch."""
cloudwatch.put_metric_data(
Namespace='TradingBot',
MetricData=[
{
'MetricName': 'TradesExecuted',
'Value': metrics.get('trades_executed', 0),
'Unit': 'Count'
},
{
'MetricName': 'PortfolioValue',
'Value': metrics.get('portfolio_value', 0),
'Unit': 'None'
},
{
'MetricName': 'DailyReturn',
'Value': metrics.get('daily_return', 0),
'Unit': 'Percent'
}
]
)
def _send_alert(self, message: str):
"""Send alert via SNS."""
sns_client.publish(
TopicArn=os.getenv('ALERT_TOPIC_ARN'),
Subject='Trading Bot Alert',
Message=message
)
async def handler(self, event: Dict, context: Any) -> Dict:
"""Main Lambda handler."""
try:
logger.info(f"Event received: {json.dumps(event)}")
strategy_id = event.get('strategy_id', 'default')
symbol = event.get('symbol', 'AAPL')
# Load configuration
config = self.load_strategy_config(strategy_id)
# Fetch market data
market_data = await self.fetch_market_data(symbol)
# Evaluate strategy
signal = self.evaluate_strategy(symbol, market_data, config)
if signal['signal'] and signal['confidence'] > config.get('min_confidence', 0.5):
# Execute trade
order = self.submit_order(
symbol=symbol,
side=signal['signal'],
quantity=config.get('position_size', 10),
price=signal['price']
)
logger.info(f"Order submitted: {order['order_id']}")
# Publish metrics
self.publish_metrics({
'trades_executed': 1,
'portfolio_value': 100000,
'daily_return': 0.5
})
return {
'statusCode': 200,
'body': json.dumps({
'signal': signal['signal'],
'confidence': signal['confidence'],
'timestamp': datetime.utcnow().isoformat()
})
}
except Exception as e:
logger.error(f"Handler error: {e}")
return {
'statusCode': 500,
'body': json.dumps({'error': str(e)})
}
Lambda handler entry point
def lambda_handler(event, context):
"""Entry point for AWS Lambda."""
bot = TradingBotLambda()
return asyncio.run(bot.handler(event, context))
`
Infrastructure as Code (Terraform)
`hcl
main.tf - AWS Lambda trading bot infrastructure
provider "aws" {
region = var.aws_region
}
IAM role for Lambda
resource "aws_iam_role" "trading_bot_role" {
name = "trading-bot-lambda-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "lambda.amazonaws.com"
}
}]
})
}
Attach policies
resource "aws_iam_role_policy_attachment" "basic_execution" {
role = aws_iam_role.trading_bot_role.name
policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}
resource "aws_iam_role_policy" "dynamodb_policy" {
name = "trading-bot-dynamodb"
role = aws_iam_role.trading_bot_role.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Action = [
"dynamodb:PutItem",
"dynamodb:UpdateItem",
"dynamodb:GetItem",
"dynamodb:Query"
]
Resource = [
aws_dynamodb_table.trades.arn,
aws_dynamodb_table.positions.arn
]
}]
})
}
Lambda function
resource "aws_lambda_function" "trading_bot" {
filename = "lambda_package.zip"
function_name = "trading-bot"
role = aws_iam_role.trading_bot_role.arn
handler = "index.lambda_handler"
source_code_hash = filebase64sha256("lambda_package.zip")
timeout = 60
memory_size = 512
environment {
variables = {
TRADES_TABLE = aws_dynamodb_table.trades.name
POSITIONS_TABLE = aws_dynamodb_table.positions.name
CONFIG_BUCKET = aws_s3_bucket.config.id
ALERT_TOPIC_ARN = aws_sns_topic.alerts.arn
ALPACA_KEY = var.alpaca_key
}
}
}
DynamoDB tables
resource "aws_dynamodb_table" "trades" {
name = "trading-bot-trades"
billing_mode = "PAY_PER_REQUEST"
hash_key = "order_id"
range_key = "timestamp"
attribute {
name = "order_id"
type = "S"
}
attribute {
name = "timestamp"
type = "S"
}
ttl {
attribute_name = "expiration"
enabled = true
}
}
resource "aws_dynamodb_table" "positions" {
name = "trading-bot-positions"
billing_mode = "PAY_PER_REQUEST"
hash_key = "symbol"
attribute {
name = "symbol"
type = "S"
}
}
S3 bucket for configuration
resource "aws_s3_bucket" "config" {
bucket = "trading-bot-config-${var.aws_account_id}"
}
SNS topic for alerts
resource "aws_sns_topic" "alerts" {
name = "trading-bot-alerts"
}
EventBridge rule for scheduled execution
resource "aws_cloudwatch_event_rule" "trading_schedule" {
name = "trading-bot-schedule"
description = "Trigger trading bot every minute"
schedule_expression = "rate(1 minute)"
}
resource "aws_cloudwatch_event_target" "lambda_target" {
rule = aws_cloudwatch_event_rule.trading_schedule.name
target_id = "TradingBotLambda"
arn = aws_lambda_function.trading_bot.arn
input = jsonencode({
strategy_id = "default"
symbol = "AAPL"
})
}
resource "aws_lambda_permission" "allow_eventbridge" {
statement_id = "AllowExecutionFromEventBridge"
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.trading_bot.function_name
principal = "events.amazonaws.com"
source_arn = aws_cloudwatch_event_rule.trading_schedule.arn
}
CloudWatch alarms
resource "aws_cloudwatch_metric_alarm" "lambda_errors" {
alarm_name = "trading-bot-lambda-errors"
comparison_operator = "GreaterThanThreshold"
evaluation_periods = "1"
metric_name = "Errors"
namespace = "AWS/Lambda"
period = "300"
statistic = "Sum"
threshold = "1"
alarm_actions = [aws_sns_topic.alerts.arn]
dimensions = {
FunctionName = aws_lambda_function.trading_bot.function_name
}
}
`
Deployment Script
`bash
#!/bin/bash
deploy.sh - Deploy Lambda trading bot
set -e
Build Lambda package
pip install -r requirements.txt -t python/
zip -r lambda_package.zip index.py python/
Deploy with Terraform
terraform init
terraform plan
terraform apply -auto-approve
Verify deployment
echo "Lambda function deployed successfully"
aws lambda invoke \
--function-name trading-bot \
--payload '{"strategy_id":"default","symbol":"AAPL"}' \
response.json
echo "Response:"
cat response.json
``
Key Advantages
- No Server Management: Automatic scaling and patching
- Cost-Efficient: Pay only for execution time
- High Availability: Built-in redundancy and fault tolerance
- Easy Integration: Direct AWS service integration
- Monitoring: CloudWatch integration for logs and metrics
Limitations and Considerations
- 15-minute execution timeout
- Cold start latency (critical for HFT)
- Memory/CPU constraints
- State management via external services (DynamoDB)
Best Practices
- Use VPC endpoints for security
- Implement circuit breakers for API failures
- Cache configurations to minimize cold starts
- Use Lambda layers for dependencies
- Monitor and alert on execution metrics