Building a Multi-Source Reconciliation Report in pandas: Comparing Actuals vs. Targets, Flagging Variances, and Exporting a Stakeholder-Ready Excel Summary

Introduction

Here's a situation every analyst knows: it's Monday morning, you have sales actuals from your CRM, targets from a planning spreadsheet, and adjustments sitting in a separate finance file. Your manager wants a reconciliation report before the 9 AM leadership call — something that shows exactly where each region, product line, or sales rep landed relative to their number, flags the meaningful gaps, and looks clean enough to drop in front of a CFO.

You could do this in Excel. You've probably done it in Excel. But that means manually copying data between sheets, running VLOOKUPs that break when columns shift, and spending twenty minutes reformatting cells before you're done. And then doing all of it again next Monday. This lesson shows you how to build that entire reconciliation workflow in pandas — from ingesting multiple source files, through alignment, variance calculation, and flag logic, all the way to an openpyxl-formatted Excel workbook that stakeholders can open without any explanation.

By the end of this lesson, you will have a working, production-quality reconciliation engine. It will handle messy source data, gracefully surface rows that appear in one source but not another, apply tiered variance flags, and write a multi-sheet Excel file with conditional formatting, frozen panes, and a summary scorecard. More importantly, you'll understand why each design decision exists, so you can adapt the pattern to your own data environment.

What you'll learn:

• How to load, normalize, and merge actuals, targets, and adjustment data from multiple heterogeneous sources
• How to diagnose and handle join anomalies — rows that appear in one source but not another — before they silently corrupt your output
• How to calculate absolute and percentage variances and apply tiered flag logic using vectorized pandas operations
• How to write a stakeholder-ready Excel workbook with multiple sheets, cell formatting, conditional color-coding, and summary statistics using openpyxl
• How to wrap the entire pipeline in a reusable function structure so re-running the report takes a single command

---

Prerequisites

You should be comfortable with pandas DataFrames, merging, and groupby aggregation before diving in. If you need to get oriented, the lessons on joining DataFrames with pandas merge and grouping and aggregating in pandas cover the building blocks. You should also have pandas, openpyxl, and numpy installed in your environment. If you're setting up fresh, see setting up Python for data analysis.

---

Designing the Dataset: What We're Working With

Before writing a single line of code, we need to understand the shape of our three source files. Reconciliation reports fail not because the math is hard, but because the inputs are messier than expected. We'll model a realistic scenario: quarterly sales performance for a mid-size company.

actuals.csv — pulled from the CRM nightly, one row per sales rep per product line per quarter:

rep_id, rep_name,    region,    product_line, quarter, actual_revenue
R001,   Ana Ruiz,    Northeast, Enterprise,   Q1,      412000
R001,   Ana Ruiz,    Northeast, SMB,          Q1,       87500
R002,   Dev Patel,   Midwest,   Enterprise,   Q1,      388000
...

targets.xlsx — comes from Finance, built in a planning tool, exported monthly. The key difference: Finance uses a slightly different rep ID scheme and their product line names aren't always consistent with the CRM.

employee_id, territory,  segment,    period, target_revenue
EMP-R001,    Northeast,  Enterprise, Q1,     450000
EMP-R001,    Northeast,  SMB,        Q1,      95000
EMP-R002,    Midwest,    Enterprise, Q1,     400000
...

adjustments.csv — submitted by regional managers to account for deals that closed late, were credited to wrong reps, or had returns. These are additive corrections to actual revenue.

rep_id, product_line, quarter, adjustment
R001,   Enterprise,   Q1,       15000
R003,   SMB,          Q1,      -8000
...

This setup captures the real problems: mismatched key formats, inconsistent category names, and a supplementary file that doesn't have a row for every combination.

Let's start building.

---

Step 1: Loading and Normalizing Each Source

The first rule of reconciliation work is: normalize your keys before you touch anything else. A join that looks like it should work perfectly will silently produce a Cartesian explosion or drop rows because "EMP-R001" doesn't match "R001", or "Enterprise" doesn't match "enterprise".

import pandas as pd
import numpy as np
from pathlib import Path

DATA_DIR = Path("data")

# ── Load actuals ──────────────────────────────────────────────────────────────
actuals = pd.read_csv(DATA_DIR / "actuals.csv")

# Strip whitespace from all string columns — CRM exports are notorious for this
actuals.columns = actuals.columns.str.strip()
str_cols = actuals.select_dtypes("object").columns
actuals[str_cols] = actuals[str_cols].apply(lambda s: s.str.strip())

# Normalize key columns to lowercase for consistent joining
actuals["rep_id_key"] = actuals["rep_id"].str.upper()
actuals["product_key"] = actuals["product_line"].str.lower().str.replace(" ", "_")
actuals["period_key"] = actuals["quarter"].str.upper()

print(actuals.shape)
actuals.head()

# ── Load targets ──────────────────────────────────────────────────────────────
targets = pd.read_excel(DATA_DIR / "targets.xlsx")

targets.columns = targets.columns.str.strip()
str_cols = targets.select_dtypes("object").columns
targets[str_cols] = targets[str_cols].apply(lambda s: s.str.strip())

