Building a Dynamic pandas Report Pipeline with Parameters, Filters, and Excel Output: Running the Same Analysis Across Any Segment, Date Range, or Region Without Rewriting Code

Introduction

Here's a situation every analyst knows intimately: your manager asks for a regional sales report. You build it in Python, it looks great, and everyone is happy. Then three days later they need the same report but for a different region. Then a different date range. Then filtered to a specific product category. Each time, you open the script, find the hardcoded values buried somewhere in the middle of the code, change them, re-run, rename the output file, and send it. You're not writing new analysis — you're performing surgery on your own script, over and over again, hoping you don't accidentally break something.

This is a fixable problem, and the fix is what this article is about. A parameterized report pipeline separates what analysis to run from which slice of data to run it on. You define the logic once. The parameters — segment, date range, region, threshold, whatever varies — get passed in at runtime. You run the same script against the North region, the South region, and the West region without touching a single line of analysis code. The output is a formatted Excel workbook for each run, named automatically, with the parameters baked in so stakeholders always know exactly what they're looking at.

By the end of this lesson, you'll have built a complete, production-ready report pipeline that you can actually deploy. You'll understand the architectural decisions behind it, the failure modes to watch for, and the patterns that scale when the requirements inevitably get more complex.

What you'll learn:

• How to design a parameter system using Python dictionaries and function signatures that makes your pipeline self-documenting
• How to build composable filter functions that handle date ranges, categorical segments, and numeric thresholds without collapsing under edge cases
• How to chain aggregation, ranking, and derived metrics into a reusable analysis core that runs identically on any filtered slice
• How to write structured, multi-sheet Excel output with formatting tailored to each report variant
• How to run batch report generation across a grid of parameter combinations with proper error handling and logging

---

Prerequisites

This is an expert-level lesson. You should already be comfortable with pandas DataFrames, boolean filtering, groupby aggregations, and basic Python functions. If you need a refresher on any of those, the articles on selecting and filtering data in pandas and grouping and aggregating in pandas will catch you up quickly.

You'll also need pandas, openpyxl, and optionally argparse installed in your environment. We'll assume you're working in a project structure with a scripts directory and an outputs directory. If you're just starting to set up your Python environment, see setting up Python for data analysis.

---

The Dataset We're Working With

Before we write a single function, we need something real to work with. We'll use a simulated retail sales dataset — the kind you'd get by exporting a transactions table from a database or combining multiple monthly files.

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import random

# Reproducible fake data — realistic enough to stress-test our pipeline
np.random.seed(42)
random.seed(42)

n = 50_000

regions = ["North", "South", "East", "West"]
segments = ["Enterprise", "Mid-Market", "SMB"]
categories = ["Hardware", "Software", "Services", "Consumables"]
reps = [f"Rep_{i:03d}" for i in range(1, 51)]

start_date = datetime(2023, 1, 1)
dates = [start_date + timedelta(days=random.randint(0, 729)) for _ in range(n)]

df_raw = pd.DataFrame({
    "transaction_id": range(1, n + 1),
    "date": dates,
    "region": np.random.choice(regions, n, p=[0.3, 0.25, 0.25, 0.2]),
    "segment": np.random.choice(segments, n, p=[0.2, 0.35, 0.45]),
    "category": np.random.choice(categories, n),
    "rep": np.random.choice(reps, n),
    "units": np.random.randint(1, 50, n),
    "unit_price": np.round(np.random.uniform(10, 2000, n), 2),
    "discount_pct": np.random.choice([0, 0.05, 0.10, 0.15, 0.20], n, p=[0.5, 0.2, 0.15, 0.1, 0.05]),
    "cost": np.round(np.random.uniform(5, 1500, n), 2),
})

df_raw["revenue"] = np.round(df_raw["units"] * df_raw["unit_price"] * (1 - df_raw["discount_pct"]), 2)
df_raw["gross_profit"] = np.round(df_raw["revenue"] - (df_raw["units"] * df_raw["cost"]), 2)
df_raw["date"] = pd.to_datetime(df_raw["date"])

print(df_raw.shape)
print(df_raw.dtypes)
print(df_raw.head(3))

This gives us 50,000 transactions across two years, four regions, three customer segments, and four product categories. It's rich enough to make our filters actually meaningful — filtering to "Enterprise + North + Q1 2024" produces a meaningfully different dataset from "SMB + South + Full Year 2023."

---

Designing the Parameter System

The most important decision in building a parameterized pipeline is choosing where parameters live and how they flow through your code. There are three common approaches, each with trade-offs:

Option 1: Global constants at the top of the script. Simple, but you have to edit the file for every run. This is the pattern you're trying to escape.

Option 2: A config dictionary passed through every function. Clean, explicit, and easy to serialize to a log file. This is what we'll use.

Option 3: A config class with validation. More overhead, but gives you type checking and default values. Worth it for complex pipelines.

We'll use Option 2 as our primary approach, with elements of Option 3 for validation. Here's the parameter schema:

from dataclasses import dataclass, field
from typing import Optional, List

