Module 6Capstone 20 min

Capstone: Build & harden

Implement the design: generate the returns feed, extend the pipeline and gate, add the metrics — then survive the three planted incidents.

Build day. You'll extend generate_raw.py to emit the returns feed (return ~6% of completed orders, 1–10 days after purchase, reasons from a small enum — plus, deliberately: one duplicate return_id, one return referencing a nonexistent order, one arriving 12 days late). Your own generator now plants your incidents, which is exactly how you know your defenses are real.

You don't have to invent the structure from a blank file. Here's a skeleton in the same style as generate_raw.py — the plumbing is done; the TODOs are where your judgment goes. Paste it at the bottom of generate_raw.py (it reuses the random, csv, date, timedelta imports already at the top) and call generate_returns() at the end of the file:

generate_returns() — fill in the TODOspython

def generate_returns():
    '''Emit data/raw/returns_YYYY-MM-DD.csv from the orders already written.
    Reads back the order exports so returns reference real orders.'''
    REASONS = ['damaged', 'wrong_item', 'changed_mind', 'late_delivery']

    # Collect (order_id, order_date) for COMPLETED orders from the exports.
    completed = []
    start = date(2026, 5, 1)
    for d in range(30):
        day = start + timedelta(days=d)
        with open(f'data/raw/orders_{day.isoformat()}.csv', newline='') as f:
            for r in csv.DictReader(f):
                if r['status'] == 'completed':
                    completed.append((r['order_id'], day))

    # Return ~6% of completed orders. Group the resulting return rows by the
    # DATE the return happened (purchase date + 1..10 days) so each day gets
    # its own returns_<date>.csv, matching the daily-export pattern.
    by_day = {}   # date -> list of [return_id, order_id, returned_at, reason]
    rid = 5000
    for order_id, order_day in completed:
        if random.random() < 0.06:
            rid += 1
            # TODO 1: pick a return date 1..10 days after order_day
            #         (use timedelta(days=random.randint(1, 10))).
            returned = order_day   # <- replace with the real return date
            reason = random.choice(REASONS)
            row = [f'R{rid}', order_id, returned.isoformat(), reason]
            by_day.setdefault(returned, []).append(row)

    # TODO 2 (planted incident: DUPLICATE return_id): append one extra row
    #        that reuses an existing R#### id, on any day. Your uniqueness
    #        check must catch it.
    # TODO 3 (planted incident: ORPHAN return): append one row whose order_id
    #        does not exist in any export (e.g. 'O0'). Your RI check decides
    #        whether that blocks or quarantines.
    # TODO 4 (planted incident: 12-DAYS-LATE return): append one row whose
    #        returned date is 12 days after its order_day, so it lands in a
    #        file far from the purchase. This is the late-arriving-data case.

    for day, rows in by_day.items():
        with open(f'data/raw/returns_{day.isoformat()}.csv', 'w', newline='') as f:
            w = csv.writer(f)
            w.writerow(['return_id', 'order_id', 'returned_at', 'reason'])
            w.writerows(rows)
    print(f'returns written: {sum(len(r) for r in by_day.values())} rows '
          f'across {len(by_day)} files')

Build checklist

1Model + pipeline: stg_returns view; fct_returns (grain: one row per return event) loaded with partition overwrite on returned_at::DATE. Run twice; row counts identical, or stop and fix.
2Gate: extend checks.py per your design — uniqueness, RI to fct_order_lines, accepted reasons, freshness, volume. The orphan-return check should implement your design ruling (block, or quarantine-and-count?), not a default.
3Resolution untouched, and prove it: wholesale refs (B####) must not enter dim_customer. Add the negative check: zero wholesale refs in customer_xwalk.
4Metrics: return_rate and net_revenue_after_returns in metrics.yaml per your definitions; consumer metrics gain the mandatory channel != 'wholesale' filter (or your v2 plan does it more gracefully). weekly_report.py gains the two new certified numbers.
5The survival test: run the full sequence — generate, pipeline, checks — and confirm the gate catches all three planted incidents by name. Fix data, not thresholds, and re-run to green.

the whole system, one command eachbash

python3 generate_raw.py     # sources (now with returns + planted incidents)
python3 pipeline.py         # raw -> staging -> marts, idempotent
python3 resolve.py          # golden customers + crosswalk
python3 checks.py           # the gate: expect 3 FAILs naming your incidents
#  ...triage, repair, re-run...
python3 checks.py           # green
python3 weekly_report.py    # Monday numbers, through the layer

That five-command sequence is a data platform — small, but architecturally identical to what runs at companies a thousand times Harbor Lane's size: layered ELT, idempotent loads, blocking quality gates, continuous resolution, governed metrics. Scale changes the tools; it doesn't change the shape.

The regression trap

Most points are lost not on the new feature but on breaking the old ones: the wholesale filter that silently changes historical net_revenue, the returns join that duplicates fact rows (grain violation → every metric inflates), the resolve.py rerun that reshuffles golden keys. Re-run Module 5's weekly_report.py and diff against pre-capstone output — consumer metrics for May must not move. That diff is your regression test.

PreviousCapstone: Design the foundation Next Capstone: Validate & document