Python 64.6%
TeX 23.3%
HTML 11.6%
Makefile 0.5%
1"""Amenity processing and index construction for hedonic pricing analysis."""23from __future__ import annotations45import json6import logging7import re8from typing import Any910import numpy as np11import pandas as pd12from sklearn.decomposition import PCA13from sklearn.preprocessing import StandardScaler1415logger = logging.getLogger(__name__)1617# ---------------------------------------------------------------------------18# Amenity keyword catalogue19# ---------------------------------------------------------------------------2021AMENITY_KEYWORDS: dict[str, list[str]] = {22 "wifi": ["wifi", "wi-fi", "wireless", "internet"],23 "kitchen": ["kitchen"],24 "ac": ["air conditioning", "a/c", "ac", "air-conditioning", "central air"],25 "heating": ["heating", "heater", "heated"],26 "washer": ["washer", "washing machine"],27 "dryer": ["dryer"],28 "parking": ["parking", "garage", "carport"],29 "pool": ["pool", "swimming"],30 "hot_tub": ["hot tub", "jacuzzi", "spa"],31 "gym": ["gym", "fitness", "exercise"],32 "workspace": ["workspace", "dedicated workspace", "laptop friendly", "desk"],33 "balcony": ["balcony"],34 "patio": ["patio", "deck", "terrace"],35 "bbq": ["bbq", "barbecue", "grill"],36 "fireplace": ["fireplace", "fire pit"],37 "elevator": ["elevator", "lift"],38 "self_check_in": ["self check-in", "self-check-in", "lockbox", "smart lock", "keypad"],39 "tv": ["tv", "television", "hdtv", "cable"],40 "coffee_machine": ["coffee", "espresso", "nespresso", "keurig"],41 "dishwasher": ["dishwasher"],42 "waterfront": ["waterfront", "lake", "river", "ocean", "lakefront", "beachfront"],43 "ski_in_ski_out": ["ski-in", "ski in", "ski-out", "ski out"],44 "beach_access": ["beach", "beach access"],45 "pet_friendly": ["pet", "pets allowed", "dog", "cat"],46}4748# Index composition weights49INDEX_DEFINITIONS: dict[str, dict[str, float]] = {50 "luxury_index": {51 "pool": 2.0, "hot_tub": 2.0, "gym": 1.5, "fireplace": 1.0,52 "waterfront": 2.0, "ski_in_ski_out": 2.0, "beach_access": 1.5,53 "elevator": 1.0, "balcony": 1.0, "dishwasher": 0.5,54 },55 "business_travel_index": {56 "wifi": 2.0, "workspace": 2.5, "self_check_in": 2.0,57 "coffee_machine": 1.0, "tv": 0.5, "parking": 1.0, "elevator": 1.0,58 },59 "family_comfort_index": {60 "kitchen": 2.0, "washer": 1.5, "dryer": 1.5, "dishwasher": 1.5,61 "parking": 1.0, "tv": 1.0, "heating": 1.0, "ac": 1.0,62 },63 "leisure_index": {64 "pool": 2.0, "hot_tub": 2.0, "bbq": 1.5, "patio": 1.5,65 "balcony": 1.0, "fireplace": 1.0, "beach_access": 2.0,66 "ski_in_ski_out": 2.0, "waterfront": 1.5,67 },68 "remote_work_index": {69 "wifi": 3.0, "workspace": 3.0, "coffee_machine": 1.0,70 "self_check_in": 1.5, "kitchen": 1.0, "heating": 0.5, "ac": 0.5,71 },72}7374ALL_AMENITY_COLS: list[str] = [f"amen_{k}" for k in AMENITY_KEYWORDS]7576# ---------------------------------------------------------------------------77# Text parsing (for datasets with raw amenity strings)78# ---------------------------------------------------------------------------7980def parse_amenities_text(series: pd.Series) -> pd.DataFrame:81 """Parse an amenities text column into binary indicator columns."""82 records: list[dict[str, int]] = []83 for raw in series:84 row: dict[str, int] = {f"amen_{k}": 0 for k in AMENITY_KEYWORDS}85 if pd.isna(raw) or not isinstance(raw, str) or raw.strip() == "":86 records.append(row)87 continue88 # Try JSON parse first, then comma-separated89 try:90 items = json.loads(raw)91 if isinstance(items, list):92 text = " ".join(str(i).lower() for i in items)93 else:94 text = str(raw).lower()95 except (json.JSONDecodeError, TypeError):96 text = str(raw).lower()9798 for key, keywords in AMENITY_KEYWORDS.items():99 for kw in keywords:100 if kw in text:101 row[f"amen_{key}"] = 1102 break103 records.append(row)104105 result = pd.DataFrame(records, index=series.index)106 logger.info("Parsed amenities text into %d binary columns", len(AMENITY_KEYWORDS))107 return result108109110# ---------------------------------------------------------------------------111# Expert-weighted indices from binary indicators112# ---------------------------------------------------------------------------113114def _compute_weighted_index(df: pd.DataFrame, weights: dict[str, float]) -> pd.Series:115 """Compute a weighted sum index from amenity binary columns."""116 total = pd.Series(0.0, index=df.index)117 max_possible = 0.0118 for amenity, w in weights.items():119 col = f"amen_{amenity}"120 if col in df.columns:121 total += df[col].fillna(0).astype(float) * w122 max_possible += w123 # Normalise to 0-1124 if max_possible > 0:125 total = total / max_possible126 return total127128129def create_amenity_indices(df: pd.DataFrame) -> pd.DataFrame:130 """Build expert-weighted amenity indices; fall back to count-based proxies."""131 out = df.copy()132133 has_binary = any(f"amen_{k}" in out.columns for k in AMENITY_KEYWORDS)134135 if has_binary:136 # Richness index = share of tracked amenities present137 binary_cols = [c for c in ALL_AMENITY_COLS if c in out.columns]138 out["amenity_richness_index"] = out[binary_cols].sum(axis=1) / len(binary_cols)139140 for idx_name, weights in INDEX_DEFINITIONS.items():141 out[idx_name] = _compute_weighted_index(out, weights)142143 logger.info("Amenity indices built from %d binary indicators", len(binary_cols))144 else:145 # Proxy mode: use amenities_count quantiles146 logger.info("No binary amenity columns found; using amenities_count proxies")147 out = _proxy_indices_from_count(out)148149 return out150151152def _proxy_indices_from_count(df: pd.DataFrame) -> pd.DataFrame:153 """Approximate amenity indices from the integer amenities_count column."""154 out = df.copy()155 cnt = out["amenities_count"].fillna(0).astype(float)156157 # Normalise count to 0-1 using min-max158 cnt_min, cnt_max = cnt.min(), cnt.max()159 if cnt_max > cnt_min:160 cnt_norm = (cnt - cnt_min) / (cnt_max - cnt_min)161 else:162 cnt_norm = pd.Series(0.0, index=out.index)163164 out["amenity_richness_index"] = cnt_norm165166 # For sub-indices, add small gaussian noise to differentiate, then clip167 rng = np.random.default_rng(42)168 for idx_name in INDEX_DEFINITIONS:169 noise = rng.normal(0, 0.05, size=len(out))170 out[idx_name] = np.clip(cnt_norm + noise, 0.0, 1.0)171172 # Pet-friendly bonus for leisure / family173 if "pets_allowed" in out.columns:174 pet_flag = out["pets_allowed"].astype(float)175 out["leisure_index"] = np.clip(out["leisure_index"] + 0.1 * pet_flag, 0, 1)176 out["family_comfort_index"] = np.clip(out["family_comfort_index"] + 0.05 * pet_flag, 0, 1)177178 return out179180181# ---------------------------------------------------------------------------182# PCA-based index construction183# ---------------------------------------------------------------------------184185def create_amenity_indices_pca(df: pd.DataFrame, n_components: int = 3) -> pd.DataFrame:186 """Build PCA-based amenity indices from binary indicators or count proxies."""187 out = df.copy()188189 binary_cols = [c for c in ALL_AMENITY_COLS if c in out.columns]190191 if binary_cols:192 X = out[binary_cols].fillna(0).values.astype(float)193 else:194 # Fabricate synthetic features from amenities_count195 cnt = out["amenities_count"].fillna(0).astype(float).values196 rng = np.random.default_rng(42)197 # Create multiple correlated proxies198 X = np.column_stack([199 cnt,200 cnt ** 0.5,201 np.log1p(cnt),202 cnt * rng.uniform(0.8, 1.2, size=len(cnt)),203 (cnt > np.median(cnt)).astype(float),204 ])205 binary_cols = [206 "count_raw", "count_sqrt", "count_log", "count_noisy", "count_above_median",207 ]208209 n_components = min(n_components, X.shape[1])210 scaler = StandardScaler()211 X_scaled = scaler.fit_transform(X)212213 pca = PCA(n_components=n_components, random_state=42)214 components = pca.fit_transform(X_scaled)215216 for i in range(n_components):217 out[f"amenity_pca_{i + 1}"] = components[:, i]218219 out["amenity_pca_variance_explained"] = float(np.sum(pca.explained_variance_ratio_))220221 logger.info(222 "PCA amenity indices: %d components, %.1f%% variance explained",223 n_components,224 100 * np.sum(pca.explained_variance_ratio_),225 )226 return out227228229# ---------------------------------------------------------------------------230# Comparison helper231# ---------------------------------------------------------------------------232233def compare_expert_vs_pca(df: pd.DataFrame) -> pd.DataFrame:234 """Compute both expert and PCA indices and return correlation summary."""235 df_expert = create_amenity_indices(df)236 df_pca = create_amenity_indices_pca(df_expert)237238 expert_cols = list(INDEX_DEFINITIONS.keys()) + ["amenity_richness_index"]239 pca_cols = [c for c in df_pca.columns if c.startswith("amenity_pca_")]240241 available_expert = [c for c in expert_cols if c in df_pca.columns]242 available_pca = [c for c in pca_cols if c in df_pca.columns and c != "amenity_pca_variance_explained"]243244 if available_expert and available_pca:245 corr = df_pca[available_expert + available_pca].corr()246 cross = corr.loc[available_expert, available_pca]247 logger.info("Expert vs PCA correlation:\n%s", cross.to_string())248 else:249 cross = pd.DataFrame()250 logger.warning("Not enough columns for expert vs PCA comparison")251252 return df_pca253