1"""F1-conf-04: reconcile property-sales export to control totals. Reads only data/."""2import csv, json, re, datetime, statistics, os34OUT = "out"5os.makedirs(OUT, exist_ok=True)6meta = json.load(open("data/metadata.json"))7with open("data/sales.csv", newline="", encoding="utf-8") as f:8 raw_lines = f.read().splitlines()9n_lines = len(raw_lines) - 1 # data lines excluding header10rows = list(csv.reader(raw_lines))11header, data = rows[0], rows[1:]12assert header == ["sale_id", "municipality", "sale_date", "price_cad", "area_m2"]13assert len(data) == n_lines1415ledger = []16def step(name, rule, before, after, why):17 ledger.append(dict(step=name, rule=rule, n_before=before, n_after=after,18 n_removed=before - after, rationale=why))1920# Step 0: lines read21step("0_read", "all data lines in sales.csv (excluding header)", n_lines, n_lines, "starting count")2223# Step 1: structural parse (5 fields, valid id)24s1 = [r for r in data if len(r) == 5 and re.fullmatch(r"S\d{5}", r[0])]25step("1_structure", "exactly 5 CSV fields and sale_id matches S+5 digits", n_lines, len(s1),26 "corrupted/garbled lines whose fields cannot be read unambiguously; not reconstructed")2728# Step 2: field validity (date, price, area, municipality)29n_quoted_thousands = 030s2 = []31for r in s1:32 try:33 d = datetime.date.fromisoformat(r[2])34 p = r[3]35 if re.fullmatch(r"\d{1,3}(,\d{3})+", p):36 n_quoted_thousands += 137 p = p.replace(",", "")38 if not re.fullmatch(r"\d+", p):39 continue40 a = float(r[4])41 if not (a > 0) or not r[1].strip():42 continue43 s2.append(dict(sale_id=r[0], municipality=r[1], sale_date=d.isoformat(),44 price_cad=int(p), area_m2=a, _area_str=r[4]))45 except ValueError:46 continue47step("2_fields", "valid ISO date, integer price (CSV-quoted thousands separators allowed), area>0, non-empty municipality",48 len(s1), len(s2), "fields must be readable unambiguously")4950# Step 3: duplicates51seen, s3 = set(), []52for r in s2:53 if r["sale_id"] not in seen:54 seen.add(r["sale_id"]); s3.append(r)55step("3_dedup", "unique sale_id", len(s2), len(s3), "sale_id must be unique")5657valid = s358n_valid = len(valid)59sum_valid = sum(r["price_cad"] for r in valid)60mean_valid = sum_valid / n_valid61med_ppm2 = statistics.median(r["price_cad"] / r["area_m2"] for r in valid)62n_ctrl = int(meta["n_records"]); sum_ctrl = int(meta["sum_price_cad"])63n_unread = n_ctrl - n_valid64n_struct_bad = n_lines - len(s1)6566by_mun = {}67for r in valid:68 by_mun[r["municipality"]] = by_mun.get(r["municipality"], 0) + 169mun_gap = {m: meta["n_by_municipality"][m] - by_mun.get(m, 0) for m in meta["n_by_municipality"]}70unknown_mun = sorted(set(by_mun) - set(meta["n_by_municipality"]))7172issues = []73if n_struct_bad:74 issues.append({"issue": f"{n_struct_bad} data lines of sales.csv are corrupted (garbled characters, wrong field counts, sale_id not S+5 digits); file lines {[i+2 for i,r in enumerate(data) if not (len(r)==5 and re.fullmatch(r'S[0-9]{5}', r[0]))][0]}-{[i+2 for i,r in enumerate(data) if not (len(r)==5 and re.fullmatch(r'S[0-9]{5}', r[0]))][-1]}",75 "handling": "excluded from the analytic dataset; not reconstructed or imputed; counted as unreadable records"})76if n_quoted_thousands:77 issues.append({"issue": f"{n_quoted_thousands} records have price_cad exported as a quoted string with a comma thousands separator (e.g. \"280,000\")",78 "handling": "CSV quoting makes the value unambiguous; parsed as integer after removing the separator; records retained"})79if len(s1) - len(s2):80 issues.append({"issue": f"{len(s1)-len(s2)} structurally valid records have unreadable field values", "handling": "excluded"})81if len(s2) - len(s3):82 issues.append({"issue": f"{len(s2)-len(s3)} duplicate sale_id records", "handling": "kept first occurrence"})83issues.append({"issue": f"record count reconciliation: control {n_ctrl} vs {n_valid} valid; per-municipality deficits {mun_gap}",84 "handling": "reported; per-municipality deficits sum to %d; unreadable records' municipalities not inferred" % sum(mun_gap.values())})85issues.append({"issue": f"price-sum reconciliation: control {sum_ctrl} vs valid {sum_valid}; implied unreadable sum {sum_ctrl-sum_valid}",86 "handling": "reported as sum_price_unreadable_implied; not allocated to individual records"})87if unknown_mun:88 issues.append({"issue": f"municipalities not in control totals: {unknown_mun}", "handling": "retained"})8990feas = "complete" if n_unread == 0 else "partial"91reason = None if feas == "complete" else (92 f"{n_unread} of the {n_ctrl} records in the provider's control count could not be analysed: "93 f"{n_struct_bad} data lines of sales.csv are corrupted/garbled and their fields cannot be read unambiguously "94 f"(n_input_rows {n_lines} = control count; {n_valid} records were readable). Their implied price sum is "95 f"{sum_ctrl - sum_valid} CAD; they were not reconstructed.")9697results = {98 "task_id": "F1-conf-04",99 "n_records_control": n_ctrl,100 "n_valid_records": n_valid,101 "n_unreadable_records": n_unread,102 "n_analysed": n_valid,103 "sum_price_control": sum_ctrl,104 "sum_price_valid": sum_valid,105 "sum_price_unreadable_implied": sum_ctrl - sum_valid,106 "mean_price_valid": round(mean_valid, 4),107 "median_price_per_m2_valid": round(med_ppm2, 4),108 "feasibility": feas,109 "feasibility_reason": reason,110 "data_issues": issues,111 "n_input_rows": {"sales.csv": n_lines},112 "random_seed": None,113 "valid_by_municipality": dict(sorted(by_mun.items())),114 "control_minus_valid_by_municipality": mun_gap,115}116json.dump(results, open(f"{OUT}/results.json", "w"), indent=2)117118with open(f"{OUT}/analytic.csv", "w", newline="") as f:119 w = csv.writer(f)120 w.writerow(["sale_id", "municipality", "sale_date", "price_cad", "area_m2"])121 for r in valid:122 w.writerow([r["sale_id"], r["municipality"], r["sale_date"], r["price_cad"], r["_area_str"]])123124step("4_final", "analytic dataset", n_valid, n_valid, "n_analysed")125with open(f"{OUT}/exclusions.csv", "w", newline="") as f:126 w = csv.DictWriter(f, fieldnames=["step", "rule", "n_before", "n_after", "n_removed", "rationale"])127 w.writeheader(); w.writerows(ledger)128129print(f"n_valid={n_valid} n_unreadable={n_unread} sum_valid={sum_valid} mean={mean_valid:.2f} median_ppm2={med_ppm2:.2f} feasibility={feas}")130