Build a Forecast-Ready Sales ETL

Build a local Python and DuckDB pipeline for forecast-ready sales data.

Introduction

30 Second Summary

Sales histories often contain inconsistent labels, missing fields, or numbers stored as text. Those hidden problems can weaken a forecast before anyone notices.

In this project, you will build a repeatable sales extract-transform-load pipeline that converts messy sales history into clean analytical tables in DuckDB. Each run preserves rejected rows for review before printing six months of sales trends.

What You'll Build

Picture running pipeline.py and seeing a six-month sales trend backed by clean records plus an audit trail for every rejected row.

By the end of this project, you'll have:

  • A repeatable ETL workflow you can rerun with one command without accumulating duplicate records.
  • A rejected-row quarantine that preserves source line numbers plus failure reasons for every bad record.
  • A forecast-ready sales report showing six months of unit plus revenue trends for forecasting research.
  • Secret Mission: Add previous-month revenue plus revenue-change features with a SQL window function.

Are there any prerequisites?

No prior data engineering experience is required. The guide walks you through setting up Python plus DuckDB from scratch.

You need a Mac with Terminal plus a code editor. Everything runs locally with free open-source tools.

Before We Start

Before the hands-on work begins, this checkpoint connects your cleaned monthly dataset to a sales decision you care about. That question gives every later quality check a clear purpose.

Set Up a Reproducible Workspace

Your sales decision question needs a pipeline whose results can be trusted. A changing Python runtime could make the same source data behave differently on another run.

A local virtual environment keeps this project's Python packages separate from the rest of your Mac. A pinned dependency makes the workspace reproducible.

In this step, get ready to:
  • Confirm Python 3.14.7 is installed.
  • Create an isolated .venv inside sales-forecast-etl.
  • Install the pinned DuckDB 1.5.5 dependency.
Check the Python runtime

The project uses Python 3.14.7 so every later command runs against the same language version. Start by checking what your Mac currently provides.

  • Press Cmd+Space to open Spotlight.
  • Type Terminal into Spotlight.
  • Press Enter to open Terminal.
  • Ask Python to print its installed version by running this command:
python3 --version

What does this command do?

The python3 command starts your installed Python 3 interpreter. The --version option prints its version before exiting.

✔️ I see the required Python version

Python 3.14.7 is ready for this project.

  • Keep the Terminal window open for the workspace commands below.

ⓧ I see an older Python version

Your current Python version is older than the one this project uses. Install Python 3.14.7 before creating the virtual environment.

Install the required Python release

The installer may request your Mac password. That approval only allows Python to be installed on your computer.

  • Click Continue in the installer.
  • Click Agree to accept the license.
  • Click Install to complete the installation.
Install Python's certificates

The certificate helper prepares Python to make secure network connections. This prevents certificate problems when pip downloads DuckDB.

  • Select Applications in the Finder sidebar.
  • Open the Python 3.14 folder.
  • Double-click Install Certificates.command.
  • Return to the Terminal window from earlier.
  • Check the newly installed version by running this command:
python3 --version

What does this check prove?

The output identifies the Python interpreter that python3 now launches. You need to see Python 3.14.7 before continuing.

Still seeing the older version?

Close Terminal after the installer finishes. Open a fresh Terminal window before checking again.

If the older version remains active, use Help me find the Python 3.14 interpreter installed by the official macOS installer..

ⓧ Command not found

Your shell cannot find Python yet. Install Python 3.14.7 from the official macOS package.

Install the required Python release

The installer may request your Mac password. That approval only allows Python to be installed on your computer.

  • Click Continue in the installer.
  • Click Agree to accept the license.
  • Click Install to complete the installation.
Install Python's certificates

The certificate helper prepares Python to make secure network connections. This allows pip to download packages safely.

  • Select Applications in the Finder sidebar.
  • Open the Python 3.14 folder.
  • Double-click Install Certificates.command.
  • Return to the Terminal window from earlier.
  • Confirm Python is now available by running this command:
python3 --version

What does this check prove?

The command confirms that Terminal can locate the installed Python interpreter. You need to see Python 3.14.7 before continuing.

Still seeing command not found?

Close Terminal after the installer finishes. Open a fresh Terminal window so it can detect the new installation.

If Python remains unavailable, use Help me troubleshoot why Terminal cannot find Python after the official macOS installation..

That's the runtime secured. Python 3.14.7 is now ready to support the same pipeline behavior on each run.

Create an isolated project workspace

The sales-forecast-etl folder keeps every project file in one known location. Its .venv environment gives this project its own Python packages.

  • Create the project folder on your Desktop. Enter it and activate its virtual environment by running these commands:
cd ~/Desktop
mkdir sales-forecast-etl
cd sales-forecast-etl
python3 -m venv .venv
source .venv/bin/activate

What do these commands set up?

  • The cd command moves Terminal to your Desktop.
  • The mkdir command creates the sales-forecast-etl folder.
  • The venv module creates the isolated .venv environment.
  • The source command activates that environment in your current Terminal session.
  • Confirm which Python interpreter is active by running this command:
which python

What does this path prove?

The which command prints the executable used when you type python. A path ending in .venv/bin/python proves the project environment is active.

You should see a path ending in .venv/bin/python.

Does the path point outside .venv?

Make sure Terminal is inside the sales-forecast-etl folder. Run the activation command again if the path does not end in .venv/bin/python.

Use Help me activate the .venv environment in my sales-forecast-etl folder. if the interpreter still points elsewhere.

Pin and install DuckDB

The requirements.txt file records the exact package release used by the project. Pinning DuckDB to 1.5.5 prevents a future release from changing the pipeline unexpectedly.

Why DuckDB for this project?