# Finance uses "EMP-R001" format — strip the prefix to align with CRM rep IDs
targets["rep_id_key"] = targets["employee_id"].str.replace("EMP-", "", regex=False).str.upper()

# Finance calls it "segment"; CRM calls it "product_line"
# Also normalize the values — Finance uses "Ent" as shorthand sometimes
segment_map = {
    "enterprise": "enterprise",
    "ent":        "enterprise",
    "smb":        "smb",
    "mid-market": "mid_market",
    "midmarket":  "mid_market",
}
targets["product_key"] = (
    targets["segment"].str.lower().str.replace(" ", "_").map(segment_map).fillna(
        targets["segment"].str.lower().str.replace(" ", "_")
    )
)
targets["period_key"] = targets["period"].str.upper()

print(targets.shape)
targets.head()

# ── Load adjustments ─────────────────────────────────────────────────────────
adjustments = pd.read_csv(DATA_DIR / "adjustments.csv")

adjustments.columns = adjustments.columns.str.strip()
str_cols = adjustments.select_dtypes("object").columns
adjustments[str_cols] = adjustments[str_cols].apply(lambda s: s.str.strip())

adjustments["rep_id_key"] = adjustments["rep_id"].str.upper()
adjustments["product_key"] = adjustments["product_line"].str.lower().str.replace(" ", "_")
adjustments["period_key"] = adjustments["quarter"].str.upper()

# Multiple adjustment rows for the same key are valid — sum them now
adjustments = (
    adjustments
    .groupby(["rep_id_key", "product_key", "period_key"], as_index=False)["adjustment"]
    .sum()
)

print(adjustments.shape)
adjustments.head()

The .select_dtypes("object") approach for bulk whitespace stripping is safer than listing column names manually, because source files change. For deeper treatment of string normalization techniques, see text cleanup at scale with pandas string methods and regular expressions.

---

Step 2: Diagnosing Join Health Before Merging

This step separates analysts who build robust pipelines from those who build fragile ones. Before merging three datasets, you need to understand which keys exist in which sources. If targets has reps that actuals doesn't, or vice versa, a simple inner join will silently hide those rows. An outer join will surface them but fill with NaN in ways that propagate errors downstream.

def audit_key_coverage(df_left, df_right, keys, label_left, label_right):
    """
    Compare key coverage between two DataFrames.
    Returns a summary dict and prints a diagnostic report.
    """
    left_keys = set(df_left.set_index(keys).index)
    right_keys = set(df_right.set_index(keys).index)
    
    only_left  = left_keys - right_keys
    only_right = right_keys - left_keys
    both       = left_keys & right_keys
    
    print(f"\n{'='*60}")
    print(f"Key Coverage: {label_left} vs {label_right}")
    print(f"{'='*60}")
    print(f"  Keys in both:          {len(both):>5}")
    print(f"  Only in {label_left:<12} {len(only_left):>5}")
    print(f"  Only in {label_right:<12} {len(only_right):>5}")
    
    if only_left:
        print(f"\n  Sample keys only in {label_left}:")
        for k in list(only_left)[:5]:
            print(f"    {k}")
    if only_right:
        print(f"\n  Sample keys only in {label_right}:")
        for k in list(only_right)[:5]:
            print(f"    {k}")
    
    return {
        "only_left":  pd.DataFrame(list(only_left),  columns=keys),
        "only_right": pd.DataFrame(list(only_right), columns=keys),
    }

JOIN_KEYS = ["rep_id_key", "product_key", "period_key"]

coverage = audit_key_coverage(actuals, targets, JOIN_KEYS, "actuals", "targets")

Running this before your merge gives you the information you need to make an intentional decision. If a rep appears in targets but not actuals, that's either a data quality problem or a new hire who hasn't closed anything yet — both outcomes deserve different handling in the report, not silent omission.

# Rows only in actuals (rep closed deals but has no target — unusual, needs flagging)
actuals_no_target = coverage["only_left"]

# Rows only in targets (target was set but no activity recorded)
targets_no_actuals = coverage["only_right"]

print(f"\nReps with actuals but no target: {len(actuals_no_target)}")
print(f"Reps with target but no actuals: {len(targets_no_actuals)}")

For a comprehensive treatment of all the things that go wrong at merge time, detecting and resolving data quality issues across merged DataFrames is required reading.

---

Step 3: Merging the Three Sources

With a clear picture of key coverage, we can now make intentional choices about merge strategy.

# ── Merge actuals + targets ───────────────────────────────────────────────────
# Use outer join to preserve rows from both sides
# We'll classify them explicitly rather than letting NaN propagate silently

recon = pd.merge(
    actuals [["rep_id_key", "rep_name", "region", "product_key", "period_key", "actual_revenue"]],
    targets [["rep_id_key", "product_key", "period_key", "target_revenue"]],
    on=JOIN_KEYS,
    how="outer",
    indicator=True,   # adds "_merge" column: "left_only", "right_only", "both"
)

print(recon["_merge"].value_counts())

The indicator=True parameter on pd.merge() is one of the most useful and underused features in pandas for this kind of work. It gives you a categorical column that explicitly identifies which source each row came from — no guessing from NaN patterns.

