from __future__ import annotations import csv import hashlib import json import random import shutil import zipfile from collections import defaultdict from datetime import date, datetime, timedelta from pathlib import Path from openpyxl import Workbook, load_workbook from openpyxl.chart import BarChart, LineChart, Reference from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.worksheet.table import Table, TableStyleInfo ROOT = Path(__file__).resolve().parent INPUT = ROOT / "input" OUTPUT = ROOT / "output" PUBLIC_DOWNLOADS = ROOT.parent.parent / "site" / "proof" / "analytics-reconciliation" / "downloads" SEED = 772026 REGIONS = ["East", "Central", "West"] CHANNELS = ["Direct", "Partner", "Web"] HEADER_FILL = PatternFill("solid", fgColor="171717") ACCENT_FILL = PatternFill("solid", fgColor="EAF4FF") WHITE_FONT = Font(color="FFFFFF", bold=True) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(65536), b""): digest.update(block) return digest.hexdigest() def write_inputs() -> None: random.seed(SEED) shutil.rmtree(INPUT, ignore_errors=True) INPUT.mkdir(parents=True) start = date(2026, 5, 1) rows: list[dict[str, object]] = [] for index in range(175): day = start + timedelta(days=index % 90) region = REGIONS[index % len(REGIONS)] channel = CHANNELS[(index * 2) % len(CHANNELS)] orders = 4 + (index * 7) % 29 unit_value = 78 + (index * 11) % 94 revenue = round(orders * unit_value + random.uniform(-20, 20), 2) refunds = round(revenue * ((index % 9) / 200), 2) rows.append({ "event_id": f"EVT-{index + 1:04d}", "occurred_at": day.isoformat(), "region": region, "channel": channel, "orders": orders, "revenue": revenue, "refunds": refunds, "processing_minutes": 12 + (index * 13) % 83, }) # Five intentionally excluded rows: three duplicates and two invalid records. rows.extend([dict(rows[7]), dict(rows[48]), dict(rows[109])]) rows.append({ "event_id": "EVT-BAD-REGION", "occurred_at": "2026-07-30", "region": "Unknown", "channel": "Web", "orders": 8, "revenue": 812.50, "refunds": 0, "processing_minutes": 22, }) rows.append({ "event_id": "EVT-BAD-ORDERS", "occurred_at": "2026-07-31", "region": "East", "channel": "Direct", "orders": -4, "revenue": 455.00, "refunds": 0, "processing_minutes": 19, }) for file_index in range(3): chunk = rows[file_index * 60 : (file_index + 1) * 60] path = INPUT / f"operations_{file_index + 1}.csv" revenue_header = "gross_revenue" if file_index == 1 else "revenue" headers = [ "event_id", "occurred_at", "region", "channel", "orders", revenue_header, "refunds", "processing_minutes", ] if file_index == 2: headers.append("campaign") with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=headers) writer.writeheader() for row in chunk: emitted = dict(row) if file_index == 1: emitted["gross_revenue"] = emitted.pop("revenue") if file_index == 2: emitted["campaign"] = "summer-ops" writer.writerow(emitted) def parse_inputs(): clean: list[dict[str, object]] = [] exceptions: list[dict[str, object]] = [] seen: set[str] = set() schemas: dict[str, list[str]] = {} source_rows = 0 source_revenue = 0.0 source_orders = 0 for path in sorted(INPUT.glob("*.csv")): with path.open(encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) schemas[path.name] = list(reader.fieldnames or []) for raw in reader: source_rows += 1 normalized = dict(raw) if "gross_revenue" in normalized: normalized["revenue"] = normalized.pop("gross_revenue") revenue = float(normalized["revenue"]) orders = int(normalized["orders"]) source_revenue += revenue source_orders += orders event_id = normalized["event_id"].strip() reason = None if event_id in seen: reason = "duplicate_event_id" elif normalized["region"] not in REGIONS: reason = "invalid_region" elif orders < 0: reason = "negative_orders" if reason: exceptions.append({ "source_file": path.name, "event_id": event_id, "reason": reason, "orders": orders, "revenue": revenue, }) continue seen.add(event_id) clean.append({ "event_id": event_id, "occurred_at": normalized["occurred_at"], "region": normalized["region"], "channel": normalized["channel"], "orders": orders, "revenue": revenue, "refunds": float(normalized["refunds"]), "net_revenue": round(revenue - float(normalized["refunds"]), 2), "processing_minutes": int(normalized["processing_minutes"]), "source_file": path.name, }) excluded_revenue = round(sum(float(row["revenue"]) for row in exceptions), 2) excluded_orders = sum(int(row["orders"]) for row in exceptions) clean_revenue = round(sum(float(row["revenue"]) for row in clean), 2) clean_orders = sum(int(row["orders"]) for row in clean) reconciliation = { "source_rows": source_rows, "accepted_rows": len(clean), "excluded_rows": len(exceptions), "duplicate_rows": sum(row["reason"] == "duplicate_event_id" for row in exceptions), "invalid_rows": sum(row["reason"] != "duplicate_event_id" for row in exceptions), "source_orders": source_orders, "accepted_orders": clean_orders, "excluded_orders": excluded_orders, "orders_control_difference": source_orders - clean_orders - excluded_orders, "source_gross_revenue": round(source_revenue, 2), "accepted_gross_revenue": clean_revenue, "excluded_gross_revenue": excluded_revenue, "revenue_control_difference": round(source_revenue - clean_revenue - excluded_revenue, 2), "schema_aliases_applied": {"operations_2.csv": "gross_revenue -> revenue"}, "extra_columns_ignored": {"operations_3.csv": ["campaign"]}, "input_schemas": schemas, } if reconciliation["orders_control_difference"] != 0: raise AssertionError("orders control total failed") if abs(reconciliation["revenue_control_difference"]) > 0.01: raise AssertionError("revenue control total failed") return clean, exceptions, reconciliation def group_metrics(clean): by_region = defaultdict(lambda: {"orders": 0, "gross": 0.0, "net": 0.0}) by_channel = defaultdict(lambda: {"orders": 0, "gross": 0.0, "net": 0.0}) by_day = defaultdict(float) for row in clean: for group, key in ((by_region, row["region"]), (by_channel, row["channel"])): group[key]["orders"] += int(row["orders"]) group[key]["gross"] += float(row["revenue"]) group[key]["net"] += float(row["net_revenue"]) by_day[row["occurred_at"]] += float(row["net_revenue"]) return by_region, by_channel, by_day def style_header(sheet, row=1): for cell in sheet[row]: cell.fill = HEADER_FILL cell.font = WHITE_FONT cell.alignment = Alignment(horizontal="center") def add_table(sheet, name): if sheet.max_row < 2 or sheet.max_column < 1: return ref = f"A1:{sheet.cell(sheet.max_row, sheet.max_column).coordinate}" table = Table(displayName=name, ref=ref) table.tableStyleInfo = TableStyleInfo(name="TableStyleMedium2", showRowStripes=True) sheet.add_table(table) def build_workbook(clean, exceptions, reconciliation, by_region, by_channel, by_day): workbook = Workbook() dashboard = workbook.active dashboard.title = "Executive Summary" dashboard.sheet_view.showGridLines = False dashboard["A1"] = "Synthetic Operations Analytics" dashboard["A1"].font = Font(size=20, bold=True) dashboard["A2"] = "Self-owned portfolio sample · synthetic data · not client work" dashboard["A2"].font = Font(italic=True, color="666666") metrics = [ ("Accepted rows", len(clean)), ("Excluded rows", len(exceptions)), ("Total orders", sum(int(row["orders"]) for row in clean)), ("Gross revenue", round(sum(float(row["revenue"]) for row in clean), 2)), ("Net revenue", round(sum(float(row["net_revenue"]) for row in clean), 2)), ("Revenue control difference", reconciliation["revenue_control_difference"]), ] for index, (label, value) in enumerate(metrics, start=4): dashboard.cell(index, 1, label).fill = ACCENT_FILL dashboard.cell(index, 1).font = Font(bold=True) dashboard.cell(index, 2, value) dashboard["B7"].number_format = "$#,##0.00" dashboard["B8"].number_format = "$#,##0.00" dashboard["B9"].number_format = "$#,##0.00" dashboard["A12"] = "Region" dashboard["B12"] = "Orders" dashboard["C12"] = "Gross Revenue" dashboard["D12"] = "Net Revenue" for r_index, region in enumerate(REGIONS, start=13): values = by_region[region] dashboard.cell(r_index, 1, region) dashboard.cell(r_index, 2, values["orders"]) dashboard.cell(r_index, 3, round(values["gross"], 2)).number_format = "$#,##0.00" dashboard.cell(r_index, 4, round(values["net"], 2)).number_format = "$#,##0.00" style_header(dashboard, 12) region_chart = BarChart() region_chart.title = "Net Revenue by Region" region_chart.y_axis.title = "USD" region_chart.add_data(Reference(dashboard, min_col=4, min_row=12, max_row=15), titles_from_data=True) region_chart.set_categories(Reference(dashboard, min_col=1, min_row=13, max_row=15)) region_chart.height = 7 region_chart.width = 12 dashboard.add_chart(region_chart, "F4") dashboard["A19"] = "Date" dashboard["B19"] = "Net Revenue" for row_index, (day, value) in enumerate(sorted(by_day.items()), start=20): dashboard.cell(row_index, 1, day) dashboard.cell(row_index, 2, round(value, 2)).number_format = "$#,##0.00" style_header(dashboard, 19) line_chart = LineChart() line_chart.title = "Daily Net Revenue" line_chart.add_data(Reference(dashboard, min_col=2, min_row=19, max_row=19 + len(by_day)), titles_from_data=True) line_chart.set_categories(Reference(dashboard, min_col=1, min_row=20, max_row=19 + len(by_day))) line_chart.height = 8 line_chart.width = 17 dashboard.add_chart(line_chart, "F20") dashboard.column_dimensions["A"].width = 28 dashboard.column_dimensions["B"].width = 18 dashboard.column_dimensions["C"].width = 18 dashboard.column_dimensions["D"].width = 18 clean_sheet = workbook.create_sheet("Clean Data") clean_headers = list(clean[0].keys()) clean_sheet.append(clean_headers) for row in clean: clean_sheet.append([row[column] for column in clean_headers]) style_header(clean_sheet) clean_sheet.freeze_panes = "A2" clean_sheet.auto_filter.ref = clean_sheet.dimensions add_table(clean_sheet, "CleanOperations") for column in range(1, clean_sheet.max_column + 1): clean_sheet.column_dimensions[clean_sheet.cell(1, column).column_letter].width = 20 exception_sheet = workbook.create_sheet("Exceptions") exception_headers = list(exceptions[0].keys()) exception_sheet.append(exception_headers) for row in exceptions: exception_sheet.append([row[column] for column in exception_headers]) style_header(exception_sheet) add_table(exception_sheet, "RejectedRows") exception_sheet.freeze_panes = "A2" reconciliation_sheet = workbook.create_sheet("Reconciliation") reconciliation_sheet.append(["Control", "Value"]) for key in ( "source_rows", "accepted_rows", "excluded_rows", "duplicate_rows", "invalid_rows", "source_orders", "accepted_orders", "excluded_orders", "orders_control_difference", "source_gross_revenue", "accepted_gross_revenue", "excluded_gross_revenue", "revenue_control_difference", ): reconciliation_sheet.append([key, reconciliation[key]]) style_header(reconciliation_sheet) reconciliation_sheet.column_dimensions["A"].width = 34 reconciliation_sheet.column_dimensions["B"].width = 22 for row in range(11, 15): reconciliation_sheet.cell(row, 2).number_format = "$#,##0.00" workbook.calculation.fullCalcOnLoad = True output = OUTPUT / "synthetic-operations-analytics.xlsx" workbook.save(output) return output def write_csv(path: Path, rows): headers = list(rows[0].keys()) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=headers) writer.writeheader() writer.writerows(rows) def build_public_dashboard(clean, exceptions, reconciliation, by_region, by_channel): total_orders = sum(int(row["orders"]) for row in clean) gross = round(sum(float(row["revenue"]) for row in clean), 2) net = round(sum(float(row["net_revenue"]) for row in clean), 2) max_net = max(values["net"] for values in by_region.values()) bars = "".join( f'
{region}
${values["net"]:,.0f}
' for region, values in sorted(by_region.items()) ) channels = "".join( f'{channel}{values["orders"]:,}${values["gross"]:,.2f}${values["net"]:,.2f}' for channel, values in sorted(by_channel.items()) ) exception_rows = "".join( f'{row["event_id"]}{row["reason"]}{row["source_file"]}${row["revenue"]:,.2f}' for row in exceptions ) html = f''' Analytics Reconciliation Proof — Reliable Workflows
Self-owned · synthetic data · executed pipeline