A pandas plus SQLite workflow would split transformation work from analytical storage. That adds connector setup to a small local project.

DuckDB provides analytical SQL. It also persists tables in a local database file without running a database server.

  • Create requirements.txt with the pinned dependency by running this command:
echo "duckdb==1.5.5" > requirements.txt

What does this command create?

The echo command produces the pinned dependency line. The output redirection writes that line into requirements.txt.

  • Confirm the new file exists by running this command:
ls

What does this check show?

The ls command lists the files in sales-forecast-etl.

You should see requirements.txt in the output.

✔️ Awesome, I've got everything!

Your requirements.txt file now pins the project to DuckDB 1.5.5.

ⓧ I'd like to double check the full code

Compare your requirements.txt file with this complete version.

duckdb==1.5.5

What should this file contain?

The package name is duckdb. The double equals sign pins its version to 1.5.5.

Don't see requirements.txt?

Confirm that the command ran while Terminal was inside sales-forecast-etl.

Use Help me create requirements.txt inside my sales-forecast-etl folder. if the file appears in another location.

  • Install the dependency recorded in requirements.txt by running this command:
python -m pip install -r requirements.txt

What does this installation command do?

The active python interpreter runs pip inside .venv. Pip reads requirements.txt before installing its pinned package.

The installation should finish without an error before Terminal returns to its prompt.

Did the package installation fail?

Confirm that which python ends in .venv/bin/python. An interpreter outside that folder means the virtual environment is inactive.

Use Help me troubleshoot the DuckDB installation inside my virtual environment. for help with the output you received.

Before you check, which DuckDB version do you expect pip to report?

  • Verify the installed DuckDB package metadata by running this command:
python -m pip show duckdb

What does this verification show?

The command asks pip for metadata about the installed duckdb package. Its output includes the package version installed inside the active environment.

You should see Version: 1.5.5 in the output. You've locked the workspace to the exact dependency this pipeline uses.

Are the package details missing?

Check that requirements.txt contains the exact pinned line from the comparison tab. Run the installation command again after activating .venv.

Use Help me verify DuckDB 1.5.5 inside my active virtual environment. if the displayed version is missing or different.

Your isolated workspace is ready. Next, you'll give the pipeline messy sales history and inspect exactly what it receives.

Extract and Preview Sales History

Your isolated Python environment is active. DuckDB 1.5.5 is pinned inside it.

Forecasting depends on knowing exactly what arrived from the source. A raw CSV preview creates that checkpoint before any transformation changes the values.

In this step, get ready to:
  • Create a sales history source with realistic inconsistencies.
  • Build an extractor that checks the source header.
  • Preview the raw row count plus five unchanged records.
Create the raw sales file

The source includes inconsistent labels plus three records that fail later quality checks. Keeping these flaws visible makes the pipeline's data-quality work measurable.

  • Create sales_history.csv inside the sales-forecast-etl directory by using your code editor's new-file control.
  • Paste these raw sales records into sales_history.csv:
sale_month,region,channel,units,unit_price
2026-01,north,online,100,20.00
2026-01,South,Retail,80,25.00
2026-01,west,PARTNER,60,30.00
2026-02,North,Online,110,20.00
2026-02,south,retail,82,25.00
2026-02,West,Partner,63,30.00
2026-03,NORTH,online,115,21.00
2026-03,South,Retail,85,25.00
2026-03,west,partner,65,31.00
2026-03,North,Online,n/a,21.00
2026-04,North,Online,123,21.00
2026-04,south,RETAIL,88,26.00
2026-04,West,Partner,68,31.00
2026-04,,Retail,90,26.00
2026-05,north,online,130,22.00
2026-05,South,Retail,92,26.00
2026-05,West,Partner,72,32.00
2026/05,West,Partner,74,$32.00
2026-06,North,Online,138,22.00
2026-06,South,Retail,96,27.00
2026-06,West,Partner,75,32.00

What Does This Data Preserve?

  • The header defines the five fields expected by the extractor.
  • Mixed capitalization preserves inconsistent region and channel labels.
  • The n/a value preserves an invalid units field.
  • The empty April region preserves a missing required label.
  • The 2026/05 value preserves a malformed month.
  • Compare the first line with the five required field names.
  • Save sales_history.csv.
  • Check the editor line number beside the final June record.

You should see the final record on line 22. That confirms the file contains one header plus 21 raw records.

Does the CSV Look Misaligned?

  • Check that the header contains five comma-separated field names.
  • Keep the empty region in the April record because that blank value is deliberate.
  • Compare the file structure with help from this CSV checker.

✔️ Awesome, I've got everything!

Your saved sales_history.csv contains one header plus 21 raw sales records.

ⓧ I'd like to double check the full code

Compare your complete sales_history.csv with this reference:

sale_month,region,channel,units,unit_price
2026-01,north,online,100,20.00
2026-01,South,Retail,80,25.00
2026-01,west,PARTNER,60,30.00
2026-02,North,Online,110,20.00
2026-02,south,retail,82,25.00
2026-02,West,Partner,63,30.00
2026-03,NORTH,online,115,21.00
2026-03,South,Retail,85,25.00
2026-03,west,partner,65,31.00
2026-03,North,Online,n/a,21.00
2026-04,North,Online,123,21.00
2026-04,south,RETAIL,88,26.00
2026-04,West,Partner,68,31.00
2026-04,,Retail,90,26.00
2026-05,north,online,130,22.00
2026-05,South,Retail,92,26.00
2026-05,West,Partner,72,32.00
2026/05,West,Partner,74,$32.00
2026-06,North,Online,138,22.00
2026-06,South,Retail,96,27.00
2026-06,West,Partner,75,32.00
Build the command-line foundation