@dataclass
class ReportConfig:
    """
    Defines a single report run. Every parameter has a default so you can 
    construct a config with only the overrides you care about.
    """
    # Segmentation filters
    regions: Optional[List[str]] = None          # None = all regions
    segments: Optional[List[str]] = None         # None = all segments
    categories: Optional[List[str]] = None       # None = all categories
    
    # Date range
    date_start: Optional[str] = None             # "YYYY-MM-DD" or None
    date_end: Optional[str] = None               # "YYYY-MM-DD" or None
    
    # Thresholds and grouping
    min_revenue_threshold: float = 0.0           # Exclude transactions below this
    group_by: List[str] = field(default_factory=lambda: ["region", "segment"])
    top_n_reps: int = 10                         # How many reps to include in leaderboard
    
    # Output behavior
    output_dir: str = "outputs"
    report_label: Optional[str] = None          # Human-readable label for file naming
    include_charts: bool = False                # Whether to embed charts (future extension)
    
    def __post_init__(self):
        """Validate and normalize inputs."""
        if self.date_start is not None:
            self.date_start = pd.Timestamp(self.date_start)
        if self.date_end is not None:
            self.date_end = pd.Timestamp(self.date_end)
        
        if self.date_start and self.date_end:
            if self.date_start > self.date_end:
                raise ValueError(f"date_start ({self.date_start}) must be before date_end ({self.date_end})")
        
        if self.top_n_reps < 1:
            raise ValueError("top_n_reps must be at least 1")
        
        # Normalize string lists to consistent case
        if self.regions:
            self.regions = [r.strip().title() for r in self.regions]
        if self.segments:
            self.segments = [s.strip().title() for s in self.segments]
        if self.categories:
            self.categories = [c.strip().title() for c in self.categories]
    
    def to_label(self) -> str:
        """Generate a descriptive label for file naming if none was provided."""
        if self.report_label:
            return self.report_label.replace(" ", "_")
        
        parts = []
        if self.regions:
            parts.append("_".join(self.regions))
        if self.segments:
            parts.append("_".join(self.segments))
        if self.date_start:
            parts.append(self.date_start.strftime("%Y%m%d"))
        if self.date_end:
            parts.append(self.date_end.strftime("%Y%m%d"))
        
        return "_".join(parts) if parts else "all_data"

Using a dataclass here is a deliberate choice. The __post_init__ hook gives us a single validation point — you can't accidentally construct an invalid config and pass it downstream without knowing. The to_label() method ensures file names are always descriptive and consistent without manual effort.

---

Building the Filter Layer

The filter layer takes your raw DataFrame and a ReportConfig and returns a filtered DataFrame. The critical design principle here is that each filter should be independent and composable. Don't write one giant filter expression — write small filter functions that chain together.

def apply_date_filter(df: pd.DataFrame, config: ReportConfig) -> pd.DataFrame:
    """Filter to the configured date range."""
    if config.date_start is None and config.date_end is None:
        return df
    
    mask = pd.Series(True, index=df.index)
    
    if config.date_start is not None:
        mask &= df["date"] >= config.date_start
    if config.date_end is not None:
        mask &= df["date"] <= config.date_end
    
    result = df.loc[mask]
    
    if result.empty:
        raise ValueError(
            f"Date filter ({config.date_start} to {config.date_end}) "
            f"returned 0 rows. Check your date range."
        )
    
    return result


def apply_categorical_filters(df: pd.DataFrame, config: ReportConfig) -> pd.DataFrame:
    """Filter to specified regions, segments, and categories."""
    mask = pd.Series(True, index=df.index)
    
    if config.regions:
        valid_regions = set(config.regions) & set(df["region"].unique())
        invalid = set(config.regions) - valid_regions
        if invalid:
            print(f"  [WARNING] Regions not found in data: {invalid}")
        mask &= df["region"].isin(valid_regions)
    
    if config.segments:
        valid_segments = set(config.segments) & set(df["segment"].unique())
        invalid = set(config.segments) - valid_segments
        if invalid:
            print(f"  [WARNING] Segments not found in data: {invalid}")
        mask &= df["segment"].isin(valid_segments)
    
    if config.categories:
        valid_categories = set(config.categories) & set(df["category"].unique())
        invalid = set(config.categories) - valid_categories
        if invalid:
            print(f"  [WARNING] Categories not found in data: {invalid}")
        mask &= df["category"].isin(valid_categories)
    
    result = df.loc[mask]
    
    if result.empty:
        raise ValueError(
            "Categorical filters returned 0 rows. "
            f"regions={config.regions}, segments={config.segments}, categories={config.categories}"
        )
    
    return result


def apply_threshold_filter(df: pd.DataFrame, config: ReportConfig) -> pd.DataFrame:
    """Exclude transactions below the revenue threshold."""
    if config.min_revenue_threshold <= 0:
        return df
    
    result = df.loc[df["revenue"] >= config.min_revenue_threshold]
    
    pct_removed = (len(df) - len(result)) / len(df) * 100
    if pct_removed > 50:
        print(
            f"  [WARNING] Revenue threshold of {config.min_revenue_threshold:,.2f} "
            f"removed {pct_removed:.1f}% of rows. Is this intentional?"
        )
    
    return result


