# Task 09 — Cleanse & Normalize → Canonical Dataset

## Goal
Streamed transformation of all raw CSV into the canonical dataset; anomalies quarantined, lineage recorded.

## Input
- Mappings (Task 07), schema + instrument ranges (Task 08), `config/anomaly_rules.json`, tokens (Task 03).

## Actions
1. Write `scripts/cleanse.py` — streaming, constant memory, parallel by file, resumable (checkpoint by inventory offset).
2. Per file: apply column mapping → coerce dtypes → normalize units → standardize codes (atmosphere, instrument) → resolve hierarchy keys from tokens (natural keys: batch/coupon/sample/run/test/cycle numbers).
3. Anomaly checks per row: out-of-instrument-range, non-coercible type, missing required, duplicate key, sequence gap (per ordered series), malformed row. Reasons coded per `anomaly_rules.json`.
4. Clean rows → `canonical/<measurement_type>/…` partitioned files (JSONL or Parquet; partition friction points by test/cycle per Phase-1 volume estimates). Rejected rows → `rejects/<type>.jsonl` with `{source_file, row_offset, reason_code, raw_payload}`.
5. Lineage per output row: `{source_file_id, row_offset, mapping_version, schema_version}` → compact companion files in `lineage/`.

## Intermediate Result (done when)
- Every in-scope file processed; row counts reconcile: `source ≈ canonical + rejects` per file.
- Canonical partitions written; lineage present for every canonical row.