# ── Merge in adjustments ──────────────────────────────────────────────────────
# Adjustments don't exist for every row, so left join is correct here
recon = pd.merge(
    recon,
    adjustments[["rep_id_key", "product_key", "period_key", "adjustment"]],
    on=JOIN_KEYS,
    how="left",
)

# Missing adjustment = 0, not NaN
recon["adjustment"] = recon["adjustment"].fillna(0)

# ── Compute adjusted actuals ──────────────────────────────────────────────────
recon["adjusted_actual"] = recon["actual_revenue"].fillna(0) + recon["adjustment"]

print(recon.shape)
recon.head(10)

Now let's reconstruct rep metadata for rows that came from targets only, where rep_name and region will be NaN because those fields live in the actuals file:

# For right_only rows (target exists, no actuals), rep_name/region are NaN
# Try to backfill from any other actuals row for the same rep
rep_lookup = (
    actuals[["rep_id_key", "rep_name", "region"]]
    .drop_duplicates("rep_id_key")
    .set_index("rep_id_key")
)

recon["rep_name"] = recon.apply(
    lambda row: rep_lookup.loc[row["rep_id_key"], "rep_name"]
    if (pd.isna(row["rep_name"]) and row["rep_id_key"] in rep_lookup.index)
    else row["rep_name"],
    axis=1,
)

recon["region"] = recon.apply(
    lambda row: rep_lookup.loc[row["rep_id_key"], "region"]
    if (pd.isna(row["region"]) and row["rep_id_key"] in rep_lookup.index)
    else row["region"],
    axis=1,
)

# Any remaining NaN rep_name means truly unknown — mark explicitly
recon["rep_name"] = recon["rep_name"].fillna("UNKNOWN")
recon["region"]   = recon["region"].fillna("UNKNOWN")

---

Step 4: Calculating Variances and Applying Flag Logic

Now we get to the heart of the report. Variance calculation is simple arithmetic; the design decisions are all in how you classify and communicate those variances.

# ── Variance calculations ─────────────────────────────────────────────────────

# Absolute variance: positive = beat target, negative = missed target
recon["variance_abs"] = recon["adjusted_actual"] - recon["target_revenue"].fillna(0)

# Percentage variance: relative to target
# Guard against division by zero (target = 0 or missing)
recon["variance_pct"] = np.where(
    recon["target_revenue"].notna() & (recon["target_revenue"] != 0),
    recon["variance_abs"] / recon["target_revenue"] * 100,
    np.nan,
)

# Attainment rate: adjusted_actual as % of target
recon["attainment_pct"] = np.where(
    recon["target_revenue"].notna() & (recon["target_revenue"] != 0),
    recon["adjusted_actual"] / recon["target_revenue"] * 100,
    np.nan,
)

Now for flag logic. Stakeholders don't want to read through 500 rows looking for problems — they want the report to tell them where to look. We'll apply a tiered flag system:

# ── Tiered variance flags ─────────────────────────────────────────────────────

def classify_variance(row):
    """
    Apply tiered performance flags based on attainment percentage.
    Handles edge cases: no target, no actuals.
    """
    merge_status = row["_merge"]
    
    # Structural issues take priority over performance flags
    if merge_status == "right_only":
        return "⚠ NO ACTUALS"         # target set, nothing recorded
    if merge_status == "left_only":
        return "⚠ NO TARGET"          # actuals exist, no target was set
    
    att = row["attainment_pct"]
    
    if pd.isna(att):
        return "⚠ TARGET IS ZERO"
    elif att >= 100:
        return "✓ ON/ABOVE TARGET"
    elif att >= 90:
        return "△ MINOR MISS (<10%)"
    elif att >= 75:
        return "▽ MODERATE MISS (10-25%)"
    else:
        return "✗ SIGNIFICANT MISS (>25%)"

recon["flag"] = recon.apply(classify_variance, axis=1)

print(recon["flag"].value_counts())

The classify_variance function uses apply because the logic depends on multiple columns with branching conditions. This is one of the legitimate use cases for apply — not raw performance problems but genuine multi-column conditional logic. If performance is critical, you can replicate the same logic with nested np.where calls, which is faster at scale:

# ── Vectorized equivalent (faster at scale) ───────────────────────────────────

recon["flag_v2"] = np.where(
    recon["_merge"] == "right_only", "⚠ NO ACTUALS",
    np.where(
        recon["_merge"] == "left_only", "⚠ NO TARGET",
        np.where(
            recon["attainment_pct"].isna(), "⚠ TARGET IS ZERO",
            np.where(
                recon["attainment_pct"] >= 100, "✓ ON/ABOVE TARGET",
                np.where(
                    recon["attainment_pct"] >= 90, "△ MINOR MISS (<10%)",
                    np.where(
                        recon["attainment_pct"] >= 75, "▽ MODERATE MISS (10-25%)",
                        "✗ SIGNIFICANT MISS (>25%)"
                    )
                )
            )
        )
    )
)

# Verify both approaches agree
assert (recon["flag"] == recon["flag_v2"]).all(), "Flag mismatch between methods!"
recon.drop(columns=["flag_v2"], inplace=True)

Using np.where with the assertion check is a good production pattern: develop with the readable apply version, then swap to the vectorized version and confirm they agree before deploying. For more on conditional column patterns, see conditional columns and bucketing in pandas.

