Reconciliation Engines and Double-Entry Ledger Design

If the payment gateway is the engine, the ledger is the black box flight recorder. Every financial event — authorization, capture, refund, chargeback, fee — is recorded as a pair of balanced entries that maintain an invariant: debits always equal credits. When they don’t, money has appeared from nowhere or disappeared, and you have a financial incident.

The diagram shows the double-entry recording for a typical payment flow: customer pays merchant, gateway collects fees, PSP settles.

Double-Entry Ledger

Double-entry bookkeeping is a 700-year-old technology that remains the foundation of all financial systems. Every transaction produces exactly two entries: a debit and a credit, always equal in amount. The system is self-auditing — if debits don’t equal credits, something is wrong.

from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional
import uuid

class EntryType(Enum):
    DEBIT = "debit"
    CREDIT = "credit"

class AccountType(Enum):
    """
    Account types in a payment gateway's chart of accounts.
    
    Asset accounts (debit-normal):
      - Cash/bank accounts increase with debits
      - Receivables increase with debits
    
    Liability accounts (credit-normal):
      - Payables increase with credits
      - Customer balances increase with credits
    
    Revenue accounts (credit-normal):
      - Fees earned increase with credits
    """
    ASSET = "asset"
    LIABILITY = "liability"
    REVENUE = "revenue"
    EXPENSE = "expense"

@dataclass
class LedgerEntry:
    """
    A single ledger entry (one side of a double-entry transaction).
    """
    entry_id: str
    journal_id: str          # Groups related entries
    account_id: str
    entry_type: EntryType
    amount: Decimal
    currency: str
    
    # Reference data
    payment_id: str = ""
    description: str = ""
    
    # Timestamps
    created_at: datetime = field(default_factory=datetime.utcnow)
    effective_date: datetime = field(default_factory=datetime.utcnow)
    
    # Reconciliation
    reconciled: bool = False
    reconciled_at: Optional[datetime] = None
    settlement_id: str = ""

@dataclass
class JournalEntry:
    """
    A complete journal entry: a balanced set of debits and credits.
    
    Invariant: sum(debits) == sum(credits) for each journal entry.
    This invariant is enforced at creation time and can never be
    violated — there is no API to create an unbalanced entry.
    """
    journal_id: str
    entries: list[LedgerEntry]
    description: str
    created_at: datetime = field(default_factory=datetime.utcnow)
    
    def validate(self) -> bool:
        """Verify the fundamental accounting equation: debits = credits."""
        debits = sum(
            e.amount for e in self.entries if e.entry_type == EntryType.DEBIT
        )
        credits = sum(
            e.amount for e in self.entries if e.entry_type == EntryType.CREDIT
        )
        return debits == credits