CSV/Excel analytics with control totals, not dashboard theater.

A deterministic Python pipeline ingests three intentionally inconsistent CSV exports, applies schema aliases, rejects duplicate and invalid records, reconciles every excluded value, and produces a styled Excel workbook with charts and clean-data handoff.

Download Excel workbookClean CSVMachine-readable evidenceInspect pipeline source

Executed result

Source rows{reconciliation["source_rows"]}
Accepted rows{len(clean)}
Total orders{total_orders:,}
Net revenue${net:,.0f}

Net revenue by region

{bars}

Channel totals

{channels}
ChannelOrdersGrossNet

Exceptions are accounted for

Three replayed IDs and two invalid records are excluded. Their orders and revenue remain visible in the reconciliation ledger, producing zero control-total difference.

{exception_rows}
EventReasonSourceGross
Control results

orders_control_difference = {reconciliation["orders_control_difference"]}
revenue_control_difference = {reconciliation["revenue_control_difference"]:.2f}

What this proves—and what it does not

Demonstrated

Multi-file ingestion, a header alias, ignored extra columns, typed normalization, deduplication, validation, KPI aggregation, reconciliation, CSV output, JSON evidence, and a multi-sheet Excel workbook with charts.

Boundary

This is a self-owned portfolio demonstration created with synthetic data. It is not client work, does not reproduce a buyer's files, and does not imply Power BI or Tableau project history.

