Pandas for Trade Data Processing

A single trade that fails to tie out — a counterparty confirm whose price is off by a quantized cent, a delivery date shifted one interval by a botched timezone conversion — does not announce itself. It rides silently through position netting and mark-to-market until the T+1 settlement statement lands and the desk’s booked P&L disagrees with the ISO’s. The failure mode this component prevents is exactly that: a non-deterministic, un-audited transformation between raw counterparty feeds and the reconciled ledger, where a break can appear or disappear depending on row order, dtype coercion, or floating-point drift. Within the Trade Ingestion & Matching Workflows framework, pandas is the in-memory engine that normalizes heterogeneous submissions, aligns them against internal ETRM records on a stable composite key, and produces a matched set and an exception set that are byte-identical every time the same inputs replay — whether the run fires for the preliminary D+1 cycle or a D+90 dispute.

The flowchart below shows how heterogeneous source feeds converge into a single validated DataFrame, then feed the outer-join matching step that splits records into reconciled and exception sets.

Every stage after validation is a pure, replayable transformation over trade-indexed data: given the same source payloads and the same code revision, the matching layer must emit the same reconciled ledger and the same exception rows. The sections below define the matching data model the layer operates over, the standards that bound it, the stage-by-stage implementation, the edge cases that quietly corrupt a match, and the alerting and verification controls that keep an undetected break out of settlement.

The Matching Data Model and Composite Keys #

A trade match is a join, and the quality of the join is decided entirely by the key. Energy trades carry no globally unique cross-party identifier — the internal trade_id a desk assigns is meaningless to the counterparty and to the ISO — so reconciliation is performed on a composite key of economically meaningful attributes that both sides independently record. The canonical key is (delivery_date, product_code, counterparty_id, delivery_point), optionally widened with hour_ending for hourly power. Every column in the key must be canonicalized to a single representation before the merge, because pandas joins on exact equality: "PJM-WEST" and "pjm_west" are different keys, and a category dtype does not rescue a value that was never normalized.

Once joined, each row carries a match status. The indicator column produced by pd.merge gives the structural verdict — present on both sides, internal-only, or external-only — but the economic verdict requires a tolerance test on the quantities that both sides report. A pair reconciles only when volume and price agree within configured bands:

$$\left| V^{\text{int}}{k} - V^{\text{ext}}{k} \right| \le \tau_V \quad\text{and}\quad \left| P^{\text{int}}{k} - P^{\text{ext}}{k} \right| \le \tau_P$$

where \(k\) is the composite key, \(V\) is volume in MWh (or Dth for gas), \(P\) is the settlement price, and \(\tau_V, \tau_P\) are the volume and price tolerances. The taxonomy of outcomes the matching layer must produce is fixed:

The delivery_date and hour_ending components of the key are produced upstream by the Settlement Cycle Mapping engine, and the field-level shape of each external feed — column names, units, timestamp encoding — is enforced by the ISO/RTO Data Format Standards. The reconciled price column feeds directly into the Pricing Logic Implementation engine, so a price break that slips through here becomes a mispriced charge downstream.

Specification & Standards Reference #

Trade matching is bounded by the same messaging and reporting mandates that govern the confirmations being matched. A pipeline that ignores them can produce an arithmetically clean match that still fails an audit:

• NAESB WGQ / WEQ business practices define the EDI transaction sets — 867 (product transfer / meter usage) and 810 (invoice) — through which counterparty volumes and prices arrive, and the nomination and confirmation cycle those documents settle against.
• FERC requires, under the Open Access Transmission Tariff and the Uniform System of Accounts, that every settled trade be traceable from the confirmed record back to its source submission and forward to its financial posting — the audit lineage the matching layer must emit.
• CFTC swap-data and large-trader reporting obligations mean an unmatched or misclassified trade is not only a P&L risk but a reportable-data risk; the exception queue is part of the compliance surface.
• ISO/RTO settlement statement formats (PJM billing line items, MISO market settlement statements, CAISO settlement statement BXML, ERCOT settlement extracts) define the external side of the join and its charge-code granularity.
• SOX controls require that the transformation from raw feed to ledger be deterministic, versioned, and reproducible — the reason every match runs as a pure function over quantized inputs.