---

Step 5: Building Regional and Product-Line Summaries

The detail rows are important for drill-down, but leadership wants aggregated views. We'll build two summary DataFrames that roll up to region and product line.

# ── Helper: safe aggregation ──────────────────────────────────────────────────
# Only aggregate rows where both actuals and targets exist
# "both" rows have valid variances; one-sided rows would distort summaries

recon_both = recon[recon["_merge"] == "both"].copy()

def build_summary(df, group_cols):
    """Build a performance summary grouped by the given columns."""
    grp = df.groupby(group_cols, as_index=False).agg(
        total_actual   = ("adjusted_actual",  "sum"),
        total_target   = ("target_revenue",   "sum"),
        total_variance = ("variance_abs",     "sum"),
        rep_count      = ("rep_id_key",       "nunique"),
        row_count      = ("rep_id_key",       "count"),
    )
    
    grp["attainment_pct"] = np.where(
        grp["total_target"] != 0,
        grp["total_actual"] / grp["total_target"] * 100,
        np.nan,
    )
    grp["variance_pct"] = np.where(
        grp["total_target"] != 0,
        grp["total_variance"] / grp["total_target"] * 100,
        np.nan,
    )
    
    # Count reps with significant miss (attainment < 75%)
    miss_mask = df["attainment_pct"] < 75
    miss_counts = (
        df[miss_mask]
        .groupby(group_cols)["rep_id_key"]
        .nunique()
        .rename("reps_significant_miss")
    )
    grp = grp.merge(miss_counts, on=group_cols, how="left")
    grp["reps_significant_miss"] = grp["reps_significant_miss"].fillna(0).astype(int)
    
    grp = grp.sort_values("attainment_pct", ascending=True)  # worst first
    return grp

regional_summary  = build_summary(recon_both, ["region", "period_key"])
product_summary   = build_summary(recon_both, ["product_key", "period_key"])

print(regional_summary)
print(product_summary)

---

Step 6: Building the Scorecard

A one-page executive scorecard gives leadership the headline numbers without needing to read the detail. We'll build this as a structured DataFrame that we can write to its own Excel sheet.

# ── Overall scorecard ─────────────────────────────────────────────────────────

total_actual   = recon_both["adjusted_actual"].sum()
total_target   = recon_both["target_revenue"].sum()
total_variance = total_actual - total_target
overall_att    = total_actual / total_target * 100 if total_target != 0 else np.nan

flag_counts = recon["flag"].value_counts()

scorecard_data = {
    "Metric": [
        "Total Adjusted Actual Revenue",
        "Total Target Revenue",
        "Total Variance (Absolute)",
        "Overall Attainment %",
        "Reps On/Above Target",
        "Reps with Minor Miss (<10%)",
        "Reps with Moderate Miss (10-25%)",
        "Reps with Significant Miss (>25%)",
        "Rows with No Target Set",
        "Rows with No Actuals Recorded",
        "Total Reps Covered",
        "Total Product Lines",
        "Total Regions",
    ],
    "Value": [
        f"${total_actual:,.0f}",
        f"${total_target:,.0f}",
        f"${total_variance:+,.0f}",
        f"{overall_att:.1f}%" if not pd.isna(overall_att) else "N/A",
        int(flag_counts.get("✓ ON/ABOVE TARGET", 0)),
        int(flag_counts.get("△ MINOR MISS (<10%)", 0)),
        int(flag_counts.get("▽ MODERATE MISS (10-25%)", 0)),
        int(flag_counts.get("✗ SIGNIFICANT MISS (>25%)", 0)),
        int(flag_counts.get("⚠ NO TARGET", 0)),
        int(flag_counts.get("⚠ NO ACTUALS", 0)),
        recon["rep_id_key"].nunique(),
        recon["product_key"].nunique(),
        recon["region"].nunique(),
    ]
}

scorecard = pd.DataFrame(scorecard_data)
print(scorecard.to_string(index=False))

---

Step 7: Writing the Stakeholder-Ready Excel Workbook

This is where the report goes from correct to convincing. We'll use openpyxl directly through pandas' ExcelWriter to apply formatting that makes the Excel file look intentionally designed rather than machine-generated.

from openpyxl.styles import (
    PatternFill, Font, Alignment, Border, Side, numbers
)
from openpyxl.utils import get_column_letter
from openpyxl.formatting.rule import ColorScaleRule, CellIsRule, FormulaRule
from datetime import datetime

OUTPUT_PATH = Path("output") / f"reconciliation_report_{datetime.today().strftime('%Y%m%d')}.xlsx"
OUTPUT_PATH.parent.mkdir(exist_ok=True)

# ── Color palette ─────────────────────────────────────────────────────────────
COLORS = {
    "header_bg":    "1F4E79",   # dark navy
    "header_font":  "FFFFFF",
    "green_bg":     "C6EFCE",
    "green_font":   "276221",
    "yellow_bg":    "FFEB9C",
    "yellow_font":  "9C5700",
    "orange_bg":    "FFDCA9",
    "orange_font":  "7B3A00",
    "red_bg":       "FFC7CE",
    "red_font":     "9C0006",
    "grey_bg":      "D9D9D9",
    "grey_font":    "595959",
    "alt_row":      "F2F7FF",   # very light blue for alternating rows
}

