TeX 59%
Python 40.8%
1"""Transparent cost-approach engine for WP20.23 V_cost = L + (RCN_hard × (1 + soft) × (1 + EI) + S) × (1 − D)45All parameters come from configs/cost_schedule.yaml; every function is pure and vectorised over a DataFrame of6property-level inputs so that sensitivity analyses and Monte Carlo draws can override any parameter.78Inputs expected per property (property-level table built by src/extraction/aggregate.py):9 floor_area_m2 (ADMIN), storeys (str '1'…'3'), attached ('detached'|'semi_detached'|'row_or_townhouse'),10 grade ('basic'…'luxury'), condition (1–5 float), renovation ('none_apparent'|'partial'|'extensive'|'unsure'),11 basement ('finished'|'partially_finished'|'unfinished'), garage ('attached'|'detached'|'carport'|'none'),12 pool ('inground'|'above_ground'|'none'), age (years at 2026), cma, roll_land_value, roll_market_date, roll_building_value.13"""14from __future__ import annotations1516import json17from pathlib import Path1819import numpy as np20import pandas as pd21import yaml2223ROOT = Path(__file__).resolve().parents[2]24CFG = yaml.safe_load(open(ROOT / "configs" / "cost_schedule.yaml"))25SQFT_PER_M2 = 10.7639262728# ----------------------------------------------------------------------------- indices29def load_bcpi() -> pd.DataFrame:30 """StatCan 18-10-0289 single-detached composite series by CMA key → DataFrame[refPer, series, value]."""31 raw = json.load(open(ROOT / "data" / "raw" / "statcan" / "bcpi_18100289_raw.json"))32 inv = {v: k for k, v in CFG["temporal_regional"]["statcan_vectors"].items()}33 rows = []34 for s in raw:35 o = s["object"]36 key = inv.get(int(o["vectorId"]))37 if key is None:38 continue39 for p in o["vectorDataPoint"]:40 if p["value"] is not None:41 rows.append({"refPer": p["refPer"], "series": key, "value": float(p["value"])})42 return pd.DataFrame(rows)434445def bcpi_factor(cma: pd.Series, from_date: pd.Series, to_quarter: str | None = None) -> pd.Series:46 """Ratio BCPI(to_quarter)/BCPI(quarter of from_date) for the CMA's series (composite when the CMA has none)."""47 b = load_bcpi()48 to_q = to_quarter or CFG["valuation_quarter"]49 series_of = CFG["temporal_regional"]["cma_to_series"]50 lut = {(r.series, r.refPer): r.value for r in b.itertuples()}5152 def q_of(d):53 d = pd.Timestamp(d)54 return f"{d.year}-{(d.month - 1) // 3 * 3 + 1:02d}-01"5556 out = []57 for c, d in zip(cma, from_date):58 s = series_of.get(c, "composite")59 q = q_of(d)60 num = lut.get((s, to_q)) or lut.get(("composite", to_q))61 den = lut.get((s, q)) or lut.get(("composite", q))62 out.append(num / den if (num and den) else np.nan)63 return pd.Series(out, index=cma.index)646566def land_index_factor(roll_market_date: pd.Series, share: float | None = None) -> pd.Series:67 """Vrai Prix single-family market index ratio from the roll month to the last available month, blended with 1 by `share`."""68 mi = pd.read_csv(ROOT / "data" / "processed" / "market_index.csv")69 mi = mi[mi.type_prop == "unifamilial"].set_index("month")["idx"]70 last = mi.iloc[-1]71 share = CFG["land"]["index_share_applied"] if share is None else share72 months = pd.to_datetime(roll_market_date).dt.strftime("%Y-%m")73 f = months.map(lambda m: last / mi.get(m, mi.iloc[0]))74 return 1.0 + share * (f - 1.0)757677# ----------------------------------------------------------------------------- RCN78def anchor(p: dict | None = None) -> dict:79 p = p or {}80 return CFG["rcn"]["anchors"][p.get("anchor", CFG["rcn"]["anchor_default"])]818283def rcn_hard(df: pd.DataFrame, p: dict | None = None) -> pd.Series:84 """Hard replacement cost new (CAD, valuation quarter) before soft costs and incentive."""85 p = p or {}86 r = CFG["rcn"]87 a = anchor(p)88 grade_cost = df["grade"].map(p.get("grade_per_sqft", a["grade_per_sqft"])).fillna(a["grade_per_sqft"]["standard"])89 storey_mult = df["storeys"].astype(str).map(p.get("storey_multiplier", a["storey_multiplier"])).fillna(1.0) + p.get("storey_shift", 0.0)90 attached_mult = df["attached"].map(r["attached_multiplier"]).fillna(1.0)91 area_sqft = df["floor_area_m2"] * SQFT_PER_M292 base = grade_cost * storey_mult * attached_mult * area_sqft93 # temporal update of the published band to the valuation quarter (Montréal series) and regional level factor94 base_q = a["reference_quarter"]95 tfac = bcpi_factor(pd.Series("montreal", index=df.index), pd.Series(base_q, index=df.index)).fillna(1.0)96 level = p.get("regional_level_factor", CFG["temporal_regional"]["regional_level_factor"])97 if isinstance(level, pd.Series):98 base = base * tfac * level99 else:100 base = base * tfac * level101 addon_scale = 1.0 + p.get("add_on_shift", 0.0)102 bsmt_sqft = area_sqft / df["storeys"].astype(str).map({"1": 1, "1.5": 1.5, "2": 2, "2.5": 2.5, "3": 3}).fillna(1.0)103 bsmt = df["basement"].map(r["basement_finish_per_sqft"]).fillna(0.0) * bsmt_sqft104 garage = df["garage"].map(r["garage"]).fillna(0.0)105 return base + addon_scale * (bsmt + garage)106107108def site_improvements(df: pd.DataFrame, rcn: pd.Series, p: dict | None = None) -> pd.Series:109 p = p or {}110 s = CFG["site_improvements"]111 share = p.get("site_share", s["baseline_share_of_rcn"])112 pool = df["pool"].map(s["pool"]).fillna(0.0) * (1.0 + p.get("pool_shift", 0.0))113 return share * rcn + pool114115116# ----------------------------------------------------------------------------- depreciation117def effective_age(df: pd.DataFrame, p: dict | None = None) -> pd.Series:118 p = p or {}119 d = CFG["depreciation"]120 strength = p.get("condition_strength", 1.0)121 cond = df["condition"].clip(1, 5)122 # interpolate the factor table on a continuous condition score123 keys = np.array(sorted(d["condition_factor"]))124 vals = np.array([d["condition_factor"][k] for k in keys])125 f = np.interp(cond, keys, vals)126 f = 1.0 + strength * (f - 1.0)127 eff = df["age"].clip(lower=0) * f128 eff = eff - df["renovation"].map({"extensive": d["renovation_extensive_years_removed"],129 "partial": d["renovation_partial_years_removed"]}).fillna(0.0)130 return eff.clip(lower=d["min_effective_age"])131132133def depreciation(df: pd.DataFrame, p: dict | None = None) -> pd.Series:134 p = p or {}135 d = CFG["depreciation"]136 life = p.get("economic_life", d["economic_life_years"])137 phys = (effective_age(df, p) / life).clip(upper=d["max_physical"])138 ext = p.get("external_obsolescence", 0.0)139 return 1.0 - (1.0 - phys) * (1.0 - ext)140141142# ----------------------------------------------------------------------------- full indication143def cost_indication(df: pd.DataFrame, p: dict | None = None) -> pd.DataFrame:144 """Return a DataFrame with L, RCN_hard, RCN (incl. soft+EI), S, D, V_cost, improvements_depreciated."""145 p = p or {}146 r = CFG["rcn"]147 hard = rcn_hard(df, p)148 soft = p.get("soft_cost_share", r["soft_cost_share"])149 ei = p.get("entrepreneurial_incentive", r["entrepreneurial_incentive"])150 rcn = hard * (1.0 + soft) * (1.0 + ei)151 S = site_improvements(df, hard, p)152 D = depreciation(df, p)153 L = df["roll_land_value"] * land_index_factor(df["roll_market_date"], p.get("land_index_share"))154 impr = (rcn + S) * (1.0 - D)155 out = pd.DataFrame({"L": L, "RCN_hard": hard, "RCN": rcn, "S": S, "D": D, "effective_age": effective_age(df, p),156 "improvements_depreciated": impr, "V_cost": L + impr}, index=df.index)157 # assessor benchmark: roll building value indexed with BCPI from the roll market date to the valuation quarter158 out["assessor_building_indexed"] = df["roll_building_value"] * bcpi_factor(df["cma"], df["roll_market_date"])159 return out160