def apply_all_filters(df: pd.DataFrame, config: ReportConfig) -> pd.DataFrame:
    """
    Apply all configured filters in a defined order.
    Order matters: date first (usually most selective), then categorical, then threshold.
    """
    original_count = len(df)
    
    df = apply_date_filter(df, config)
    df = apply_categorical_filters(df, config)
    df = apply_threshold_filter(df, config)
    
    final_count = len(df)
    retention_pct = final_count / original_count * 100
    
    print(f"  Filtered: {original_count:,} → {final_count:,} rows ({retention_pct:.1f}% retained)")
    
    return df

Notice what we're doing with the invalid-value warnings. Rather than raising an error when a requested region doesn't exist in the data (which might be fine — maybe "Northwest" was just renamed to "North"), we warn and continue with the values that do match. But we raise a hard error if the result is completely empty, because an empty dataset producing a report is almost always a mistake.

---

The Analysis Core

The analysis core takes the filtered DataFrame and produces the summary tables. This is the section that runs identically regardless of what parameters were used to filter the data. It doesn't know or care whether it's operating on North region data or all-region data — it just produces the agreed-upon outputs.

def compute_summary_by_group(df: pd.DataFrame, config: ReportConfig) -> pd.DataFrame:
    """
    Aggregate revenue, profit, units, and transaction count 
    by the dimensions specified in config.group_by.
    """
    agg = df.groupby(config.group_by, observed=True).agg(
        transactions=("transaction_id", "count"),
        total_units=("units", "sum"),
        total_revenue=("revenue", "sum"),
        total_gross_profit=("gross_profit", "sum"),
        avg_deal_size=("revenue", "mean"),
        avg_discount_pct=("discount_pct", "mean"),
    ).reset_index()
    
    # Derived metrics
    agg["gp_margin_pct"] = (agg["total_gross_profit"] / agg["total_revenue"] * 100).round(2)
    agg["revenue_share_pct"] = (agg["total_revenue"] / agg["total_revenue"].sum() * 100).round(2)
    agg["avg_deal_size"] = agg["avg_deal_size"].round(2)
    agg["avg_discount_pct"] = (agg["avg_discount_pct"] * 100).round(2)
    
    # Sort by revenue descending
    agg = agg.sort_values("total_revenue", ascending=False).reset_index(drop=True)
    
    return agg


def compute_monthly_trend(df: pd.DataFrame) -> pd.DataFrame:
    """
    Aggregate revenue and GP by month. Works on whatever date range 
    is in the filtered dataset — no hardcoded periods.
    """
    df = df.copy()
    df["year_month"] = df["date"].dt.to_period("M")
    
    trend = df.groupby("year_month", observed=True).agg(
        transactions=("transaction_id", "count"),
        total_revenue=("revenue", "sum"),
        total_gross_profit=("gross_profit", "sum"),
    ).reset_index()
    
    trend["year_month_str"] = trend["year_month"].astype(str)
    trend["gp_margin_pct"] = (trend["total_gross_profit"] / trend["total_revenue"] * 100).round(2)
    
    # Month-over-month revenue change
    trend["mom_revenue_chg_pct"] = trend["total_revenue"].pct_change() * 100
    trend["mom_revenue_chg_pct"] = trend["mom_revenue_chg_pct"].round(2)
    
    # Cumulative revenue
    trend["cumulative_revenue"] = trend["total_revenue"].cumsum().round(2)
    
    trend = trend.drop(columns=["year_month"])
    
    return trend


def compute_rep_leaderboard(df: pd.DataFrame, config: ReportConfig) -> pd.DataFrame:
    """
    Top N reps by revenue with rank, revenue share, and GP margin.
    """
    by_rep = df.groupby("rep", observed=True).agg(
        transactions=("transaction_id", "count"),
        total_revenue=("revenue", "sum"),
        total_gross_profit=("gross_profit", "sum"),
        avg_discount_pct=("discount_pct", "mean"),
    ).reset_index()
    
    by_rep["gp_margin_pct"] = (by_rep["total_gross_profit"] / by_rep["total_revenue"] * 100).round(2)
    by_rep["revenue_share_pct"] = (by_rep["total_revenue"] / by_rep["total_revenue"].sum() * 100).round(2)
    by_rep["avg_discount_pct"] = (by_rep["avg_discount_pct"] * 100).round(2)
    by_rep["rank"] = by_rep["total_revenue"].rank(method="dense", ascending=False).astype(int)
    
    by_rep = by_rep.sort_values("rank").head(config.top_n_reps).reset_index(drop=True)
    
    # Reorder columns for readability
    cols = ["rank", "rep", "transactions", "total_revenue", "revenue_share_pct", 
            "total_gross_profit", "gp_margin_pct", "avg_discount_pct"]
    
    return by_rep[cols]


