How rare is your HDB flat?

"""Rebuilds this insight's numbers from HDB's two public files and writes data/flat_models.json.

    python prep/prep_flat_models.py                 download both files from data.gov.sg, then rebuild
    python prep/prep_flat_models.py --no-download   rebuild from the two files already in data/raw/ (no download)

Files (in data/raw/):
    ResaleFlatPrices2017.csv      Resale flat prices based on registration date from Jan-2017 onwards (data.gov.sg)
    HDBPropertyInformation.csv    HDB Property Information (data.gov.sg)
Also written: data/dataset_metadata.json (data.gov.sg's own record of each file: name, last updated, coverage end).
Window: every resale from Jan 2017 to Sep 2026. Oct 2026 is a month not yet finished, so it is left out.
A model is ranked only if it has at least THRESH resales; the others are shown but not ranked.
The page counts the same numbers again with plain Python (page.py); build_page.py stops if the two disagree.
"""
import json, math, re, sys, time, urllib.request
from pathlib import Path
import pandas as pd

HERE = Path(__file__).resolve().parent
DATA = HERE.parent / "data"
RAW = DATA / "raw"
DATASETS = {"resale": ("d_8b84c4ee58e3cfc0ece0d773c8ca6abc", "ResaleFlatPrices2017.csv"),
            "property": ("d_17f5382f26140b1fdae0ba2ef6239d2f", "HDBPropertyInformation.csv")}
THRESH = 50
LAST = "2026-09"
MODELS = ["Model A-Maisonette", "Adjoined flat", "Type S1", "Type S2", "Premium Apartment Loft", "Terrace",
          "Multi Generation", "3Gen", "Improved-Maisonette", "Premium Maisonette"]
ABBR = {"AVE": "Avenue", "ST": "Street", "RD": "Road", "DR": "Drive", "CRES": "Crescent", "CTRL": "Central", "STH": "South",
        "NTH": "North", "BT": "Bukit", "CL": "Close", "GDN": "Garden", "GDNS": "Gardens", "JLN": "Jalan", "LOR": "Lorong",
        "PL": "Place", "TER": "Terrace", "UPP": "Upper", "TG": "Tanjong", "KG": "Kampong", "C'WEALTH": "Commonwealth",
        "HTS": "Heights", "PK": "Park", "LK": "Link", "BLVD": "Boulevard", "MKT": "Market", "CTR": "Centre"}
SPECIAL = {"MCNAIR": "McNair", "MA'MOR": "Ma'mor", "QUEEN'S": "Queen's"}


def get(url):
    # data.gov.sg refuses requests that carry no User-Agent (HTTP 403), so send a plain one
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (CompareSG insight prep)"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return r.read()


def download():
    keep_all = {}
    RAW.mkdir(parents=True, exist_ok=True)
    for key, (ds, fname) in DATASETS.items():
        meta = json.loads(get("https://api-production.data.gov.sg/v2/public/api/datasets/" + ds + "/metadata"))["data"]
        keep_all[key] = {k: meta.get(k) for k in ("datasetId", "name", "managedBy", "lastUpdatedAt", "coverageStart", "coverageEnd", "format")}
        for _ in range(20):
            j = json.loads(get("https://api-open.data.gov.sg/v1/public/api/datasets/" + ds + "/poll-download"))["data"]
            if j.get("status") == "DOWNLOAD_SUCCESS":
                break
            time.sleep(3)
        (RAW / fname).write_bytes(get(j["url"]))
    (DATA / "dataset_metadata.json").write_text(json.dumps(keep_all, indent=1), encoding="utf-8")


if "--no-download" not in sys.argv:
    download()


def ru(x):
    return int(math.floor(x + 0.5))


def pretty(s):
    out = []
    for w in str(s).split():
        out.append(ABBR.get(w) or SPECIAL.get(w) or (w if any(c.isdigit() for c in w) else w.capitalize()))
    return " ".join(out)


def town(t):
    return {"KALLANG/WHAMPOA": "Kallang/Whampoa"}.get(t, t.title())