class PaymentLedger:
    """
    Double-entry ledger for payment processing.
    
    Chart of accounts for a payment gateway:
    
    ASSETS:
      - settlement_receivable: money the PSP owes us
      - bank_account: actual cash in bank
    
    LIABILITIES:
      - merchant_payable: money we owe merchants
      - customer_refund_payable: refunds owed to customers
    
    REVENUE:
      - processing_fees: our transaction fees
      - markup_fees: interchange markup
    
    EXPENSE:
      - psp_fees: fees charged by PSPs
      - interchange_fees: fees charged by card networks
      - chargeback_costs: chargeback penalties
    """
    
    # Standard account IDs
    SETTLEMENT_RECEIVABLE = "asset:settlement_receivable"
    BANK_ACCOUNT = "asset:bank_account"
    MERCHANT_PAYABLE = "liability:merchant_payable"
    PROCESSING_FEE_REVENUE = "revenue:processing_fees"
    PSP_FEE_EXPENSE = "expense:psp_fees"
    CHARGEBACK_EXPENSE = "expense:chargeback_costs"
    
    def __init__(self):
        self._entries: list[LedgerEntry] = []
        self._journals: list[JournalEntry] = []
    
    def record_capture(
        self, payment_id: str, amount: Decimal,
        currency: str, gateway_fee: Decimal, psp_fee: Decimal,
        merchant_id: str
    ) -> JournalEntry:
        """
        Record a payment capture in the ledger.
        
        When a payment is captured:
        1. PSP owes us the full amount (settlement receivable ↑)
        2. We owe the merchant the amount minus our fee (merchant payable ↑)
        3. We earned processing fees (revenue ↑)
        4. We owe the PSP their fee (expense ↑)
        
        Debits:
          settlement_receivable: amount (PSP will pay us)
          psp_fee_expense: psp_fee (cost of using the PSP)
        
        Credits:
          merchant_payable:amount - gateway_fee (we owe merchant)
          processing_fee_revenue: gateway_fee (our fee)
          settlement_receivable: psp_fee (netted from settlement)
        
        Note: the PSP fee is typically netted from the settlement
        amount, so the actual settlement receivable is (amount - psp_fee).
        """
        journal_id = str(uuid.uuid4())
        net_settlement = amount - psp_fee
        merchant_payout = amount - gateway_fee
        
        entries = [
            # Debit: PSP owes us the net settlement amount
            LedgerEntry(
                entry_id=str(uuid.uuid4()),
                journal_id=journal_id,
                account_id=self.SETTLEMENT_RECEIVABLE,
                entry_type=EntryType.DEBIT,
                amount=net_settlement,
                currency=currency,
                payment_id=payment_id,
                description="Settlement receivable from PSP",
            ),
            # Debit: PSP fee is an expense
            LedgerEntry(
                entry_id=str(uuid.uuid4()),
                journal_id=journal_id,
                account_id=self.PSP_FEE_EXPENSE,
                entry_type=EntryType.DEBIT,
                amount=psp_fee,
                currency=currency,
                payment_id=payment_id,
                description="PSP processing fee",
            ),
            # Credit: We owe the merchant their portion
            LedgerEntry(
                entry_id=str(uuid.uuid4()),
                journal_id=journal_id,
                account_id=f"liability:merchant_payable:{merchant_id}",
                entry_type=EntryType.CREDIT,
                amount=merchant_payout,
                currency=currency,
                payment_id=payment_id,
                description=f"Merchant payout for payment {payment_id}",
            ),
            # Credit: Our processing fee revenue
            LedgerEntry(
                entry_id=str(uuid.uuid4()),
                journal_id=journal_id,
                account_id=self.PROCESSING_FEE_REVENUE,
                entry_type=EntryType.CREDIT,
                amount=gateway_fee,
                currency=currency,
                payment_id=payment_id,
                description="Gateway processing fee",
            ),
        ]
        
        journal = JournalEntry(
            journal_id=journal_id,
            entries=entries,
            description=f"Capture payment {payment_id}: {amount} {currency}",
        )
        
        if not journal.validate():
            raise LedgerImbalance(
                f"Journal {journal_id} is unbalanced: "
                f"debits={net_settlement + psp_fee}, "
                f"credits={merchant_payout + gateway_fee}"
            )
        
        self._journals.append(journal)
        self._entries.extend(entries)
        
        return journal
    
    def record_refund(
        self, payment_id: str, refund_amount: Decimal,
        currency: str, merchant_id: str
    ) -> JournalEntry:
        """
        Record a refund.
        
        A refund reverses the capture:
        1. We no longer owe the merchant the refunded amount
        2. We owe the customer the refund (through PSP)
        3. Our settlement receivable decreases
        """
        journal_id = str(uuid.uuid4())
        
        entries = [
            # Debit: reduce merchant payable
            LedgerEntry(
                entry_id=str(uuid.uuid4()),
                journal_id=journal_id,
                account_id=f"liability:merchant_payable:{merchant_id}",
                entry_type=EntryType.DEBIT,
                amount=refund_amount,
                currency=currency,
                payment_id=payment_id,
                description=f"Refund reduces merchant payable",
            ),
            # Credit: reduce settlement receivable
            LedgerEntry(
                entry_id=str(uuid.uuid4()),
                journal_id=journal_id,
                account_id=self.SETTLEMENT_RECEIVABLE,
                entry_type=EntryType.CREDIT,
                amount=refund_amount,
                currency=currency,
                payment_id=payment_id,
                description="Refund reduces settlement receivable",
            ),
        ]
        
        journal = JournalEntry(
            journal_id=journal_id,
            entries=entries,
            description=f"Refund payment {payment_id}: {refund_amount} {currency}",
        )
        
        if not journal.validate():
            raise LedgerImbalance(f"Refund journal {journal_id} is unbalanced")
        
        self._journals.append(journal)
        self._entries.extend(entries)
        return journal
    
    def get_account_balance(self, account_id: str) -> Decimal:
        """
        Compute the current balance of an account.
        
        For debit-normal accounts (assets, expenses):
          balance = sum(debits) - sum(credits)
        
        For credit-normal accounts (liabilities, revenue):
          balance = sum(credits) - sum(debits)
        """
        debits = sum(
            e.amount for e in self._entries
            if e.account_id == account_id and e.entry_type == EntryType.DEBIT
        )
        credits = sum(
            e.amount for e in self._entries
            if e.account_id == account_id and e.entry_type == EntryType.CREDIT
        )
        
        # Determine account type from ID prefix
        if account_id.startswith("asset:") or account_id.startswith("expense:"):
            return debits - credits
        else:
            return credits - debits
    
    def trial_balance(self) -> dict:
        """
        Compute the trial balance: all account balances.
        
        The trial balance must balance: total debits == total credits.
        If it doesn't, there's a bug in the ledger.
        """
        accounts: dict[str, Decimal] = {}
        
        for entry in self._entries:
            if entry.account_id not in accounts:
                accounts[entry.account_id] = Decimal(0)
            
            if entry.entry_type == EntryType.DEBIT:
                accounts[entry.account_id] += entry.amount
            else:
                accounts[entry.account_id] -= entry.amount
        
        total_debits = sum(
            e.amount for e in self._entries if e.entry_type == EntryType.DEBIT
        )
        total_credits = sum(
            e.amount for e in self._entries if e.entry_type == EntryType.CREDIT
        )
        
        return {
            "accounts": accounts,
            "total_debits": total_debits,
            "total_credits": total_credits,
            "is_balanced": total_debits == total_credits,
        }