The pipeline needs stable file names plus an exact header contract. Its argument parser also reserves separate modes for each stage of the project.

  • Create pipeline.py inside the sales-forecast-etl directory by using your code editor's new-file control.
  • Paste this configuration plus argument parser into pipeline.py:
import argparse
import csv
from decimal import Decimal

import duckdb

SOURCE_FILE = "sales_history.csv"
DATABASE_FILE = "sales_forecast.duckdb"
EXPECTED_FIELDS = ["sale_month", "region", "channel", "units", "unit_price"]
MONEY = Decimal("0.01")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Prepare monthly sales history for forecasting research."
    )
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument("--preview", action="store_true")
    modes.add_argument("--naive", action="store_true")
    modes.add_argument("--quality", action="store_true")
    return parser.parse_args()

What Does This Code Set Up?

  • The imports provide command flag parsing plus CSV reading.
  • The decimal import prepares the pipeline for precise money calculations.
  • The DuckDB import prepares the script for the analytical database added later.
  • SOURCE_FILE points the extractor at the raw sales history.
  • EXPECTED_FIELDS defines the required header order.
  • parse_args() keeps the preview mode separate from later pipeline modes.
  • Save pipeline.py.
  • Check that the configuration starts with SOURCE_FILE.
  • Confirm the current file ends with return parser.parse_args().

Seeing Editor Errors?

  • Check that every quoted value has matching quotation marks.
  • Keep every line inside parse_args() indented by four spaces.
  • Review this chunk with help for the parser setup.
Connect and run the raw preview

The extractor uses csv.DictReader to map each source record to the five header fields. It rejects an unexpected header before later stages can misread the columns.

  • Locate return parser.parse_args() in pipeline.py.
  • Paste this extraction plus preview flow below that line:
def extract_sales():
    with open(SOURCE_FILE, newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")
        return list(reader)


def preview_sales(rows):
    print(f"Extracted {len(rows)} rows")
    for row in rows[:5]:
        print(row)


def main():
    args = parse_args()
    rows = extract_sales()

    if args.preview:
        preview_sales(rows)


if __name__ == "__main__":
    main()

How Does the Preview Work?

  • extract_sales() opens the configured source file.
  • EXPECTED_FIELDS protects the pipeline from missing or reordered columns.
  • preview_sales() reports the extracted row count.
  • rows[:5] limits the raw preview to five records.
  • main() selects the preview path when the matching flag is present.
  • Save pipeline.py.

Before you run the preview, how many records do you expect the extractor to count?

  • Run the raw preview from the activated virtual environment by entering this command:
python pipeline.py --preview

What Should You See?

You should see Extracted 21 rows followed by five raw dictionaries. The capitalization remains unchanged because the preview performs no transformation.

Good progress. Your extractor now proves exactly what the pipeline received before any values are cleaned.

Preview Command Not Working?

  • Confirm Terminal still shows the activated .venv environment.
  • Check that pipeline.py sits beside sales_history.csv inside sales-forecast-etl.
  • Compare the CSV header with EXPECTED_FIELDS in the same order.
  • Diagnose the preview with targeted debugging help.

✔️ Awesome, I've got everything!

Your terminal reports 21 extracted rows plus five raw dictionaries. Double check that pipeline.py is saved.

ⓧ I'd like to double check the full code

Compare your complete pipeline.py with this reference:

import argparse
import csv
from decimal import Decimal

import duckdb

SOURCE_FILE = "sales_history.csv"
DATABASE_FILE = "sales_forecast.duckdb"
EXPECTED_FIELDS = ["sale_month", "region", "channel", "units", "unit_price"]
MONEY = Decimal("0.01")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Prepare monthly sales history for forecasting research."
    )
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument("--preview", action="store_true")
    modes.add_argument("--naive", action="store_true")
    modes.add_argument("--quality", action="store_true")
    return parser.parse_args()


