Data EngineeringApril 20266 min read

Building a Highly Concurrent RTA Mailback Parsing Pipeline with FastAPI.

CAMS and Karvy send mutual fund transaction data in emails that are inconsistent, poorly structured, and change format without warning. Here is how I replaced hours of manual data entry with a fully automated FastAPI pipeline that runs in under 5 minutes.

Author

Faize Muhammed Basheer

Stack

FastAPI / Python / PostgreSQL

Context

Live Fintech Production

01 / What is an RTA Mailback?

RTA stands for Registrar and Transfer Agent. In India's mutual fund industry, CAMS and Karvy are the two dominant RTAs — they process every SIP transaction, redemption, dividend payout, and brokerage calculation across all fund houses.

When a transaction completes, the RTA sends a confirmation email to the registered MFD — the Mutual Fund Distributor. This email contains the transaction details: investor name, folio number, scheme code, amount, NAV, units allotted, and brokerage earned. In the industry this is called a "mailback."

The problem: these emails are the primary data source for portfolio reconciliation on our MFD platform. Without parsing them correctly, the portfolio values investors see are wrong or stale.

02 / The Manual Process (Before)

Before building this pipeline, the process was entirely manual. Someone received the RTA email, read the transaction details, and entered them into the system by hand.

For a single MFD handling a few transactions per day, this works. For a platform serving multiple MFDs with hundreds of transactions daily — it breaks immediately.

Manual entry means delayed portfolio updates, human transcription errors, and a process that doesn't scale past 2-3 MFDs without hiring people whose entire job is copying emails into a database.

The goal was zero human intervention — email arrives, pipeline runs, database is updated, investor sees accurate portfolio within 5 minutes.

03 / Why FastAPI and Not Node.js

Our primary backend is Node.js. The natural choice would have been to build the pipeline there too. I chose Python and FastAPI instead — deliberately — for three reasons.

01

Python's text processing ecosystem is superior

The email parsing, regex extraction, and tabular data processing libraries in Python — email, re, pandas, BeautifulSoup — are far more mature than their Node.js equivalents. RTA emails contain HTML tables, inline styles, and inconsistent whitespace. Python handles this cleanly.

02

FastAPI is genuinely fast for I/O-heavy workloads

FastAPI is built on Starlette and uses Python's async/await natively. For a pipeline that reads emails, runs parsing logic, and writes to a database — all I/O operations — it performs extremely well without blocking.

03

Isolation is a feature

Keeping the parsing pipeline as a separate FastAPI microservice means a bug in the parser never affects the main Node.js API. The two services communicate over HTTP. If parsing fails, the main platform keeps running.

04 / The Pipeline Architecture

The pipeline has four stages: ingestion, detection, parsing, and validation + write.

RTA Email (CAMS / Karvy) ↓ Email Server → Webhook Trigger → FastAPI /ingest endpoint ↓ Stage 1: Email body extraction (HTML + plain text) ↓ Stage 2: RTA Detection (CAMS or Karvy?) ↓ Stage 3: Format-specific parser runs ↓ Stage 4: Schema validation + deduplication check ↓ Write to PostgreSQL ledger ✅ ↓ Flag failed records for manual review ⚠️

Stage 1 — Email Body Extraction

RTA emails arrive in multipart MIME format — both HTML and plain text versions of the same content. The HTML version is more structured but harder to parse reliably due to inline styles. The plain text version is easier to parse but sometimes truncated.

The extractor pulls both versions and passes them to the detector.

import email from bs4 import BeautifulSoup def extract_email_body(raw_email: bytes) -> dict: msg = email.message_from_bytes(raw_email) html_body = "" text_body = "" for part in msg.walk(): content_type = part.get_content_type() if content_type == "text/html": html_body = part.get_payload(decode=True).decode("utf-8", errors="ignore") elif content_type == "text/plain": text_body = part.get_payload(decode=True).decode("utf-8", errors="ignore") # Clean HTML — strip styles, scripts, preserve table structure soup = BeautifulSoup(html_body, "html.parser") for tag in soup(["style", "script", "head"]): tag.decompose() return { "html": str(soup), "text": text_body, "subject": msg.get("Subject", ""), "sender": msg.get("From", ""), }

Stage 2 — RTA Detection

CAMS and Karvy use different email formats — different sender domains, different subject line patterns, different table structures. The detector reads the sender address and subject line to decide which parser to route the email to.

def detect_rta(email_data: dict) -> str: sender = email_data["sender"].lower() subject = email_data["subject"].lower() if "camsonline.com" in sender or "cams" in subject: return "CAMS" elif "karvy.com" in sender or "kfintech.com" in sender or "karvy" in subject: return "KARVY" else: # Unknown sender — route to manual review queue return "UNKNOWN"

Stage 3 — Format-Specific Parsers

Each RTA has its own parser. CAMS sends structured HTML tables with relatively consistent column headers. Karvy mixes plain text sections with HTML tables and uses different field names for the same data.