''' public_dir = PUBLIC_DOWNLOADS.parent public_dir.mkdir(parents=True, exist_ok=True) (public_dir / "index.html").write_text(html, encoding="utf-8") def main(): shutil.rmtree(OUTPUT, ignore_errors=True) OUTPUT.mkdir(parents=True) write_inputs() clean, exceptions, reconciliation = parse_inputs() by_region, by_channel, by_day = group_metrics(clean) workbook = build_workbook(clean, exceptions, reconciliation, by_region, by_channel, by_day) clean_csv = OUTPUT / "clean-operations.csv" exception_csv = OUTPUT / "exceptions.csv" evidence_json = OUTPUT / "reconciliation.json" write_csv(clean_csv, clean) write_csv(exception_csv, exceptions) evidence_json.write_text(json.dumps(reconciliation, indent=2), encoding="utf-8") with zipfile.ZipFile(OUTPUT / "synthetic-source-csvs.zip", "w", zipfile.ZIP_DEFLATED) as archive: for path in sorted(INPUT.glob("*.csv")): archive.write(path, path.name) build_public_dashboard(clean, exceptions, reconciliation, by_region, by_channel) shutil.rmtree(PUBLIC_DOWNLOADS, ignore_errors=True) PUBLIC_DOWNLOADS.mkdir(parents=True) artifacts = [workbook, clean_csv, exception_csv, evidence_json, OUTPUT / "synthetic-source-csvs.zip"] for artifact in artifacts: shutil.copy2(artifact, PUBLIC_DOWNLOADS / artifact.name) shutil.copy2(Path(__file__), PUBLIC_DOWNLOADS / "build_sample.py") # Reload the workbook and assert the artifact structure survived save. verified = load_workbook(workbook, data_only=False, read_only=False) checks = { "sheets": verified.sheetnames, "charts": len(verified["Executive Summary"]._charts), "clean_rows_including_header": verified["Clean Data"].max_row, "exception_rows_including_header": verified["Exceptions"].max_row, "source_rows": reconciliation["source_rows"], "accepted_rows": reconciliation["accepted_rows"], "excluded_rows": reconciliation["excluded_rows"], "control_totals_pass": reconciliation["orders_control_difference"] == 0 and abs(reconciliation["revenue_control_difference"]) <= 0.01, "artifact_hashes": {path.name: sha256(path) for path in artifacts}, } if checks["charts"] != 2 or checks["clean_rows_including_header"] != len(clean) + 1: raise AssertionError(checks) print(json.dumps(checks, indent=2)) if __name__ == "__main__": main()