def extract_sales():
    with open(SOURCE_FILE, newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")
        return list(reader)


def preview_sales(rows):
    print(f"Extracted {len(rows)} rows")
    for row in rows[:5]:
        print(row)


def main():
    args = parse_args()
    rows = extract_sales()

    if args.preview:
        preview_sales(rows)


if __name__ == "__main__":
    main()

Your raw-data checkpoint is repeatable. Next, you will test how direct numeric conversion handles the deliberately flawed input.

Let the Naive Transform Fail

Your Python extractor already proves that all 21 source rows arrive unchanged. The next question is whether each numeric field can pass through a simple ETL transform.

Direct conversion assumes every units value is numeric. It makes the same assumption about every price value.

In this step, get ready to:
  • Add naive_transform to convert source numbers directly.
  • Route --naive to the new transform.
  • Run the naive mode to test every source row.
Add the naive transform

A transform converts raw text into values that calculations can use. This first version applies direct conversions to every row.

  • In pipeline.py, find the preview_sales function.
  • Define the direct conversion loop between preview_sales and main by copying this code:
def naive_transform(rows):
    for row in rows:
        units = int(row["units"])
        unit_price = Decimal(row["unit_price"])
        _ = units * unit_price
    print("Naive transform completed")

What does this code do?

  • The for loop sends each raw row through the same conversion.
  • The int(row["units"]) expression converts units into an integer.
  • The Decimal(row["unit_price"]) expression converts the price into a decimal value.
  • The final print call is reached only after the loop handles every row.
  • Save pipeline.py.
  • Confirm the existing preview remains intact by running this command:
python pipeline.py --preview

What should you see?

You should still see Extracted 21 rows. Five raw dictionaries should follow.

Good, your raw-data checkpoint remains intact.

Preview no longer runs?

  • Make sure naive_transform sits between preview_sales and main.
  • Check that every line inside naive_transform uses the indentation shown above.
  • Help me restore the preview output.
Connect the naive mode

The argument parser already recognizes --naive. The main() function now needs a branch that sends extracted rows into the new transform.

  • Return to pipeline.py from earlier.
  • Scroll to the main() function.
  • Find this preview branch:
    if args.preview:
        preview_sales(rows)

What are you locating?

This branch currently sends extracted rows to preview_sales when the learner selects --preview.

  • Replace that branch with this version:
    if args.preview:
        preview_sales(rows)
        return

    if args.naive:
        naive_transform(rows)

What changed?

The return statement ends the preview path immediately. The second branch passes extracted rows to naive_transform when --naive is active.

Run the naive mode

Both modes are now reachable from the same script. The naive run tests every raw numeric value through direct conversion.

  • Save pipeline.py.

Before you run it, do you think all 21 rows will reach the completion message?

  • Test every raw row by running this command:
python pipeline.py --naive

This failure is intentional

The script stops when int(row["units"]) receives n/a. You do not see Naive transform completed.

One malformed field blocks every later sale from reaching a completed dataset.

  • Switch back to sales_history.csv from earlier.
  • Count the CSV header as source line 1.
  • Locate the row 2026-03,North,Online,n/a,21.00.

The malformed units value sits on source line 11. Every record after that line remains unprocessed because the conversion exception ends the run.

Did the naive transform complete?

  • Confirm sales_history.csv still contains 2026-03,North,Online,n/a,21.00.
  • Check that main() calls naive_transform(rows) under if args.naive:.
  • Help me reproduce the intended naive transform failure.

✔️ Awesome, I've got everything!

You have reproduced the direct-conversion failure. Your pipeline now shows exactly how one malformed value can stop every later record.

ⓧ I'd like to double check the full code

The complete cumulative pipeline.py appears below for comparison.

import argparse
import csv
from decimal import Decimal

import duckdb

SOURCE_FILE = "sales_history.csv"
DATABASE_FILE = "sales_forecast.duckdb"
EXPECTED_FIELDS = ["sale_month", "region", "channel", "units", "unit_price"]
MONEY = Decimal("0.01")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Prepare monthly sales history for forecasting research."
    )
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument("--preview", action="store_true")
    modes.add_argument("--naive", action="store_true")
    modes.add_argument("--quality", action="store_true")
    return parser.parse_args()


def extract_sales():
    with open(SOURCE_FILE, newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")
        return list(reader)


def preview_sales(rows):
    print(f"Extracted {len(rows)} rows")
    for row in rows[:5]:
        print(row)


def naive_transform(rows):
    for row in rows:
        units = int(row["units"])
        unit_price = Decimal(row["unit_price"])
        _ = units * unit_price
    print("Naive transform completed")


def main():
    args = parse_args()
    rows = extract_sales()

    if args.preview:
        preview_sales(rows)
        return

    if args.naive:
        naive_transform(rows)


if __name__ == "__main__":
    main()

You found the exact point where direct conversion breaks. Next, you will keep valid sales moving by placing malformed rows in an auditable rejected-row quarantine.

Validate and Quarantine Sales Rows

The naive transform stopped at one malformed units value. Every later month was blocked from the analytical dataset.

Row-level validation checks each record independently. One bad value can no longer block usable sales history.

The transform normalizes valid records into a clean collection. It places invalid records in a rejected-row quarantine for investigation.

In this step, get ready to:
  • Convert each valid source row into a normalized sales record.
  • Preserve auditable details for every rejected row.
  • Print quality counts plus line-specific rejection reasons.
Build the clean and reject paths

A row-level try block gives every source record its own chance to pass. The exception path captures failed records without ending the loop.

  • In pipeline.py, locate the import lines at the top of the file.
  • Replace from decimal import Decimal with these two lines:
from datetime import datetime
from decimal import Decimal, InvalidOperation

Why These Imports?

  • The datetime class checks whether each month matches the required date format.
  • The InvalidOperation exception identifies values that cannot become valid decimal numbers.

The next two snippets form one continuous transform_sales function. The first section prepares valid records for the clean collection.

  • In pipeline.py, find the end of naive_transform.
  • Start transform_sales beneath naive_transform by copying this first section:
def transform_sales(rows):
    clean_rows = []
    rejected_rows = []

    for row_number, row in enumerate(rows, start=2):
        try:
            sale_month = datetime.strptime(
                row["sale_month"].strip(), "%Y-%m"
            ).date()
            region = row["region"].strip().title()
            channel = row["channel"].strip().title()

            if not region or not channel:
                raise ValueError("region and channel are required")

            units = int(row["units"].strip())
            unit_price = Decimal(
                row["unit_price"].strip().replace("$", "")
            ).quantize(MONEY)

How Does the Clean Path Work?

  • The clean_rows collection holds normalized records that pass every check.
  • The rejected_rows collection keeps records that fail a check.
  • The enumerate call starts at 2 because the header occupies the first line of the CSV file.
  • The strptime call accepts months that match %Y-%m.
  • The title calls normalize inconsistent capitalization in region and channel labels.
  • The price conversion removes a currency symbol before rounding the value to two decimal places.
  • Complete transform_sales by copying this second section immediately below ).quantize(MONEY):
            if units <= 0 or unit_price <= 0:
                raise ValueError("units and unit_price must be positive")

            revenue = (unit_price * units).quantize(MONEY)
            clean_rows.append(
                (sale_month, region, channel, units, unit_price, revenue)
            )
        except (InvalidOperation, KeyError, TypeError, ValueError) as error:
            raw_row = " | ".join(
                f"{field}={row.get(field)!r}" for field in EXPECTED_FIELDS
            )
            rejected_rows.append((row_number, str(error), raw_row))

    return clean_rows, rejected_rows