def compute_category_breakdown(df: pd.DataFrame) -> pd.DataFrame:
    """Category-level summary with share of revenue."""
    by_cat = df.groupby("category", observed=True).agg(
        transactions=("transaction_id", "count"),
        total_revenue=("revenue", "sum"),
        total_gross_profit=("gross_profit", "sum"),
        avg_unit_price=("unit_price", "mean"),
    ).reset_index()
    
    by_cat["gp_margin_pct"] = (by_cat["total_gross_profit"] / by_cat["total_revenue"] * 100).round(2)
    by_cat["revenue_share_pct"] = (by_cat["total_revenue"] / by_cat["total_revenue"].sum() * 100).round(2)
    by_cat["avg_unit_price"] = by_cat["avg_unit_price"].round(2)
    by_cat = by_cat.sort_values("total_revenue", ascending=False).reset_index(drop=True)
    
    return by_cat


def run_analysis(df: pd.DataFrame, config: ReportConfig) -> dict:
    """
    Orchestrate all analysis functions and return a dict of named DataFrames.
    This is the single entry point for the analysis core.
    """
    print("  Running analysis...")
    
    results = {
        "summary_by_group": compute_summary_by_group(df, config),
        "monthly_trend": compute_monthly_trend(df),
        "rep_leaderboard": compute_rep_leaderboard(df, config),
        "category_breakdown": compute_category_breakdown(df),
    }
    
    # Metadata row: useful for the cover sheet
    results["metadata"] = pd.DataFrame([{
        "report_generated": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "date_range": f"{config.date_start or 'All'} to {config.date_end or 'All'}",
        "regions": ", ".join(config.regions) if config.regions else "All",
        "segments": ", ".join(config.segments) if config.segments else "All",
        "categories": ", ".join(config.categories) if config.categories else "All",
        "total_rows_analyzed": len(df),
        "group_by_dimensions": ", ".join(config.group_by),
    }])
    
    return results

The run_analysis function returns a dictionary of DataFrames, keyed by name. This is a clean interface for the output layer — it can iterate over the dictionary and write each DataFrame to its own Excel sheet without knowing anything about how the analysis works. This separation is what makes the pipeline genuinely reusable.

For more on the kinds of derived metrics we computed here — cumulative sums, percent-of-total, and ranked aggregations — see ranking and window calculations in pandas and weighted averages, percent of total, and custom aggregations in pandas.

---

The Output Layer: Structured Excel Workbooks

The output layer takes the results dictionary and writes it to a formatted Excel workbook. We'll use openpyxl through pandas' ExcelWriter context manager. The key principle here is that formatting should serve comprehension — not just look pretty.

import os
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter
from openpyxl.styles.numbers import FORMAT_PERCENTAGE_00

# Color palette — consistent across all reports
COLORS = {
    "header_fill": "1F4E79",    # Dark blue
    "header_font": "FFFFFF",    # White
    "accent_fill": "D6E4F0",    # Light blue for alternating rows
    "subheader_fill": "2E75B6", # Medium blue
    "warning_fill": "FFC000",   # Amber for warnings
    "positive_fill": "E2EFDA",  # Light green
    "negative_fill": "FCE4D6",  # Light red/salmon
}


def make_header_style():
    return {
        "font": Font(bold=True, color=COLORS["header_font"], size=11),
        "fill": PatternFill("solid", fgColor=COLORS["header_fill"]),
        "alignment": Alignment(horizontal="center", vertical="center", wrap_text=True),
    }


def apply_table_formatting(ws, df: pd.DataFrame, start_row: int = 2):
    """
    Apply header formatting and column widths to a worksheet.
    Assumes row 1 is the header row written by pandas.
    """
    header_styles = make_header_style()
    
    # Format header row
    for col_idx, col_name in enumerate(df.columns, start=1):
        cell = ws.cell(row=start_row - 1, column=col_idx)
        cell.value = col_name.replace("_", " ").title()
        cell.font = header_styles["font"]
        cell.fill = header_styles["fill"]
        cell.alignment = header_styles["alignment"]
    
    # Auto-fit column widths
    for col_idx, col_name in enumerate(df.columns, start=1):
        col_letter = get_column_letter(col_idx)
        header_len = len(str(col_name).replace("_", " ").title())
        
        max_data_len = df[col_name].astype(str).str.len().max() if len(df) > 0 else 0
        optimal_width = min(max(header_len, max_data_len) + 4, 40)
        ws.column_dimensions[col_letter].width = optimal_width
    
    # Freeze the header row
    ws.freeze_panes = ws.cell(row=start_row, column=1)


def format_currency_columns(ws, df: pd.DataFrame, currency_cols: list, start_row: int = 2):
    """Apply currency formatting to specified columns by name."""
    for col_idx, col_name in enumerate(df.columns, start=1):
        if col_name in currency_cols:
            col_letter = get_column_letter(col_idx)
            for row in range(start_row, start_row + len(df)):
                ws[f"{col_letter}{row}"].number_format = '$#,##0.00'