r_all = pd.read_csv(RAW / "ResaleFlatPrices2017.csv")
raw_rows = len(r_all)
r = r_all[r_all.month <= LAST].copy()
N = len(r)
months = (2026 - 2017) * 12 + 9
years = months / 12
r["town"] = r.town.map(town)
r["street"] = r.street_name.map(pretty)
r["blk"] = r.block.astype(str)
r["key"] = r.blk + "|" + r.street
r["flat_type"] = r.flat_type.str.title().str.replace(" Room", " room", regex=False)
r["year"] = r.month.str[:4].astype(int)
YRS = [str(y) for y in range(2017, 2027)]
all_per_year = [int((r.year == y).sum()) for y in range(2017, 2027)]
assert sum(all_per_year) == N
towns_all = sorted(r.town.unique())

p = pd.read_csv(RAW / "HDBPropertyInformation.csv")
sold_cols = [c for c in p.columns if c.endswith("_sold")]
p["sold"] = p[sold_cols].sum(axis=1)
stock_total = int(p.sold.sum())
p["street_p"] = p.street.map(pretty)
p["blk"] = p.blk_no.astype(str)
p["key"] = p.blk + "|" + p.street_p

avg_per_home = N / stock_total   # resales per home over the whole window, for the average flat

out = {"N": N, "months": months, "years": years, "stock_total": stock_total, "last_full": LAST, "thresh": THRESH, "yrs": YRS,
       "all_per_year": all_per_year, "median_area_all": float(r.floor_area_sqm.median()), "raw_rows": raw_rows,
       "oct_rows": int((r_all.month > LAST).sum()), "prop_rows": len(p), "n_towns_all": len(towns_all), "towns_all": towns_all,
       "avg_per_home": avg_per_home}

# ---- exact duplicates (same row twice) : counted, not removed, because HDB gives no unit number ----
dup = r.duplicated(subset=["month", "town", "flat_type", "blk", "street_name", "storey_range", "floor_area_sqm", "flat_model",
                           "lease_commence_date", "resale_price"], keep=False)
out["dups_all"] = int(dup.sum())

models = []
for name in MODELS:
    x = r[r.flat_model == name]
    n = len(x)
    g = x.groupby("key")
    blocks = g.ngroups
    blocks_once = int((g.size() == 1).sum())
    # ties are broken by name, so the order does not depend on the pandas version
    tl = x.groupby("town").agg(n=("key", "size"), b=("key", "nunique"))
    tl = tl.loc[sorted(tl.index, key=lambda t: (-tl.n[t], -tl.b[t], t))]
    sl = x.groupby("street").size()
    sl = sl.loc[sorted(sl.index, key=lambda t: (-sl[t], t))]
    py = [int((x.year == y).sum()) for y in range(2017, 2027)]
    first3, last3 = sum(py[:3]), sum(py[-3:])
    ramp = (last3 / (33 / 12)) / max(first3 / 3, 0.0001) > 3 or first3 == 0
    types = sorted(x.flat_type.unique(), key=lambda s: (s == "Executive", s))
    # same-type comparison: median area of this model vs every other resale of the same flat type
    cmp = []
    for t in types:
        a = x[x.flat_type == t].floor_area_sqm
        b = r[(r.flat_type == t) & (r.flat_model != name)].floor_area_sqm
        if len(a) >= 20 and len(b) >= 20:
            cmp.append([t, float(a.median()), float(b.median()), int(len(a))])
    m = dict(name=name, n=int(n), one_in=N / n, blocks=int(blocks), blocks_once=blocks_once, towns=int(len(tl)), streets=int(len(sl)),
             town_list=[[t, int(v.n), int(v.b)] for t, v in tl.iterrows()], street_list=[[s, int(v)] for s, v in sl.items()],
             first3=first3, last3=last3, ramp=bool(ramp), types=types, area=[float(x.floor_area_sqm.min()), float(x.floor_area_sqm.max())],
             area_med=float(x.floor_area_sqm.median()), lease=[int(x.lease_commence_date.min()), int(x.lease_commence_date.max())],
             per_year=py, first_year=int(x.year.min()), ranked=bool(n >= THRESH), cmp=cmp, stock=None)
    models.append(m)
byname = {m["name"]: m for m in models}