How Does Quarantine Work?

  • The positive-value check rejects zero or negative units and prices.
  • The revenue calculation uses MONEY to keep two decimal places.
  • The except block handles one failed row before the loop continues to the next record.
  • The raw_row string preserves the original field values for later investigation.
  • Each rejected tuple stores the source line. It also stores the failure reason and original values.
  • Save pipeline.py.
  • Confirm the raw preview still works by running this command:
python pipeline.py --preview

What Should You See?

You should still see Extracted 21 rows followed by five raw row dictionaries. This confirms the existing extract path remains intact.

Preview Stopped Working?

  • Check that both import lines sit with the other imports at the top of pipeline.py.
  • Check that the second half of transform_sales continues at the indentation shown above.
  • Confirm that transform_sales appears below naive_transform.
  • Help me debug my transform function.

Good progress. Your raw preview still works with the new row-level transform in place.

Connect the quality report

A quality report makes the two transform outcomes visible. It shows how much data can continue through the pipeline and where each rejected record came from.

  • Add print_quality below transform_sales by copying this function:
def print_quality(clean_rows, rejected_rows):
    print(f"Clean rows: {len(clean_rows)}")
    print(f"Rejected rows: {len(rejected_rows)}")
    for row_number, reason, _ in rejected_rows:
        print(f"Line {row_number}: {reason}")

What Does the Report Show?

  • The first two lines compare the number of clean records with the number of rejected records.
  • The loop prints the original source line for each reject.
  • The underscore ignores the stored raw values because this compact report only needs the line number and reason.
  • In main, find the condition that checks args.naive.
  • Replace the end of main from that condition through the call to main() with this code:
    if args.naive:
        naive_transform(rows)
        return

    clean_rows, rejected_rows = transform_sales(rows)

    if args.quality:
        print_quality(clean_rows, rejected_rows)


if __name__ == "__main__":
    main()

How Does the Quality Path Work?

  • The new return keeps the deliberately naive path separate from the robust transform.
  • The default flow now passes every extracted row through transform_sales.
  • The quality condition prints the counts and rejection reasons without loading a database.
  • Save pipeline.py.

Before you run the quality path, do you expect one malformed row to stop the loop again?

  • Run the quality report by entering this command:
python pipeline.py --quality

What Should You See?

You should see Clean rows: 18 followed by Rejected rows: 3.

The rejected entries point to source lines 11, 15, and 19. Their reasons identify the invalid units value, missing region, and invalid month format.

Counts Do Not Match?

  • Confirm that sales_history.csv still contains all 21 source records.
  • Check that enumerate uses start=2 so the reported lines match the source file.
  • Check that the exception block appends to rejected_rows at the same indentation level as the try block.
  • Help me find why my clean and rejected counts differ.

That recovery path is working. Eighteen valid records now continue while three flawed records remain available for investigation.

✔️ Awesome, I've got everything!

Your pipeline.py now reports 18 clean rows and three line-specific rejects. Make sure the file is saved before moving on.

ⓧ I'd like to double check the full code

Compare your complete pipeline.py with this version. The existing preview and naive paths remain in place.

import argparse
import csv
from datetime import datetime
from decimal import Decimal, InvalidOperation

import duckdb

SOURCE_FILE = "sales_history.csv"
DATABASE_FILE = "sales_forecast.duckdb"
EXPECTED_FIELDS = ["sale_month", "region", "channel", "units", "unit_price"]
MONEY = Decimal("0.01")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Prepare monthly sales history for forecasting research."
    )
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument("--preview", action="store_true")
    modes.add_argument("--naive", action="store_true")
    modes.add_argument("--quality", action="store_true")
    return parser.parse_args()


