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))

# 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
  }
}

#!/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