# Author: Simon-Pierre Boucher — contact@spboucher.ai """ 04_merge_data.py ================ Merge Airbnb and rental listing data using three linkage strategies: 1. Spatial buffer merge (Haversine distance at 250m, 500m, 1km, 2km) 2. City/borough-level aggregation 3. Combined analysis file Inputs: data/processed/airbnb_clean.parquet data/processed/rent_clean.parquet Outputs: data/processed/merged_spatial.parquet data/processed/merged_neighborhood.parquet data/processed/merged_analysis.parquet """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import numpy as np import pandas as pd from src.config import ( AIRBNB_CLEAN, RENT_CLEAN, MERGED_SPATIAL, MERGED_NEIGHBORHOOD, MERGED_ANALYSIS, BUFFER_KM, CHUNK_SIZE, require, ) from src.geo import haversine_matrix PROCESSED_HINT = ("Run scripts/02_clean_airbnb.py and scripts/03_clean_rent.py " "first (requires the raw data), or restore the committed " "parquet files in data/processed/.") # =================================================================== # Strategy 1 — spatial buffer merge # =================================================================== def spatial_buffer_merge(rent: pd.DataFrame, airbnb: pd.DataFrame) -> pd.DataFrame: """ For each rental listing, compute Airbnb exposure metrics within multiple distance buffers. """ n_rent = len(rent) n_airbnb = len(airbnb) # Pre-extract numpy arrays for speed rent_lat = rent["lat"].values.astype(np.float64) rent_lon = rent["lon"].values.astype(np.float64) airbnb_lat = airbnb["lat"].values.astype(np.float64) airbnb_lon = airbnb["lon"].values.astype(np.float64) airbnb_price = airbnb["price_numeric"].values.astype(np.float64) airbnb_entire = airbnb["is_entire_home"].values.astype(np.float64) airbnb_rating = airbnb["rating"].values.astype(np.float64) airbnb_superhost = airbnb["is_superhost"].values.astype(np.float64) # Initialise result columns result_cols = {} for buf in BUFFER_KM: tag = f"{int(buf * 1000)}m" result_cols[f"airbnb_count_{tag}"] = np.zeros(n_rent, dtype=np.int32) result_cols[f"airbnb_density_{tag}"] = np.zeros(n_rent, dtype=np.float64) result_cols[f"mean_airbnb_price_{tag}"] = np.full(n_rent, np.nan, dtype=np.float64) result_cols[f"share_entire_home_{tag}"] = np.full(n_rent, np.nan, dtype=np.float64) result_cols[f"mean_rating_{tag}"] = np.full(n_rent, np.nan, dtype=np.float64) result_cols[f"superhost_share_{tag}"] = np.full(n_rent, np.nan, dtype=np.float64) n_chunks = int(np.ceil(n_rent / CHUNK_SIZE)) print(f" Processing {n_rent} rental listings in {n_chunks} chunks " f"against {n_airbnb} Airbnb listings ...") for c in range(n_chunks): start = c * CHUNK_SIZE end = min(start + CHUNK_SIZE, n_rent) idx = slice(start, end) # Distance matrix: (chunk_size, n_airbnb) dist = haversine_matrix(rent_lat[idx], rent_lon[idx], airbnb_lat, airbnb_lon) for buf in BUFFER_KM: tag = f"{int(buf * 1000)}m" within = dist <= buf # boolean mask (chunk, n_airbnb) counts = within.sum(axis=1) result_cols[f"airbnb_count_{tag}"][idx] = counts area = np.pi * buf ** 2 result_cols[f"airbnb_density_{tag}"][idx] = counts / area # For each rental in the chunk, compute means over nearby Airbnb for i_local in range(end - start): mask = within[i_local] if mask.sum() == 0: continue i_global = start + i_local prices = airbnb_price[mask] valid_prices = prices[~np.isnan(prices)] if len(valid_prices) > 0: result_cols[f"mean_airbnb_price_{tag}"][i_global] = valid_prices.mean() result_cols[f"share_entire_home_{tag}"][i_global] = airbnb_entire[mask].mean() ratings = airbnb_rating[mask] valid_ratings = ratings[~np.isnan(ratings)] if len(valid_ratings) > 0: result_cols[f"mean_rating_{tag}"][i_global] = valid_ratings.mean() result_cols[f"superhost_share_{tag}"][i_global] = airbnb_superhost[mask].mean() if (c + 1) % 20 == 0 or (c + 1) == n_chunks: print(f" Chunk {c + 1}/{n_chunks} done.") # Attach to rent DataFrame spatial = rent.copy() for col, arr in result_cols.items(): spatial[col] = arr return spatial # =================================================================== # Strategy 2 — city / borough aggregation # =================================================================== def city_borough_aggregation(rent: pd.DataFrame, airbnb: pd.DataFrame) -> pd.DataFrame: """ Aggregate Airbnb metrics at the city level (and borough level for Montreal) and merge onto rental data. """ # --- City-level aggregation --- city_key = "city_clean" if "city_clean" in airbnb.columns else "city" city_agg = airbnb.groupby(city_key).agg( airbnb_count_city=("price_numeric", "size"), mean_airbnb_price_city=("price_numeric", "mean"), median_airbnb_price_city=("price_numeric", "median"), share_entire_home_city=("is_entire_home", "mean"), mean_reviews_city=("num_reviews", "mean"), mean_rating_city=("rating", "mean"), ).reset_index() # Rename the grouping column for merging city_agg = city_agg.rename(columns={city_key: "city"}) # Count rentals per city to compute density rent_city_counts = rent.groupby("city").size().reset_index(name="_n_rent_city") city_agg = city_agg.merge(rent_city_counts, on="city", how="left") city_agg["airbnb_density_per_1000_rentals_city"] = np.where( city_agg["_n_rent_city"] > 0, city_agg["airbnb_count_city"] / city_agg["_n_rent_city"] * 1000, np.nan, ) city_agg = city_agg.drop(columns=["_n_rent_city"]) # Merge onto rent (the rent file's 'city' column aligns with the # standardized Airbnb city names) neighborhood = rent.copy() neighborhood = neighborhood.merge(city_agg, on="city", how="left") # --- Borough-level aggregation (Montreal only) --- if "borough" in rent.columns: # Identify Montreal Airbnb listings based on city name mtl_variants = ["montreal", "montréal", "mtl"] if city_key in airbnb.columns: airbnb_city_lower = airbnb[city_key].str.lower().str.strip() else: airbnb_city_lower = airbnb["city"].str.lower().str.strip() airbnb_mtl = airbnb[airbnb_city_lower.isin(mtl_variants)].copy() if len(airbnb_mtl) > 0 and "borough" not in airbnb_mtl.columns: # Airbnb data might not have borough info; skip borough merge print(" Note: Airbnb data does not have 'borough' column; " "borough-level aggregation skipped.") elif len(airbnb_mtl) > 0 and "borough" in airbnb_mtl.columns: borough_agg = airbnb_mtl.groupby("borough").agg( airbnb_count_borough=("price_numeric", "size"), mean_airbnb_price_borough=("price_numeric", "mean"), median_airbnb_price_borough=("price_numeric", "median"), share_entire_home_borough=("is_entire_home", "mean"), mean_reviews_borough=("num_reviews", "mean"), mean_rating_borough=("rating", "mean"), ).reset_index() rent_borough_counts = ( rent[rent["borough"].notna()] .groupby("borough") .size() .reset_index(name="_n_rent_borough") ) borough_agg = borough_agg.merge(rent_borough_counts, on="borough", how="left") borough_agg["airbnb_density_per_1000_rentals_borough"] = np.where( borough_agg["_n_rent_borough"] > 0, borough_agg["airbnb_count_borough"] / borough_agg["_n_rent_borough"] * 1000, np.nan, ) borough_agg = borough_agg.drop(columns=["_n_rent_borough"]) neighborhood = neighborhood.merge(borough_agg, on="borough", how="left") return neighborhood # =================================================================== # Main # =================================================================== def main(): print("=" * 70) print("04 MERGE DATA") print("=" * 70) # ------------------------------------------------------------------ # Load data # ------------------------------------------------------------------ print("\n[1] Loading cleaned data ...") airbnb = pd.read_parquet(require(AIRBNB_CLEAN, PROCESSED_HINT)) rent = pd.read_parquet(require(RENT_CLEAN, PROCESSED_HINT)) # Ensure coordinate column names are consistent if "long" in airbnb.columns and "lon" not in airbnb.columns: airbnb = airbnb.rename(columns={"long": "lon"}) print(f" Airbnb : {airbnb.shape[0]:,} rows x {airbnb.shape[1]} cols") print(f" Rent : {rent.shape[0]:,} rows x {rent.shape[1]} cols") # Drop rows with missing coordinates airbnb_valid = airbnb.dropna(subset=["lat", "lon"]) rent_valid = rent.dropna(subset=["lat", "lon"]) print(f" Airbnb with valid coords: {len(airbnb_valid):,}") print(f" Rent with valid coords: {len(rent_valid):,}") # ------------------------------------------------------------------ # Strategy 1: Spatial buffer merge # ------------------------------------------------------------------ print("\n[2] Strategy 1 — Spatial buffer merge ...") spatial = spatial_buffer_merge(rent_valid, airbnb_valid) for buf in BUFFER_KM: tag = f"{int(buf * 1000)}m" col = f"airbnb_count_{tag}" print(f" Buffer {tag}: " f"mean count = {spatial[col].mean():.2f}, " f"median = {spatial[col].median():.0f}, " f"max = {spatial[col].max()}") spatial.to_parquet(MERGED_SPATIAL, index=False) print(f" Saved: {MERGED_SPATIAL}") # ------------------------------------------------------------------ # Strategy 2: City/borough aggregation # ------------------------------------------------------------------ print("\n[3] Strategy 2 — City/borough-level aggregation ...") neighborhood = city_borough_aggregation(rent_valid, airbnb_valid) city_cols = [c for c in neighborhood.columns if c.endswith("_city")] if city_cols: print(f" City-level columns added: {city_cols}") print(f" Rentals with city match: " f"{neighborhood['airbnb_count_city'].notna().sum():,} / {len(neighborhood):,}") borough_cols = [c for c in neighborhood.columns if c.endswith("_borough")] if borough_cols: print(f" Borough-level columns added: {borough_cols}") print(f" Rentals with borough match: " f"{neighborhood['airbnb_count_borough'].notna().sum():,} / {len(neighborhood):,}") neighborhood.to_parquet(MERGED_NEIGHBORHOOD, index=False) print(f" Saved: {MERGED_NEIGHBORHOOD}") # ------------------------------------------------------------------ # Combined analysis file # ------------------------------------------------------------------ print("\n[4] Creating combined analysis file ...") # Start from spatial (which already has rent + buffer vars) and add # neighborhood-level columns that are not already present. Both frames # are row-aligned because they started from the same rent_valid. spatial_cols = set(spatial.columns) extra_cols = [c for c in neighborhood.columns if c not in spatial_cols] analysis = spatial.copy() for col in extra_cols: analysis[col] = neighborhood[col].values analysis.to_parquet(MERGED_ANALYSIS, index=False) print(f" Saved: {MERGED_ANALYSIS}") print(f" Final shape: {analysis.shape[0]:,} rows x {analysis.shape[1]} cols") # ------------------------------------------------------------------ # Summary diagnostics # ------------------------------------------------------------------ print("\n" + "=" * 70) print("MERGE DIAGNOSTICS") print("=" * 70) print(f" Total rental listings: {len(rent):>8,}") print(f" With valid coordinates: {len(rent_valid):>8,}") print(f" Total Airbnb listings: {len(airbnb):>8,}") print(f" With valid coordinates: {len(airbnb_valid):>8,}") print() for buf in BUFFER_KM: tag = f"{int(buf * 1000)}m" col = f"airbnb_count_{tag}" n_zero = (analysis[col] == 0).sum() n_nonzero = (analysis[col] > 0).sum() print(f" Buffer {tag:>5s}: {n_nonzero:,} rentals have >= 1 Airbnb nearby, " f"{n_zero:,} have 0") print() print(" Columns in merged_analysis.parquet:") for i, col in enumerate(analysis.columns): print(f" {i + 1:3d}. {col}") print("\nDone.\n") if __name__ == "__main__": main()