"""Amenity processing and index construction for hedonic pricing analysis.""" from __future__ import annotations import json import logging import re from typing import Any import numpy as np import pandas as pd from sklearn.decomposition import PCA from sklearn.preprocessing import StandardScaler logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Amenity keyword catalogue # --------------------------------------------------------------------------- AMENITY_KEYWORDS: dict[str, list[str]] = { "wifi": ["wifi", "wi-fi", "wireless", "internet"], "kitchen": ["kitchen"], "ac": ["air conditioning", "a/c", "ac", "air-conditioning", "central air"], "heating": ["heating", "heater", "heated"], "washer": ["washer", "washing machine"], "dryer": ["dryer"], "parking": ["parking", "garage", "carport"], "pool": ["pool", "swimming"], "hot_tub": ["hot tub", "jacuzzi", "spa"], "gym": ["gym", "fitness", "exercise"], "workspace": ["workspace", "dedicated workspace", "laptop friendly", "desk"], "balcony": ["balcony"], "patio": ["patio", "deck", "terrace"], "bbq": ["bbq", "barbecue", "grill"], "fireplace": ["fireplace", "fire pit"], "elevator": ["elevator", "lift"], "self_check_in": ["self check-in", "self-check-in", "lockbox", "smart lock", "keypad"], "tv": ["tv", "television", "hdtv", "cable"], "coffee_machine": ["coffee", "espresso", "nespresso", "keurig"], "dishwasher": ["dishwasher"], "waterfront": ["waterfront", "lake", "river", "ocean", "lakefront", "beachfront"], "ski_in_ski_out": ["ski-in", "ski in", "ski-out", "ski out"], "beach_access": ["beach", "beach access"], "pet_friendly": ["pet", "pets allowed", "dog", "cat"], } # Index composition weights INDEX_DEFINITIONS: dict[str, dict[str, float]] = { "luxury_index": { "pool": 2.0, "hot_tub": 2.0, "gym": 1.5, "fireplace": 1.0, "waterfront": 2.0, "ski_in_ski_out": 2.0, "beach_access": 1.5, "elevator": 1.0, "balcony": 1.0, "dishwasher": 0.5, }, "business_travel_index": { "wifi": 2.0, "workspace": 2.5, "self_check_in": 2.0, "coffee_machine": 1.0, "tv": 0.5, "parking": 1.0, "elevator": 1.0, }, "family_comfort_index": { "kitchen": 2.0, "washer": 1.5, "dryer": 1.5, "dishwasher": 1.5, "parking": 1.0, "tv": 1.0, "heating": 1.0, "ac": 1.0, }, "leisure_index": { "pool": 2.0, "hot_tub": 2.0, "bbq": 1.5, "patio": 1.5, "balcony": 1.0, "fireplace": 1.0, "beach_access": 2.0, "ski_in_ski_out": 2.0, "waterfront": 1.5, }, "remote_work_index": { "wifi": 3.0, "workspace": 3.0, "coffee_machine": 1.0, "self_check_in": 1.5, "kitchen": 1.0, "heating": 0.5, "ac": 0.5, }, } ALL_AMENITY_COLS: list[str] = [f"amen_{k}" for k in AMENITY_KEYWORDS] # --------------------------------------------------------------------------- # Text parsing (for datasets with raw amenity strings) # --------------------------------------------------------------------------- def parse_amenities_text(series: pd.Series) -> pd.DataFrame: """Parse an amenities text column into binary indicator columns.""" records: list[dict[str, int]] = [] for raw in series: row: dict[str, int] = {f"amen_{k}": 0 for k in AMENITY_KEYWORDS} if pd.isna(raw) or not isinstance(raw, str) or raw.strip() == "": records.append(row) continue # Try JSON parse first, then comma-separated try: items = json.loads(raw) if isinstance(items, list): text = " ".join(str(i).lower() for i in items) else: text = str(raw).lower() except (json.JSONDecodeError, TypeError): text = str(raw).lower() for key, keywords in AMENITY_KEYWORDS.items(): for kw in keywords: if kw in text: row[f"amen_{key}"] = 1 break records.append(row) result = pd.DataFrame(records, index=series.index) logger.info("Parsed amenities text into %d binary columns", len(AMENITY_KEYWORDS)) return result # --------------------------------------------------------------------------- # Expert-weighted indices from binary indicators # --------------------------------------------------------------------------- def _compute_weighted_index(df: pd.DataFrame, weights: dict[str, float]) -> pd.Series: """Compute a weighted sum index from amenity binary columns.""" total = pd.Series(0.0, index=df.index) max_possible = 0.0 for amenity, w in weights.items(): col = f"amen_{amenity}" if col in df.columns: total += df[col].fillna(0).astype(float) * w max_possible += w # Normalise to 0-1 if max_possible > 0: total = total / max_possible return total def create_amenity_indices(df: pd.DataFrame) -> pd.DataFrame: """Build expert-weighted amenity indices; fall back to count-based proxies.""" out = df.copy() has_binary = any(f"amen_{k}" in out.columns for k in AMENITY_KEYWORDS) if has_binary: # Richness index = share of tracked amenities present binary_cols = [c for c in ALL_AMENITY_COLS if c in out.columns] out["amenity_richness_index"] = out[binary_cols].sum(axis=1) / len(binary_cols) for idx_name, weights in INDEX_DEFINITIONS.items(): out[idx_name] = _compute_weighted_index(out, weights) logger.info("Amenity indices built from %d binary indicators", len(binary_cols)) else: # Proxy mode: use amenities_count quantiles logger.info("No binary amenity columns found; using amenities_count proxies") out = _proxy_indices_from_count(out) return out def _proxy_indices_from_count(df: pd.DataFrame) -> pd.DataFrame: """Approximate amenity indices from the integer amenities_count column.""" out = df.copy() cnt = out["amenities_count"].fillna(0).astype(float) # Normalise count to 0-1 using min-max cnt_min, cnt_max = cnt.min(), cnt.max() if cnt_max > cnt_min: cnt_norm = (cnt - cnt_min) / (cnt_max - cnt_min) else: cnt_norm = pd.Series(0.0, index=out.index) out["amenity_richness_index"] = cnt_norm # For sub-indices, add small gaussian noise to differentiate, then clip rng = np.random.default_rng(42) for idx_name in INDEX_DEFINITIONS: noise = rng.normal(0, 0.05, size=len(out)) out[idx_name] = np.clip(cnt_norm + noise, 0.0, 1.0) # Pet-friendly bonus for leisure / family if "pets_allowed" in out.columns: pet_flag = out["pets_allowed"].astype(float) out["leisure_index"] = np.clip(out["leisure_index"] + 0.1 * pet_flag, 0, 1) out["family_comfort_index"] = np.clip(out["family_comfort_index"] + 0.05 * pet_flag, 0, 1) return out # --------------------------------------------------------------------------- # PCA-based index construction # --------------------------------------------------------------------------- def create_amenity_indices_pca(df: pd.DataFrame, n_components: int = 3) -> pd.DataFrame: """Build PCA-based amenity indices from binary indicators or count proxies.""" out = df.copy() binary_cols = [c for c in ALL_AMENITY_COLS if c in out.columns] if binary_cols: X = out[binary_cols].fillna(0).values.astype(float) else: # Fabricate synthetic features from amenities_count cnt = out["amenities_count"].fillna(0).astype(float).values rng = np.random.default_rng(42) # Create multiple correlated proxies X = np.column_stack([ cnt, cnt ** 0.5, np.log1p(cnt), cnt * rng.uniform(0.8, 1.2, size=len(cnt)), (cnt > np.median(cnt)).astype(float), ]) binary_cols = [ "count_raw", "count_sqrt", "count_log", "count_noisy", "count_above_median", ] n_components = min(n_components, X.shape[1]) scaler = StandardScaler() X_scaled = scaler.fit_transform(X) pca = PCA(n_components=n_components, random_state=42) components = pca.fit_transform(X_scaled) for i in range(n_components): out[f"amenity_pca_{i + 1}"] = components[:, i] out["amenity_pca_variance_explained"] = float(np.sum(pca.explained_variance_ratio_)) logger.info( "PCA amenity indices: %d components, %.1f%% variance explained", n_components, 100 * np.sum(pca.explained_variance_ratio_), ) return out # --------------------------------------------------------------------------- # Comparison helper # --------------------------------------------------------------------------- def compare_expert_vs_pca(df: pd.DataFrame) -> pd.DataFrame: """Compute both expert and PCA indices and return correlation summary.""" df_expert = create_amenity_indices(df) df_pca = create_amenity_indices_pca(df_expert) expert_cols = list(INDEX_DEFINITIONS.keys()) + ["amenity_richness_index"] pca_cols = [c for c in df_pca.columns if c.startswith("amenity_pca_")] available_expert = [c for c in expert_cols if c in df_pca.columns] available_pca = [c for c in pca_cols if c in df_pca.columns and c != "amenity_pca_variance_explained"] if available_expert and available_pca: corr = df_pca[available_expert + available_pca].corr() cross = corr.loc[available_expert, available_pca] logger.info("Expert vs PCA correlation:\n%s", cross.to_string()) else: cross = pd.DataFrame() logger.warning("Not enough columns for expert vs PCA comparison") return df_pca