Portfolio Data Processing Solution

A serverless solution for processing daily portfolio CSV uploads from S3 and generating JSON messages for a message queue.

Overview

This solution processes four daily CSV files uploaded to S3:

• clients_YYYYMMDD.csv - Client information
• portfolios_YYYYMMDD.csv - Portfolio details
• accounts_YYYYMMDD.csv - Bank account information
• transactions_YYYYMMDD.csv - Transaction records

Once all four files are uploaded for a given date, the system automatically:

1. Validates and processes the data
2. Joins information across files
3. Generates JSON messages (client messages, portfolio messages, error messages)
4. Sends messages to an SQS queue for downstream consumption

Architecture

High-Level Architecture

S3 Bucket (CSV Upload)
    ↓ (S3 Event Notification)
Coordinator Lambda
    ↓ (Tracks uploads in DynamoDB)
    ↓ (Triggers when all 4 files present)
Processor Lambda
    ↓ (Reads CSVs, processes data)
    ↓ (Generates messages)
SQS Queue → Downstream Consumers

Components

1. S3 Bucket: Receives daily CSV file uploads
2. Coordinator Lambda: Triggered on each file upload, tracks completion status
3. DynamoDB Table: Tracks which files have been uploaded for each date
4. Processor Lambda: Processes all four files when ready and generates messages
5. SQS Queue: Stores generated JSON messages for consumption
6. Dead Letter Queue (DLQ): Captures failed messages for investigation

Data Flow

1. External client uploads CSV file to S3
2. S3 event notification triggers Coordinator Lambda
3. Coordinator updates DynamoDB tracking table
4. When all 4 files are present, Coordinator invokes Processor Lambda
5. Processor reads and validates all CSV files
6. Processor joins data and generates messages:
   • One client message per client (with aggregated taxes)
   • One portfolio message per portfolio (with transaction metrics)
   • Error messages for any validation failures
7. Messages are sent to SQS in batches
8. Downstream consumers read from SQS queue

Services Used

AWS Lambda

Why: Serverless compute eliminates server management and automatically scales with load.

What it achieves:

• Event-driven processing triggered by S3 uploads
• Pay-per-use pricing (no idle costs)
• Automatic scaling to handle varying upload volumes
• Isolated execution environment per invocation

Configuration:

• Coordinator: 60s timeout, 128MB memory (lightweight tracking)
• Processor: 300s timeout, 512MB memory (data processing)

Amazon S3

Why: Highly durable, scalable object storage designed for data lakes.

What it achieves:

• Unlimited storage capacity
• 99.999999999% (11 9's) durability
• Event notifications for automation
• Versioning enabled for data recovery
• Server-side encryption by default

Amazon SQS

Why: Fully managed message queue service with high throughput and reliability.

What it achieves:

• Decouples data processing from consumption
• Guarantees at-least-once delivery
• Message retention up to 14 days
• Dead Letter Queue for failed messages
• Supports batch operations (up to 10 messages)

Amazon DynamoDB

Why: Fast, scalable NoSQL database with single-digit millisecond performance.

What it achieves:

• Tracks file upload status per date
• On-demand billing (pay per request)
• Automatic scaling without provisioning
• TTL support for automatic data cleanup
• Consistent reads for coordination logic

AWS CloudWatch

Why: Centralized logging and monitoring service.

What it achieves:

• Lambda execution logs with 7-day retention
• Error tracking and alerting capabilities
• Performance metrics and dashboards
• Troubleshooting and audit trail

Quick Start

The project includes a Makefile for common operations:

# Show all available commands
make help

# Set up Python environment (or use setup.sh)
./setup.sh

# Install dependencies
make install

# Run tests
make test

# Deploy infrastructure
make deploy

# Upload test data to S3
make upload-test-data

# Clean up build artifacts
make clean

# Destroy infrastructure
make destroy

Setup and Deployment

Prerequisites

• AWS Account with appropriate permissions
• Terraform >= 1.0
• Python 3.11
• AWS CLI configured with credentials
• Make (optional, but recommended)

Installation Steps

Option 1: Using setup script (recommended)

./setup.sh
source venv/bin/activate

Option 2: Manual setup

1. Install Python dependencies

   pip install -r requirements.txt

2. Run tests (optional)

   make test
   # or
   pytest tests/ -v

3. Configure Terraform variables

   Edit infrastructure/staging.tfvars or infrastructure/prod.tfvars:

   aws_region   = "us-east-1"
   environment  = "staging"
   bucket_name  = "your-unique-bucket-name"

4. Deploy infrastructure

   # Using Makefile
   make deploy
   
   # Or manually
   cd infrastructure
   terraform init
   terraform apply -var-file="staging.tfvars"

5. Note the outputs

   After deployment, Terraform will output:

   • S3 bucket name
   • SQS queue URL
   • Lambda function ARNs

Testing

1. Upload test data

   # Using Makefile (recommended)
   make upload-test-data
   
   # Or manually
   aws s3 cp example_data/clients_20200130.csv s3://your-bucket-name/
   aws s3 cp example_data/portfolios_20200130.csv s3://your-bucket-name/
   aws s3 cp example_data/accounts_20200130.csv s3://your-bucket-name/
   aws s3 cp example_data/transactions_20200130.csv s3://your-bucket-name/

2. Check CloudWatch Logs

   aws logs tail /aws/lambda/staging-csv-processor --follow

3. Read messages from SQS

   # Get queue URL from Terraform output
   QUEUE_URL=$(cd infrastructure && terraform output -raw sqs_queue_url)
   
   # Use the provided script to read all messages
   python scripts/read_all_messages.py $QUEUE_URL
   
   # Or delete messages after reading
   python scripts/read_all_messages.py $QUEUE_URL --delete

Scalability

Current Scalability Features

1. Automatic Lambda Scaling

   • Lambdas automatically scale to 1000 concurrent executions per region
   • Each invocation processes independently
   • No state shared between executions

2. S3 Unlimited Storage

   • Supports unlimited file uploads
   • No performance degradation with data growth
   • Automatic partitioning by AWS

3. SQS Throughput

   • Standard queue supports nearly unlimited TPS
   • Batch operations reduce API calls
   • Messages distributed across multiple servers

4. DynamoDB On-Demand

   • Automatically scales read/write capacity
   • No capacity planning required
   • Handles sudden traffic spikes

Scalability Considerations

Current Limitations

1. Single Lambda for Processing

   • Current design processes all four files in one Lambda invocation
   • Limited by Lambda timeout (300s) and memory (512MB)
   • Large files (> 100MB) could approach limits

2. Sequential File Processing

   • Files are read and processed one after another
   • Not optimized for very large datasets

3. In-Memory Processing

   • All CSV data loaded into memory
   • Limited by Lambda memory allocation

Recommendations for Scale

1. For Larger Files (100MB+)

   • Use AWS Batch or ECS Fargate for longer processing times
   • Implement streaming CSV parsing instead of loading full files
   • Consider Amazon EMR for big data processing

2. For Higher Throughput

   • Implement parallel processing per file type
   • Use Step Functions to orchestrate multiple Lambdas
   • Process files in chunks using byte-range requests

3. For Very Large Datasets

   • Migrate to AWS Glue ETL jobs
   • Use Amazon Athena for SQL-based processing
   • Implement Apache Spark on EMR

4. For Global Scale

   • Deploy multi-region infrastructure
   • Use S3 Cross-Region Replication
   • Implement geo-based routing

Security

Current Security Measures

1. S3 Security

   • Server-side encryption (AES-256) enabled
   • Public access blocked by default
   • Versioning enabled for data recovery
   • Bucket policies restrict access to specific IAM roles

2. IAM Least Privilege

   • Lambda functions have minimal required permissions
   • Separate roles for Coordinator and Processor
   • No wildcard permissions in policies

3. Network Security

   • Lambdas run in AWS-managed VPC
   • No public endpoints exposed
   • All communication over AWS internal network

4. Data in Transit

   • All AWS service communication uses TLS 1.2+
   • S3 to Lambda uses AWS PrivateLink

5. Logging and Monitoring

   • All Lambda executions logged to CloudWatch
   • 7-day log retention for audit trail
   • Failed messages routed to DLQ for investigation

Security Concerns and Improvements

Current Concerns

1. No Data Validation

   • CSV content not validated for malicious code
   • No schema enforcement on uploaded files

2. No Access Control on S3 Uploads

   • Any IAM principal with access can upload files
   • No authentication on external client uploads

3. Plain Text Data

   • No encryption at application level
   • Sensitive financial data stored as-is

4. No Secrets Management

   • Environment variables visible in Lambda console
   • No rotation mechanism

Recommended Improvements

1. Authentication and Authorization

   • Implement S3 pre-signed URLs for uploads with expiration
   • Use API Gateway with authentication for upload control
   • Enable S3 Object Lock for WORM compliance

2. Enhanced Encryption

   • Use AWS KMS Customer Managed Keys (CMK)
   • Implement field-level encryption for sensitive data
   • Enable CloudTrail for KMS key usage auditing

3. Data Validation

   • Add CSV schema validation before processing
   • Implement virus scanning (Lambda or S3 Object Lambda)
   • Add checksum verification for file integrity

4. Network Security

   • Deploy Lambdas in private VPC subnets
   • Use VPC endpoints for S3, SQS, DynamoDB access
   • Implement AWS WAF if adding API Gateway

5. Secrets Management

   • Store queue URLs and sensitive config in AWS Secrets Manager
   • Implement automatic secret rotation
   • Use IAM authentication for SQS/DynamoDB instead of URLs

6. Compliance and Auditing

   • Enable AWS CloudTrail for all API calls
   • Implement S3 access logging
   • Add compliance frameworks (PCI DSS, GDPR) controls
   • Enable AWS Config for resource compliance

7. DDoS Protection

   • Implement S3 event filtering to prevent event flood
   • Add rate limiting on uploads
   • Use AWS Shield for DDoS protection

Future Improvements

High Priority

1. Error Handling and Retry Logic

   • Implement exponential backoff for transient failures
   • Add circuit breaker pattern for downstream services
   • Detailed error categorization and alerting

2. Data Quality Validation

   • Schema validation using JSON Schema or Pydantic
   • Business rule validation (positive balances)
   • Duplicate detection across days

3. Monitoring and Alerting

   • CloudWatch alarms for Lambda errors
   • SQS queue depth monitoring
   • Processing time metrics and SLAs
   • Dashboard for operational visibility

4. Testing

   • Unit tests for processing logic
   • Integration tests with LocalStack
   • Load testing with synthetic data
   • Chaos engineering tests

Medium Priority

5. Performance Optimization

   • Parallel file processing using asyncio
   • Streaming CSV parsing for large files
   • Lambda SnapStart for faster cold starts

6. Data Pipeline Features

   • Idempotency keys to prevent duplicate processing
   • Exactly-once semantics using SQS FIFO queues
   • Late-arriving data handling (process out-of-order dates)

7. Operations

   • Infrastructure as Code validation (tfsec, checkov)
   • CI/CD pipeline with automated testing
   • Blue-green deployment for zero downtime

8. Cost Optimization

   • S3 Intelligent-Tiering for old files
   • DynamoDB on-demand to provisioned analysis
   • Lambda reserved concurrency for cost control

Low Priority

9. Advanced Features

   • Real-time streaming with Kinesis Data Streams
   • Machine learning for anomaly detection
   • GraphQL API for message querying

10. Multi-tenancy

    • Partition by client in S3 prefixes
    • Separate queues per client
    • Client-specific processing rules

Assumptions

1. File Upload Pattern

   • All four files for a date are uploaded within a reasonable time window (1 hour)
   • File names strictly follow the pattern TYPE_YYYYMMDD.csv
   • Each date's files are uploaded exactly once (no re-uploads of the same date)

2. Data Constraints

   • CSV files are well-formed
   • File sizes are reasonable (<100MB per file)
   • Client references are unique and consistent across files
   • One account per portfolio (as stated in requirements)

3. Processing Requirements

   • Processing can occur asynchronously (eventual consistency acceptable)
   • Message delivery order is not critical (using standard SQS)
   • At-least-once delivery is acceptable (consumers handle duplicates)

4. Operational

   • AWS region deployment is US-based
   • 7-day log retention is sufficient for troubleshooting
   • Daily processing volume is manageable by Lambda (< 1000 files/day)

5. Business Logic

   • Taxes paid aggregation sums across all client portfolios
   • Transaction count includes all transaction types
   • Deposits are identified by keyword "DEPOSIT" only
   • Missing relationships (portfolio without client) generate error messages

Project Structure

.
├── README.md                      # This file
├── requirements.txt               # Python dependencies
├── .gitignore                     # Git ignore rules
├── Makefile                       # Common commands for development
├── setup.sh                       # Python environment setup script
├── docs/
│   ├── index.md                   # Problem statement and data examples
│   ├── deployment.md              # Deployment guide
│   └── message_formats.md         # Message format specifications
├── src/
│   ├── coordinator.py             # Tracks file uploads, triggers processor
│   └── processor.py               # Main data processing logic
├── infrastructure/
│   ├── main.tf                    # Main infrastructure definitions
│   ├── variables.tf               # Terraform input variables
│   ├── outputs.tf                 # Terraform outputs (bucket, queue URLs)
│   ├── providers.tf               # AWS provider configuration
│   ├── versions.tf                # Terraform version constraints
│   ├── staging.tfvars             # Staging environment variables
│   └── prod.tfvars                # Production environment variables
├── scripts/
│   └── read_all_messages.py       # Utility to read SQS messages
├── example_data/
│   ├── clients_20200130.csv       # Sample client data
│   ├── portfolios_20200130.csv    # Sample portfolio data
│   ├── accounts_20200130.csv      # Sample account data
│   └── transactions_20200130.csv  # Sample transaction data
└── tests/
    └── test_processor.py          # Unit tests for processor logic

Documentation

• docs/index.md - Overview of the problem statement and data format examples
• docs/deployment.md - Detailed deployment instructions
• docs/message_formats.md - Complete message format specifications