Bank Reconciliation in Python: Building a Plaid Integration from Scratch

Edwards Tech Innovations — Sun, 12 Apr 2026 23:11:55 +0000

Bank Reconciliation in Python: Building a Plaid Integration from Scratch

If you're building a fintech product, a business dashboard, or any application that needs to pull bank data, you're going to need Plaid. But Plaid's API surface is large, and the official SDK has a lot of ceremony around request/response objects.

In this tutorial, I'll walk through a complete Plaid integration module -- 337 lines of Python -- that handles the full lifecycle: link tokens, account access, transaction sync, reconciliation, and spending categorization.

Prerequisites

You will need:

A Plaid developer account (sandbox is free)
Python 3.9+
The plaid-python SDK

pip install plaid-python python-dotenv

Set up your .env:

PLAID_CLIENT_ID=your_client_id
PLAID_SECRET=your_sandbox_secret
PLAID_ENV=sandbox

Step 1: Initialize the Client

The first design decision is handling the case where Plaid credentials aren't available. In a multi-integration system, not every deployment will have every service configured. We use a try/except on import and an initialized flag.

import os
from typing import Dict, Any, List
from datetime import datetime, timedelta

try:
    import plaid
    from plaid.api import plaid_api
    from plaid.model.products import Products
    from plaid.model.country_code import CountryCode
    from plaid.model.link_token_create_request import LinkTokenCreateRequest
    from plaid.model.link_token_create_request_user import LinkTokenCreateRequestUser
    from plaid.model.item_public_token_exchange_request import ItemPublicTokenExchangeRequest
    from plaid.model.transactions_get_request import TransactionsGetRequest
    from plaid.model.transactions_get_request_options import TransactionsGetRequestOptions
    from plaid.model.accounts_get_request import AccountsGetRequest
    from plaid.model.accounts_balance_get_request import AccountsBalanceGetRequest
    PLAID_AVAILABLE = True
except ImportError:
    PLAID_AVAILABLE = False


class PlaidIntegration:
    def __init__(self):
        self.client_id = os.getenv("PLAID_CLIENT_ID")
        self.secret = os.getenv("PLAID_SECRET")
        self.env = os.getenv("PLAID_ENV", "sandbox")
        self.initialized = bool(self.client_id and self.secret and PLAID_AVAILABLE)

        if self.initialized:
            if self.env == "production":
                host = plaid.Environment.Production
            elif self.env == "development":
                host = plaid.Environment.Development
            else:
                host = plaid.Environment.Sandbox

            configuration = plaid.Configuration(
                host=host,
                api_key={
                    "clientId": self.client_id,
                    "secret": self.secret
                }
            )
            api_client = plaid.ApiClient(configuration)
            self.client = plaid_api.PlaidApi(api_client)

The environment toggle (sandbox / development / production) is read from an env var with a safe default. In production, you never want to accidentally hit sandbox.

Step 2: Link Tokens -- The Entry Point

Plaid Link is a drop-in frontend component that lets users securely connect their bank accounts. Your backend creates a link token, your frontend uses it to open the Link flow, and then you exchange the resulting public token for a persistent access token.

