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, hashlib5from datetime import date6from pathlib import Path7import numpy as np8import pandas as pd910DATA, OUT = Path("data"), Path("out")11OUT.mkdir(exist_ok=True)12meta = json.loads((DATA / "metadata.json").read_text())13MUNIS = set(meta["n_by_municipality"])1415with open(DATA / "sales.csv", newline="", encoding="utf-8") as f:16 rows = list(csv.reader(f))17header, lines = rows[0], rows[1:]18assert header == ["sale_id", "municipality", "sale_date", "price_cad", "area_m2"], header19n_lines = len(lines)2021def to_num(s, thousands=False):22 s = s.strip()23 pat = r"\d{1,3}(,\d{3})+|\d+" if thousands else r"\d+(\.\d+)?"24 if not re.fullmatch(pat, s):25 return None26 return float(s.replace(",", ""))2728def iso(s):29 if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", s):30 return None31 try:32 return date.fromisoformat(s)33 except ValueError:34 return None3536ledger, cur = [], lines37def step(name, rule, keep, rationale):38 global cur39 before = len(cur)40 cur = [r for r in cur if keep(r)]41 ledger.append(dict(step=name, rule=rule, n_before=before, n_after=len(cur),42 n_removed=before - len(cur), rationale=rationale))4344step("1_field_count", "exactly 5 CSV fields (RFC-4180 quoting respected)", lambda r: len(r) == 5,45 "line cannot be split into the 5 declared columns; content unreadable, not reconstructed")46step("2_sale_id", "sale_id matches ^S\\d{5}$", lambda r: re.fullmatch(r"S\d{5}", r[0]) is not None,47 "identifier not in documented format; record cannot be identified unambiguously")48step("3_municipality", "municipality is one of the 5 control-total municipalities", lambda r: r[1] in MUNIS,49 "unknown municipality string; not guessed")50step("4_sale_date", "sale_date is a valid ISO YYYY-MM-DD date", lambda r: iso(r[2]) is not None,51 "date unreadable")52step("5_price", "price_cad is a positive integer (plain digits, or a quoted value with standard 3-digit thousands separators)",53 lambda r: (to_num(r[3], True) or 0) > 0, "price unreadable or non-positive")54step("6_area", "area_m2 is a positive decimal number", lambda r: (to_num(r[4]) or 0) > 0,55 "area unreadable or non-positive")56ids = [r[0] for r in cur]57dup = {i for i in ids if ids.count(i) > 1}58step("7_unique_id", "sale_id not duplicated among remaining rows", lambda r: r[0] not in dup,59 "duplicate identifiers would be ambiguous")6061df = pd.DataFrame(cur, columns=header)62n_thousands = int(df["price_cad"].str.contains(",").sum())63df["price_cad"] = df["price_cad"].map(lambda s: int(to_num(s, True)))64df["area_m2"] = df["area_m2"].astype(float)65df = df.sort_values("sale_id").reset_index(drop=True)66df.to_csv(OUT / "analytic.csv", index=False)67pd.DataFrame(ledger).to_csv(OUT / "exclusions.csv", index=False)6869n_valid = len(df)70n_ctrl = int(meta["n_records"])71sum_ctrl = int(meta["sum_price_cad"])72sum_valid = int(df["price_cad"].sum())73n_unread = n_ctrl - n_valid74missing_ids = sorted(set(f"S{i:05d}" for i in range(1, n_ctrl + 1)) - set(df["sale_id"]))75by_muni = df["municipality"].value_counts().to_dict()76muni_gap = {m: int(meta["n_by_municipality"][m] - by_muni.get(m, 0)) for m in meta["n_by_municipality"]}77removed = {e["step"]: e["n_removed"] for e in ledger}7879issues = []80if removed["1_field_count"] or removed["2_sale_id"]:81 issues.append({"issue": f"{removed['1_field_count'] + removed['2_sale_id']} data lines are garbled (random characters, wrong number of fields, malformed sale_id); "82 f"sale_ids {missing_ids[0]}..{missing_ids[-1]} ({len(missing_ids)} ids) are absent from the readable rows" if missing_ids else "garbled lines",83 "handling": "excluded from the analytic dataset (not reconstructed or imputed); counted as unreadable; their implied price total is derived only as control sum minus valid sum"})84if n_thousands:85 issues.append({"issue": f"{n_thousands} rows have price_cad written as a quoted string with thousands separators (e.g. \"280,000\")",86 "handling": "the CSV quoting makes the value unambiguous; separators removed and value parsed as integer CAD; rows retained"})87for k in ["3_municipality", "4_sale_date", "5_price", "6_area", "7_unique_id"]:88 if removed[k]:89 issues.append({"issue": f"{removed[k]} rows failed rule {k}", "handling": "excluded"})90if n_unread:91 issues.append({"issue": f"record count ({n_valid}) and price sum ({sum_valid}) of readable rows fall short of control totals ({n_ctrl}, {sum_ctrl}); per-municipality shortfall {muni_gap}",92 "handling": "reported as reconciliation gap; feasibility set to partial; summary statistics computed on readable records only"})9394feas = "complete" if n_unread == 0 else "partial"95reason = None if feas == "complete" else (96 f"{n_unread} of the {n_ctrl} control records could not be analysed: {n_lines} data lines were read, of which "97 f"{n_lines - n_valid} are corrupted/garbled export lines (wrong field count or malformed fields; ids {missing_ids[0]}-{missing_ids[-1]}) "98 f"whose content cannot be read unambiguously; they were not reconstructed. Their implied price total is {sum_ctrl - sum_valid} CAD.")99100res = {101 "task_id": "F1-conf-04",102 "n_records_control": n_ctrl,103 "n_valid_records": n_valid,104 "n_unreadable_records": n_unread,105 "n_analysed": n_valid,106 "sum_price_control": sum_ctrl,107 "sum_price_valid": sum_valid,108 "sum_price_unreadable_implied": sum_ctrl - sum_valid,109 "mean_price_valid": round(float(df["price_cad"].mean()), 4),110 "median_price_per_m2_valid": round(float(np.median(df["price_cad"] / df["area_m2"])), 4),111 "feasibility": feas,112 "feasibility_reason": reason,113 "data_issues": issues,114 "n_input_rows": {"sales.csv": n_lines},115 "random_seed": None,116 # supplementary keys117 "n_by_municipality_valid": {m: int(by_muni.get(m, 0)) for m in meta["n_by_municipality"]},118 "n_by_municipality_shortfall": muni_gap,119 "n_price_thousands_separator_parsed": n_thousands,120 "missing_sale_id_range": [missing_ids[0], missing_ids[-1]] if missing_ids else None,121 "sale_date_range_valid": [df["sale_date"].min(), df["sale_date"].max()],122 "mean_price_unreadable_implied": round((sum_ctrl - sum_valid) / n_unread, 4) if n_unread else None,123 "input_sha256": {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(DATA.iterdir())},124}125(OUT / "results.json").write_text(json.dumps(res, indent=2, ensure_ascii=False) + "\n")126print(json.dumps({k: res[k] for k in ["n_valid_records", "n_unreadable_records", "sum_price_valid",127 "mean_price_valid", "median_price_per_m2_valid", "feasibility"]}))128