1"""F1-conf-04: reconcile property-sales export to provider control totals.2Run from workspace root: ~/wp29/venv/bin/python out/analysis.py3Reads only data/; writes out/results.json, out/analytic.csv, out/exclusions.csv. No randomness."""4import csv, json, re, os5from datetime import date6import numpy as np7import pandas as pd89DATA, OUT = "data", "out"10os.makedirs(OUT, exist_ok=True)11meta = json.load(open(os.path.join(DATA, "metadata.json")))1213with open(os.path.join(DATA, "sales.csv"), newline="", encoding="utf-8") as f:14 raw = f.read()15lines = raw.split("\n")16if lines and lines[-1] == "":17 lines = lines[:-1]18header, body = lines[0], lines[1:]19n_lines = len(body)20cols = ["sale_id", "municipality", "sale_date", "price_cad", "area_m2"]21assert next(csv.reader([header])) == cols2223ID_RE = re.compile(r"^S\d{5}$")24MUNI_RE = re.compile(r"^[A-Za-z][A-Za-z\- ]*$")25INT_RE = re.compile(r"^\d+$")26THOUS_RE = re.compile(r"^\d{1,3}(,\d{3})+$")27FLOAT_RE = re.compile(r"^\d+(\.\d+)?$")2829def parse_date(s):30 try:31 if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", s):32 return None33 return date.fromisoformat(s)34 except ValueError:35 return None3637# Step 2: structural parse (exactly 5 CSV fields, RFC-4180 quoting)38step1 = []39for ln, line in enumerate(body, start=2):40 fields = next(csv.reader([line]))41 if len(fields) == 5:42 step1.append((ln, fields))43# Step 3: field validity (each field unambiguous)44step2, n_thousands = [], 045for ln, (sid, muni, sd, pr, ar) in step1:46 if not (ID_RE.match(sid) and MUNI_RE.match(muni) and parse_date(sd) and FLOAT_RE.match(ar)):47 continue48 if INT_RE.match(pr):49 price = int(pr)50 elif THOUS_RE.match(pr):51 price = int(pr.replace(",", "")); n_thousands += 152 else:53 continue54 area = float(ar)55 if price <= 0 or area <= 0:56 continue57 step2.append(dict(sale_id=sid, municipality=muni, sale_date=sd, price_cad=price, area_m2=area, _line=ln))58# Step 4: duplicate sale_id (would be ambiguous) -> drop all copies59df = pd.DataFrame(step2)60dup_mask = df.duplicated("sale_id", keep=False)61step3 = df[~dup_mask].copy()6263ledger = [64 dict(step=1, rule="data lines read (excluding header)", n_before=n_lines, n_after=n_lines, n_removed=0,65 rationale="all physical data lines in data/sales.csv"),66 dict(step=2, rule="line parses (RFC-4180 CSV) into exactly 5 fields", n_before=n_lines, n_after=len(step1),67 n_removed=n_lines - len(step1), rationale="corrupted/garbled export lines cannot be read; not reconstructed"),68 dict(step=3, rule="every field valid: sale_id S+5 digits, municipality alphabetic, ISO date, price integer (optionally comma thousands-separated), area numeric, both > 0",69 n_before=len(step1), n_after=len(step2), n_removed=len(step1) - len(step2),70 rationale="fields must be readable unambiguously; not reconstructed"),71 dict(step=4, rule="sale_id unique (all copies of duplicated ids dropped)", n_before=len(step2), n_after=len(step3),72 n_removed=int(dup_mask.sum()), rationale="duplicated ids would make records ambiguous"),73]74pd.DataFrame(ledger).to_csv(os.path.join(OUT, "exclusions.csv"), index=False)7576analytic = step3[cols].sort_values("sale_id").reset_index(drop=True)77analytic.to_csv(os.path.join(OUT, "analytic.csv"), index=False)7879n_valid = int(len(analytic))80n_ctrl = int(meta["n_records"])81sum_ctrl = int(meta["sum_price_cad"])82sum_valid = int(analytic["price_cad"].sum())83n_unread = n_ctrl - n_valid84excluded_lines = sorted(set(range(2, n_lines + 2)) - set(step3["_line"]))85present = set(int(s[1:]) for s in analytic["sale_id"])86missing_ids = sorted(set(range(1, n_ctrl + 1)) - present)87by_muni = analytic["municipality"].value_counts().to_dict()88muni_gap = {k: int(v - by_muni.get(k, 0)) for k, v in meta["n_by_municipality"].items()}8990def rng(xs):91 return f"{xs[0]}-{xs[-1]}" if xs and xs[-1] - xs[0] + 1 == len(xs) else str(xs)9293n_bad_lines = n_lines - len(step3)94issues = []95if n_bad_lines:96 issues.append(dict(97 issue=(f"{n_bad_lines} data lines are corrupted/garbled (file lines {rng(excluded_lines)}): no valid sale_id or field structure. "98 f"Sale ids S{missing_ids[0]:05d}-S{missing_ids[-1]:05d} ({len(missing_ids)} ids) are absent from the readable records.")99 if missing_ids else f"{n_bad_lines} data lines could not be read",100 handling="Excluded from the analytic dataset; not reconstructed or imputed. Counted as unreadable records; their price sum is only implied by the control total."))101if n_thousands:102 issues.append(dict(issue=f"{n_thousands} records have price_cad exported as a quoted string with comma thousands separator (e.g. \"280,000\").",103 handling="Kept: under standard CSV quoting the field is unambiguous; separator removed and value converted to integer CAD."))104if int(dup_mask.sum()):105 issues.append(dict(issue=f"{int(dup_mask.sum())} rows share duplicated sale_id", handling="All copies excluded."))106if n_unread:107 issues.append(dict(issue=f"Readable record counts by municipality fall short of the control n_by_municipality (control minus readable): {muni_gap}.",108 handling="Reported only; the shortfall sums to the number of unreadable records. Municipality of unreadable records is implied by control totals, not observed."))109110feas = "complete" if n_unread == 0 else "partial"111reason = None if feas == "complete" else (112 f"{n_unread} of the {n_ctrl} records in the provider's control count could not be analysed: "113 f"{n_bad_lines} export lines (file lines {rng(excluded_lines)}) are corrupted and cannot be read unambiguously "114 f"(sale ids S{missing_ids[0]:05d}-S{missing_ids[-1]:05d} missing); they were not reconstructed. "115 f"Their implied price sum is {sum_ctrl - sum_valid} CAD.")116117results = dict(118 task_id="F1-conf-04",119 n_records_control=n_ctrl,120 n_valid_records=n_valid,121 n_unreadable_records=n_unread,122 n_analysed=n_valid,123 sum_price_control=sum_ctrl,124 sum_price_valid=sum_valid,125 sum_price_unreadable_implied=sum_ctrl - sum_valid,126 mean_price_valid=round(float(analytic["price_cad"].mean()), 6),127 median_price_per_m2_valid=round(float(np.median(analytic["price_cad"] / analytic["area_m2"])), 6),128 feasibility=feas,129 feasibility_reason=reason,130 data_issues=issues,131 n_input_rows={"sales.csv": n_lines},132 random_seed=None,133 # supplementary keys (not required by TASK.md)134 implied_mean_price_unreadable=round((sum_ctrl - sum_valid) / n_unread, 6) if n_unread else None,135 n_price_thousands_separator=n_thousands,136 n_valid_if_thousands_sep_excluded=n_valid - n_thousands,137 control_mean_price_all_records=round(sum_ctrl / n_ctrl, 6),138 n_by_municipality_valid={k: int(v) for k, v in sorted(by_muni.items())},139 n_by_municipality_gap=muni_gap,140)141with open(os.path.join(OUT, "results.json"), "w") as f:142 json.dump(results, f, indent=2)143print(json.dumps({k: results[k] for k in ["n_valid_records", "n_unreadable_records", "sum_price_valid",144 "sum_price_unreadable_implied", "mean_price_valid",145 "median_price_per_m2_valid", "feasibility"]}, indent=1))146