The upstream contract enforcement that guarantees a payload is even eligible for matching is the job of the Schema Validation Frameworks component; this page assumes structurally valid records and focuses on the economic match.

Step-by-Step Implementation #

The pipeline is five pure stages: ingest and validate, normalize and canonicalize the key, deterministic outer merge, Decimal-exact tolerance classification, and audited persistence. Financial quantities are read as strings and carried as Decimal end to end — never as float — so that a price never drifts across a rounding boundary between runs.

1. Ingest and validate the raw feed #

Read money and volume columns as strings so no binary float ever touches a settlement quantity, declare explicit dtypes for everything else, and validate the frame against a declarative contract before any matching logic runs. Records that fail validation are collected lazily and routed to the exception queue rather than silently dropped.

import pandas as pd
import pandera as pa
from pandera.typing import DataFrame, Series


class TradeFeedSchema(pa.DataFrameModel):
    trade_id: Series[str] = pa.Field(str_matches=r"^TRD-\d{8}$")
    delivery_date: Series[pd.DatetimeTZDtype] = pa.Field(
        dtype_kwargs={"unit": "ns", "tz": "UTC"}
    )
    product_code: Series[str] = pa.Field(nullable=False)
    counterparty_id: Series[str] = pa.Field(nullable=False)
    delivery_point: Series[str] = pa.Field(nullable=False)
    # Money and volume arrive as strings to preserve Decimal precision.
    volume_mwh: Series[str] = pa.Field(nullable=False)
    price_usd: Series[str] = pa.Field(nullable=False)
    source_system: Series[str] = pa.Field(isin=["ETRM", "COUNTERPARTY", "ISO"])


def ingest_and_validate(raw_path: str) -> DataFrame[TradeFeedSchema]:
    trades_df = pd.read_csv(
        raw_path,
        dtype={
            "trade_id": "string",
            "product_code": "string",
            "counterparty_id": "string",
            "delivery_point": "string",
            "volume_mwh": "string",   # keep as text -> Decimal later
            "price_usd": "string",
            "source_system": "category",
        },
        parse_dates=["delivery_date"],
    )
    # date_parser was removed in pandas 2.0; normalize to UTC explicitly.
    trades_df["delivery_date"] = pd.to_datetime(trades_df["delivery_date"], utc=True)
    return TradeFeedSchema.validate(trades_df, lazy=True)

2. Normalize types and canonicalize the composite key #

Coerce every key column to one canonical representation so the join sees exact equality, convert the string money columns to Decimal, and cast the finished key columns to category for join-time memory and cache efficiency. Canonicalization is the single highest-leverage step: most “mysterious” unmatched trades are a case-fold or whitespace mismatch, not a real break.

from decimal import Decimal

KEY_COLS = ["delivery_date", "product_code", "counterparty_id", "delivery_point"]

# Delivery-point aliases resolved to internal canonical names.
POINT_ALIASES = {"pjm_west": "PJM-WEST", "hh": "NG-HENRY-HUB", "socal": "SOCAL-CITYGATE"}


def to_decimal(series: pd.Series) -> pd.Series:
    # Empty / null -> None so a missing quantity never becomes Decimal('0').
    return series.map(lambda v: Decimal(v) if isinstance(v, str) and v.strip() else None)


def normalize_trades(trades_df: pd.DataFrame) -> pd.DataFrame:
    trades_df = trades_df.copy()
    trades_df["product_code"] = trades_df["product_code"].str.strip().str.upper()
    trades_df["counterparty_id"] = trades_df["counterparty_id"].str.strip().str.upper()
    trades_df["delivery_point"] = (
        trades_df["delivery_point"].str.strip().str.lower().map(POINT_ALIASES)
        .fillna(trades_df["delivery_point"].str.strip().str.upper())
    )
    trades_df["volume_mwh"] = to_decimal(trades_df["volume_mwh"])
    trades_df["price_usd"] = to_decimal(trades_df["price_usd"])
    for col in KEY_COLS:
        trades_df[col] = trades_df[col].astype("category")
    return trades_df