def make_fill(hex_color):
    return PatternFill("solid", fgColor=hex_color)

def make_font(hex_color, bold=False, size=11):
    return Font(color=hex_color, bold=bold, size=size, name="Calibri")

def apply_header_style(ws, row_num, col_count):
    """Apply dark navy header style to a row."""
    for col in range(1, col_count + 1):
        cell = ws.cell(row=row_num, column=col)
        cell.fill = make_fill(COLORS["header_bg"])
        cell.font = make_font(COLORS["header_font"], bold=True)
        cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)

def auto_width(ws, min_width=10, max_width=40):
    """Set column widths based on content."""
    for col in ws.columns:
        max_len = max(
            (len(str(cell.value)) if cell.value is not None else 0)
            for cell in col
        )
        col_letter = get_column_letter(col[0].column)
        ws.column_dimensions[col_letter].width = min(max(max_len + 2, min_width), max_width)

def write_df_to_sheet(ws, df, start_row=1, number_formats=None):
    """
    Write a DataFrame to a worksheet starting at start_row.
    Returns the last row written.
    """
    # Write headers
    for col_idx, col_name in enumerate(df.columns, 1):
        ws.cell(row=start_row, column=col_idx, value=col_name)
    apply_header_style(ws, start_row, len(df.columns))
    
    # Write data rows
    for row_idx, row in enumerate(df.itertuples(index=False), start_row + 1):
        for col_idx, value in enumerate(row, 1):
            cell = ws.cell(row=row_idx, column=col_idx, value=value)
            
            # Alternate row shading
            if (row_idx - start_row) % 2 == 0:
                cell.fill = make_fill(COLORS["alt_row"])
            
            # Apply number formats if specified
            if number_formats and col_idx in number_formats:
                cell.number_format = number_formats[col_idx]
    
    return row_idx  # last row written

Now we open the writer and populate each sheet:

# ── Prepare the detail sheet ──────────────────────────────────────────────────
detail_cols = [
    "rep_id_key", "rep_name", "region", "product_key", "period_key",
    "actual_revenue", "adjustment", "adjusted_actual",
    "target_revenue", "variance_abs", "variance_pct", "attainment_pct", "flag"
]

detail_output = recon[detail_cols].copy()

# Round numeric columns for cleaner display
for col in ["actual_revenue", "adjustment", "adjusted_actual", "target_revenue", "variance_abs"]:
    detail_output[col] = detail_output[col].round(0)

detail_output["variance_pct"]  = detail_output["variance_pct"].round(1)
detail_output["attainment_pct"] = detail_output["attainment_pct"].round(1)

# Rename columns for stakeholder-friendly display
detail_output.columns = [
    "Rep ID", "Rep Name", "Region", "Product Line", "Period",
    "Actual Revenue", "Adjustment", "Adjusted Actual",
    "Target Revenue", "Variance ($)", "Variance (%)", "Attainment (%)", "Status Flag"
]

# Sort: worst performers first, then by region
detail_output = detail_output.sort_values(
    ["Attainment (%)", "Region", "Rep Name"],
    ascending=[True, True, True],
    na_position="first",
)