def parse_cams(email_data: dict) -> list[dict]: soup = BeautifulSoup(email_data["html"], "html.parser") transactions = [] # CAMS always uses a specific table class tables = soup.find_all("table", {"class": lambda c: c and "transaction" in c.lower()}) for table in tables: rows = table.find_all("tr") headers = [th.get_text(strip=True).lower() for th in rows[0].find_all(["th", "td"])] for row in rows[1:]: cells = [td.get_text(strip=True) for td in row.find_all("td")] if len(cells) != len(headers): continue # Skip malformed rows record = dict(zip(headers, cells)) transactions.append({ "folio": record.get("folio no", "").strip(), "scheme_code": record.get("scheme code", "").strip(), "amount": parse_amount(record.get("amount", "0")), "units": parse_float(record.get("units", "0")), "nav": parse_float(record.get("nav", "0")), "txn_ref": record.get("transaction ref", "").strip(), "txn_date": parse_date(record.get("date", "")), "rta": "CAMS", }) return transactions def parse_amount(raw: str) -> float: # Handle formats: "1,23,456.78" "Rs.1234" "INR 1234.56" cleaned = re.sub(r"[^\d.]", "", raw.replace(",", "")) try: return float(cleaned) except ValueError: return 0.0

Stage 4 — Validation and Deduplication

Every parsed record goes through schema validation before touching the database. Amounts must be positive. Folio numbers must be non-empty. Scheme codes must match known codes in our master list. Dates must be parseable.

Then deduplication — the transaction reference number is checked against existing records. RTA emails get forwarded and CC'd; the same transaction arriving twice must not be written twice.

async def validate_and_write(transactions: list[dict], db: AsyncSession): written = 0 skipped = 0 flagged = [] for txn in transactions: # Schema validation if not txn["folio"] or not txn["scheme_code"]: flagged.append({**txn, "reason": "missing_required_fields"}) continue if txn["amount"] <= 0 or txn["amount"] > 10_000_000: flagged.append({**txn, "reason": "amount_out_of_range"}) continue # Deduplication — check transaction reference existing = await db.execute( select(Transaction).where( Transaction.txn_ref == txn["txn_ref"], Transaction.rta == txn["rta"] ) ) if existing.scalar_one_or_none(): skipped += 1 continue # Safe to write db.add(Transaction(**txn)) written += 1 await db.commit() # Route flagged records to manual review — never silently drop if flagged: await send_to_review_queue(flagged) return {"written": written, "skipped": skipped, "flagged": len(flagged)}

05 / The Hardest Part: Format Changes

CAMS and Karvy update their email formats without any notification. A column gets renamed. A table structure changes. A new section appears before the transaction table, shifting all the row indices.

The naive approach — hardcoded column indices — breaks silently. The parser runs, extracts nothing, writes nothing, and the system shows no error. Portfolio data goes stale without anyone knowing.

The solution: explicit zero-record detection. If the parser returns an empty list from a non-empty email, that is flagged as a parsing failure — not treated as "no transactions today."

@app.post("/ingest") async def ingest_email(request: EmailPayload, db: AsyncSession = Depends(get_db)): email_data = extract_email_body(request.raw_email) rta = detect_rta(email_data) if rta == "UNKNOWN": await flag_for_review(email_data, reason="unknown_rta") return {"status": "flagged", "reason": "unknown_rta"} parser = PARSERS[rta] transactions = parser(email_data) # Critical: empty result from non-empty email = format change email_has_content = len(email_data["text"].strip()) > 200 if len(transactions) == 0 and email_has_content: await flag_for_review(email_data, reason="parser_returned_empty") return {"status": "flagged", "reason": "possible_format_change"} result = await validate_and_write(transactions, db) return {"status": "ok", **result}

This means every format change gets caught immediately and routed to a review queue — where someone can inspect the email, update the parser if needed, and reprocess. No silent data loss.

06 / Results

After deploying the pipeline:

01

Manual data entry time: ~0 hours per day

Previously took hours per MFD per day. Now fully automated for all MFDs on the platform simultaneously.

02

Email to DB time: under 5 minutes

From RTA email arriving to portfolio data updated in the database. Previously same-day or next-day depending on when someone manually processed it.

03

Zero silent failures

Every parsing anomaly is flagged and routed to review. The pipeline either succeeds loudly or fails loudly — never silently produces wrong data.

07 / What I Would Do Differently

If I built this again, I would add a format fingerprinting system — a hash of the email structure (table count, column count, column names) stored alongside each parsed record. When a format change happens, the fingerprint mismatch triggers an immediate alert before even attempting to parse — rather than detecting the problem after an empty parse result.

I would also build a format versioning system — CAMS_v1, CAMS_v2 — so that when a format changes, the old parser keeps working for older emails in the queue while the new parser handles fresh ones. Currently a format update requires redeploying the parser.