3. Run the deterministic outer merge #

Sort both frames by the composite key first — a sorted merge is cache-friendly and, more importantly, makes row order in the output reproducible — then perform a single outer merge with indicator=True. The outer join is deliberate: it keeps internal-only and external-only rows so no trade can vanish from the reconciliation surface.

def match_trades(internal_df: pd.DataFrame, external_df: pd.DataFrame) -> pd.DataFrame:
    internal_df = internal_df.sort_values(KEY_COLS).reset_index(drop=True)
    external_df = external_df.sort_values(KEY_COLS).reset_index(drop=True)

    merged_df = pd.merge(
        internal_df,
        external_df,
        on=KEY_COLS,
        how="outer",
        indicator="structural_status",
        suffixes=("_int", "_ext"),
    )
    return merged_df

4. Classify against Decimal-exact tolerances #

The structural indicator only tells you a key matched. The economic verdict compares the internal and external volume and price with Decimal subtraction and configurable bands, then writes the final match_status from the taxonomy table. Because both operands are Decimal, the comparison is exact — a price_break is a real disagreement, never a float artifact.

from decimal import Decimal

VOLUME_TOLERANCE = Decimal("0.001")   # MWh
PRICE_TOLERANCE = Decimal("0.01")     # USD


def _abs_diff(a, b):
    if a is None or b is None:
        return None
    return abs(a - b)


def classify_matches(merged_df: pd.DataFrame) -> pd.DataFrame:
    merged_df = merged_df.copy()
    vol_gap = merged_df.apply(
        lambda r: _abs_diff(r["volume_mwh_int"], r["volume_mwh_ext"]), axis=1
    )
    price_gap = merged_df.apply(
        lambda r: _abs_diff(r["price_usd_int"], r["price_usd_ext"]), axis=1
    )

    def verdict(row_idx: int) -> str:
        structural = merged_df.at[row_idx, "structural_status"]
        if structural == "left_only":
            return "unmatched_internal"
        if structural == "right_only":
            return "unmatched_external"
        vg, pg = vol_gap.iat[row_idx], price_gap.iat[row_idx]
        if vg is not None and vg > VOLUME_TOLERANCE:
            return "volume_break"
        if pg is not None and pg > PRICE_TOLERANCE:
            return "price_break"
        return "reconciled"

    merged_df["match_status"] = [verdict(i) for i in range(len(merged_df))]
    return merged_df

5. Persist with an audit hash and route exceptions #

Split the classified frame into the reconciled ledger and the routed exceptions, and stamp each output with a deterministic content hash so any two runs over identical inputs are provably identical. The hash is computed over the canonicalized, sorted rows — it is the SOX-grade fingerprint that lets an auditor confirm the ledger was not tampered with between calculation and posting.

import hashlib