with pd.ExcelWriter(OUTPUT_PATH, engine="openpyxl") as writer:
    
    # ── Sheet 1: Scorecard ────────────────────────────────────────────────────
    scorecard.to_excel(writer, sheet_name="📊 Scorecard", index=False)
    ws_sc = writer.sheets["📊 Scorecard"]
    apply_header_style(ws_sc, 1, len(scorecard.columns))
    
    # Bold the values column for readability
    for row in range(2, len(scorecard) + 2):
        ws_sc.cell(row=row, column=2).font = Font(bold=True, name="Calibri", size=11)
    
    # Freeze header row
    ws_sc.freeze_panes = "A2"
    ws_sc.column_dimensions["A"].width = 38
    ws_sc.column_dimensions["B"].width = 22
    
    # Add a title above the scorecard data
    ws_sc.insert_rows(1)
    title_cell = ws_sc.cell(row=1, column=1, value="Quarterly Sales Reconciliation — Executive Scorecard")
    title_cell.font = Font(bold=True, size=14, name="Calibri", color=COLORS["header_bg"])
    title_cell.alignment = Alignment(horizontal="left")
    ws_sc.merge_cells("A1:B1")
    ws_sc.row_dimensions[1].height = 24
    
    # ── Sheet 2: Detail Report ─────────────────────────────────────────────────
    detail_output.to_excel(writer, sheet_name="🔍 Detail", index=False)
    ws_det = writer.sheets["🔍 Detail"]
    apply_header_style(ws_det, 1, len(detail_output.columns))
    ws_det.freeze_panes = "A2"
    auto_width(ws_det)
    
    # Number formatting for currency columns
    currency_fmt = '#,##0'
    pct_fmt = '0.0"%"'
    
    currency_cols = [6, 7, 8, 9, 10]   # Actual Revenue through Variance ($)
    pct_cols      = [11, 12]            # Variance (%) and Attainment (%)
    
    for row in ws_det.iter_rows(min_row=2, max_row=len(detail_output) + 1):
        for cell in row:
            if cell.column in currency_cols:
                cell.number_format = currency_fmt
            elif cell.column in pct_cols:
                cell.number_format = pct_fmt
    
    # ── Conditional formatting for Status Flag column (column 13) ─────────────
    flag_col_letter = get_column_letter(13)
    data_range = f"{flag_col_letter}2:{flag_col_letter}{len(detail_output) + 1}"
    
    flag_formats = {
        "✓ ON/ABOVE TARGET":          (COLORS["green_bg"],  COLORS["green_font"]),
        "△ MINOR MISS (<10%)":        (COLORS["yellow_bg"], COLORS["yellow_font"]),
        "▽ MODERATE MISS (10-25%)":   (COLORS["orange_bg"], COLORS["orange_font"]),
        "✗ SIGNIFICANT MISS (>25%)":  (COLORS["red_bg"],    COLORS["red_font"]),
        "⚠ NO ACTUALS":              (COLORS["grey_bg"],   COLORS["grey_font"]),
        "⚠ NO TARGET":               (COLORS["grey_bg"],   COLORS["grey_font"]),
        "⚠ TARGET IS ZERO":          (COLORS["grey_bg"],   COLORS["grey_font"]),
    }
    
    # Apply cell-by-cell formatting (more reliable than formula-based CF for exact string match)
    for row in ws_det.iter_rows(
        min_row=2, max_row=len(detail_output) + 1,
        min_col=13, max_col=13
    ):
        for cell in row:
            flag_val = cell.value
            if flag_val in flag_formats:
                bg, fg = flag_formats[flag_val]
                cell.fill = make_fill(bg)
                cell.font = Font(color=fg, bold=True, name="Calibri", size=10)
                cell.alignment = Alignment(horizontal="center")
    
    # ── Sheet 3: Regional Summary ─────────────────────────────────────────────
    regional_summary.to_excel(writer, sheet_name="🗺 Regional", index=False)
    ws_reg = writer.sheets["🗺 Regional"]
    apply_header_style(ws_reg, 1, len(regional_summary.columns))
    ws_reg.freeze_panes = "A2"
    auto_width(ws_reg)
    
    # Color scale on attainment_pct column
    att_col_idx = regional_summary.columns.get_loc("attainment_pct") + 1
    att_col_letter = get_column_letter(att_col_idx)
    att_range = f"{att_col_letter}2:{att_col_letter}{len(regional_summary) + 1}"
    
    ws_reg.conditional_formatting.add(
        att_range,
        ColorScaleRule(
            start_type="num", start_value=50,  start_color="F8696B",  # red
            mid_type="num",   mid_value=100,   mid_color="FFEB84",    # yellow
            end_type="num",   end_value=125,   end_color="63BE7B",    # green
        )
    )
    
    # ── Sheet 4: Product Summary ───────────────────────────────────────────────
    product_summary.to_excel(writer, sheet_name="📦 By Product", index=False)
    ws_prod = writer.sheets["📦 By Product"]
    apply_header_style(ws_prod, 1, len(product_summary.columns))
    ws_prod.freeze_panes = "A2"
    auto_width(ws_prod)
    
    att_col_idx = product_summary.columns.get_loc("attainment_pct") + 1
    att_col_letter = get_column_letter(att_col_idx)
    att_range = f"{att_col_letter}2:{att_col_letter}{len(product_summary) + 1}"
    
    ws_prod.conditional_formatting.add(
        att_range,
        ColorScaleRule(
            start_type="num", start_value=50,  start_color="F8696B",
            mid_type="num",   mid_value=100,   mid_color="FFEB84",
            end_type="num",   end_value=125,   end_color="63BE7B",
        )
    )
    
    # ── Sheet 5: Data Quality Log ──────────────────────────────────────────────
    # Surface the structural issues in their own sheet so they don't get lost
    quality_issues = recon[recon["_merge"].isin(["left_only", "right_only"])][
        ["rep_id_key", "rep_name", "region", "product_key", "period_key",
         "actual_revenue", "target_revenue", "flag"]
    ].copy()
    
    quality_issues.columns = [
        "Rep ID", "Rep Name", "Region", "Product Line", "Period",
        "Actual Revenue", "Target Revenue", "Issue Type"
    ]
    
    quality_issues.to_excel(writer, sheet_name="⚠ Data Quality", index=False)
    ws_dq = writer.sheets["⚠ Data Quality"]
    apply_header_style(ws_dq, 1, len(quality_issues.columns))
    ws_dq.freeze_panes = "A2"
    auto_width(ws_dq)

print(f"Report written to: {OUTPUT_PATH}")

---

Step 8: Wrapping It in a Reusable Pipeline Function

Everything we've built so far is excellent for a notebook. To make it production-worthy — something you schedule to run every Monday morning without touching it — wrap it in functions with a clean entry point.

# reconciliation_report.py

import pandas as pd
import numpy as np
from pathlib import Path
from openpyxl.styles import PatternFill, Font, Alignment
from openpyxl.utils import get_column_letter
from openpyxl.formatting.rule import ColorScaleRule
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)