class LedgerImbalance(Exception):
    pass

Settlement File Reconciliation

PSPs send daily settlement files listing all transactions they processed. Reconciliation matches these against the gateway’s records to find discrepancies:

@dataclass
class SettlementRecord:
    """A single record from a PSP settlement file."""
    psp_reference: str
    transaction_date: datetime
    settlement_date: datetime
    gross_amount: Decimal
    fee_amount: Decimal
    net_amount: Decimal
    currency: str
    transaction_type: str   # "capture", "refund", "chargeback"
    
@dataclass
class ReconciliationResult:
    """Result of reconciling a single transaction."""
    psp_reference: str
    status: str              # "matched", "amount_mismatch", "missing_in_gateway",
                            #  "missing_in_settlement", "fee_mismatch"
    gateway_amount: Optional[Decimal] = None
    settlement_amount: Optional[Decimal] = None
    difference: Optional[Decimal] = None
    resolution: str = ""     # "auto_resolved", "manual_review", "adjusted"

class ReconciliationEngine:
    """
    Three-way reconciliation engine.
    
    Reconciliation levels:
    1. Transaction-level: match individual transactions
    2. Settlement-level: match settlement batch totals
    3. Bank-level: match bank statement against settlement totals
    
    This implements level 1 (transaction-level) matching.
    """
    
    def __init__(self, tolerance: Decimal = Decimal("0.01")):
        """
        tolerance: maximum acceptable difference between amounts
        before flagging as a mismatch (handles rounding differences).
        """
        self._tolerance = tolerance
    
    def reconcile(
        self, gateway_records: list[dict],
        settlement_records: list[SettlementRecord]
    ) -> list[ReconciliationResult]:
        """
        Match gateway records against settlement file records.
        
        Algorithm:
        1. Index both sets by PSP reference
        2. For each reference in either set:
           - Present in both → compare amounts
           - Present only in gateway → "missing_in_settlement"
           - Present only in settlement → "missing_in_gateway"
        """
        # Index by PSP reference
        gateway_by_ref = {r["psp_reference"]: r for r in gateway_records}
        settlement_by_ref = {r.psp_reference: r for r in settlement_records}
        
        all_refs = set(gateway_by_ref.keys()) | set(settlement_by_ref.keys())
        results = []
        
        for ref in all_refs:
            gw = gateway_by_ref.get(ref)
            st = settlement_by_ref.get(ref)
            
            if gw and st:
                # Both exist — compare amounts
                gw_amount = Decimal(str(gw["amount"]))
                st_amount = st.gross_amount
                difference = abs(gw_amount - st_amount)
                
                if difference <= self._tolerance:
                    # Check fees
                    gw_fee = Decimal(str(gw.get("psp_fee", 0)))
                    fee_diff = abs(gw_fee - st.fee_amount)
                    
                    if fee_diff <= self._tolerance:
                        status = "matched"
                    else:
                        status = "fee_mismatch"
                else:
                    status = "amount_mismatch"
                
                results.append(ReconciliationResult(
                    psp_reference=ref,
                    status=status,
                    gateway_amount=gw_amount,
                    settlement_amount=st_amount,
                    difference=difference if status != "matched" else None,
                ))
            
            elif gw and not st:
                results.append(ReconciliationResult(
                    psp_reference=ref,
                    status="missing_in_settlement",
                    gateway_amount=Decimal(str(gw["amount"])),
                ))
            
            elif st and not gw:
                results.append(ReconciliationResult(
                    psp_reference=ref,
                    status="missing_in_gateway",
                    settlement_amount=st.gross_amount,
                ))
        
        return results
    
    def generate_report(
        self, results: list[ReconciliationResult]
    ) -> dict:
        """Generate a reconciliation summary report."""
        total = len(results)
        matched = sum(1 for r in results if r.status == "matched")
        
        mismatches = [r for r in results if r.status != "matched"]
        
        total_discrepancy = sum(
            r.difference or Decimal(0) for r in mismatches
        )
        
        return {
            "total_transactions": total,
            "matched": matched,
            "match_rate": f"{matched/total*100:.2f}%" if total > 0 else "N/A",
            "mismatches": {
                "amount_mismatch": sum(
                    1 for r in results if r.status == "amount_mismatch"
                ),
                "fee_mismatch": sum(
                    1 for r in results if r.status == "fee_mismatch"
                ),
                "missing_in_settlement": sum(
                    1 for r in results if r.status == "missing_in_settlement"
                ),
                "missing_in_gateway": sum(
                    1 for r in results if r.status == "missing_in_gateway"
                ),
            },
            "total_discrepancy": str(total_discrepancy),
        }