def write_cover_sheet(wb, metadata_df: pd.DataFrame, config: ReportConfig):
    """Write a human-readable cover/summary sheet."""
    ws = wb.create_sheet("Report Info", 0)
    ws.sheet_view.showGridLines = False
    
    # Title
    ws["B2"] = "Sales Analysis Report"
    ws["B2"].font = Font(bold=True, size=18, color=COLORS["header_fill"])
    
    # Parameters
    row = 4
    ws[f"B{row}"] = "Report Parameters"
    ws[f"B{row}"].font = Font(bold=True, size=12, color=COLORS["subheader_fill"])
    
    row += 1
    params = {
        "Generated At": metadata_df["report_generated"].iloc[0],
        "Date Range": metadata_df["date_range"].iloc[0],
        "Regions": metadata_df["regions"].iloc[0],
        "Segments": metadata_df["segments"].iloc[0],
        "Categories": metadata_df["categories"].iloc[0],
        "Rows Analyzed": f"{metadata_df['total_rows_analyzed'].iloc[0]:,}",
        "Group By": metadata_df["group_by_dimensions"].iloc[0],
        "Top N Reps": config.top_n_reps,
        "Min Revenue Threshold": f"${config.min_revenue_threshold:,.2f}",
    }
    
    for label, value in params.items():
        ws[f"B{row}"] = label
        ws[f"B{row}"].font = Font(bold=True)
        ws[f"C{row}"] = str(value)
        row += 1
    
    ws.column_dimensions["A"].width = 3
    ws.column_dimensions["B"].width = 28
    ws.column_dimensions["C"].width = 45


def write_results_to_excel(results: dict, config: ReportConfig) -> str:
    """
    Write all result DataFrames to a multi-sheet Excel workbook.
    Returns the path to the written file.
    """
    os.makedirs(config.output_dir, exist_ok=True)
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    label = config.to_label()
    filename = f"sales_report_{label}_{timestamp}.xlsx"
    filepath = os.path.join(config.output_dir, filename)
    
    # Sheet configuration: which sheets to write and how
    sheet_config = {
        "summary_by_group": {
            "sheet_name": "Summary by Group",
            "currency_cols": ["total_revenue", "total_gross_profit", "avg_deal_size"],
        },
        "monthly_trend": {
            "sheet_name": "Monthly Trend",
            "currency_cols": ["total_revenue", "total_gross_profit", "cumulative_revenue"],
        },
        "rep_leaderboard": {
            "sheet_name": "Rep Leaderboard",
            "currency_cols": ["total_revenue", "total_gross_profit"],
        },
        "category_breakdown": {
            "sheet_name": "Category Breakdown",
            "currency_cols": ["total_revenue", "total_gross_profit", "avg_unit_price"],
        },
    }
    
    with pd.ExcelWriter(filepath, engine="openpyxl") as writer:
        for key, cfg in sheet_config.items():
            if key not in results or results[key] is None:
                continue
            
            df = results[key]
            sheet_name = cfg["sheet_name"]
            
            df.to_excel(writer, sheet_name=sheet_name, index=False, startrow=0)
            
            ws = writer.sheets[sheet_name]
            apply_table_formatting(ws, df, start_row=2)
            format_currency_columns(ws, df, cfg.get("currency_cols", []), start_row=2)
        
        # Cover sheet — add after data sheets so we can insert at position 0
        wb = writer.book
        write_cover_sheet(wb, results["metadata"], config)
    
    print(f"  Written: {filepath}")
    return filepath

The sheet_config dictionary inside write_results_to_excel is a compact way to define sheet-specific behavior — sheet name, currency columns, and any future options — without duplicating the formatting loop. When you need to add a new analysis table, you add it to run_analysis() and to this dictionary. The loop handles the rest.

---

The Pipeline Orchestrator

Now we tie it all together in a single run_report function that accepts a config, executes the pipeline, and returns the output path:

def run_report(df_source: pd.DataFrame, config: ReportConfig) -> str:
    """
    Full pipeline: filter → analyze → export.
    
    Args:
        df_source: The raw, unfiltered source DataFrame.
        config: A ReportConfig describing what to run and how to filter.
    
    Returns:
        Path to the written Excel file.
    """
    label = config.to_label()
    print(f"\n{'='*60}")
    print(f"Running report: {label}")
    print(f"{'='*60}")
    
    try:
        # Step 1: Filter
        df_filtered = apply_all_filters(df_source, config)
        
        # Step 2: Analyze
        results = run_analysis(df_filtered, config)
        
        # Step 3: Export
        output_path = write_results_to_excel(results, config)
        
        print(f"  ✓ Report complete: {output_path}")
        return output_path
    
    except ValueError as e:
        print(f"  ✗ Report failed (data issue): {e}")
        return None
    except Exception as e:
        print(f"  ✗ Report failed (unexpected error): {e}")
        raise  # Re-raise unexpected errors so they're visible


# ---- RUNNING A SINGLE REPORT ----

config_north_enterprise = ReportConfig(
    regions=["North"],
    segments=["Enterprise"],
    date_start="2024-01-01",
    date_end="2024-12-31",
    group_by=["category", "segment"],
    top_n_reps=10,
    report_label="North_Enterprise_2024",
    output_dir="outputs/sales_reports",
)

path = run_report(df_raw, config_north_enterprise)

Running this produces output like:

