"""Publication figures (vector PDF + PNG preview) from saved results only. Palette: validated default palette of the dataviz reference (categorical fixed order; sequential one-hue blue ramp for ordered levels). Thin marks, direct labels, no dual axes, recessive grid.""" from __future__ import annotations import json from pathlib import Path import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 import numpy as np # noqa: E402 import pandas as pd # noqa: E402 ROOT = Path(__file__).resolve().parents[2] P = ROOT / "data" / "processed" T = ROOT / "results" / "tables" F = ROOT / "results" / "figures" F.mkdir(parents=True, exist_ok=True) CAT = {"haiku": "#2a78d6", "sonnet": "#eb6834", "opus": "#1baf7a", "text": "#eda100", "visual": "#2a78d6", "gray": "#9a9892"} SEQ = {"B0": "#9ec5f4", "B1": "#6da7ec", "B2": "#3987e5", "B3": "#1c5cab", "B4": "#0d366b"} SCENE_COL = {"exterior_front": "#2a78d6", "exterior_rear_or_side": "#6da7ec", "exterior_aerial_or_street": "#9ec5f4", "living_or_dining": "#eb6834", "kitchen": "#1baf7a", "bedroom": "#eda100", "bathroom": "#e87ba4", "basement": "#008300", "yard_pool_or_deck": "#6250d6", "garage": "#e34948", "other": "#c3c2b7", "floor_plan_or_document": "#9a9892", "mechanical_or_utility": "#52514e", "unusable": "#0b0b0b"} plt.rcParams.update({"font.family": "serif", "font.size": 9, "axes.spines.top": False, "axes.spines.right": False, "axes.grid": True, "grid.color": "#e6e5e1", "grid.linewidth": 0.5, "axes.axisbelow": True, "legend.frameon": False, "figure.dpi": 150, "pdf.fonttype": 42}) LEVEL_LAB = {"B0": "B0 roll only", "B1": "B1 + text", "B2": "B2 + images", "B3": "B3 multimodal", "B4": "B4 + admin recon."} def save(fig, name): fig.tight_layout() fig.savefig(F / f"{name}.pdf"); fig.savefig(F / f"{name}.png", dpi=170) plt.close(fig) def tiers(): return [t for t in ["haiku", "sonnet", "opus"] if (P / f"images_{t}.parquet").exists()] # ---------------------------------------------------------------- F2 sample composition def fig_sample(): frame = pd.read_parquet(P / "frame.parquet"); smp = pd.read_parquet(P / "sample.parquet") fig, axes = plt.subplots(1, 3, figsize=(7.2, 2.4)) for ax, col, title in zip(axes, ["cma", "age_band", "price_tier"], ["Metropolitan area", "Building age (years)", "Asking-price quartile"]): f = frame[col].astype(str).value_counts(normalize=True); s = smp[col].astype(str).value_counts(normalize=True) cats = list(f.sort_values(ascending=False).index) if col == "cma" else sorted(f.index) x = np.arange(len(cats)); w = 0.38 ax.bar(x - w / 2, [f.get(c, 0) * 100 for c in cats], w, color=CAT["gray"], label="frame (n = %d)" % len(frame)) ax.bar(x + w / 2, [s.get(c, 0) * 100 for c in cats], w, color=CAT["haiku"], label="sample (n = %d)" % len(smp)) ax.set_xticks(x); ax.set_xticklabels(cats, rotation=30, ha="right"); ax.set_title(title, fontsize=9); ax.set_ylabel("% of listings") axes[0].legend(loc="upper right", fontsize=7) save(fig, "f02_sample_composition") # ---------------------------------------------------------------- F3 scene by position (MNAR evidence) def fig_scene_position(tier="haiku"): img = pd.read_parquet(P / f"images_{tier}.parquet"); img = img[img.ok] ct = pd.crosstab(img.image_index, img.scene, normalize="index") order = [c for c in SCENE_COL if c in ct.columns] + [c for c in ct.columns if c not in SCENE_COL] ct = ct[order] fig, ax = plt.subplots(figsize=(7.2, 3.0)) bottom = np.zeros(len(ct)) for c in ct.columns: ax.bar(ct.index + 1, ct[c] * 100, bottom=bottom, color=SCENE_COL.get(c, "#c3c2b7"), width=0.8, label=c.replace("_", " "), linewidth=0.5, edgecolor="white") bottom += ct[c].values * 100 ax.set_xlabel("Position of the photograph in the listing (seller order)"); ax.set_ylabel("% of photographs") ax.set_xticks(ct.index + 1); ax.set_ylim(0, 100) ax.legend(ncol=3, fontsize=6.5, loc="upper center", bbox_to_anchor=(0.5, -0.22)) save(fig, "f03_scene_by_position") # ---------------------------------------------------------------- F4 scene presence vs K def fig_scene_vs_k(): df = pd.read_csv(T / "exp3_image_count.csv") fig, ax = plt.subplots(figsize=(4.6, 2.8)) for tier in df.tier.unique(): d = df[df.tier == tier] for col, lab, ls in [("share_with_kitchen", "kitchen", "-"), ("share_with_bathroom", "bathroom", "--"), ("share_with_basement", "basement", ":")]: ax.plot(d.K, d[col] * 100, ls, color=CAT[tier], lw=1.6, marker="o", ms=3.5, label=f"{tier}: {lab}" if tier == df.tier.unique()[0] or True else None) ax.set_xlabel("Images analysed per listing (K, seller order)"); ax.set_ylabel("% of listings with ≥ 1 photo of the room") ax.set_ylim(0, 100); ax.legend(fontsize=6.5, ncol=2) save(fig, "f04_scene_presence_vs_k") # ---------------------------------------------------------------- F5 agreement dot plot def fig_agreement(): df = pd.read_csv(T / "exp1_agreement.csv") df = df[df.channel == "visual"] attrs = [a for a in df.attribute.unique() if "kitchen" not in a] fig, axes = plt.subplots(1, 2, figsize=(7.2, 3.2), sharey=True) y = np.arange(len(attrs)) for ax, metric, title in zip(axes, ["coverage", "accuracy"], ["Coverage (informative outputs)", "Agreement, informative cases"]): for tier in df.tier.unique(): d = df[df.tier == tier].set_index("attribute").reindex(attrs) ax.scatter(d[metric], y, color=CAT[tier], s=22, label=tier, zorder=3) ax.set_xlim(0, 1.02); ax.set_title(title, fontsize=8.5); ax.set_xlabel("share") axes[0].set_yticks(y); axes[0].set_yticklabels([a.replace(" vs ", "\nvs ").replace("∈", "in") for a in attrs], fontsize=6.5) axes[1].legend(fontsize=7, loc="lower left") save(fig, "f05_agreement") # ---------------------------------------------------------------- F6 confusion heatmaps def fig_confusion(): tier = json.load(open(ROOT / "results" / "raw" / "exp_agreement.json"))["exp1_confusion_tier"] fig, axes = plt.subplots(1, 2, figsize=(7.2, 3.0)) for ax, name, title in zip(axes, ["roof", "storeys"], ["Roof covering: broker field (rows) vs image (cols)", "Storeys: roll (rows) vs image (cols)"]): ct = pd.read_csv(T / f"exp1_confusion_{name}_{tier}.csv", index_col=0) ct = ct.loc[:, [c for c in ct.columns if c != "not_applicable"]] norm = ct.div(ct.sum(axis=1), axis=0) im = ax.imshow(norm.values, cmap="Blues", vmin=0, vmax=1, aspect="auto") ax.set_xticks(range(ct.shape[1])); ax.set_xticklabels([c.replace("_", "\n") for c in ct.columns], fontsize=6.5) ax.set_yticks(range(ct.shape[0])); ax.set_yticklabels(ct.index, fontsize=7) for i in range(ct.shape[0]): for j in range(ct.shape[1]): ax.text(j, i, int(ct.values[i, j]), ha="center", va="center", fontsize=6.5, color="white" if norm.values[i, j] > 0.6 else "#0b0b0b") ax.set_title(title, fontsize=8); ax.grid(False) save(fig, "f06_confusion") # ---------------------------------------------------------------- F7 cross-model agreement def fig_cross_model(): f = T / "exp4_cross_model.csv" if not f.exists(): return df = pd.read_csv(f) cat = df[df.alpha_nominal.notna()].copy() fig, axes = plt.subplots(1, 2, figsize=(7.2, 3.0), gridspec_kw={"width_ratios": [1.3, 1]}) ax = axes[0] y = np.arange(len(cat)) ax.barh(y, cat.alpha_nominal, color=CAT["haiku"], height=0.6) ax.set_yticks(y); ax.set_yticklabels([a.replace("_", " ") for a in cat.attribute], fontsize=7) ax.axvline(0.67, color="#52514e", lw=0.8, ls="--"); ax.axvline(0.8, color="#52514e", lw=0.8, ls=":") ax.set_xlabel("Krippendorff's α (nominal) across extractor tiers"); ax.set_xlim(0, 1) ax = axes[1] ts = tiers() if len(ts) >= 2: a = pd.read_parquet(P / f"props_{ts[0]}_K10.parquet").set_index("uid"); b = pd.read_parquet(P / f"props_{ts[1]}_K10.parquet").set_index("uid") c = a[["condition_composite"]].join(b[["condition_composite"]], lsuffix="_a", rsuffix="_b").dropna() ax.scatter(c.iloc[:, 0] + np.random.default_rng(1).normal(0, 0.03, len(c)), c.iloc[:, 1] + np.random.default_rng(2).normal(0, 0.03, len(c)), s=5, alpha=0.35, color=CAT["sonnet"], edgecolor="none") ax.plot([1, 5], [1, 5], color="#52514e", lw=0.8) ax.set_xlabel(f"composite condition, {ts[0]}"); ax.set_ylabel(f"composite condition, {ts[1]}") ax.set_title(f"r = {np.corrcoef(c.iloc[:, 0], c.iloc[:, 1])[0, 1]:.2f}, n = {len(c)}", fontsize=8) save(fig, "f07_cross_model") # ---------------------------------------------------------------- F8 calibration def fig_calibration(): df = pd.read_csv(T / "exp12_calibration.csv"); sel = pd.read_csv(T / "exp12_selective.csv") fig, axes = plt.subplots(1, 2, figsize=(7.2, 2.9)) ax = axes[0] for tier in df.tier.unique(): d = df[df.tier == tier] ax.plot(d.conf_mean, d.roof_acc, "-o", color=CAT[tier], ms=4, lw=1.4, label=f"{tier}: roof covering") ax.plot(d.conf_mean, d.clad_acc, "--s", color=CAT[tier], ms=3.5, lw=1.2, label=f"{tier}: cladding") ax.plot([0.5, 1], [0.5, 1], color="#52514e", lw=0.8) ax.set_xlabel("stated confidence (per image, bin mean)"); ax.set_ylabel("agreement with broker field"); ax.set_xlim(0.5, 1); ax.set_ylim(0.5, 1.02) ax.legend(fontsize=6.5) ax = axes[1] for tier in sel.tier.unique(): d = sel[sel.tier == tier] ax.plot(d.kept_share * 100, d.roof_accuracy, "-o", color=CAT[tier], ms=4, lw=1.4, label=tier) for _, r in d.iterrows(): ax.annotate(f"≥{r.threshold:.1f}", (r.kept_share * 100, r.roof_accuracy), fontsize=6, xytext=(2, 3), textcoords="offset points") ax.set_xlabel("% of exterior images retained (confidence threshold)"); ax.set_ylabel("roof agreement among retained") ax.legend(fontsize=7) save(fig, "f08_calibration") # ---------------------------------------------------------------- F9 decomposition by level and age band def fig_decomposition(): df = pd.read_csv(T / "exp5_levels_summary.csv"); main = "sonnet" if "sonnet" in df.tier.values else df.tier.iloc[0] d = df[(df.tier == main) & (df.anchor == "B")].set_index("level").reindex(["B0", "B1", "B2", "B3", "B4"]) dec = pd.read_csv(T / "exp5_decomposition.csv"); dec = dec[(dec.tier == main) & (dec.level == "B3") & (dec.group == "age_band")] fig, axes = plt.subplots(1, 2, figsize=(7.2, 3.0)) ax = axes[0] x = np.arange(len(d)) L = d["V_median"] * d["L_share_median"]; I = d["V_median"] - L ax.bar(x, L / 1000, color="#9ec5f4", label="land (roll value, indexed)") ax.bar(x, I / 1000, bottom=L / 1000, color=[SEQ[l] for l in d.index], label="depreciated improvements") for i, (l, v) in enumerate(zip(d.index, d["V_median"])): ax.text(i, v / 1000 + 8, f"${v/1000:,.0f}k", ha="center", fontsize=7) ax.set_xticks(x); ax.set_xticklabels([LEVEL_LAB[l] for l in d.index], fontsize=6.5, rotation=20, ha="right") ax.set_ylabel("median cost indication (k$)"); ax.legend(fontsize=6.5, loc="lower right") ax = axes[1] dec = dec.set_index("value").reindex(["0-10", "11-25", "26-50", "51-75", "76+"]) x = np.arange(len(dec)); w = 0.26 ax.bar(x - w, dec["L"] / 1000, w, color="#9ec5f4", label="land") ax.bar(x, dec["RCN"] / 1000, w, color="#1c5cab", label="RCN (incl. soft, incentive)") ax.bar(x + w, dec["improvements"] / 1000, w, color="#eb6834", label="depreciated improvements") ax.set_xticks(x); ax.set_xticklabels(dec.index); ax.set_xlabel("building age (years)"); ax.set_ylabel("median (k$), B3") ax.legend(fontsize=6.5) save(fig, "f09_decomposition") # ---------------------------------------------------------------- F10 ratio distributions by level def fig_ratios(): main = "sonnet" if (P / "cost_sonnet_B3.parquet").exists() else "haiku" fig, axes = plt.subplots(1, 2, figsize=(7.2, 2.9), sharey=False) for ax, col, title, xlim in zip(axes, ["ratio_price", "ratio_assessor"], ["Cost indication / asking price", "Depreciated improvements / assessor building value"], [(0, 2), (0, 3)]): data, labels = [], [] for lvl in ["B0", "B1", "B2", "B3", "B4"]: f = P / f"cost_{main}_{lvl}.parquet" if f.exists(): d = pd.read_parquet(f)[col].replace([np.inf, -np.inf], np.nan).dropna() data.append(d.clip(*xlim)); labels.append(LEVEL_LAB[lvl]) parts = ax.violinplot(data, showmedians=True, showextrema=False, widths=0.85) for b, lvl in zip(parts["bodies"], ["B0", "B1", "B2", "B3", "B4"][:len(data)]): b.set_facecolor(SEQ[lvl]); b.set_alpha(0.85); b.set_edgecolor("white") parts["cmedians"].set_color("#0b0b0b"); parts["cmedians"].set_linewidth(1) ax.axhline(1.0, color="#52514e", lw=0.8, ls="--") ax.set_xticks(range(1, len(labels) + 1)); ax.set_xticklabels(labels, fontsize=6.5, rotation=20, ha="right"); ax.set_ylim(*xlim) ax.set_title(title, fontsize=8) save(fig, "f10_ratio_distributions") # ---------------------------------------------------------------- F11 sensitivity tornado def fig_sensitivity(): df = pd.read_csv(T / "exp7_8_sensitivity.csv") base = df[df.scenario.str.startswith("baseline")].iloc[0] d = df[~df.scenario.str.startswith("baseline") & ~df.scenario.str.startswith("no depreciation")].copy() d["delta"] = (d["V_median"] / base["V_median"] - 1) * 100 d = d.reindex(d.delta.abs().sort_values().index) fig, ax = plt.subplots(figsize=(6.8, 4.2)) y = np.arange(len(d)) ax.barh(y, d.delta, color=np.where(d.delta >= 0, "#2a78d6", "#eb6834"), height=0.62) ax.set_yticks(y); ax.set_yticklabels(d.scenario, fontsize=7) ax.axvline(0, color="#0b0b0b", lw=0.8) for yi, v in zip(y, d.delta): inside = v < -15 ax.text(v + (0.6 if v >= 0 else (1.0 if inside else -0.6)), yi, f"{v:+.1f}%", va="center", ha="left" if (v >= 0 or inside) else "right", fontsize=6.5, color="white" if inside else "#0b0b0b") ax.set_xlabel("change in median cost indication vs baseline (B3, anchor B), %") ax.set_xlim(min(d.delta.min() * 1.05, -10), max(d.delta.max() * 1.6, 10)) save(fig, "f11_sensitivity") # ---------------------------------------------------------------- F12 effect of visual grade and condition def fig_grade_condition(): main = "sonnet" if (P / "exp6_effects_sonnet.parquet").exists() else "haiku" eff = pd.read_parquet(P / f"exp6_effects_{main}.parquet") fig, axes = plt.subplots(1, 2, figsize=(7.2, 2.9)) ax = axes[0] order = ["basic", "standard", "upgraded", "luxury"] data = [eff.loc[eff.grade == g, "delta_impr_pct"].clip(-80, 150) for g in order] bp = ax.boxplot(data, widths=0.55, showfliers=False, patch_artist=True, medianprops={"color": "#0b0b0b"}) for patch, c in zip(bp["boxes"], ["#9ec5f4", "#6da7ec", "#3987e5", "#1c5cab"]): patch.set_facecolor(c); patch.set_edgecolor("white") ax.set_xticks(range(1, 5)); ax.set_xticklabels([f"{g}\n(n={int((eff.grade == g).sum())})" for g in order], fontsize=7) ax.axhline(0, color="#52514e", lw=0.8, ls="--"); ax.set_ylabel("Δ depreciated improvements, B2 vs B0 (%)"); ax.set_title("by visual finish grade", fontsize=8) ax = axes[1] ax.scatter(eff.condition + np.random.default_rng(3).normal(0, 0.03, len(eff)), eff.delta_D_pts, s=6, alpha=0.4, color="#eb6834", edgecolor="none") ax.set_xlabel("composite visual condition (1–5)"); ax.set_ylabel("Δ depreciation, B2 vs B0 (percentage points)") ax.axhline(0, color="#52514e", lw=0.8, ls="--") save(fig, "f12_grade_condition_effects") # ---------------------------------------------------------------- F13 Monte Carlo intervals def fig_mc(): main = "sonnet" if (P / "exp14_mc_sonnet.parquet").exists() else "haiku" mc = pd.read_parquet(P / f"exp14_mc_{main}.parquet") fig, axes = plt.subplots(1, 2, figsize=(7.2, 3.0)) ax = axes[0] ax.hist(mc.rel_width.clip(0, 1) * 100, bins=40, color="#2a78d6", edgecolor="white", linewidth=0.4) ax.axvline(mc.rel_width.median() * 100, color="#0b0b0b", lw=1); ax.set_xlabel("relative width of the 10–90 % interval (% of median)"); ax.set_ylabel("listings") ax = axes[1] rng = np.random.default_rng(4); idx = rng.choice(mc.index, size=min(250, len(mc)), replace=False) s = mc.loc[idx].sort_values("V_mc_median") x = np.arange(len(s)) ax.fill_between(x, s.V_p10 / 1e3, s.V_p90 / 1e3, color="#9ec5f4", alpha=0.8, label="10–90 % interval") ax.plot(x, s.V_mc_median / 1e3, color="#1c5cab", lw=1, label="median indication") ax.scatter(x, s.price / 1e3, s=5, color="#eb6834", label="asking price", zorder=3) ax.set_yscale("log"); ax.set_xlabel("listings (random subset of 250, sorted by indication)"); ax.set_ylabel("k$ (log scale)") ax.legend(fontsize=6.5, loc="upper left") save(fig, "f13_monte_carlo") # ---------------------------------------------------------------- F14 subgroup heatmap def fig_subgroups(): main = "sonnet" if (P / "cost_sonnet_B3.parquet").exists() else "haiku" d = pd.read_parquet(P / f"cost_{main}_B3.parquet") d["ape"] = (d.ratio_price - 1).abs() * 100 piv = d.pivot_table(index="age_band", columns="price_tier", values="ape", aggfunc="median", observed=True).reindex(["0-10", "11-25", "26-50", "51-75", "76+"]) cnt = d.pivot_table(index="age_band", columns="price_tier", values="ape", aggfunc="size", observed=True).reindex(piv.index) fig, ax = plt.subplots(figsize=(4.6, 2.9)) im = ax.imshow(piv.values, cmap="Blues", aspect="auto") ax.set_xticks(range(piv.shape[1])); ax.set_xticklabels(piv.columns); ax.set_yticks(range(piv.shape[0])); ax.set_yticklabels(piv.index) for i in range(piv.shape[0]): for j in range(piv.shape[1]): ax.text(j, i, f"{piv.values[i, j]:.0f}%\n(n={int(cnt.values[i, j])})", ha="center", va="center", fontsize=6.5, color="white" if piv.values[i, j] > np.nanpercentile(piv.values, 60) else "#0b0b0b") ax.set_xlabel("asking-price quartile"); ax.set_ylabel("building age (years)"); ax.set_title("median |cost indication / price − 1|, B3", fontsize=8); ax.grid(False) save(fig, "f14_subgroups") # ---------------------------------------------------------------- F15 recorded sales def fig_sales(): main = "sonnet" sales = pd.read_parquet(P / "frame_sales_after_listing.parquet").sort_values("date").drop_duplicates("uid", keep="last").set_index("uid") pts = [] for lvfile in [P / f"levels_{main}_sold.parquet", P / f"levels_{main}.parquet"]: if not lvfile.exists(): continue import sys sys.path.insert(0, str(ROOT)) from src.cost_model import cost L = pd.read_parquet(lvfile); d = L[L.level == "B3"].set_index("uid"); d = d[d.index.isin(sales.index)] if len(d): out = cost.cost_indication(d); pts.append(pd.DataFrame({"V": out.V_cost, "sale": sales.loc[d.index, "amount"], "ask": d.price})) if not pts: return s = pd.concat(pts) fig, ax = plt.subplots(figsize=(3.8, 3.4)) ax.scatter(s.sale / 1e3, s.V / 1e3, s=10, color="#2a78d6", alpha=0.7, edgecolor="none", label="cost indication (B3)") ax.scatter(s.sale / 1e3, s.ask / 1e3, s=10, color="#eb6834", alpha=0.6, marker="x", label="asking price") lim = [s.sale.min() / 1e3 * 0.8, s.sale.max() / 1e3 * 1.1] ax.plot(lim, lim, color="#52514e", lw=0.8); ax.set_xscale("log"); ax.set_yscale("log") ax.set_xlabel("recorded sale price after listing (k$, log)"); ax.set_ylabel("k$ (log)"); ax.legend(fontsize=6.5) ax.set_title(f"n = {len(s)} units sold after listing", fontsize=8) save(fig, "f15_recorded_sales") # ---------------------------------------------------------------- F16 image count sensitivity def fig_k_sensitivity(): df = pd.read_csv(T / "exp3_image_count.csv") fig, ax = plt.subplots(figsize=(4.6, 2.8)) for tier in df.tier.unique(): d = df[df.tier == tier] ax.plot(d.K, d.grade_available * 100, "-o", color=CAT[tier], ms=4, lw=1.5, label=f"{tier}: finish grade available") ax.plot(d.K, d.grade_stable_vs_full * 100, "--s", color=CAT[tier], ms=3.5, lw=1.2, label=f"{tier}: grade equal to full-K grade") ax.set_xlabel("images analysed per listing (K)"); ax.set_ylabel("% of listings"); ax.set_ylim(0, 102); ax.legend(fontsize=6.5) save(fig, "f16_image_count") def all_figures(): for fn in [fig_sample, fig_scene_position, fig_scene_vs_k, fig_agreement, fig_confusion, fig_cross_model, fig_calibration, fig_decomposition, fig_ratios, fig_sensitivity, fig_grade_condition, fig_mc, fig_subgroups, fig_sales, fig_k_sensitivity]: try: fn() print("ok ", fn.__name__) except Exception as e: # noqa: BLE001 print("FAIL", fn.__name__, repr(e)) if __name__ == "__main__": all_figures()