def load_and_normalize_sources(data_dir: Path, quarter: str) -> tuple:
    """
    Load actuals, targets, and adjustments for a given quarter.
    Returns normalized DataFrames ready for merging.
    """
    logger.info(f"Loading source data for {quarter} from {data_dir}")
    
    actuals     = _load_actuals(data_dir / "actuals.csv", quarter)
    targets     = _load_targets(data_dir / "targets.xlsx", quarter)
    adjustments = _load_adjustments(data_dir / "adjustments.csv", quarter)
    
    return actuals, targets, adjustments


def build_reconciliation(actuals, targets, adjustments) -> pd.DataFrame:
    """
    Merge and reconcile the three source DataFrames.
    Returns the full reconciliation detail table.
    """
    logger.info("Running reconciliation merge")
    JOIN_KEYS = ["rep_id_key", "product_key", "period_key"]
    
    recon = pd.merge(
        actuals [["rep_id_key", "rep_name", "region", "product_key", "period_key", "actual_revenue"]],
        targets [["rep_id_key", "product_key", "period_key", "target_revenue"]],
        on=JOIN_KEYS, how="outer", indicator=True,
    )
    
    recon = pd.merge(recon, adjustments, on=JOIN_KEYS, how="left")
    recon["adjustment"]      = recon["adjustment"].fillna(0)
    recon["adjusted_actual"] = recon["actual_revenue"].fillna(0) + recon["adjustment"]
    
    recon = _compute_variances(recon)
    recon = _apply_flags(recon)
    
    logger.info(
        f"Reconciliation complete: {len(recon)} rows, "
        f"{recon['_merge'].value_counts().to_dict()}"
    )
    return recon


def export_report(recon: pd.DataFrame, output_path: Path) -> None:
    """Write the full reconciliation workbook to output_path."""
    logger.info(f"Exporting workbook to {output_path}")
    output_path.parent.mkdir(parents=True, exist_ok=True)
    
    # Build summary tables
    recon_both       = recon[recon["_merge"] == "both"].copy()
    regional_summary = build_summary(recon_both, ["region", "period_key"])
    product_summary  = build_summary(recon_both, ["product_key", "period_key"])
    scorecard        = build_scorecard(recon, recon_both)
    detail_output    = prepare_detail(recon)
    
    _write_workbook(
        output_path, scorecard, detail_output,
        regional_summary, product_summary, recon
    )
    logger.info(f"Report exported successfully: {output_path}")


def run_reconciliation(
    data_dir:   str | Path = "data",
    output_dir: str | Path = "output",
    quarter:    str = "Q1",
) -> Path:
    """
    End-to-end entry point. Load → Reconcile → Export.
    Returns the path to the generated workbook.
    """
    data_dir   = Path(data_dir)
    output_dir = Path(output_dir)
    
    actuals, targets, adjustments = load_and_normalize_sources(data_dir, quarter)
    recon = build_reconciliation(actuals, targets, adjustments)
    
    filename    = f"reconciliation_{quarter}_{datetime.today().strftime('%Y%m%d')}.xlsx"
    output_path = output_dir / filename
    
    export_report(recon, output_path)
    return output_path


# ── CLI entry point ───────────────────────────────────────────────────────────
if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="Run quarterly reconciliation report")
    parser.add_argument("--data-dir",   default="data",   help="Source data directory")
    parser.add_argument("--output-dir", default="output", help="Output directory")
    parser.add_argument("--quarter",    default="Q1",     help="Quarter to reconcile (e.g. Q1)")
    args = parser.parse_args()
    
    output_path = run_reconciliation(
        data_dir=args.data_dir,
        output_dir=args.output_dir,
        quarter=args.quarter,
    )
    print(f"Report ready: {output_path}")

With this structure, running the full report is:

python reconciliation_report.py --quarter Q2 --data-dir /data/q2 --output-dir /reports

Or from Python:

from reconciliation_report import run_reconciliation
path = run_reconciliation(quarter="Q2")

---

Hands-On Exercise

Now it's your turn to extend the pipeline with two additions that are commonly requested in production reconciliation environments.

Exercise Part 1: Rolling Quarter-over-Quarter Comparison

Modify build_summary() to accept data from two quarters and add columns showing the quarter-over-quarter change in attainment percentage. The output should show, for each region, whether attainment is improving or declining relative to the prior period.

Hint: Merge the two summary DataFrames on region, suffix the columns, then compute the delta. If you have time-series data available, the working with dates and time series in pandas lesson covers resampling patterns that generalize this nicely.

Exercise Part 2: Threshold-Based Email Alert Logic

Add a function generate_alert_list(recon, threshold=0.75) that returns a DataFrame of reps whose attainment is below the threshold, sorted by the severity of the miss. The DataFrame should include the rep's manager (assume a managers.csv lookup file with rep_id and manager_email columns). The function should be designed so a downstream notification system can consume it directly — clean columns, no NaN, one row per rep-product combination.

Hint: You'll need a pd.merge() to bring in manager data. Think about whether you want inner or left join here, and what happens if a rep doesn't appear in the managers file.

---

Common Mistakes & Troubleshooting

Mistake 1: Using inner join for the main merge

The default pd.merge() behavior is how="inner". For reconciliation work, this silently drops every row that exists in only one source. You will never see an error — the row count will just be lower than it should be, and you'll have a report that looks complete but isn't.