def extract_sales():
    with open(SOURCE_FILE, newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")
        return list(reader)


def preview_sales(rows):
    print(f"Extracted {len(rows)} rows")
    for row in rows[:5]:
        print(row)


def naive_transform(rows):
    for row in rows:
        units = int(row["units"])
        unit_price = Decimal(row["unit_price"])
        _ = units * unit_price
    print("Naive transform completed")


def transform_sales(rows):
    clean_rows = []
    rejected_rows = []

    for row_number, row in enumerate(rows, start=2):
        try:
            sale_month = datetime.strptime(
                row["sale_month"].strip(), "%Y-%m"
            ).date()
            region = row["region"].strip().title()
            channel = row["channel"].strip().title()

            if not region or not channel:
                raise ValueError("region and channel are required")

            units = int(row["units"].strip())
            unit_price = Decimal(
                row["unit_price"].strip().replace("$", "")
            ).quantize(MONEY)

            if units <= 0 or unit_price <= 0:
                raise ValueError("units and unit_price must be positive")

            revenue = (unit_price * units).quantize(MONEY)
            clean_rows.append(
                (sale_month, region, channel, units, unit_price, revenue)
            )
        except (InvalidOperation, KeyError, TypeError, ValueError) as error:
            raw_row = " | ".join(
                f"{field}={row.get(field)!r}" for field in EXPECTED_FIELDS
            )
            rejected_rows.append((row_number, str(error), raw_row))

    return clean_rows, rejected_rows


def print_quality(clean_rows, rejected_rows):
    print(f"Clean rows: {len(clean_rows)}")
    print(f"Rejected rows: {len(rejected_rows)}")
    for row_number, reason, _ in rejected_rows:
        print(f"Line {row_number}: {reason}")


def main():
    args = parse_args()
    rows = extract_sales()

    if args.preview:
        preview_sales(rows)
        return

    if args.naive:
        naive_transform(rows)
        return

    clean_rows, rejected_rows = transform_sales(rows)

    if args.quality:
        print_quality(clean_rows, rejected_rows)


if __name__ == "__main__":
    main()

How to Compare the File

Check the import group first. Then compare transform_sales, print_quality, and the final branches in main.

Your ETL pipeline can now separate usable sales history from auditable rejects. Next, you will load both collections into DuckDB and build the monthly trend table.

Load Monthly Sales into DuckDB

Your quality checks now keep 18 usable sales records moving through the pipeline. They also quarantine three flawed rows with evidence you can inspect.

Durable forecasting research needs consistent monthly tables. In this step, DuckDB writes each ETL result to sales_forecast.duckdb before the terminal prints the sales trend.

In this step, get ready to:
  • Persist clean sales records plus quarantined rows in separate database tables.
  • Aggregate sales by month, region, and channel.
  • Print the six-month sales trend twice to prove the load is repeatable.
Create the database tables

A persistent database keeps the transformed rows available after the script finishes. Separate tables preserve trusted sales data alongside the rejected-row audit trail.

  • In pipeline.py, find the end of print_quality().
  • Add the first part of load_pipeline() above main() by pasting this code:
def load_pipeline(clean_rows, rejected_rows):
    with duckdb.connect(DATABASE_FILE) as connection:
        connection.execute(
            """
            CREATE OR REPLACE TABLE clean_sales (
                sale_month DATE,
                region VARCHAR,
                channel VARCHAR,
                units INTEGER,
                unit_price DECIMAL(10, 2),
                revenue DECIMAL(12, 2)
            )
            """
        )
        connection.execute(
            """
            CREATE OR REPLACE TABLE rejected_sales (
                source_line INTEGER,
                reason VARCHAR,
                raw_row VARCHAR
            )
            """
        )

What does this code do?

  • DuckDB opens sales_forecast.duckdb as a file-backed database. Closing the connection leaves its tables on disk.
  • The clean_sales schema gives dates, counts, labels, prices, and revenue explicit database types.
  • The rejected_sales schema stores each source line with its failure reason plus the original values.
  • The CREATE OR REPLACE TABLE statements rebuild both tables on every run. This prevents old records from accumulating.

The default path must call the loader. The quality-only path needs to stop before that call so its existing behavior stays unchanged.

  • In main(), find this block near the bottom of pipeline.py:
    if args.quality:
        print_quality(clean_rows, rejected_rows)

Why change this block?

This branch currently prints the quality summary. The new default database path belongs immediately after it.

  • Replace that block with this expanded version:
    if args.quality:
        print_quality(clean_rows, rejected_rows)
        return

    load_pipeline(clean_rows, rejected_rows)

How does the default path work?

The return keeps --quality focused on validation output. Running the script without a mode now sends the transformed collections to load_pipeline().

  • Save pipeline.py.
  • Create the initial database file by running this command:
python pipeline.py

What does this run prove?

The command extracts the CSV rows before transforming them. It then creates clean_sales plus rejected_sales inside the database file.

You return to the terminal prompt without an exception. The sales_forecast.duckdb file is now present in the sales-forecast-etl directory.

Database file missing?

  • Confirm the terminal prompt still shows the activated .venv in the sales-forecast-etl directory.
  • Check that load_pipeline(clean_rows, rejected_rows) is indented inside main().
  • Check the terminal for the first exception shown above the prompt.
  • Help me find why the DuckDB file was not created.
Load records and print the report

The schemas provide empty destinations. Parameterized inserts now load both transformed collections before a SQL query materializes consistent monthly totals.

  • In load_pipeline(), find the closing line of the rejected_sales table statement.
  • Continue the indented connection block by pasting this code below that line:
        if clean_rows:
            connection.executemany(
                "INSERT INTO clean_sales VALUES (?, ?, ?, ?, ?, ?)", clean_rows
            )
        if rejected_rows:
            connection.executemany(
                "INSERT INTO rejected_sales VALUES (?, ?, ?)", rejected_rows
            )

        connection.execute(
            """
            CREATE OR REPLACE TABLE monthly_sales AS
            SELECT
                sale_month,
                region,
                channel,
                SUM(units) AS units,
                SUM(revenue) AS revenue
            FROM clean_sales
            GROUP BY sale_month, region, channel
            """
        )

How are the rows loaded?

  • The first executemany() call inserts every valid tuple into clean_sales with parameter placeholders.
  • The second executemany() call preserves each rejected tuple in rejected_sales.
  • The final query groups trusted rows by sale_month, region, and channel.
  • Replacing monthly_sales on every run keeps the aggregate aligned with the current source file.
  • Save pipeline.py.
  • Populate all three database tables by running this command:
python pipeline.py

What changed in the database?

The existing tables are replaced before 18 clean rows plus three rejected rows are inserted. The run also materializes monthly_sales at the month, region, and channel grain.

You return to the terminal prompt without an exception. The database now contains the loaded records plus the monthly analytical table.

Seeing a database error?

  • Check that both executemany() calls remain indented inside the with duckdb.connect(DATABASE_FILE) as connection: block.
  • Confirm the clean_sales insert contains six question-mark placeholders.
  • Confirm the rejected_sales insert contains three question-mark placeholders.
  • Help me debug the DuckDB load error.

A database becomes useful for forecasting research when you can read its results clearly. The report queries stored table counts before rolling the monthly table into one trend row per month.

  • Add print_report() below load_pipeline() by pasting this code:
def print_report():
    with duckdb.connect(DATABASE_FILE) as connection:
        clean_count, rejected_count = connection.execute(
            """
            SELECT
                (SELECT COUNT(*) FROM clean_sales),
                (SELECT COUNT(*) FROM rejected_sales)
            """
        ).fetchone()

        trend_rows = connection.execute(
            """
            SELECT
                sale_month,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue
            FROM monthly_sales
            GROUP BY sale_month
            ORDER BY sale_month
            """
        ).fetchall()

    print("\nData quality report")
    print(f"Clean rows: {clean_count}")
    print(f"Rejected rows: {rejected_count}")
    print("\nMonthly sales trend")
    for sale_month, total_units, total_revenue in trend_rows:
        print(
            f"{sale_month} | units={total_units} | revenue=${total_revenue:,.2f}"
        )

What does the report show?

  • The first query reads the clean plus rejected counts from the persisted tables.
  • The second query totals units plus revenue across all regions and channels for each month.
  • The ORDER BY sale_month clause keeps the trend chronological.
  • The final loop formats revenue to two decimal places so each month is easy to compare.
  • In main(), find this loader call:
    load_pipeline(clean_rows, rejected_rows)

Where does the report belong?

The loader finishes before the report reads the database. This order ensures every printed count comes from the newly rebuilt tables.

  • Replace the loader call with these two lines:
    load_pipeline(clean_rows, rejected_rows)
    print_report()

How does the workflow finish?

The first call persists the latest transformed collections. The second call reads that completed database state into the terminal report.

Before you run this, which month do you expect to finish with the highest unit total?

  • Save pipeline.py.
  • Generate the full database report by running this command:
python pipeline.py

What should the completed report prove?

The report reads 18 trusted records plus three quarantined records from DuckDB. Its six chronological rows show the monthly totals available for forecasting research.

You should see Clean rows: 18 plus Rejected rows: 3. The report ends with 2026-06-01 | units=309 | revenue=$8,028.00.

Report missing or incomplete?

  • Confirm print_report() appears directly after load_pipeline(clean_rows, rejected_rows) inside main().
  • Check that the trend query reads from monthly_sales.
  • Check that GROUP BY sale_month appears before ORDER BY sale_month.
  • Help me debug the missing monthly report.
Prove the load is repeatable

A repeatable load produces the same database state from the same source file. Replacing the tables before inserting rows protects the report from duplicate accumulation.

Before you rerun the pipeline, do you expect either row count to increase?

  • Run the completed pipeline again with the same source data:
python pipeline.py

Why do the results stay stable?

Each database table is rebuilt from the current transformed collections. The second run therefore reports the same counts plus the same monthly totals.

You should still see Clean rows: 18 plus Rejected rows: 3. The final line should remain 2026-06-01 | units=309 | revenue=$8,028.00.

Counts changed after the rerun?

  • Check that all three table statements begin with CREATE OR REPLACE TABLE.
  • Confirm the insert statements run after the replacement statements.
  • Compare sales_history.csv with the original 21-record source file.
  • Help me find why rerunning the ETL pipeline changes the row counts.

✔️ Awesome, I've got everything!

Your pipeline now rebuilds three DuckDB tables before printing the same six-month report on every run.

ⓧ I'd like to double check the full code

Compare your saved pipeline.py with this completed reference.

import argparse
import csv
from datetime import datetime
from decimal import Decimal, InvalidOperation

import duckdb

SOURCE_FILE = "sales_history.csv"
DATABASE_FILE = "sales_forecast.duckdb"
EXPECTED_FIELDS = ["sale_month", "region", "channel", "units", "unit_price"]
MONEY = Decimal("0.01")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Prepare monthly sales history for forecasting research."
    )
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument("--preview", action="store_true")
    modes.add_argument("--naive", action="store_true")
    modes.add_argument("--quality", action="store_true")
    return parser.parse_args()