def content_hash(df: pd.DataFrame) -> str:
    # Hash the sorted, stringified rows so the fingerprint is order-stable.
    canonical = df.sort_values(KEY_COLS).astype(str).to_csv(index=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def persist_results(classified_df: pd.DataFrame, run_id: str) -> dict:
    reconciled_df = classified_df[classified_df["match_status"] == "reconciled"]
    exceptions_df = classified_df[classified_df["match_status"] != "reconciled"]

    reconciled_df.to_parquet(f"ledger/reconciled_{run_id}.parquet", index=False)
    exceptions_df.to_parquet(f"exceptions/routed_{run_id}.parquet", index=False)

    return {
        "run_id": run_id,
        "reconciled_rows": len(reconciled_df),
        "exception_rows": len(exceptions_df),
        "reconciled_hash": content_hash(reconciled_df),
        "exception_hash": content_hash(exceptions_df),
    }

The pipeline reads as five pure stages across the top; below them, the money column is traced as a single highlighted lane to show that the same Decimal value crosses every stage untouched — never coerced to float, never rounded — so a tolerance comparison at stage 4 is exact and the stage 5 fingerprint is reproducible.

Edge Cases and Failure Modes #

The happy path is one merge; the reconciliation risk lives entirely in the cases below, each of which silently produces a wrong match unless explicitly handled.

• DST boundaries. On the spring-forward day an hourly power schedule has 23 delivery hours; on fall-back it has 25, with a repeated wall-clock hour. Keying on a naive local timestamp collapses or duplicates the repeated hour. Carry delivery_date as UTC (as in step 1) and derive hour_ending from the market’s interval grid, never from a local clock, so the composite key stays one-to-one across the transition.
• Float money drift. Reading price_usd as float64 means 19.99 is stored as 19.989999…; scaled across thousands of rows the accumulated error eventually trips a tolerance test and manufactures a phantom price_break. Carrying money as Decimal from parse to persist (steps 1, 2, 4) removes the failure class entirely.
• Partial fills and amendments. One internal trade may correspond to two external confirms (a split delivery), or an amended contract may reuse the key with a new economic value. A plain merge either fans out into a Cartesian product or picks an arbitrary row. Detect duplicate keys before the merge and aggregate or version them explicitly.
• Duplicate trade_id. A re-sent feed replays the same trades; without dedup the merge double-counts volume. Drop exact-duplicate rows on ingest and treat same-key-different-value rows as amendments, not duplicates.
• Schema drift. A counterparty renames price_usd to unit_price or starts sending volume in Dth instead of MWh. The pandera contract in step 1 fails fast and routes the whole feed to the exception queue rather than matching on a misaligned column.
• Zero-volume and null quantities. A 0 volume is a legitimate curtailed trade; a missing volume is a defect. to_decimal maps blanks to None so the tolerance test skips them and the row lands as a break to investigate, never as a silent reconciled on Decimal('0').

def guard_duplicate_keys(trades_df: pd.DataFrame) -> pd.DataFrame:
    dup_mask = trades_df.duplicated(subset=KEY_COLS, keep=False)
    if dup_mask.any():
        # Route colliding keys out; a fan-out merge is never allowed to run.
        collisions = trades_df[dup_mask]
        collisions.to_parquet("exceptions/key_collisions.parquet", index=False)
        return trades_df[~dup_mask]
    return trades_df

Threshold & Alerting Configuration #

Tolerances and break rates are operational parameters, not constants baked into code, so the desk can tighten them for month-end close and relax them for illiquid products without a deploy. The critical alert is not any single break but the break rate — the fraction of matched pairs falling outside tolerance — which signals a systemic feed or mapping problem rather than a one-off dispute.

from decimal import Decimal

ALERT_THRESHOLDS = {
    "break_rate_warn": Decimal("0.01"),
    "break_rate_crit": Decimal("0.05"),
    "unmatched_external_crit": 10,
}


def evaluate_alerts(classified_df: pd.DataFrame) -> str:
    total = len(classified_df)
    breaks = (classified_df["match_status"].isin(["price_break", "volume_break"])).sum()
    unmatched_ext = (classified_df["match_status"] == "unmatched_external").sum()
    break_rate = Decimal(breaks) / Decimal(total) if total else Decimal("0")

    if break_rate > ALERT_THRESHOLDS["break_rate_crit"] or (
        unmatched_ext > ALERT_THRESHOLDS["unmatched_external_crit"]
    ):
        return "critical"
    if break_rate > ALERT_THRESHOLDS["break_rate_warn"]:
        return "warning"
    return "info"

The escalation routing and dedup logic behind these tiers — suppressing repeat pages, grouping breaks by counterparty — is the domain of the Threshold Tuning & Alerts engine, which consumes the match_status column this layer emits.

Testing & Reconciliation Verification #

Verification is a shadow calculation: an independent reference implementation runs the same inputs and its output is diffed against production. Because every quantity is Decimal and every output is content-hashed, the pass condition is exact equality, not approximate agreement — any non-zero diff is a regression, never a rounding artifact.

def test_reconciled_hash_is_stable():
    # Same inputs, two runs -> identical fingerprint (SOX reproducibility).
    result_a = persist_results(classify_matches(match_trades(INT_FIXTURE, EXT_FIXTURE)), "a")
    result_b = persist_results(classify_matches(match_trades(INT_FIXTURE, EXT_FIXTURE)), "b")
    assert result_a["reconciled_hash"] == result_b["reconciled_hash"]


def test_dst_fallback_keeps_25_hours():
    # Fall-back day: 25 distinct hour_ending keys, no collapsed duplicate.
    keyed = normalize_trades(FALLBACK_DAY_FIXTURE)
    assert keyed.duplicated(subset=KEY_COLS).sum() == 0


def test_decimal_price_avoids_phantom_break():
    # 19.99 vs 19.99 must reconcile; a float pipeline would drift and break it.
    row = classify_matches(match_trades(PRICE_FIXTURE_INT, PRICE_FIXTURE_EXT))
    assert (row["match_status"] == "reconciled").all()


def test_missing_volume_routes_to_exception():
    # Blank volume -> None -> break, never a silent reconcile on Decimal('0').
    row = classify_matches(match_trades(NULL_VOL_INT, NULL_VOL_EXT))
    assert "reconciled" not in set(row["match_status"])

Shadow reconciliation runs as a diff: production reconciled_hash must equal the reference run’s hash for the same inputs, and the exception set must contain exactly the injected breaks and no others. The child page Parsing CSV vs XML trade feeds with pandas covers the format-specific ingestion that feeds step 1, and the concurrency needed to run this match across dozens of counterparty feeds without blocking is handled by the Async Batch Processing Pipelines component. For the vectorized joins and group-by aggregations this layer depends on, the official pandas documentation is the reference.

Frequently Asked Questions #

Why match on a composite key instead of a trade ID? #

Because no single identifier is shared across all three parties. The internal trade_id is assigned by the desk and is unknown to the counterparty and the ISO. Reconciliation therefore joins on economically meaningful attributes that every side records independently — delivery_date, product_code, counterparty_id, and delivery_point — after each of those columns has been canonicalized to one exact representation. Matching on trade_id would leave every external record unmatched.

Why use Python’s decimal module for trade prices instead of float? #

Settlement prices reconcile to the cent and are scaled across thousands of rows. Binary floating point cannot represent most decimal fractions exactly, so a value like 19.99 is stored as 19.989999…, and accumulated drift eventually pushes a genuinely equal pair outside the price tolerance and manufactures a phantom break. Reading money columns as strings and carrying them as Decimal from ingest through classification makes the tolerance comparison exact and the output reproducible.

How does the pipeline keep two runs over the same data identical? #

Three properties combine to guarantee it: every stage is a pure function with no hidden state, both frames are sorted by the composite key before the merge so row order is deterministic, and each output is fingerprinted with a SHA-256 hash over the sorted rows. Two runs over identical inputs produce identical hashes; a differing hash is a regression to investigate, which is exactly the SOX reproducibility control auditors expect.

When should the match move off pandas to polars? #

Keep pandas for the reconciliation reporting and the Decimal-exact tolerance logic, where correctness and auditability dominate. For daily settlement runs whose composite-key joins exceed roughly ten million rows, converting the heaviest merge to polars or an Arrow-backed frame can yield large throughput gains while the final classification and audit-hash step stays in pandas. The decision hinges on join cardinality and memory headroom, not on line count alone.

Related #

• Parsing CSV vs XML trade feeds with pandas
• Schema Validation Frameworks
• ETRM API Integration Patterns
• Async Batch Processing Pipelines
• Trade Ingestion & Matching Workflows