Fix: Always use how="outer" for the actuals-targets merge, and always run the coverage audit before merging.

Mistake 2: Percentage variance on zero or NaN targets

If target_revenue is zero or NaN, dividing by it produces inf, -inf, or NaN. All three values will either crash your Excel export or display as #DIV/0! in ways that frighten stakeholders.

Fix: Always wrap percentage calculations in np.where guards, as shown in Step 4. And always check your results with recon.replace([np.inf, -np.inf], np.nan).isna().sum() before exporting.

Mistake 3: Key normalization after the merge

Normalizing rep_id_key in one DataFrame but forgetting it in another means your outer join produces double rows instead of matched rows. You'll get R001 from actuals and EMP-R001 from targets both appearing in the output as separate rows, neither with a match.

Fix: Build a short validation function that checks the intersection of key columns before merging:

def assert_keys_overlap(df_left, df_right, keys, min_overlap_pct=0.5):
    left_set  = set(df_left.set_index(keys).index)
    right_set = set(df_right.set_index(keys).index)
    overlap   = len(left_set & right_set) / max(len(left_set), 1)
    if overlap < min_overlap_pct:
        raise ValueError(
            f"Key overlap is only {overlap:.1%} — check normalization. "
            f"Sample left keys: {list(left_set)[:3]}. "
            f"Sample right keys: {list(right_set)[:3]}"
        )

Mistake 4: Forgetting to handle duplicate keys before merging

If actuals has two rows for the same (rep_id_key, product_key, period_key) — perhaps because the CRM exported a correction row alongside the original — an outer merge will create a Cartesian product for that key, doubling or tripling actuals rows.

Fix: Always check for duplicates on your join keys before merging:

dupe_check = actuals.duplicated(subset=JOIN_KEYS, keep=False)
if dupe_check.any():
    print(f"WARNING: {dupe_check.sum()} duplicate key rows in actuals:")
    print(actuals[dupe_check].to_string())

This connects directly to patterns covered in detecting and resolving data quality issues across merged DataFrames.

Mistake 5: openpyxl formatting applied before data is written

If you apply cell formatting in a loop before calling to_excel, the styles get overwritten when pandas writes the data. Always apply formatting after to_excel writes the data to the sheet, inside the with pd.ExcelWriter(...) as writer: block.

Mistake 6: Using .apply() on large reconciliation DataFrames

If your reconciliation covers tens of thousands of rep-product-period combinations, apply(classify_variance, axis=1) will be noticeably slow. Replace it with the nested np.where pattern shown in Step 4. For datasets above ~1 million rows, consider moving to polars for the transformation layer. See handling large datasets in Python for the tradeoff analysis.

---

Performance and Scaling Considerations

For most business reconciliation use cases — a few thousand rep-product-period combinations — everything in this lesson runs in under a second. But as the dataset grows, a few specific choices matter:

Memory layout: The _merge indicator column from pd.merge() is returned as a CategoricalDtype by default, which is memory-efficient. Keep it as a category rather than converting to string.

Groupby summaries: The build_summary() function uses as_index=False on groupby, which is slightly faster than using reset_index() afterward and avoids MultiIndex complications. For deep dives on aggregation patterns including custom agg functions, see weighted averages, percent of total, and custom aggregations in pandas.

Excel export limits: openpyxl writing speed degrades quadratically for very wide workbooks with lots of conditional formatting rules. If your detail sheet exceeds 50,000 rows, consider writing the detail as a separate CSV (faster) and keeping the Excel workbook for the scorecard and summaries only.

Incremental runs: If you're running this report for many quarters, building an incremental approach — loading only the new quarter rather than reprocessing history — requires a slightly different architecture. The building a reusable ETL pipeline in pandas lesson covers the extract-transform-load pattern that supports this well.

---

Summary & Next Steps

You've built a complete reconciliation pipeline that handles the full lifecycle of a real-world reporting problem: ingesting heterogeneous sources, normalizing keys, diagnosing join health, merging with structural integrity, computing tiered variances, and exporting a polished, formatted Excel workbook across five dedicated sheets.

The most important principles to carry forward:

1. Normalize before you merge. Key mismatches are silent killers in reconciliation work.
2. Audit before you trust. The coverage diagnostic in Step 2 should be non-negotiable in any pipeline that joins external sources.
3. Use indicator=True. It costs nothing and prevents an entire class of debugging nightmares.
4. Surface problems prominently. Sort by attainment ascending, put the data quality sheet right there in the workbook, and use color coding that actually communicates severity.
5. Wrap it in functions. A report that requires manual steps to run is a report that will eventually be run incorrectly.

Where to go next:

• To automate this report to run on a schedule without human intervention, see building and automating recurring reports with pandas
• To deepen your Excel export formatting — adding charts, sparklines, and more sophisticated conditional formatting — automating Excel reports with pandas and openpyxl covers the full openpyxl API
• If your actuals come from a database rather than flat files, reading from SQL databases into pandas with SQLAlchemy covers the connection layer

The reconciliation pattern you've built here isn't just for sales data. The same architecture applies to budget vs. actuals in finance, inventory levels vs. safety stock in operations, headcount plans vs. actual hiring in HR, and KPIs vs. targets in any business function. Master this pattern once and you'll reach for it constantly.