def create_link_token(self, user_id: str, products: List[str] = None) -> Dict[str, Any]:
    """Create a link token for Plaid Link"""
    try:
        product_list = [Products(p) for p in (products or ["transactions"])]

        request = LinkTokenCreateRequest(
            products=product_list,
            client_name="Your App Name",
            country_codes=[CountryCode("US")],
            language="en",
            user=LinkTokenCreateRequestUser(client_user_id=user_id)
        )

        response = self.client.link_token_create(request)
        return {
            "success": True,
            "link_token": response.link_token,
            "expiration": response.expiration.isoformat()
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

The products parameter controls what data you can access. Common options:

"transactions" -- account balances and transaction history
"auth" -- account and routing numbers (for ACH)
"identity" -- account holder name, address, email

You can request multiple products in one link session.

Step 3: Token Exchange

When the user completes the Link flow, Plaid gives your frontend a public_token. This is temporary. You exchange it for a permanent access_token that you store server-side.

def exchange_public_token(self, public_token: str) -> Dict[str, Any]:
    """Exchange public token for access token"""
    try:
        request = ItemPublicTokenExchangeRequest(public_token=public_token)
        response = self.client.item_public_token_exchange(request)
        return {
            "success": True,
            "access_token": response.access_token,
            "item_id": response.item_id
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

Important: The access_token is a credential. Store it encrypted. Never log it. Never send it to the frontend.

Step 4: Pulling Transactions

This is where it gets useful. Plaid's transaction endpoint supports date ranges, pagination, and account filtering.

def get_transactions(self, access_token: str, start_date: str = None,
                    end_date: str = None, account_ids: List[str] = None,
                    count: int = 100, offset: int = 0) -> Dict[str, Any]:
    """Get transactions for date range"""
    try:
        if not start_date:
            start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
        if not end_date:
            end_date = datetime.now().strftime("%Y-%m-%d")

        options = TransactionsGetRequestOptions(count=count, offset=offset)
        if account_ids:
            options.account_ids = account_ids

        request = TransactionsGetRequest(
            access_token=access_token,
            start_date=datetime.strptime(start_date, "%Y-%m-%d").date(),
            end_date=datetime.strptime(end_date, "%Y-%m-%d").date(),
            options=options
        )

        response = self.client.transactions_get(request)

        return {
            "success": True,
            "total_transactions": response.total_transactions,
            "transactions": [{
                "transaction_id": t.transaction_id,
                "account_id": t.account_id,
                "date": t.date.isoformat(),
                "name": t.name,
                "merchant_name": t.merchant_name,
                "amount": t.amount,
                "currency": t.iso_currency_code,
                "category": t.category,
                "pending": t.pending,
                "payment_channel": t.payment_channel.value if t.payment_channel else None
            } for t in response.transactions]
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

Plaid returns up to 500 transactions per request. For accounts with heavy activity, you need to paginate:

def get_all_transactions(self, access_token: str, start_date: str = None,
                        end_date: str = None) -> List[Dict]:
    """Get all transactions (handles pagination)"""
    all_transactions = []
    offset = 0
    count = 500

    while True:
        result = self.get_transactions(
            access_token, start_date, end_date,
            count=count, offset=offset
        )
        if not result.get("success"):
            break

        transactions = result.get("transactions", [])
        all_transactions.extend(transactions)

        if len(transactions) < count:
            break
        offset += count

    return all_transactions

Step 5: Bank Reconciliation

Reconciliation is matching your internal records against what the bank actually shows. This is where business applications diverge from simple "show me my transactions" dashboards.

The approach here is straightforward: match by amount (within a penny tolerance), then report what matched, what's in the bank but not in your records, and what's in your records but not in the bank.

def reconcile_transactions(self, access_token: str, expected_transactions: List[Dict],
                          start_date: str = None, end_date: str = None) -> Dict[str, Any]:
    """Reconcile bank transactions against expected transactions"""
    try:
        bank_txns = self.get_all_transactions(access_token, start_date, end_date)

        matched = []
        unmatched_bank = []
        unmatched_expected = list(expected_transactions)

        for bank_txn in bank_txns:
            found_match = False
            for i, exp_txn in enumerate(unmatched_expected):
                if abs(bank_txn["amount"] - exp_txn.get("amount", 0)) < 0.01:
                    matched.append({
                        "bank_transaction": bank_txn,
                        "expected_transaction": exp_txn,
                        "status": "matched"
                    })
                    unmatched_expected.pop(i)
                    found_match = True
                    break

            if not found_match:
                unmatched_bank.append(bank_txn)

        return {
            "success": True,
            "matched_count": len(matched),
            "unmatched_bank_count": len(unmatched_bank),
            "unmatched_expected_count": len(unmatched_expected),
            "matched": matched,
            "unmatched_bank": unmatched_bank,
            "unmatched_expected": unmatched_expected,
            "reconciliation_status": "complete" if not unmatched_bank and not unmatched_expected else "needs_review"
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

The reconciliation_status field gives you a quick check: "complete" means everything lines up, "needs_review" means a human should look at the discrepancies.

Step 6: Spending Categorization

Plaid provides category data with transactions, but the raw categories are nested lists. This method rolls them up into a summary with totals:

def categorize_transactions(self, transactions: List[Dict]) -> Dict[str, Any]:
    """Categorize transactions by type"""
    categories = {}
    for txn in transactions:
        cat = txn.get("category", ["Uncategorized"])[0] if txn.get("category") else "Uncategorized"
        if cat not in categories:
            categories[cat] = {"count": 0, "total": 0, "transactions": []}
        categories[cat]["count"] += 1
        categories[cat]["total"] += txn.get("amount", 0)
        categories[cat]["transactions"].append(txn)

    return {
        "success": True,
        "categories": categories,
        "summary": {cat: {"count": data["count"], "total": data["total"]}
                   for cat, data in categories.items()}
    }

And for a full spending report with top merchants:

def get_spending_summary(self, access_token: str, start_date: str = None,
                        end_date: str = None) -> Dict[str, Any]:
    """Get spending summary by category"""
    transactions = self.get_all_transactions(access_token, start_date, end_date)

    # In Plaid, positive amounts are debits (money out)
    expenses = [t for t in transactions if t["amount"] > 0]
    income = [t for t in transactions if t["amount"] < 0]

    categorized = self.categorize_transactions(expenses)

    return {
        "success": True,
        "period": {"start": start_date, "end": end_date},
        "total_expenses": sum(t["amount"] for t in expenses),
        "total_income": abs(sum(t["amount"] for t in income)),
        "transaction_count": len(transactions),
        "expense_categories": categorized.get("summary", {}),
        "top_merchants": self._get_top_merchants(expenses, limit=10)
    }

One thing that catches people: Plaid uses positive numbers for debits and negative numbers for credits. This is the opposite of what most people expect. The code above handles this correctly by filtering on sign.

Putting It Together

With the singleton pattern, using this in your application is straightforward:

from dotenv import load_dotenv
load_dotenv()

from eti_integrations import get_plaid

plaid = get_plaid()

# Check if Plaid is configured
if not plaid.initialized:
    print("Set PLAID_CLIENT_ID and PLAID_SECRET in .env")
else:
    # Create a link token for your frontend
    link = plaid.create_link_token(user_id="user_123")
    print(f"Link token: {link['link_token']}")

    # After the user completes Link and you have an access token:
    summary = plaid.get_spending_summary(access_token, "2025-01-01", "2025-03-31")
    print(f"Total expenses: ${summary['total_expenses']:.2f}")
    print(f"Total income: ${summary['total_income']:.2f}")
    for category, data in summary["expense_categories"].items():
        print(f"  {category}: ${data['total']:.2f} ({data['count']} transactions)")

What I'd Build Next

This module covers the core Plaid use cases. If I were extending it, I'd add:

Plaid Investments for brokerage account data
Plaid Liabilities for credit card and loan balances
Webhook handling for real-time transaction updates instead of polling
Caching for balance requests that don't need real-time data

This Plaid module is one of 9 production Python modules in the AI Integration Pack. The other 8 cover Stripe, PayPal, Twilio, Airtable, ClickUp, Outlook, and e-commerce platforms (Gumroad, LemonSqueezy, Paddle). 4,488 lines of code total, 152 tests, security-audited.

If you're interested: AI Integration Pack on Gumroad -- $49 for the full pack.

DEV Community: Edwards Tech Innovations

Bank Reconciliation in Python: Building a Plaid Integration from Scratch

Bank Reconciliation in Python: Building a Plaid Integration from Scratch

Prerequisites

Step 1: Initialize the Client

Step 2: Link Tokens -- The Entry Point

Step 3: Token Exchange

Step 4: Pulling Transactions

Step 5: Bank Reconciliation

Step 6: Spending Categorization

Putting It Together

What I'd Build Next