"""Module 04 worked analysis. Read-only, local files, Python 3.9+ standard library. Save as worked_analysis.py beside the CSV downloads. Run: python3 worked_analysis.py Inspect the code, attempt the exercises first, then compare the printed results. """ import csv import json import math from collections import Counter, defaultdict from datetime import date from decimal import Decimal from pathlib import Path from statistics import mean, median FIELDS = ['order_id', 'order_date', 'channel', 'status', 'order_value_eur', 'delivery_days'] BASE = Path(__file__).resolve().parent def read_rows(name): with (BASE / name).open(newline='', encoding='utf-8') as handle: reader = csv.DictReader(handle) if reader.fieldnames != FIELDS: raise ValueError('Unexpected schema: ' + name) rows = list(reader) if not rows or any(None in r or any(v is None for v in r.values()) for r in rows): raise ValueError('Empty or malformed input: ' + name) return rows def prepare(raw, reference): """Use only the declared format rule and named, reference-backed field fixes. New duplicate groups or unexplained differences are exposed, never guessed. """ refs = {r['order_id']: r for r in reference} if len(refs) != len(reference): raise ValueError('Reference IDs must be unique') groups = defaultdict(list) for row in raw: groups[row['order_id']].append(row) if set(groups) != set(refs): raise ValueError('Source/reference identifier coverage differs') fix_fields = {'A008': 'order_date', 'A009': 'order_value_eur', 'Y003': 'channel', 'Y005': 'order_value_eur', 'Y006': 'delivery_days'} prepared = [] for order_id, candidates in groups.items(): if len(candidates) > 1: if order_id == 'A004' and len(candidates) == 2 and candidates[0] == candidates[1]: row = dict(candidates[0]) elif order_id == 'Y007' and len(candidates) == 2: matches = [r for r in candidates if r == refs[order_id]] if len(matches) != 1: raise ValueError('Y007 conflict lacks one authoritative match') row = dict(matches[0]) else: raise ValueError('New duplicate group needs investigation: ' + order_id) else: row = dict(candidates[0]) if order_id == 'A005': row['channel'] = row['channel'].strip().lower() if order_id in fix_fields: field = fix_fields[order_id] row[field] = refs[order_id][field] if row != refs[order_id]: raise ValueError('Unexplained source/reference difference: ' + order_id) prepared.append(row) return prepared def typed(rows): output = [] for row in rows: parsed_date = date.fromisoformat(row['order_date']) if parsed_date.isoformat() != row['order_date'] or not date(2026, 4, 1) <= parsed_date < date(2026, 6, 1): raise ValueError('Date outside declared contract: ' + row['order_id']) value = Decimal(row['order_value_eur']) if not value.is_finite() or value < 0: raise ValueError('Invalid required value: ' + row['order_id']) if row['channel'] not in ('web', 'store') or row['status'] not in ('completed', 'returned', 'cancelled'): raise ValueError('Category outside declared contract') delivery = None if row['delivery_days'] == '' else int(row['delivery_days']) if delivery is not None and delivery < 0: raise ValueError('Invalid delivery duration') output.append({**row, 'value': float(value), 'delivery': delivery}) return output def percentile_inc(values, p): if not values: raise ValueError('Percentile requires observations') xs = sorted(values) position = (len(xs) - 1) * p lower = int(position) upper = min(lower + 1, len(xs) - 1) return xs[lower] + (xs[upper] - xs[lower]) * (position - lower) def analyse(rows): completed = [r for r in rows if r['status'] == 'completed'] if not completed: raise ValueError('No completed orders') values = [r['value'] for r in completed] q1, q3 = percentile_inc(values, .25), percentile_inc(values, .75) low, high = q1 - 1.5 * (q3 - q1), q3 + 1.5 * (q3 - q1) flagged = [r for r in completed if r['value'] < low or r['value'] > high] unflagged = [r for r in completed if low <= r['value'] <= high] monthly, channels = [], [] for month in ('2026-04', '2026-05'): all_month = [r for r in rows if r['order_date'].startswith(month)] cs = [r for r in all_month if r['status'] == 'completed'] known = [r['delivery'] for r in cs if r['delivery'] is not None] total = sum(r['value'] for r in cs) monthly.append(dict(month=month, all_status_orders=len(all_month), completed_orders=len(cs), completed_sales_eur=total, aov_eur=total / len(cs) if cs else None, median_completed_value_eur=median(r['value'] for r in cs) if cs else None, known_delivery_completed=len(known), missing_delivery_completed=len(cs)-len(known), observed_mean_delivery_days=mean(known) if known else None)) for ch in ('web', 'store'): seg = [r for r in cs if r['channel'] == ch] channels.append(dict(month=month, channel=ch, completed_orders=len(seg), completed_sales_eur=sum(r['value'] for r in seg), aov_eur=mean(r['value'] for r in seg) if seg else None, order_share=len(seg)/len(cs) if cs else None)) bins = [(0, 100), (100, 150), (150, 200), (200, 300), (300, 1100)] counts = [sum(lo <= v < hi or (i == 4 and lo <= v <= hi) for v in values) for i, (lo, hi) in enumerate(bins)] # Values outside the published display bins must not silently disappear. if sum(counts) != len(values): raise ValueError('Histogram display bounds need review') return dict(monthly=monthly, channels=channels, distribution=dict(count=len(values), total_eur=sum(values), mean_eur=mean(values), median_eur=median(values), q1_inc_eur=q1, q3_inc_eur=q3, lower_fence_eur=low, upper_fence_eur=high, flagged_order_ids=[r['order_id'] for r in flagged], histogram_counts=counts), exclusion_scenario=dict(label='Excluding all IQR-flagged verified orders; baseline retained', count=len(unflagged), total_eur=sum(r['value'] for r in unflagged), mean_eur=mean(r['value'] for r in unflagged) if unflagged else None), equal_mix_scenario=[dict(month=m, aov_eur=sum(r['aov_eur']*.5 for r in channels if r['month']==m)) if all(r['aov_eur'] is not None for r in channels if r['month']==m) else dict(month=m, aov_eur=None) for m in ('2026-04','2026-05')]) def compare_expected(name, actual): with (BASE / name).open(newline='', encoding='utf-8') as handle: expected = list(csv.DictReader(handle)) # Compare by stable month/channel keys, not CSV row order. key = lambda r: (r['month'], r.get('channel', '')) expected = sorted(expected, key=key) actual = sorted(actual, key=key) if len(expected) != len(actual): raise ValueError('Expected result row count differs: ' + name) for exp, got in zip(expected, actual): if set(exp) != set(got): raise ValueError('Expected result columns differ: ' + name) for field, value in got.items(): if isinstance(value, (int, float)): if not math.isclose(float(exp[field]), value, rel_tol=1e-10, abs_tol=1e-10): raise ValueError('Expected numeric result differs: ' + name + '/' + field) elif exp[field] != value: raise ValueError('Expected result differs: ' + name + '/' + field) def main(): raw, reference = read_rows('orders_raw.csv'), read_rows('orders_verified.csv') clean = typed(prepare(raw, reference)) result = analyse(clean) compare_expected('monthly_results.csv', result['monthly']) compare_expected('channel_results.csv', result['channels']) result['profile'] = dict(raw_rows=len(raw), distinct_source_ids=len({r['order_id'] for r in raw}), verified_rows=len(clean), duplicate_groups={k:v for k,v in Counter(r['order_id'] for r in raw).items() if v>1}) print(json.dumps(result, indent=2, allow_nan=False)) if __name__ == '__main__': main()