""" THAMAN API — Spatial Lookup Module ==================================== Loads all spatial reference data at startup and computes distance / density / NTA-level features for any lat/lng coordinate. Data loaded (all from local raw/ files): - MTA Subway stations → dist_subway_m, dist_express_subway_m, nearest_station_is_express - MTA Bus stops → dist_bus_m - High schools → dist_school_m - Elementary schools → dist_elem_school_m - Parks → dist_park_m - Airbnb listings → airbnb_count_500m (BallTree 500m radius) - NTA boundaries → point-in-polygon → NTA code - NTA stats (from features.csv) → crime_rate, income, school_district, etc. - Mortgage rates CSV → mortgage_rate_30yr (latest) """ import os import json import math import numpy as np import polars as pl import geopandas as gpd from scipy.spatial import cKDTree from sklearn.neighbors import BallTree from datetime import datetime BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) RAW = os.path.join(BASE, "data", "raw") PROC = os.path.join(BASE, "data", "processed") # Express subway routes in NYC (trunk lines that run express segments) EXPRESS_ROUTES = {"A", "C", "E", "2", "3", "4", "5", "D", "F", "B", "N", "Q", "J", "Z"} # Feature descriptions for human-readable explanations FEATURE_DESCRIPTIONS = { "gross_square_feet": "Building size (sq ft)", "bldgclass": "Building class / type", "land_square_feet": "Land area (sq ft)", "longitude": "East-West location (Manhattan premium)", "latitude": "North-South location", "school_district": "School district number", "median_income_nta": "Neighborhood median income", "residential_units": "Number of residential units", "building_age": "Age of building (years)", "airbnb_count_500m": "Airbnb density within 500m", "poi_count_500m": "Points of interest within 500m", "price_appreciation": "Prior price appreciation rate", "crime_rate_nta": "Neighborhood crime rate (per 1k residents)", "borough_income_deviation": "Income deviation from borough average", "numfloors": "Number of floors", "residfar": "Residential floor area ratio allowance", "prior_sale_price": "Prior recorded sale price", "builtfar": "Actual built floor area ratio", "borough": "Borough (1=Manhattan … 5=Staten Island)", "dist_waterfront_m": "Distance to waterfront (m)", "dist_subway_m": "Distance to nearest subway station (m)", "dist_park_m": "Distance to nearest park (m)", "dist_school_m": "Distance to nearest high school (m)", "dist_elem_school_m": "Distance to nearest elementary school (m)", "dist_bus_m": "Distance to nearest bus stop (m)", "dist_hospital_m": "Distance to nearest hospital/clinic (m)", "dist_bike_lane_m": "Distance to nearest bike lane (m)", "dist_express_subway_m": "Distance to nearest express subway (m)", "noise_density_nta": "Neighborhood noise complaint density", "livability_complaint_rate":"Neighborhood livability complaint rate", "population_2020": "NTA population (2020 Census)", "mortgage_rate_30yr": "30-year fixed mortgage rate (%)", "sale_year": "Year of sale", "has_elevator": "Building has elevator", "is_condo": "Condo unit (R-class)", "is_multifamily": "Multi-family elevator building (D-class)", "is_single_fam": "Single family home (A-class)", "is_mixed_use": "Mixed-use building (S-class)", "renovated_since_2018": "Major permit/renovation since 2018", "years_since_renovation": "Years since last major renovation", "far_utilization": "FAR utilization (built / max allowed)", "commfar": "Commercial floor area ratio allowance", "facilfar": "Community facility FAR allowance", "maxallwfar": "Maximum allowable FAR", "district_avg_score": "Average school quality score in district", "district_school_count": "Number of schools in district", "has_prior_sale": "Property has a prior sale record", "is_flip": "Resold within 2 years (flip)", "years_since_prior_sale": "Years since prior sale", "nearest_station_is_express":"Nearest subway station runs express trains", # v2 new features "log_dist_subway_m": "Log-distance to nearest subway station", "log_dist_school_m": "Log-distance to nearest high school", "log_dist_park_m": "Log-distance to nearest park", "log_dist_hospital_m": "Log-distance to nearest hospital", "log_dist_bus_m": "Log-distance to nearest bus stop", "log_dist_waterfront_m": "Log-distance to waterfront", "log_dist_bike_lane_m": "Log-distance to nearest bike lane", "log_dist_elem_school_m": "Log-distance to nearest elementary school", "log_dist_express_subway_m":"Log-distance to nearest express subway", "dist_midtown_manhattan_m": "Distance to Midtown Manhattan (gravity centre)", "dist_downtown_manhattan_m":"Distance to Downtown Manhattan (gravity centre)", "dist_downtown_brooklyn_m": "Distance to Downtown Brooklyn (gravity centre)", "dist_long_island_city_m": "Distance to Long Island City (gravity centre)", "is_manhattan": "Property is in Manhattan (borough 1)", "crime_x_manhattan": "Crime rate × Manhattan flag interaction", "crime_x_non_manhattan": "Crime rate × non-Manhattan flag interaction", "walk_score_proxy": "Walkability/transit composite score (0–100)", "bldgclass_encoded": "Building class target-encoded mean log-price", "borough_bldg_encoded": "Borough × building-class target-encoded mean log-price", "sale_month_sin": "Cyclical month encoding — sine component", "sale_month_cos": "Cyclical month encoding — cosine component", } OVERTURE_BUCKETS = { "cafe": {"cafe", "coffee_shop"}, "restaurant": {"restaurant", "casual_eatery", "fast_food_restaurant", "pizzaria"}, "gym": {"gym", "fitness_center", "yoga_studio", "martial_arts_club"}, "grocery": {"grocery_store", "supermarket", "convenience_store"}, "bar": {"bar", "cocktail_bar", "night_club"}, "pharmacy": {"pharmacy", "drug_store"}, # New QoL categories (May 2026) "atm": {"atm"}, "urgent_care": {"urgent_care", "medical_clinic", "healthcare_location"}, "cinema": {"movie_theater", "cinema"}, "library": {"library", "public_library"}, "childcare": {"childcare", "child_care_facility", "preschool"}, "beauty": {"beauty_salon", "hair_salon", "nail_salon"}, "hotel": {"hotel", "hostel", "motel"}, } class SpatialLookup: """ Load all spatial reference data once at startup; serve fast lookups at request time. """ def __init__(self): print("[SpatialLookup] Loading spatial reference data...") self._load_nta_stats() self._load_subway() self._load_bus() self._load_schools() self._load_parks() self._load_airbnb() self._load_citibike() self._load_commuter_rail() self._load_waterfront() self._load_bike_lanes() self._load_poi_buckets() self._load_mortgage_rate() print("[SpatialLookup] All data loaded. Ready.") # ── Loaders ──────────────────────────────────────────────────────── def _load_nta_stats(self): """Precompute per-NTA medians from features_v6.csv (latest) + load NTA boundaries.""" for _p in ["features_v6.csv", "features_v5.csv", "features_v4.csv", "features.csv"]: feat_path = os.path.join(PROC, _p) if os.path.exists(feat_path): break features = pl.read_csv(feat_path) NTA_STAT_COLS = [ "crime_rate_nta", "noise_density_nta", "livability_complaint_rate", "population_2020", "median_income_nta", "borough_income_deviation", "school_district", "district_avg_score", "district_school_count", "poi_count_500m", "dist_hospital_m", "dist_waterfront_m", "dist_bike_lane_m", "poi_cafe_500m", "poi_restaurant_500m", "poi_gym_500m", "poi_grocery_500m", "poi_bar_500m", "poi_pharmacy_500m", # New Overture POI categories (v6) "poi_atm_500m", "poi_urgent_care_500m", "poi_cinema_500m", "poi_library_500m", "poi_childcare_500m", "poi_beauty_500m", "poi_hotel_500m", # Citi Bike access (v14) "citibike_500m", "dist_citibike_m", # LPC historic district + FEMA flood zone (v16) "is_historic_dist", "in_flood_zone", "is_landmark", "builtfar", "residfar", "commfar", "facilfar", "maxallwfar", "far_utilization", ] available = [c for c in NTA_STAT_COLS if c in features.columns] nta_stats = features.group_by("ntacode").agg( [pl.col(c).median() for c in available] ) self._nta_stats = { row["ntacode"]: {k: v for k, v in row.items() if k != "ntacode"} for row in nta_stats.iter_rows(named=True) } self._global_medians = {c: float(features[c].median() or 0.0) for c in available} # Borough-level income for borough_income_deviation fallback self._borough_median_income = { row["borough"]: row["median_income_nta"] for row in features.group_by("borough") .agg(pl.col("median_income_nta").median()) .iter_rows(named=True) } # Merge v16 NTA lookup (historic district + flood zone rates from meta.json) _meta_path = os.path.join(BASE, "models", "meta.json") if os.path.exists(_meta_path): with open(_meta_path) as _f: _meta_j = json.load(_f) _v16_lu = _meta_j.get("v16_nta_lookup", {}) _v16_glob_hist = _meta_j.get("v16_global_hist_rate", 0.0838) _v16_glob_flood = _meta_j.get("v16_global_flood_rate", 0.0834) for _nta, _vals in _v16_lu.items(): if _nta not in self._nta_stats: self._nta_stats[_nta] = {} self._nta_stats[_nta].update(_vals) # Fill global medians for fallback self._global_medians.setdefault("is_historic_dist", _v16_glob_hist) self._global_medians.setdefault("in_flood_zone", _v16_glob_flood) self._global_medians.setdefault("is_landmark", 0.0) # v17: tax exemption NTA lookup _v17_lu = _meta_j.get("v17_nta_log_exempt", {}) _v17_glob = _meta_j.get("v17_global_log_exempt", 0.0) for _nta, _val in _v17_lu.items(): if _nta not in self._nta_stats: self._nta_stats[_nta] = {} self._nta_stats[_nta]["log_exempt_amount"] = _val self._global_medians.setdefault("log_exempt_amount", _v17_glob) # v18: census tract median household income NTA lookup _v18_lu = _meta_j.get("v18_nta_log_income", {}) _v18_glob = _meta_j.get("v18_global_log_income", 10.8) for _nta, _val in _v18_lu.items(): if _nta not in self._nta_stats: self._nta_stats[_nta] = {} self._nta_stats[_nta]["log_tract_median_income"] = _val self._global_medians.setdefault("log_tract_median_income", _v18_glob) # NTA GeoDataFrame for point-in-polygon self._nta_gdf = gpd.read_file(os.path.join(RAW, "nta_boundaries.geojson")) print(f" NTA stats: {len(self._nta_stats)} NTAs | global medians computed") def _load_subway(self): subway = ( pl.read_csv(os.path.join(RAW, "MTA_Subway_Stations_20260308.csv")) .with_columns([ pl.col("GTFS Latitude").cast(pl.Float64, strict=False), pl.col("GTFS Longitude").cast(pl.Float64, strict=False), ]) .drop_nulls(subset=["GTFS Latitude", "GTFS Longitude"]) ) all_coords = subway.select(["GTFS Latitude", "GTFS Longitude"]).to_numpy() self._subway_tree = cKDTree(all_coords) # Express: any station whose Daytime Routes overlap with EXPRESS_ROUTES def is_express(routes_str): if routes_str is None: return False return bool(set(str(routes_str).split()) & EXPRESS_ROUTES) routes = subway["Daytime Routes"].to_list() express_mask = pl.Series([is_express(r) for r in routes]) express_coords = ( subway.filter(express_mask) .select(["GTFS Latitude", "GTFS Longitude"]) .to_numpy() ) self._express_tree = cKDTree(express_coords) if len(express_coords) > 0 else self._subway_tree print(f" Subway: {len(all_coords)} stations | express: {len(express_coords)}") def _load_bus(self): bus = ( pl.read_csv(os.path.join(RAW, "mta_bus_stops.csv")) .with_columns([ pl.col("latitude").cast(pl.Float64, strict=False), pl.col("longitude").cast(pl.Float64, strict=False), ]) .drop_nulls(subset=["latitude", "longitude"]) ) self._bus_tree = cKDTree(bus.select(["latitude", "longitude"]).to_numpy()) print(f" Bus stops: {len(bus)}") def _load_schools(self): hs = pl.read_csv(os.path.join(RAW, "schools.csv")).drop_nulls(subset=["latitude", "longitude"]) elem = pl.read_csv(os.path.join(RAW, "elementary_schools.csv")).drop_nulls(subset=["latitude", "longitude"]) self._hs_tree = cKDTree(hs.select(["latitude", "longitude"]).to_numpy()) self._elem_tree = cKDTree(elem.select(["latitude", "longitude"]).to_numpy()) print(f" Schools: {len(hs)} HS | {len(elem)} elementary") def _load_parks(self): parks = pl.read_csv(os.path.join(RAW, "parks_with_coords.csv"), ignore_errors=True).drop_nulls(subset=["latitude", "longitude"]) self._park_tree = cKDTree(parks.select(["latitude", "longitude"]).to_numpy()) # v19: size-stratified park trees _parks_v19 = parks.with_columns( pl.col("ACRES").cast(pl.Float64, strict=False).fill_null(0.0) ) _lrg = _parks_v19.filter(pl.col("ACRES") >= 10.0).select(["latitude","longitude"]).to_numpy() _flag = _parks_v19.filter(pl.col("ACRES") >= 100.0).select(["latitude","longitude"]).to_numpy() self._park_large_tree = cKDTree(_lrg) if len(_lrg) > 0 else None self._park_flagship_tree = cKDTree(_flag) if len(_flag) > 0 else None print(f" Parks: {len(parks)} total | large≥10ac: {len(_lrg)} | flagship≥100ac: {len(_flag)}") def _load_airbnb(self): airbnb = ( pl.read_csv(os.path.join(RAW, "airbnb_listings.csv")) .with_columns([ pl.col("latitude").cast(pl.Float64, strict=False), pl.col("longitude").cast(pl.Float64, strict=False), ]) .drop_nulls(subset=["latitude", "longitude"]) ) coords_rad = np.radians(airbnb.select(["latitude", "longitude"]).to_numpy()) self._airbnb_balltree = BallTree(coords_rad, metric="haversine") print(f" Airbnb: {len(airbnb)} listings") def _load_citibike(self): cb_path = os.path.join(RAW, "nyc_citibike_stations.csv") self._citibike_tree = None if not os.path.exists(cb_path): return cb = ( pl.read_csv(cb_path) .with_columns([ pl.col("lat").cast(pl.Float64, strict=False), pl.col("lon").cast(pl.Float64, strict=False), ]) .drop_nulls(subset=["lat", "lon"]) ) coords_rad = np.radians(cb.select(["lat", "lon"]).to_numpy()) self._citibike_tree = BallTree(coords_rad, metric="haversine") print(f" Citi Bike: {len(cb)} stations") def _load_commuter_rail(self): cr_path = os.path.join(RAW, "nyc_commuter_rail_stations.csv") self._commuter_rail_tree = None if not os.path.exists(cr_path): print(" Commuter rail: file missing — dist_commuter_rail_m will use NTA fallback") return cr = ( pl.read_csv(cr_path) .with_columns([ pl.col("lat").cast(pl.Float64, strict=False), pl.col("lon").cast(pl.Float64, strict=False), ]) .drop_nulls(subset=["lat", "lon"]) ) coords_rad = np.radians(cr.select(["lat", "lon"]).to_numpy()) self._commuter_rail_tree = BallTree(coords_rad, metric="haversine") print(f" Commuter rail: {len(cr)} stations (LIRR/Metro-North/SIR)") def _load_waterfront(self): wf_path = os.path.join(RAW, "nyc_coastline_pts.npy") if os.path.exists(wf_path): pts = np.load(wf_path) self._waterfront_tree = cKDTree(pts) print(f" Waterfront: {len(pts)} coastline points") else: self._waterfront_tree = None print(" Waterfront: coastline file missing — will use NTA median fallback") def _load_bike_lanes(self): import json as _json bike_path = os.path.join(RAW, "nyc_bike_lanes.geojson") if os.path.exists(bike_path): with open(bike_path) as f: gj = _json.load(f) pts = [] for feat in gj.get("features", []): geom = feat.get("geometry") or {} coords = geom.get("coordinates", []) gtype = geom.get("type", "") if gtype == "LineString": for c in coords[::3]: pts.append((c[1], c[0])) elif gtype == "MultiLineString": for line in coords: for c in line[::3]: pts.append((c[1], c[0])) if pts: self._bike_tree = cKDTree(np.array(pts, dtype=np.float64)) print(f" Bike lanes: {len(pts)} sampled vertices") else: self._bike_tree = None print(" Bike lanes: no features parsed — will use NTA median fallback") else: self._bike_tree = None print(" Bike lanes: file missing — will use NTA median fallback") def _load_poi_buckets(self): import json as _json op_path = os.path.join(RAW, "overture_places.geojson") npz_path = os.path.join(RAW, "overture_poi_buckets.npz") self._poi_balltrees: dict[str, BallTree | None] = {} if os.path.exists(npz_path): # compact precomputed bucket points (1.2 MB vs 619 MB geojson) data = np.load(npz_path) for bname in OVERTURE_BUCKETS: arr = data[bname] if bname in data.files else np.zeros((0, 2)) self._poi_balltrees[bname] = ( BallTree(np.radians(arr), metric="haversine") if len(arr) else None ) print(f" POI {bname}: {len(arr):,}") return if not os.path.exists(op_path): print(" POI buckets: overture_places.geojson missing — counts will be 0") for bname in OVERTURE_BUCKETS: self._poi_balltrees[bname] = None return with open(op_path) as f: op = _json.load(f) for bname, cats in OVERTURE_BUCKETS.items(): bpts = [] for feat in op["features"]: bc = feat.get("properties", {}).get("basic_category", "") if bc in cats: c = feat.get("geometry", {}).get("coordinates", []) if c and len(c) >= 2: bpts.append([c[1], c[0]]) if bpts: arr = np.array(bpts, dtype=np.float64) self._poi_balltrees[bname] = BallTree(np.radians(arr), metric="haversine") else: self._poi_balltrees[bname] = None print(f" POI {bname}: {len(bpts):,}") def _load_mortgage_rate(self): mort = pl.read_csv(os.path.join(RAW, "mortgage_rates.csv")) self._mortgage_rate = float(mort["mortgage_rate_30yr"].drop_nulls()[-1]) print(f" Mortgage rate (latest): {self._mortgage_rate}%") # ── KD-tree helper ───────────────────────────────────────────────── @staticmethod def _kdtree_dist_m(tree: cKDTree, lat: float, lon: float) -> float: """Nearest neighbor distance in meters using degree approximation.""" d, _ = tree.query([[lat, lon]], k=1) return float(d[0]) * 111_000 # 1 degree ≈ 111 km # ── NTA lookup ───────────────────────────────────────────────────── def _nta_for_point(self, lat: float, lon: float) -> str | None: """Find the NTA code for a given lat/lon via point-in-polygon.""" try: pt = gpd.GeoDataFrame( geometry=gpd.points_from_xy([lon], [lat]), crs="EPSG:4326" ) joined = gpd.sjoin(pt, self._nta_gdf[["ntacode", "geometry"]], how="left", predicate="within") ntacode = joined["ntacode"].iloc[0] return ntacode if (ntacode is not None and not (isinstance(ntacode, float) and math.isnan(ntacode))) else None except Exception: return None # ── Main lookup ──────────────────────────────────────────────────── def lookup(self, lat: float, lon: float) -> dict: """ Compute all auto-derived spatial features for the given lat/lon. Returns a dict of feature_name → value (matching feature_names in meta.json). """ feats = {} # ── Distance features ────────────────────────────────────────── feats["dist_subway_m"] = self._kdtree_dist_m(self._subway_tree, lat, lon) feats["dist_express_subway_m"] = self._kdtree_dist_m(self._express_tree, lat, lon) # nearest_station_is_express: true when express station is as close as any station feats["nearest_station_is_express"] = int( abs(feats["dist_express_subway_m"] - feats["dist_subway_m"]) < 150 ) feats["dist_bus_m"] = self._kdtree_dist_m(self._bus_tree, lat, lon) feats["dist_school_m"] = self._kdtree_dist_m(self._hs_tree, lat, lon) feats["dist_elem_school_m"] = self._kdtree_dist_m(self._elem_tree, lat, lon) feats["dist_park_m"] = self._kdtree_dist_m(self._park_tree, lat, lon) # v19: size-stratified park distances _d_lp = self._kdtree_dist_m(self._park_large_tree, lat, lon) if self._park_large_tree else float("nan") _d_fp = self._kdtree_dist_m(self._park_flagship_tree, lat, lon) if self._park_flagship_tree else float("nan") feats["dist_large_park_m"] = _d_lp feats["dist_flagship_park_m"] = _d_fp feats["log_dist_large_park_m"] = float(np.log1p(_d_lp)) if not np.isnan(_d_lp) else float("nan") feats["log_dist_flagship_park_m"] = float(np.log1p(_d_fp)) if not np.isnan(_d_fp) else float("nan") # ── Airbnb density within 500m ───────────────────────────────── radius_rad = 500.0 / 6_371_000.0 cnt = self._airbnb_balltree.query_radius( np.radians([[lat, lon]]), r=radius_rad, count_only=True ) feats["airbnb_count_500m"] = int(cnt[0]) # ── NTA lookup (needed by fallbacks below) ───────────────────── ntacode = self._nta_for_point(lat, lon) nta_data = self._nta_stats.get(ntacode, {}) if ntacode else {} # ── Citi Bike stations within 500m + nearest distance ────────── if self._citibike_tree is not None: pt_rad = np.radians([[lat, lon]]) cb_cnt = self._citibike_tree.query_radius(pt_rad, r=radius_rad, count_only=True) feats["citibike_500m"] = int(cb_cnt[0]) cb_dist, _ = self._citibike_tree.query(pt_rad, k=1) feats["dist_citibike_m"] = round(float(cb_dist[0][0]) * 6_371_000, 1) else: feats["citibike_500m"] = int(nta_data.get("citibike_500m", 0)) feats["dist_citibike_m"] = float(nta_data.get("dist_citibike_m", 9999.0)) # ── Commuter rail distance + 1km flag ───────────────────────── if self._commuter_rail_tree is not None: pt_rad = np.radians([[lat, lon]]) cr_dist, _ = self._commuter_rail_tree.query(pt_rad, k=1) cr_dist_m = round(float(cr_dist[0][0]) * 6_371_000, 1) feats["dist_commuter_rail_m"] = cr_dist_m feats["commuter_rail_1km"] = int(cr_dist_m <= 1000) else: feats["dist_commuter_rail_m"] = 9999.0 feats["commuter_rail_1km"] = 0 # ── Waterfront distance (real per-point) ─────────────────────── if self._waterfront_tree is not None: feats["dist_waterfront_m"] = self._kdtree_dist_m(self._waterfront_tree, lat, lon) else: feats["dist_waterfront_m"] = nta_data.get("dist_waterfront_m", self._global_medians.get("dist_waterfront_m", 1414.0)) feats["log_dist_waterfront_m"] = float(np.log1p(feats["dist_waterfront_m"])) feats["waterfront_200m"] = int(feats["dist_waterfront_m"] < 200) # ── Bike lane distance (real per-point) ──────────────────────── if self._bike_tree is not None: feats["dist_bike_lane_m"] = self._kdtree_dist_m(self._bike_tree, lat, lon) else: feats["dist_bike_lane_m"] = nta_data.get("dist_bike_lane_m", self._global_medians.get("dist_bike_lane_m", 152.0)) # ── POI category counts within 500m (real per-point) ────────── radius_rad_poi = 500.0 / 6_371_000.0 for bname, bt in self._poi_balltrees.items(): col = f"poi_{bname}_500m" if bt is not None: cnt = bt.query_radius(np.radians([[lat, lon]]), r=radius_rad_poi, count_only=True) feats[col] = int(cnt[0]) else: feats[col] = int(nta_data.get(col, self._global_medians.get(col, 0))) # ── NTA-level features ───────────────────────────────────────── for col in [ "crime_rate_nta", "noise_density_nta", "livability_complaint_rate", "population_2020", "median_income_nta", "borough_income_deviation", "school_district", "district_avg_score", "district_school_count", "poi_count_500m", "dist_hospital_m", "builtfar", "residfar", "commfar", "facilfar", "maxallwfar", "far_utilization", # v16: LPC historic district rate + FEMA flood zone rate (NTA-level averages) "is_historic_dist", "in_flood_zone", "is_landmark", # v17: tax exemption NTA-level mean "log_exempt_amount", # v18: census tract median household income NTA-level mean "log_tract_median_income", ]: feats[col] = nta_data.get(col, self._global_medians.get(col, 0.0)) # ── Macro features ───────────────────────────────────────────── feats["mortgage_rate_30yr"] = self._mortgage_rate now = datetime.now() feats["sale_year"] = now.year # Note: sale_month removed in v2 — main.py computes sale_month_sin/cos instead return feats def get_feature_description(self, feature_name: str) -> str: return FEATURE_DESCRIPTIONS.get(feature_name, feature_name.replace("_", " ").title()) # ══════════════════════════════════════════════════════════════════════════════ # Riyadh Spatial Lookup # ══════════════════════════════════════════════════════════════════════════════ import unicodedata as _ucd import pandas as _pd RIYADH_COMMERCIAL_BUCKETS = { "hypermarket": {"HypMkt"}, "supermarket": {"SupMkt", "MktS", "GroS"}, "bank": {"Bank"}, "restaurant": {"Res"}, "hotel": {"Hot", "HotAp"}, "gas_station": {"GasStation", "PetStation"}, "commercial_complex": {"ComC", "ComX"}, } # English station name → (lat, lon) mapping for air quality CSV _AQ_STATION_COORDS = { "At-Taawun": (24.762272, 46.650878), "Al-Muruj": (24.758315, 46.671171), "Al-Jazeera": (24.700139, 46.678500), "Al-Uraija": (24.685105, 46.703063), "Al-Khalidiya": (24.766047, 46.761886), "Ar-Rawabi": (24.751314, 46.868278), "Ad-Dhubbat": (24.723857, 46.756673), "Al-Ghurabi": (24.648444, 46.721056), "Al-Khaleej": (24.598469, 46.744378), } _LINE_ORDER = {"Line1": 1, "Line2": 2, "Line3": 3, "Line4": 4, "Line5": 5, "Line6": 6} def _normalize_ar(s: str) -> str: """Normalize Arabic string: NFKC, strip tatweel and harakat.""" if not isinstance(s, str): return "" s = _ucd.normalize("NFKC", s) s = s.replace("ـ", "") # tatweel s = "".join(c for c in s if not (0x064B <= ord(c) <= 0x065F)) return s.strip() class RiyadhSpatialLookup: """ Parallel spatial index for Riyadh-specific data sources. Loaded once at API startup; lookup() returns ~25 features for any lat/lon. """ def __init__(self): print("[RiyadhSpatialLookup] Loading Riyadh spatial data...") self._load_metro() self._load_bus() self._load_intersections() self._load_commercial() self._load_poi_csvs() self._load_air_quality() self._load_district_stats() print("[RiyadhSpatialLookup] Ready.") # ── Loaders ─────────────────────────────────────────────────────────────── def _load_metro(self): path = os.path.join(RAW, "metro-stations-in-riyadh-by-metro-line-and-station-type-2024.geojson") with open(path) as f: gj = json.load(f) coords, lines, types = [], [], [] line1 = [] for feat in gj["features"]: c = feat["geometry"]["coordinates"] # [lon, lat] lat, lon = c[1], c[0] p = feat["properties"] coords.append([lat, lon]) lines.append(p["metro_line_cd"]) types.append(int(p["metro_station_type_cd"])) if p["metro_line_cd"] == "Line1": line1.append([lat, lon]) self._metro_arr = np.array(coords, dtype=np.float64) self._metro_tree = cKDTree(self._metro_arr) self._metro_ball = BallTree(np.radians(self._metro_arr), metric="haversine") self._metro_lines = lines self._metro_types = types self._line1_tree = cKDTree(np.array(line1, dtype=np.float64)) if line1 else self._metro_tree print(f" Metro: {len(coords)} stations | Line1: {len(line1)}") def _load_bus(self): path = os.path.join(RAW, "bus-stops-in-riyadh-by-bus-route-direction-and-shelter-type-2024.geojson") with open(path) as f: gj = json.load(f) coords, shelters = [], [] for feat in gj["features"]: geo = feat["properties"].get("geo_point_2d") or {} lat, lon = geo.get("lat"), geo.get("lon") if lat is None: continue coords.append([lat, lon]) btype = str(feat["properties"].get("bsheltertypecode", "")) shelters.append(btype.startswith("A")) arr = np.array(coords, dtype=np.float64) self._bus_tree = cKDTree(arr) self._bus_ball = BallTree(np.radians(arr), metric="haversine") shelter_arr = arr[np.array(shelters)] self._shelter_ball = BallTree(np.radians(shelter_arr), metric="haversine") if len(shelter_arr) else None print(f" Bus: {len(coords)} stops | BRT: {shelter_arr.shape[0]}") def _load_intersections(self): path = os.path.join(RAW, "traffic-intersections-by-main-street-and-cross-street-2024.geojson") with open(path) as f: gj = json.load(f) coords = [] for feat in gj["features"]: geo = feat["properties"].get("geo_point_2d") or {} lat, lon = geo.get("lat"), geo.get("lon") if lat is None: continue coords.append([lat, lon]) arr = np.array(coords, dtype=np.float64) self._int_tree = cKDTree(arr) self._int_ball = BallTree(np.radians(arr), metric="haversine") print(f" Intersections: {len(coords)}") def _load_commercial(self): path = os.path.join(RAW, "commercial-services-by-category-sub-municipality-and-district-2024.geojson") with open(path) as f: gj = json.load(f) all_coords = [] bucket_coords = {k: [] for k in RIYADH_COMMERCIAL_BUCKETS} for feat in gj["features"]: geo = feat["properties"].get("geo_point_2d") or {} lat, lon = geo.get("lat"), geo.get("lon") if lat is None: continue all_coords.append([lat, lon]) cat = feat["properties"].get("comcatcode", "") for bname, codes in RIYADH_COMMERCIAL_BUCKETS.items(): if cat in codes: bucket_coords[bname].append([lat, lon]) all_arr = np.array(all_coords, dtype=np.float64) self._comm_ball_all = BallTree(np.radians(all_arr), metric="haversine") self._comm_balls = {} for bname, pts in bucket_coords.items(): if pts: self._comm_balls[bname] = BallTree( np.radians(np.array(pts, dtype=np.float64)), metric="haversine" ) else: self._comm_balls[bname] = None print(f" Commercial {bname}: {len(pts)}") def _load_poi_csvs(self): """Load mosque/mall/school/hospital/park/entertainment POI files from saudi_thaman.""" _POI_FILES = { "mosque": os.path.join(RAW, "riyadh_mosques.csv"), "mall": os.path.join(RAW, "riyadh_malls.csv"), "school": os.path.join(RAW, "riyadh_schools.csv"), "hospital": os.path.join(RAW, "riyadh_hospitals.csv"), "park": os.path.join(RAW, "riyadh_parks.csv"), "entertain": os.path.join(RAW, "rcrc_entertainment.csv"), # New QoL POIs (OSM, May 2026) — used in v3 model features "pharmacy": os.path.join(RAW, "riyadh_qol_pharmacies.csv"), "gym": os.path.join(RAW, "riyadh_qol_gyms.csv"), "coffee": os.path.join(RAW, "riyadh_qol_coffee_shops.csv"), "clinic": os.path.join(RAW, "riyadh_qol_clinics.csv"), "university": os.path.join(RAW, "riyadh_qol_universities.csv"), "supermarket":os.path.join(RAW, "riyadh_qol_supermarkets.csv"), "cinema": os.path.join(RAW, "riyadh_qol_cinemas.csv"), "sports": os.path.join(RAW, "riyadh_qol_sports_centres.csv"), # Batch-2 QoL POIs — display only (not in v3 model training) "restaurant": os.path.join(RAW, "riyadh_qol_restaurants.csv"), "library": os.path.join(RAW, "riyadh_qol_libraries.csv"), "atm": os.path.join(RAW, "riyadh_qol_atms.csv"), "kindergarten": os.path.join(RAW, "riyadh_qol_kindergartens.csv"), "swimming_pool": os.path.join(RAW, "riyadh_qol_swimming_pools.csv"), } self._poi_trees: dict = {} self._poi_balls: dict = {} for poi_name, path in _POI_FILES.items(): if not os.path.exists(path): continue df = _pd.read_csv(path, encoding="utf-8-sig") df.columns = [c.lstrip("").strip() for c in df.columns] lat_col = next((c for c in df.columns if "lat" in c.lower()), None) lon_col = next((c for c in df.columns if "lon" in c.lower()), None) if lat_col is None or lon_col is None: continue df = df[[lat_col, lon_col]].dropna() df[lat_col] = _pd.to_numeric(df[lat_col], errors="coerce") df[lon_col] = _pd.to_numeric(df[lon_col], errors="coerce") df = df.dropna() # Filter Riyadh bbox df = df[(df[lat_col] > 23.5) & (df[lat_col] < 26.0) & (df[lon_col] > 45.5) & (df[lon_col] < 48.0)] if len(df) < 2: continue arr = df[[lat_col, lon_col]].values.astype(np.float64) self._poi_trees[poi_name] = cKDTree(arr) self._poi_balls[poi_name] = BallTree(np.radians(arr), metric="haversine") print(f" POI {poi_name}: {len(arr)}") def _load_air_quality(self): aq_csv = os.path.join(RAW, "air-quality.csv") aq_df = _pd.read_csv(aq_csv, sep=";") aq_avg = aq_df[aq_df["Indicator"] == "Avg / Hourly"].copy() aq_means = ( aq_avg[aq_avg["Component"].isin(["NO2", "SO2", "PM10", "O3"])] .groupby(["Station", "Component"])["Value"] .mean() .unstack(fill_value=0) .reset_index() ) aq_means["lat"] = aq_means["Station"].map(lambda s: _AQ_STATION_COORDS.get(s, (None, None))[0]) aq_means["lon"] = aq_means["Station"].map(lambda s: _AQ_STATION_COORDS.get(s, (None, None))[1]) aq_means = aq_means[aq_means["lat"].notna()].copy() self._aq_arr = aq_means[["lat", "lon"]].values.astype(np.float64) self._aq_tree = cKDTree(self._aq_arr) self._aq_means_df = aq_means self._aq_components = [c for c in ["NO2", "SO2", "PM10", "O3"] if c in aq_means.columns] print(f" Air quality: {len(self._aq_arr)} stations | {self._aq_components}") def _load_district_stats(self): """Load per-district aggregated stats from features_riyadh.csv.""" feat_path = os.path.join(PROC, "features_riyadh.csv") if not os.path.exists(feat_path): self._district_stats = {} self._district_feat_df = None self._district_centroid_tree = None self._district_names = [] print(" District stats: features_riyadh.csv not found — run riyadh_feature_engineering.py") return df = _pd.read_csv(feat_path) # Choropleth stats (median per district across all columns) STAT_COLS = [ "district_lat", "district_lon", "dist_metro_m", "metro_stations_1km", "commercial_count_1km", "hypermarket_count_1km", "bus_stops_500m", "no2_nearest_mean", "pm10_nearest_mean", "air_quality_score", "rei_residential_qtr_idx", "district_median_price_sqm", "district_price_trend_slope", "district_commercial_mix", "riyadh_connectivity_score", ] available = [c for c in STAT_COLS if c in df.columns] agg = df.groupby("district_ar")[available].median().reset_index() self._district_stats = agg.set_index("district_ar").to_dict("index") # Full feature medians per district (for prediction) num_cols = df.select_dtypes(include="number").columns.tolist() feat_cols = [c for c in num_cols if c not in ("year", "quarter", "quarter_id", "sale_year", "sale_quarter", "sale_price_sar_sqm", "is_apartment", "is_villa", "is_residential_plot", "is_building", "district_encoded", "district_type_encoded")] self._district_feat_df = ( df.groupby("district_ar")[feat_cols] .median() .reset_index() ) # ── Haraj override from riyadh_meta.json (fresher May 2026 snapshot) ────── _rmeta_path = os.path.join(os.path.dirname(PROC), "models", "riyadh_meta.json") _haraj_lu = {} if os.path.exists(_rmeta_path): try: with open(_rmeta_path) as _rf: _rmeta = json.load(_rf) _haraj_lu = _rmeta.get("haraj_district_lookup", {}) except Exception: pass if _haraj_lu: _haraj_cols = ["haraj_listing_count","haraj_median_psqm","haraj_p25_psqm","haraj_p75_psqm","haraj_iqr_psqm"] for _col in _haraj_cols: if _col not in self._district_feat_df.columns: self._district_feat_df[_col] = float("nan") for _idx, _drow in self._district_feat_df.iterrows(): _dname = _drow["district_ar"] if _dname in _haraj_lu: _lu_row = _haraj_lu[_dname] for _col in _haraj_cols: _val = _lu_row.get(_col) if _val is not None: self._district_feat_df.at[_idx, _col] = float(_val) # Recompute haraj_asking_premium from updated median if "haraj_median_psqm" in self._district_feat_df.columns and "district_median_price_sqm" in self._district_feat_df.columns: _dmed = self._district_feat_df["district_median_price_sqm"].replace(0, float("nan")) self._district_feat_df["haraj_asking_premium"] = self._district_feat_df["haraj_median_psqm"] / _dmed print(f" Haraj override: {len(_haraj_lu)} districts from riyadh_meta.json") # Separate lookup for target-encoded district features (excluded from feat_cols # to avoid leakage during training, but valid and important at inference time). _enc_cols = [c for c in [ "district_encoded", "district_type_encoded", "district_apt_encoded", "district_recent_encoded", "district_apt_recent_encoded", ] if c in df.columns] _enc_agg = df.groupby("district_ar")[_enc_cols].median() self._district_encoded_map: dict = _enc_agg.to_dict("index") # district_ar → {col: val} # KDTree on district centroids — use district_centroids.csv + a hardcoded # fallback table for 13 districts whose geocodes defaulted to city-centre # (24.7136, 46.6753) in features_riyadh.csv and are absent from the CSV. cent_csv = os.path.join(PROC, "district_centroids.csv") _cent_override: dict = {} if os.path.exists(cent_csv): import pandas as _pd2 _c = _pd2.read_csv(cent_csv) _cent_override = { row["district_ar"]: (float(row["district_lat"]), float(row["district_lon"])) for _, row in _c.iterrows() if not (_pd2.isna(row["district_lat"]) or _pd2.isna(row["district_lon"])) } # Hardcoded real centroids for districts missing from district_centroids.csv # (verified against Google Maps / OSM district polygons) _HARDCODED: dict = { "ظهره العودة غرب": (24.7389, 46.5165), # Zahrat Al Awda West — Diriyah fringe "ظهرة العودة شرق": (24.7530, 46.5454), # Zahrat Al Awda East — Diriyah fringe "الشفاء": (24.5608, 46.6930), # Al Shifa — south Riyadh "الصفاء": (24.6717, 46.7700), # Al Safa — east-central Riyadh "المنصورة": (24.6091, 46.7444), # Al Mansourah — south Riyadh "الوسام": (24.6275, 46.6800), # Al Wisam — south Riyadh "وادي لبن": (24.6200, 46.5650), # Wadi Laban — west Riyadh "الرابية": (24.8000, 46.6600), # Al Rabi'a — north Riyadh "السحاب": (24.7700, 46.7150), # Al Sahab — north-east Riyadh "الملك سلمان": (24.7630, 46.6440), # King Salman — north Riyadh (KAFD area) "المرجان": (24.6600, 46.7500), # Al Murjan — east Riyadh "سدرة": (24.7650, 46.7000), # Sidra — north Riyadh "أخرى": (99.0, 0.0), # "Other" catch-all — sentinel far outside Riyadh so it never wins } _cent_override.update(_HARDCODED) _DEFAULT_LAT, _DEFAULT_LON = 24.7136, 46.6753 cent_lats, cent_lons = [], [] _fixed = 0 for _, row in self._district_feat_df.iterrows(): d_ar = row["district_ar"] feat_lat = float(row.get("district_lat", _DEFAULT_LAT) or _DEFAULT_LAT) feat_lon = float(row.get("district_lon", _DEFAULT_LON) or _DEFAULT_LON) is_default = (abs(feat_lat - _DEFAULT_LAT) < 0.001 and abs(feat_lon - _DEFAULT_LON) < 0.001) if is_default and d_ar in _cent_override: cent_lats.append(_cent_override[d_ar][0]) cent_lons.append(_cent_override[d_ar][1]) _fixed += 1 else: cent_lats.append(feat_lat) cent_lons.append(feat_lon) cents = np.array(list(zip(cent_lats, cent_lons)), dtype=np.float64) self._district_centroid_tree = cKDTree(cents) self._district_names = self._district_feat_df["district_ar"].tolist() print(f" District stats: {len(self._district_stats)} districts | predict features: {len(feat_cols)} | centroid-corrected: {_fixed}") # ── Lookup ──────────────────────────────────────────────────────────────── def lookup(self, lat: float, lon: float) -> dict: """Return all Riyadh spatial features for a (lat, lon) coordinate.""" feats = {} pt = np.array([[lat, lon]]) pt_rad = np.radians(pt) r_500 = 500.0 / 6_371_000 r_1km = 1000.0 / 6_371_000 # Metro d, idx = self._metro_tree.query(pt, k=1) feats["dist_metro_m"] = float(d[0]) * 111_000 feats["log_dist_metro_m"] = float(np.log1p(feats["dist_metro_m"])) feats["nearest_metro_line_num"] = int(_LINE_ORDER.get(self._metro_lines[int(idx[0])], 0)) feats["nearest_metro_type_cd"] = int(self._metro_types[int(idx[0])]) feats["metro_stations_1km"] = int(self._metro_ball.query_radius(pt_rad, r=r_1km, count_only=True)[0]) d1, _ = self._line1_tree.query(pt, k=1) feats["dist_metro_line1_m"] = float(d1[0]) * 111_000 # Bus d_b, _ = self._bus_tree.query(pt, k=1) feats["dist_bus_m"] = float(d_b[0]) * 111_000 feats["log_dist_bus_m"] = float(np.log1p(feats["dist_bus_m"])) feats["bus_stops_500m"] = int(self._bus_ball.query_radius(pt_rad, r=r_500, count_only=True)[0]) feats["brt_stops_500m"] = int( self._shelter_ball.query_radius(pt_rad, r=r_500, count_only=True)[0] if self._shelter_ball else 0 ) # Intersections d_i, _ = self._int_tree.query(pt, k=1) feats["dist_major_intersection_m"] = float(d_i[0]) * 111_000 feats["log_dist_intersection_m"] = float(np.log1p(feats["dist_major_intersection_m"])) feats["intersections_1km"] = int(self._int_ball.query_radius(pt_rad, r=r_1km, count_only=True)[0]) feats["intersections_500m"] = int(self._int_ball.query_radius(pt_rad, r=r_500, count_only=True)[0]) # Commercial feats["commercial_count_1km"] = int(self._comm_ball_all.query_radius(pt_rad, r=r_1km, count_only=True)[0]) for bname, bt in self._comm_balls.items(): feats[f"{bname}_count_1km"] = int(bt.query_radius(pt_rad, r=r_1km, count_only=True)[0]) if bt else 0 feats["commercial_density_score"] = ( feats.get("hypermarket_count_1km", 0) * 3 + feats.get("supermarket_count_1km", 0) * 2 + feats.get("bank_count_1km", 0) + feats.get("restaurant_count_1km", 0) + feats.get("hotel_count_1km", 0) ) # QoL POIs (mosque, mall, school, hospital, park, entertainment) for poi_name, tree in self._poi_trees.items(): d_poi, _ = tree.query(pt, k=1) dist_m = float(d_poi[0]) * 111_000 feats[f"dist_{poi_name}_m"] = dist_m feats[f"log_dist_{poi_name}_m"] = float(np.log1p(dist_m)) feats[f"{poi_name}_count_500m"] = int( self._poi_balls[poi_name].query_radius(pt_rad, r=r_500, count_only=True)[0] ) # Air quality (IDW from 2 nearest stations) d_aq, idx_aq = self._aq_tree.query(pt, k=min(2, len(self._aq_arr))) d_m = d_aq.ravel() * 111_000 d_m = np.where(d_m < 1, 1, d_m) weights = 1.0 / d_m for comp in self._aq_components: if comp in self._aq_means_df.columns: vals = self._aq_means_df.iloc[idx_aq.ravel()][comp].values feats[f"{comp.lower()}_nearest_mean"] = float(np.average(vals, weights=weights)) feats["dist_air_station_m"] = float(d_aq.ravel()[0]) * 111_000 return feats def get_district_stats(self) -> dict: """Return dict of district_ar → per-district metric medians.""" return self._district_stats def predict_features(self, lat: float, lon: float, property_type: str, year: int, quarter: int) -> dict: """ Build a complete 72-feature dict for Riyadh model prediction. Combines live spatial lookup + district medians + type flags + macro. """ feats: dict = {} # 1. Type flags feats["is_apartment"] = int(property_type == "شقة") feats["is_villa"] = int(property_type == "فيلا") feats["is_residential_plot"] = int(property_type == "قطعة أرض-سكنى") feats["is_building"] = int(property_type == "عمارة") # 2. Time features feats["sale_year"] = float(year) feats["sale_quarter_sin"] = float(np.sin(2 * np.pi * quarter / 4)) feats["sale_quarter_cos"] = float(np.cos(2 * np.pi * quarter / 4)) quarter_id = year * 10 + quarter # 3. Nearest district baseline features _matched_district: str = "" if self._district_centroid_tree is not None and self._district_feat_df is not None: _, idx = self._district_centroid_tree.query([[lat, lon]], k=1) row = self._district_feat_df.iloc[int(idx[0])].to_dict() _matched_district = str(row.get("district_ar", "")) feats.update({ k: v for k, v in row.items() if k != "district_ar" and (not isinstance(v, float) or not np.isnan(v)) }) # 4. Live spatial features from the exact lat/lon (override district centroid) live = self.lookup(lat, lon) feats.update(live) # 5. Log deed count (use district median as a proxy for a typical transaction) feats.setdefault("log_deed_count", float(np.log1p(5))) # 6. District target encoding — look up per-district medians. # These were excluded from feat_cols to avoid leakage during training but # are valid and important at inference time (they carry district price signal). _GLOBAL_ENC = 7.8 _GLOBAL_TYPE = 7.8 if _matched_district and hasattr(self, "_district_encoded_map"): _enc = self._district_encoded_map.get(_matched_district, {}) feats["district_encoded"] = float(_enc.get("district_encoded", _GLOBAL_ENC)) feats["district_type_encoded"] = float(_enc.get("district_type_encoded", _GLOBAL_TYPE)) feats["district_recent_encoded"] = float(_enc.get("district_recent_encoded", _GLOBAL_ENC)) # Always use apartment-specific encoding regardless of property type — # matches training where district_apt_encoded is a district-level feature (same value for all rows in district) feats["district_apt_encoded"] = float(_enc.get("district_apt_encoded", _GLOBAL_ENC)) feats["district_apt_recent_encoded"] = float(_enc.get("district_apt_recent_encoded", _GLOBAL_ENC)) else: feats.setdefault("district_encoded", _GLOBAL_ENC) feats.setdefault("district_type_encoded", _GLOBAL_TYPE) feats.setdefault("district_apt_encoded", _GLOBAL_ENC) feats.setdefault("district_recent_encoded", _GLOBAL_ENC) feats.setdefault("district_apt_recent_encoded", _GLOBAL_ENC) # 7. Connectivity score (recalculate from live spatial if scaler params available) # Keep the district median value from step 3 unless live override is possible feats.setdefault("riyadh_connectivity_score", 50.0) return feats