Back

import json
import boto3
import base64
import pandas as pd

# s3 = boto3.client('s3') # Uncomment if fetching or writing order flow data to S3

def calculate_order_flow_metrics(df):
    """
    Calculates aggressive buy/sell volume delta based on trade direction.
    Assumes incoming DataFrame has columns: 'Price', 'Volume', 'BidSize', 'AskSize', 'TradeType'
    TradeType: 1 = Bought on Ask (Aggressive Buyer), 2 = Sold on Bid (Aggressive Seller)
    """
    # Initialize Delta
    df['Aggressive_Buy'] = 0
    df['Aggressive_Sell'] = 0
    
    # Identify aggressive buyers and sellers
    df.loc[df['TradeType'] == 1, 'Aggressive_Buy'] = df['Volume']
    df.loc[df['TradeType'] == 2, 'Aggressive_Sell'] = df['Volume']
    
    # Calculate Volume Delta and Cumulative Volume Delta (CVD)
    df['Volume_Delta'] = df['Aggressive_Buy'] - df['Aggressive_Sell']
    df['CVD'] = df['Volume_Delta'].cumsum()
    
    # Order Book Imbalance (Bid - Ask)
    df['Imbalance'] = df['BidSize'] - df['AskSize']
    
    return df

def lambda_handler(event, context):
    """
    AWS Lambda entry point for order flow analysis.
    """
    try:
        # 1. Parse Event Data (Handle both direct JSON and API Gateway base64 payloads)
        body = event.get('body', {})
        if event.get('isBase64Encoded', False):
            body = base64.b64decode(body).decode('utf-8')
        
        if isinstance(body, str):
            data = json.loads(body)
        else:
            data = body

        # 2. Convert raw tick/order data into a Pandas DataFrame
        df_raw = pd.DataFrame(data.get('trades', []))
        if df_raw.empty:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'No trade data provided in the event.'})
            }

        # 3. Process and Analyze
        df_analyzed = calculate_order_flow_metrics(df_raw)

        # 4. Return results (or push to S3/DynamoDB)
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Order flow analysis completed successfully.',
                'latest_cvd': float(df_analyzed['CVD'].iloc[-1]),
                'data': df_analyzed.tail(10).to_dict(orient='records')
            })
        }

    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }