Serverless Portfolio Ingestion Pipeline

The Portfolio Pipeline is an event-driven serverless ingestion system. It automates the process of importing stock trade history sheets, updating portfolio positions (calculating current net holdings, average purchase prices, and realized profits), and delivering structured HTML email reports.

This project is built using the AWS Serverless Application Model (SAM) and deployed as a secure CloudFormation stack.

Architecture

This pipeline follows an asynchronous fan-out pattern to ensure loose coupling, high reliability, and scalability.

        CSV Upload
               │
               ▼
        S3 Trade Bucket
               │ (ObjectCreated trigger)
               ▼
           SNS Topic
               │ (Fan-out subscription)
               ▼
           SQS Queue ◄────► SQS Dead Letter Queue (DLQ)
               │ (Buffered trigger)
               ▼
      AWS Lambda Processor (VPC)
               │
        ┌──────┴──────┐
        ▼             ▼
   RDS MySQL      Amazon SES
 (Portfolio DB) (Email Updates)

1. Storage Layer: Trade CSV documents are uploaded to a specific folder (uploads/) inside a versioned Amazon S3 bucket.
2. Messaging Layer: S3 triggers an event to an Amazon SNS topic, which fans out the message to an Amazon SQS queue. SQS buffers the requests and provides a Dead Letter Queue (DLQ) for retries and error isolation.
3. Compute Layer: An AWS Lambda function is triggered by SQS. It downloads the CSV, parses the records, and connects to our database.
4. Network Security: The Lambda function and the Amazon RDS MySQL database reside inside a secure custom VPC (Virtual Private Cloud). The database is in isolated private subnets, while the Lambda function communicates with Amazon S3 via a cost-free VPC Gateway Endpoint.
5. Notification Layer: Once trades are saved, Lambda sends a summary report via Amazon SES (Simple Email Service).

AWS Resources Provisioned

When the CloudFormation stack is deployed, SAM creates and configures the following AWS resources:

1. Networking & VPC

• VPC (AWS::EC2::VPC): A custom virtual network partitioned into public and private subnets.
• Subnets (AWS::EC2::Subnet):
  • PublicSubnet1 (housing the NAT Gateway).
  • PrivateSubnet1 and PrivateSubnet2 (isolated subnets for database and Lambda function).
• Gateways:
  • InternetGateway and VPCGatewayAttachment (connecting the public subnet to the internet).
  • NatGateway and ElasticIP (allowing private Lambda functions outbound access for email alerting).
  • S3VPCEndpoint (a gateway endpoint that routes S3 traffic directly over the AWS network, bypassing the NAT Gateway to save costs).
• Route Tables (AWS::EC2::RouteTable): Dedicated public and private tables routing traffic through the IGW and NAT Gateway respectively.

2. Databases & Storage

• S3 Bucket (AWS::S3::Bucket): A versioned, secure bucket configured to publish file-upload events to an SNS topic for prefix uploads/.
• RDS MySQL DB Instance (AWS::RDS::DBInstance): A db.t3.micro instance running MySQL 8.0.45, located in a DB subnet group containing both private subnets.
• RDS Security Group (AWS::EC2::SecurityGroup): Configured to block all incoming traffic except SQL connections on port 3306 originating from the Lambda security group.

3. Messaging & Event Processing

• SNS Topic (AWS::SNS::Topic): Receives file upload events from the S3 bucket.
• SQS Queue (AWS::SQS::Queue): Subscribed to the SNS topic. Configured with a visibility timeout of 720 seconds (6x the Lambda timeout limit) to prevent duplicate processing.
• SQS DLQ (AWS::SQS::Queue): Dead Letter Queue that captures messages that fail processing after 3 execution retries.
• Queue and Topic Policies: Security policies permitting S3 to publish to SNS, and SNS to send messages to SQS.

4. Serverless Compute

• Lambda Function (AWS::Serverless::Function): The Python 3.12 executor mapped inside our VPC subnets. It has read access to S3, message poll permissions on SQS, and access to send emails via SES.
• Lambda Layer (AWS::Serverless::LayerVersion): A shared deployment layer packaging pymysql and DB helper libraries.

Database Design & Logic

The database is built on MySQL 8.0 hosted on RDS. It consists of three tables that are created automatically on the first Lambda run (idempotent setup).

1. trades

Stores every raw transaction (Buy/Sell) to prevent duplicates using a composite unique key.

CREATE TABLE IF NOT EXISTS trades (
    id          INT AUTO_INCREMENT PRIMARY KEY,
    trade_date  DATE               NOT NULL,
    symbol      VARCHAR(50)        NOT NULL,
    trade_type  ENUM('BUY','SELL') NOT NULL,
    quantity    INT                NOT NULL,
    price       DECIMAL(12,4)      NOT NULL,
    amount      DECIMAL(16,4)      NOT NULL,
    exchange    VARCHAR(10)        NOT NULL,
    source_file VARCHAR(512),
    loaded_at   TIMESTAMP          DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uq_trade (trade_date, symbol, trade_type, quantity, price, source_file)
);

2. portfolio_positions

Tracks the current state of held stocks.

• Net Quantity: Current shares owned (Buys minus Sells).
• Average Buy Price: Weighted average cost of active shares.
• Realized P&L: Net profits realized from selling assets.

CREATE TABLE IF NOT EXISTS portfolio_positions (
    id             INT AUTO_INCREMENT PRIMARY KEY,
    symbol         VARCHAR(50)   NOT NULL UNIQUE,
    net_qty        INT           DEFAULT 0,
    avg_buy_price  DECIMAL(12,4),
    total_invested DECIMAL(16,4) DEFAULT 0,
    realized_pnl   DECIMAL(16,4) DEFAULT 0,
    last_updated   TIMESTAMP     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

3. pnl_summary

Tracks high-level summary results for each uploaded file.

Calculating Portfolio Metrics

When a new trade is processed, the Lambda function recalculates the positions for the affected symbol. It executes the following math using aggregation:

SELECT
    SUM(CASE WHEN trade_type='BUY'  THEN  quantity ELSE -quantity END) AS net_qty,
    SUM(CASE WHEN trade_type='BUY'  THEN  quantity ELSE 0         END) AS buy_qty,
    SUM(CASE WHEN trade_type='BUY'  THEN  amount   ELSE 0         END) AS buy_amt,
    SUM(CASE WHEN trade_type='SELL' THEN  amount   ELSE 0         END) AS sell_amt
FROM trades WHERE symbol = 'INFY';

Using this aggregated data:

1. Average Buy Price: Total Buy Amount / Total Buy Quantity
2. Realized P&L: Total Sell Amount - (Average Buy Price * (Total Buy Quantity - Net Quantity))
3. Total Invested: Average Buy Price * Net Quantity

The computed results are then upserted into portfolio_positions using an ON DUPLICATE KEY UPDATE statement.

Key Configuration Decisions

Prerequisites for Deployment

To deploy this project from a Windows or Linux workspace, make sure you have:

1. AWS CLI (configured via aws configure with appropriate access keys).
2. AWS SAM CLI (used for building and deploying).
3. Python 3.12 (used locally to compile function dependencies).

How to Deploy and Run

1. Clone the Repository

Clone the repository to your local workspace and navigate into it:

git clone https://github.com/manikanta03090/Portfolio-Analysis.git
cd Porfolio-Analysis

2. Configuration Setup

Update samconfig.toml with your specific parameters:

parameter_overrides = "Environment=dev DBPassword=SecurePassword123! SenderEmail=sender@gmail.com ReceiverEmail=receiver@gmail.com"

3. Verification of Emails

Verify your sender and receiver emails in Amazon SES (required while your SES account is in sandbox mode):

aws ses verify-email-identity --email-address sender@gmail.com --region ap-south-1
aws ses verify-email-identity --email-address receiver@gmail.com --region ap-south-1

Note: Confirm the emails by clicking the verification link AWS sends to both inboxes.

4. Build & Deploy

Deploy the CloudFormation stack.

On Windows (PowerShell):

Set-ExecutionPolicy -ExecutionPolicy Bypass -Scope Process
.\scripts\deploy.ps1

On Linux/Mac:

chmod +x scripts/*.sh
bash scripts/deploy.sh

5. Pipeline Testing

To test the ingestion flow, we need to upload a trade history CSV to our S3 bucket inside the uploads/ folder. This triggers the event notifications.

Option A: Use the automated test scripts

The repository includes scripts that automatically retrieve your AWS account ID and upload data/sample_trades.csv to the correct S3 path:

• Windows (PowerShell): .\scripts\test_upload.ps1
• Linux/Mac: bash scripts/test_upload.sh

Option B: Upload manually via AWS CLI

You can copy your portfolio sheet directly to the S3 bucket using the AWS CLI:

aws s3 cp data/sample_trades.csv s3://<your-stack-name>-trades-<your-account-id>/uploads/sample_trades.csv --region ap-south-1

Step 6: Verify Execution & Output

1. Monitor Processing Logs (Optional): Tail the Lambda function's logs in real-time to watch the record parsing, database insertions, and email sending processes:

   Copy

   aws logs tail /aws/lambda/zerodha-portfolio-pipeline-trade-processor --follow --region ap-south-1

2. Verify Database Records (Optional): Once processed, the parsed trades are written to the MySQL trades table, and positions are upserted into portfolio_positions.

3. Check Your Email: Open the inbox of the configured ReceiverEmail. You will receive a beautifully formatted HTML report with:

   • Subject: 📈 Portfolio Updated: X trades | P&L: +₹Y.YY (or 📉 if in a loss).
   • Metrics Summary: Total trades loaded, BUY count, and SELL count.
   • Account Metrics Table: Total buy amount, total sell amount, realized P&L, and the list of stock symbols updated.

7. Cleanup

To delete the stack and remove all created resources to avoid ongoing costs:

• Windows: powershell -ExecutionPolicy Bypass -File .\scripts\cleanup.ps1
• Linux: bash scripts/cleanup.sh