Settlement File Parsing

PSPs use various file formats. A production reconciliation engine must parse them all:

import csv
from io import StringIO

class SettlementFileParser:
    """
    Parses settlement files from different PSPs.
    
    Common formats:
    - CSV with PSP-specific column layouts
    - Fixed-width (legacy systems)
    - XML/JSON (modern APIs)
    - ISO 20022 pain.002 (bank settlement confirmations)
    """
    
    @staticmethod
    def parse_csv_settlement(
        file_content: str,
        column_mapping: dict[str, str]
    ) -> list[SettlementRecord]:
        """
        Parse a CSV settlement file with configurable column mapping.
        
        column_mapping maps our standard field names to the PSP's
        column headers. Example:
        {
            "psp_reference": "Transaction ID",
            "transaction_date": "Created",
            "gross_amount": "Gross",
            "fee_amount": "Fee",
            "net_amount": "Net",
            "currency": "Currency",
            "transaction_type": "Type",
        }
        """
        records = []
        reader = csv.DictReader(StringIO(file_content))
        
        for row in reader:
            try:
                record = SettlementRecord(
                    psp_reference=row[column_mapping["psp_reference"]].strip(),
                    transaction_date=datetime.fromisoformat(
                        row[column_mapping["transaction_date"]].strip()
                    ),
                    settlement_date=datetime.fromisoformat(
                        row.get(
                            column_mapping.get("settlement_date", ""),
                            row[column_mapping["transaction_date"]]
                        ).strip()
                    ),
                    gross_amount=Decimal(
                        row[column_mapping["gross_amount"]]
                        .strip().replace(",", "")
                    ),
                    fee_amount=abs(Decimal(
                        row[column_mapping["fee_amount"]]
                        .strip().replace(",", "")
                    )),
                    net_amount=Decimal(
                        row[column_mapping["net_amount"]]
                        .strip().replace(",", "")
                    ),
                    currency=row[column_mapping["currency"]].strip(),
                    transaction_type=row[
                        column_mapping["transaction_type"]
                    ].strip().lower(),
                )
                records.append(record)
            except (KeyError, ValueError) as e:
                # Log parsing error but continue with remaining records
                continue
        
        return records

Reconciliation Metrics

“Missing in gateway” (PSP reports a transaction we don’t have) is the most critical exception — it means money moved that our system doesn’t know about. This is either a bug in our recording, a PSP error, or fraud. Every instance requires immediate investigation.

A well-functioning reconciliation engine is invisible — it runs daily, matches 99.9%+ of transactions automatically, and surfaces only the genuine exceptions that need human attention. The ledger underneath guarantees that no financial event goes unrecorded, and the three-way reconciliation (gateway ↔ PSP ↔ bank) provides independent verification at every layer.