# ---- homes ("stock"), only where HDB's block list lets us count them ----
# Type S1 and S2: blocks 1A-1G Cantonment Road
pin = p[(p.street_p == "Cantonment Road") & p.blk.isin(list("1" + c for c in "ABCDEFG"))]
assert len(pin) == 7 and (pin.residential == "Y").all()
pin_keys = set(pin.key)
rp = r[r.key.isin(pin_keys)]
assert set(rp.flat_model) <= {"Type S1", "Type S2"}, set(rp.flat_model)
assert set(r[r.flat_model.isin(["Type S1", "Type S2"])].key) == pin_keys
s1_homes, s2_homes = int(pin["4room_sold"].sum()), int(pin["5room_sold"].sum())
assert int(pin[["1room_sold", "2room_sold", "3room_sold", "exec_sold", "multigen_sold", "studio_apartment_sold"]].sum().sum()) == 0
assert set(rp[rp.flat_model == "Type S1"].flat_type) == {"4 room"} and set(rp[rp.flat_model == "Type S2"].flat_type) == {"5 room"}
out["pin_blocks"] = 7
out["pin_floors"] = int(pin.max_floor_lvl.max())
assert (pin.max_floor_lvl == out["pin_floors"]).all()
out["pin_tallest_in_file"] = bool(out["pin_floors"] == p[p.residential == "Y"].max_floor_lvl.max())
# Terrace: every home in two-storey residential blocks on the four streets that have Terrace resales
tr = r[r.flat_model == "Terrace"]
tstreets = sorted(tr.street.unique())
tp = p[(p.street_p.isin(tstreets)) & (p.residential == "Y") & (p.max_floor_lvl == 2)]
assert set(tr.key) <= set(tp.key), "a Terrace resale is outside the two-storey blocks"
assert set(p[p.key.isin(set(tr.key))].max_floor_lvl) == {2}
t_homes = int(tp.sold.sum())
other_in_tp = set(r[r.key.isin(set(tp.key))].flat_model) - {"Terrace"}
out["terrace_stock_blocks"] = int(len(tp))
out["terrace_streets"] = tstreets
out["terrace_other_models_in_those_blocks"] = sorted(other_in_tp)
out["terrace_towns"] = sorted(tr.town.unique())
out["terrace_max_area"] = float(tr.floor_area_sqm.max())
# Multi Generation: units HDB lists as multigen
mgp = p[p.multigen_sold > 0]
mg = r[r.flat_model == "Multi Generation"]
assert set(mg.key) <= set(mgp.key), "a Multi Generation resale is outside the blocks that list multigen units"
mg_homes = int(mgp.multigen_sold.sum())
out["mg_blocks"] = int(len(mgp))
for name, homes in [("Type S1", s1_homes), ("Type S2", s2_homes), ("Terrace", t_homes), ("Multi Generation", mg_homes)]:
    m = byname[name]
    m["stock"] = homes
    m["stock_one_in"] = stock_total / homes
    m["per_home"] = m["n"] / homes
    m["x_avg"] = m["per_home"] / avg_per_home

# ---- extras used in copy ----
out["all_rare_n"] = int(sum(m["n"] for m in models))
out["n_models_all"] = int(r.flat_model.nunique())
out["flat_types"] = sorted(r.flat_type.unique())
out["blocks_2026"] = int(((p.year_completed == 2026) & (p.residential == "Y")).sum())
out["n_models_4room"] = int(r[r.flat_type == "4 room"].flat_model.nunique())
am = r.groupby("flat_model").agg(n=("key", "size"), t=("flat_type", lambda s: sorted(set(s))))
am = am.loc[sorted(am.index, key=lambda k: (-am.n[k], k))]
out["all_models"] = [[k, int(v.n), v.t] for k, v in am.iterrows()]
out["models"] = models
out["adj_once_pct"] = ru(100 * byname["Adjoined flat"]["blocks_once"] / byname["Adjoined flat"]["blocks"])
json.dump(out, open(DATA / "flat_models.json", "w"), indent=1, default=float)
print("N", N, "stock_total", stock_total, "dups", out["dups_all"])
for m in models:
    print(f"{m['name']:24} n={m['n']:5} 1in{m['one_in']:8.0f} blocks={m['blocks']:4} towns={m['towns']:2} streets={m['streets']:3} types={m['types']} area={m['area']} med={m['area_med']} lease={m['lease']} stock={m['stock']} cmp={m['cmp']}")
print("terrace stock blocks", out["terrace_stock_blocks"], t_homes, "other models in those blocks", other_in_tp)
print("pin floors", out["pin_floors"], "tallest in file", out["pin_tallest_in_file"], "S1/S2 homes", s1_homes, s2_homes, "mg", mg_homes, out["mg_blocks"])
print("models total", out["n_models_all"], "4room models", out["n_models_4room"])