def extract_sales():
    with open(SOURCE_FILE, newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError(f"Expected columns {EXPECTED_FIELDS}, got {reader.fieldnames}")
        return list(reader)


def preview_sales(rows):
    print(f"Extracted {len(rows)} rows")
    for row in rows[:5]:
        print(row)


def naive_transform(rows):
    for row in rows:
        units = int(row["units"])
        unit_price = Decimal(row["unit_price"])
        _ = units * unit_price
    print("Naive transform completed")


def transform_sales(rows):
    clean_rows = []
    rejected_rows = []

    for row_number, row in enumerate(rows, start=2):
        try:
            sale_month = datetime.strptime(
                row["sale_month"].strip(), "%Y-%m"
            ).date()
            region = row["region"].strip().title()
            channel = row["channel"].strip().title()

            if not region or not channel:
                raise ValueError("region and channel are required")

            units = int(row["units"].strip())
            unit_price = Decimal(
                row["unit_price"].strip().replace("$", "")
            ).quantize(MONEY)

            if units <= 0 or unit_price <= 0:
                raise ValueError("units and unit_price must be positive")

            revenue = (unit_price * units).quantize(MONEY)
            clean_rows.append(
                (sale_month, region, channel, units, unit_price, revenue)
            )
        except (InvalidOperation, KeyError, TypeError, ValueError) as error:
            raw_row = " | ".join(
                f"{field}={row.get(field)!r}" for field in EXPECTED_FIELDS
            )
            rejected_rows.append((row_number, str(error), raw_row))

    return clean_rows, rejected_rows


def print_quality(clean_rows, rejected_rows):
    print(f"Clean rows: {len(clean_rows)}")
    print(f"Rejected rows: {len(rejected_rows)}")
    for row_number, reason, _ in rejected_rows:
        print(f"Line {row_number}: {reason}")


def load_pipeline(clean_rows, rejected_rows):
    with duckdb.connect(DATABASE_FILE) as connection:
        connection.execute(
            """
            CREATE OR REPLACE TABLE clean_sales (
                sale_month DATE,
                region VARCHAR,
                channel VARCHAR,
                units INTEGER,
                unit_price DECIMAL(10, 2),
                revenue DECIMAL(12, 2)
            )
            """
        )
        connection.execute(
            """
            CREATE OR REPLACE TABLE rejected_sales (
                source_line INTEGER,
                reason VARCHAR,
                raw_row VARCHAR
            )
            """
        )

        if clean_rows:
            connection.executemany(
                "INSERT INTO clean_sales VALUES (?, ?, ?, ?, ?, ?)", clean_rows
            )
        if rejected_rows:
            connection.executemany(
                "INSERT INTO rejected_sales VALUES (?, ?, ?)", rejected_rows
            )

        connection.execute(
            """
            CREATE OR REPLACE TABLE monthly_sales AS
            SELECT
                sale_month,
                region,
                channel,
                SUM(units) AS units,
                SUM(revenue) AS revenue
            FROM clean_sales
            GROUP BY sale_month, region, channel
            """
        )


def print_report():
    with duckdb.connect(DATABASE_FILE) as connection:
        clean_count, rejected_count = connection.execute(
            """
            SELECT
                (SELECT COUNT(*) FROM clean_sales),
                (SELECT COUNT(*) FROM rejected_sales)
            """
        ).fetchone()

        trend_rows = connection.execute(
            """
            SELECT
                sale_month,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue
            FROM monthly_sales
            GROUP BY sale_month
            ORDER BY sale_month
            """
        ).fetchall()

    print("\nData quality report")
    print(f"Clean rows: {clean_count}")
    print(f"Rejected rows: {rejected_count}")
    print("\nMonthly sales trend")
    for sale_month, total_units, total_revenue in trend_rows:
        print(
            f"{sale_month} | units={total_units} | revenue=${total_revenue:,.2f}"
        )


def main():
    args = parse_args()
    rows = extract_sales()

    if args.preview:
        preview_sales(rows)
        return

    if args.naive:
        naive_transform(rows)
        return

    clean_rows, rejected_rows = transform_sales(rows)

    if args.quality:
        print_quality(clean_rows, rejected_rows)
        return

    load_pipeline(clean_rows, rejected_rows)
    print_report()


if __name__ == "__main__":
    main()

How to use this reference

This file matches the completed repeatable pipeline. Compare the function order plus indentation before saving your own copy.

That closes the loop. Your messy source data now becomes an auditable database with stable monthly evidence for forecasting research.

Secret mission

Add a Month-over-Month Forecast Signal

Your monthly report shows how revenue changes across six months. This extension creates a forecast feature table that compares each month with the previous period.

Clean Up Your Resources

Clean Up Your Resources

Your ETL pipeline runs entirely on your Mac. It creates no cloud resources or service charges.

Choose whether to keep the sales-forecast-etl directory, pause by leaving its virtual environment, or delete the directory.

Resources you used:

  • Local .venv environment containing DuckDB version 1.5.5.
  • Pinned dependency file requirements.txt.
  • Synthetic source file sales_history.csv with 21 raw sales records.
  • Repeatable pipeline script pipeline.py.
  • Persistent analytics database sales_forecast.duckdb.
  • Forecast feature script forecast_features.py.

Keep everything running

No action needed. Choose this if you want to continue researching sales trends with the finished pipeline.

  • Keep the complete sales-forecast-etl directory on your Mac.
  • Retain sales_forecast.duckdb so the clean sales, quarantined rows, monthly trends, and forecast features remain available.
  • Replace the synthetic records in sales_history.csv with an approved work dataset when you are ready to extend the research.

Pause - I'll come back to this later

Leave the active environment to pause your terminal session. Every file remains available inside sales-forecast-etl.

  • Leave the activated .venv by running this command:
deactivate

What does this command do?

The command leaves the active virtual environment without deleting it. Your terminal prompt returns to its normal state.

  • Keep the sales-forecast-etl directory on disk for your next session.
  • Reactivate .venv when you return to the project.

Delete - I don't want to use this again

Remove the complete local workspace if you do not plan to reuse it. The sales-forecast-etl directory contains every artifact created in this project.

Deleting the directory is a significant cleanup step. Confirm its contents first so unrelated files stay untouched.

  • Confirm the sales-forecast-etl directory contains only the project resources listed above.
  • Leave the activated .venv by running this command:
deactivate

Why deactivate first?

Deactivating releases the terminal session from the virtual environment before you remove its directory.

  • Navigate to the directory containing sales-forecast-etl in your Mac file browser.
  • Move the complete sales-forecast-etl directory to Trash.
  • Open Trash to verify the directory was moved.

You should see sales-forecast-etl in Trash. The virtual environment, scripts, source CSV, and DuckDB database are now removed from their previous location.

Nice Work!

Nice Work!

You did it! Your repeatable sales ETL pipeline now turns messy source records into auditable monthly metrics for forecasting research.

You've learned how to:

  • Build a visible raw-data checkpoint that extracts 21 records from a CSV file. The preview keeps the first five records unchanged for inspection.
  • Turn a failed n/a conversion into row-level data validation. The quarantine preserves 18 clean records. It also records 3 rejected rows with their source lines and failure reasons.
  • Load normalized sales into a persistent DuckDB database. Repeatable table replacement prevents duplicate records across reruns. Monthly reports expose six months of units and revenue trends.
  • Secret Mission: Added previous-month revenue with a SQL window function. Used lag(total_revenue) to calculate month-over-month revenue change.

Ready to quiz yourself?