============================================================
Running report: North_Enterprise_2024
============================================================
  Filtered: 50,000 → 3,241 rows (6.5% retained)
  Running analysis...
  Written: outputs/sales_reports/sales_report_North_Enterprise_2024_20241115_142301.xlsx
  ✓ Report complete: outputs/sales_reports/sales_report_North_Enterprise_2024_20241115_142301.xlsx

---

Batch Generation: Running Across a Parameter Grid

The real power of this architecture appears when you want to run the same report across every combination of parameters. Instead of running four scripts manually for four regions, you define the parameter grid and let the pipeline iterate.

import itertools
from concurrent.futures import ThreadPoolExecutor, as_completed

def build_parameter_grid(
    regions_list: list,
    segments_list: list,
    date_ranges: list,
    base_config_kwargs: dict = None,
) -> list:
    """
    Generate a list of ReportConfig objects from a parameter grid.
    
    date_ranges should be a list of (start, end, label) tuples.
    
    Example:
        date_ranges = [
            ("2023-01-01", "2023-12-31", "FY2023"),
            ("2024-01-01", "2024-12-31", "FY2024"),
        ]
    """
    if base_config_kwargs is None:
        base_config_kwargs = {}
    
    configs = []
    
    for region, segment, (date_start, date_end, period_label) in itertools.product(
        regions_list, segments_list, date_ranges
    ):
        label = f"{region}_{segment}_{period_label}"
        cfg = ReportConfig(
            regions=[region],
            segments=[segment],
            date_start=date_start,
            date_end=date_end,
            report_label=label,
            **base_config_kwargs,
        )
        configs.append(cfg)
    
    return configs


def run_batch(
    df_source: pd.DataFrame,
    configs: list,
    max_workers: int = 1,
) -> dict:
    """
    Run a batch of report configs and return a dict of {label: output_path}.
    
    max_workers=1 runs sequentially (safe default).
    max_workers>1 runs in parallel using threads (faster, but be careful with 
    file handles and shared state).
    """
    results = {}
    
    if max_workers == 1:
        # Sequential: simple, debuggable, safe
        for config in configs:
            path = run_report(df_source, config)
            results[config.to_label()] = path
    else:
        # Parallel: thread-based (GIL doesn't hurt here because openpyxl is the bottleneck)
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_config = {
                executor.submit(run_report, df_source, cfg): cfg
                for cfg in configs
            }
            for future in as_completed(future_to_config):
                cfg = future_to_config[future]
                try:
                    path = future.result()
                    results[cfg.to_label()] = path
                except Exception as e:
                    print(f"  ✗ Failed: {cfg.to_label()} — {e}")
                    results[cfg.to_label()] = None
    
    # Summary
    successful = sum(1 for v in results.values() if v is not None)
    print(f"\nBatch complete: {successful}/{len(configs)} reports generated successfully")
    
    return results


# ---- RUNNING A BATCH ----

date_ranges = [
    ("2023-01-01", "2023-12-31", "FY2023"),
    ("2024-01-01", "2024-12-31", "FY2024"),
]

configs = build_parameter_grid(
    regions_list=["North", "South", "East", "West"],
    segments_list=["Enterprise", "SMB"],
    date_ranges=date_ranges,
    base_config_kwargs={
        "group_by": ["category"],
        "top_n_reps": 5,
        "output_dir": "outputs/batch_reports",
    }
)

print(f"Generated {len(configs)} report configs")  # 4 regions × 2 segments × 2 periods = 16

batch_results = run_batch(df_raw, configs, max_workers=4)

The itertools.product call does the combinatorial work for you. Four regions × two segments × two date ranges = 16 reports. If you later add a third period, you get 24 reports. The code doesn't change.

---

Making It Runnable From the Command Line

A pipeline that can only be invoked from a Jupyter notebook is limited. The right move is to expose it via argparse so it can be called from a shell, a scheduler, or a CI job.

# save as: run_sales_report.py

import argparse
import sys
import pandas as pd

def parse_args():
    parser = argparse.ArgumentParser(
        description="Generate a parameterized sales analysis report.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python run_sales_report.py --regions North East --date-start 2024-01-01 --date-end 2024-06-30
  python run_sales_report.py --segments Enterprise --group-by region category --top-n-reps 15
  python run_sales_report.py --batch --output-dir outputs/q4_batch
        """
    )
    
    parser.add_argument("--data-path", default="data/transactions.parquet",
                        help="Path to source data file (.csv or .parquet)")
    parser.add_argument("--regions", nargs="+", default=None,
                        help="One or more regions (e.g. North South). Default: all.")
    parser.add_argument("--segments", nargs="+", default=None,
                        help="One or more segments. Default: all.")
    parser.add_argument("--categories", nargs="+", default=None,
                        help="One or more product categories. Default: all.")
    parser.add_argument("--date-start", default=None,
                        help="Start date in YYYY-MM-DD format.")
    parser.add_argument("--date-end", default=None,
                        help="End date in YYYY-MM-DD format.")
    parser.add_argument("--group-by", nargs="+", default=["region", "segment"],
                        help="Dimensions to group by in the summary. Default: region segment.")
    parser.add_argument("--top-n-reps", type=int, default=10,
                        help="Number of reps in the leaderboard. Default: 10.")
    parser.add_argument("--min-revenue", type=float, default=0.0,
                        help="Minimum revenue per transaction to include. Default: 0.")
    parser.add_argument("--output-dir", default="outputs",
                        help="Directory for output files. Default: outputs/")
    parser.add_argument("--label", default=None,
                        help="Human-readable label for the output file name.")
    parser.add_argument("--batch", action="store_true",
                        help="Run batch mode: one report per region+segment combination.")
    parser.add_argument("--workers", type=int, default=1,
                        help="Parallel workers for batch mode. Default: 1 (sequential).")
    
    return parser.parse_args()


def load_data(path: str) -> pd.DataFrame:
    """Load data from CSV or Parquet based on file extension."""
    if path.endswith(".parquet"):
        return pd.read_parquet(path)
    elif path.endswith(".csv"):
        return pd.read_csv(path, parse_dates=["date"])
    else:
        raise ValueError(f"Unsupported file format: {path}. Use .csv or .parquet")


def main():
    args = parse_args()
    
    print(f"Loading data from {args.data_path}...")
    df = load_data(args.data_path)
    print(f"Loaded {len(df):,} rows")
    
    if args.batch:
        # Batch mode: run for every region/segment combination in the data
        regions = df["region"].unique().tolist()
        segments = df["segment"].unique().tolist()
        
        date_ranges = [
            (args.date_start, args.date_end, args.label or "custom_period")
        ]
        
        configs = build_parameter_grid(
            regions_list=regions,
            segments_list=segments,
            date_ranges=date_ranges,
            base_config_kwargs={
                "categories": args.categories,
                "group_by": args.group_by,
                "top_n_reps": args.top_n_reps,
                "min_revenue_threshold": args.min_revenue,
                "output_dir": args.output_dir,
            }
        )
        
        run_batch(df, configs, max_workers=args.workers)
    
    else:
        config = ReportConfig(
            regions=args.regions,
            segments=args.segments,
            categories=args.categories,
            date_start=args.date_start,
            date_end=args.date_end,
            group_by=args.group_by,
            top_n_reps=args.top_n_reps,
            min_revenue_threshold=args.min_revenue,
            output_dir=args.output_dir,
            report_label=args.label,
        )
        
        run_report(df, config)


if __name__ == "__main__":
    main()

With this in place, you can run:

# Single report: North region, 2024, grouped by category
python run_sales_report.py \
  --data-path data/transactions.csv \
  --regions North \
  --date-start 2024-01-01 \
  --date-end 2024-12-31 \
  --group-by category \
  --label North_2024_by_Category

# All regions and segments in batch, 4 workers
python run_sales_report.py \
  --data-path data/transactions.csv \
  --date-start 2024-01-01 \
  --date-end 2024-12-31 \
  --batch \
  --workers 4 \
  --output-dir outputs/q4_batch

This is what a real pipeline looks like. It runs from a cron job, a GitHub Actions workflow, or a simple shell script. The analysis code never changes. Only the parameters change.

For more on automating reports on a schedule, see building and automating recurring reports with pandas.

---

Performance Considerations for Larger Datasets

The pipeline we've built works beautifully on 50,000 rows. On 5 million rows, a few things become important:

Load data once, filter in memory. We already do this. Loading from disk for every report run would be catastrophic. If you can't fit all data in memory, consider filtering at the source — pass your date range and region as query parameters to your SQL query or Parquet partition filters.

Use appropriate dtypes. String columns stored as object consume far more memory than they need to. Converting region, segment, and category to pd.Categorical cuts memory usage and makes groupby operations faster.

def optimize_dtypes(df: pd.DataFrame) -> pd.DataFrame:
    """Convert low-cardinality string columns to Categorical."""
    categorical_cols = ["region", "segment", "category", "rep"]
    for col in categorical_cols:
        if col in df.columns and df[col].dtype == "object":
            df[col] = df[col].astype("category")
    return df

Profile before optimizing. Use %timeit in Jupyter or cProfile in scripts to find actual bottlenecks. The groupby aggregations are usually not the problem — it's almost always file I/O or string operations.

Consider Parquet for your source data. Loading 5 million rows from CSV takes several seconds and loads everything into memory. A Parquet file with column pruning and partition pushdown loads only what you need, in a fraction of the time.

For a deep treatment of working with large datasets efficiently, see handling large datasets in Python: chunked reading, efficient dtypes, and when to use Polars.

---

Hands-On Exercise

Build on the pipeline from this lesson and extend it with the following requirements. Each task builds on the previous one.

Task 1: Add a Discount Cohort Filter

Add a new parameter to ReportConfig called max_discount_pct (float, default 1.0 meaning "no filter"). Add a corresponding filter function apply_discount_filter that excludes transactions where discount_pct exceeds this threshold. Wire it into apply_all_filters.

Test it: run a report with max_discount_pct=0.10 and verify the resulting leaderboard shows different rep rankings than an unfiltered run.

Task 2: Add Quarter Grouping to Monthly Trend

Modify compute_monthly_trend to also produce a quarterly summary. Add it as a second sheet called "Quarterly Trend." The quarterly summary should include the same metrics as the monthly trend, plus a column showing which quarter (Q1 2024, Q2 2024, etc.) each row represents.

Hint: use df["date"].dt.to_period("Q").

Task 3: Add Conditional Formatting for Margin

In the Rep Leaderboard sheet, apply conditional coloring to the gp_margin_pct column: green fill for margins above 40%, amber for 20–40%, and red for below 20%. Use openpyxl to apply PatternFill based on the cell value in a loop after writing the DataFrame.

Task 4: Build a Cross-Region Comparison Report

Write a new function run_comparison_report that:

1. Runs the analysis separately for each region in the data (using apply_all_filters with a per-region config)
2. Stacks the summary_by_group results into a single DataFrame with a new "region" identifier column
3. Writes a single Excel file where each sheet is one region's summary, plus a "Comparison" sheet with all regions side by side

This exercise mirrors the kind of cross-segment comparison that stakeholders frequently request and is genuinely non-trivial to do cleanly.

---

Common Mistakes & Troubleshooting

The Empty Filter Problem

Symptom: You configure regions=["northwest"] and the report produces an empty result or raises a confusing error.

Cause: Case mismatch between your config value and the data values. The ReportConfig.__post_init__ we wrote normalizes to title case, but "NorthWest" → "Northwest" which still might not match "North West" with a space.

Fix: Add a print statement in apply_categorical_filters that shows which values were found vs. requested. Always inspect df["region"].unique() on your source data before constructing configs.

---

The Date Type Mismatch

Symptom: Date filtering returns unexpected rows or no rows even though the dates look right.

Cause: Your date column is object type (strings) rather than datetime64. Comparing "2024-01-15" >= "2024-01-01" works alphabetically but fails logically the moment your dates aren't zero-padded.

Fix: In your __post_init__, convert date_start and date_end to pd.Timestamp. In your data loading step, always parse dates explicitly. For a full treatment of date handling, see working with dates and time series in pandas.

---

The Shared State Bug in Parallel Mode

Symptom: Batch reports in parallel mode produce garbled or mixed results — some sheets seem to contain another report's data.

Cause: You're mutating a shared DataFrame inside the pipeline. If any filter or analysis function uses df = df[...] without .copy(), it may create a view that shares memory with the original, and concurrent modifications will stomp on each other.

Fix: In apply_all_filters and anywhere else you reshape data, always use df.copy() when you intend to modify the DataFrame independently. The filter functions we wrote already handle this by returning the result of df.loc[mask], which creates a copy in most cases, but being explicit with .copy() is safer.

---

The File Name Collision

Symptom: Batch runs with max_workers > 1 occasionally produce incomplete or empty Excel files.

Cause: Two threads generate the same timestamp (to the second) and try to write to the same filepath simultaneously.

Fix: Our to_label() method generates labels that include the config parameters, which makes collisions unlikely in practice. But to be safe, add a UUID or thread ID to the filename in parallel mode:

import uuid
filename = f"sales_report_{label}_{timestamp}_{uuid.uuid4().hex[:6]}.xlsx"

---

GroupBy With `observed=False`

Symptom: After converting columns to pd.Categorical, your groupby summaries include rows for category combinations that don't appear in the filtered data (e.g., rows of zeros for "West + Enterprise" even when your filter excludes West).

Cause: The default for groupby changed in pandas 2.0. With Categorical columns and observed=False, pandas generates all possible category combinations.

Fix: Always pass observed=True to groupby when your grouping columns might be Categorical. We already do this in all our compute_* functions. If you're using an older codebase that doesn't, add it.

---

Summary & Next Steps

We've built something genuinely production-worthy here. The core architecture — ReportConfig as the parameter contract, composable filter functions, an analysis core that returns a named dictionary of DataFrames, and a structured Excel output layer — will serve you across almost any reporting scenario you encounter.

Let's recap the key patterns:

• Separate concerns ruthlessly. Config, filtering, analysis, and output are four distinct responsibilities. Each is independently testable and replaceable.
• Validate at the boundary. Catch bad parameters in __post_init__, catch empty data in the filter layer, and raise clearly. Don't let bad inputs silently produce wrong reports.
• Warn, don't fail, on ambiguous inputs. A missing region value is usually a typo, not a catastrophe. Warn, skip, and continue.
• Design the output contract first. The dictionary of DataFrames is the interface between your analysis and your output. Get that right, and you can add new output formats without touching analysis code.
• Load data once, filter in memory. This is the primary scalability lever for batch generation.

Where do you go from here? If you're building reports that compare actuals against targets, look at building a multi-source reconciliation report in pandas. For structuring the project itself — module organization, shared utilities, separating configs into YAML files — see structuring a reusable data analysis project. And if you want to push further into rich Excel formatting beyond what we covered, automating Excel reports with pandas and openpyxl covers conditional formatting, charts, and dynamic named ranges in depth.

The goal isn't just to automate a report — it's to build something you trust enough to schedule and walk away from. That's what you have now.