Python 64.6%
TeX 23.3%
HTML 11.6%
Makefile 0.5%
1"""2data_loader.py -- Data ingestion and validation for hedonic pricing analysis.34Provides a uniform interface for loading the cleaned sample CSV, full5Inside Airbnb extracts (listings, calendar, reviews, neighbourhoods),6and any pre-processed interim artefacts. Every loader gracefully7handles missing files and returns ``None`` when data is unavailable.8"""910from __future__ import annotations1112import logging13from pathlib import Path14from typing import Any, Dict, List, Optional, Tuple, Union1516import numpy as np17import pandas as pd1819from src.config import (20 CLEANED_SAMPLE_CSV,21 LISTINGS_CSV_GZ,22 CALENDAR_CSV_GZ,23 REVIEWS_CSV_GZ,24 NEIGHBOURHOODS_CSV,25 NEIGHBOURHOODS_GEOJSON,26 DATA_INTERIM,27 DATA_PROCESSED,28 QUEBEC_LAT_RANGE,29 QUEBEC_LON_RANGE,30 LOG_LEVEL,31 LOG_FORMAT,32 LOG_DATE_FORMAT,33)3435# ---------------------------------------------------------------------------36# Logger37# ---------------------------------------------------------------------------3839logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT, level=LOG_LEVEL)40logger = logging.getLogger(__name__)4142# ---------------------------------------------------------------------------43# Schema expectations (column name -> expected dtype family)44# ---------------------------------------------------------------------------4546_EXPECTED_COLUMNS_SAMPLE: Dict[str, str] = {47 "name": "object",48 "city": "object",49 "property_type": "object",50 "price_numeric": "float",51 "price_category": "object",52 "rating": "float",53 "rating_category": "object",54 "bedrooms": "int",55 "bathrooms": "float",56 "guests_count": "int",57 "amenities_count": "int",58 "num_images": "int",59 "num_reviews": "float",60 "quality_score": "float",61 "is_superhost": "bool",62 "is_guest_favorite": "bool",63 "pets_allowed": "bool",64 "lat": "float",65 "long": "float",66}6768# Columns that must be present for the pipeline to proceed69_REQUIRED_COLUMNS: List[str] = [70 "price_numeric", "bedrooms", "bathrooms", "guests_count",71 "lat", "long", "property_type",72]7374# ---------------------------------------------------------------------------75# Internal helpers76# ---------------------------------------------------------------------------777879def _file_available(path: Path) -> bool:80 """Return True if *path* exists and is non-empty."""81 return path.is_file() and path.stat().st_size > 0828384def _infer_separator(path: Path) -> str:85 """Sniff the first line to decide between comma and tab."""86 with open(path, "r", encoding="utf-8", errors="replace") as fh:87 first_line = fh.readline()88 if "\t" in first_line:89 return "\t"90 return ","919293def _safe_read_csv(94 path: Path,95 *,96 compression: Optional[str] = None,97 nrows: Optional[int] = None,98 usecols: Optional[List[str]] = None,99 dtype: Optional[Dict[str, Any]] = None,100 low_memory: bool = False,101) -> Optional[pd.DataFrame]:102 """Read a CSV/TSV with graceful error handling.103104 Returns ``None`` when the file does not exist or cannot be parsed.105 """106 if not _file_available(path):107 logger.info("File not found or empty: %s", path)108 return None109110 sep = "," if compression else _infer_separator(path)111112 try:113 df = pd.read_csv(114 path,115 sep=sep,116 compression=compression or "infer",117 nrows=nrows,118 usecols=usecols,119 dtype=dtype,120 low_memory=low_memory,121 encoding="utf-8",122 on_bad_lines="warn",123 )124 logger.info(125 "Loaded %s rows x %s cols from %s",126 f"{len(df):,}",127 len(df.columns),128 path.name,129 )130 return df131 except Exception:132 logger.exception("Failed to read %s", path)133 return None134135136# ---------------------------------------------------------------------------137# Public loaders138# ---------------------------------------------------------------------------139140141def load_listings(142 *,143 path: Optional[Union[str, Path]] = None,144 nrows: Optional[int] = None,145 usecols: Optional[List[str]] = None,146) -> pd.DataFrame:147 """Load the primary listings dataset.148149 Resolution order:150 1. Explicit *path* argument.151 2. Full Inside Airbnb ``listings.csv.gz`` in ``data/raw/``.152 3. ``cleaned_dataset_sample.csv`` at project root.153154 Raises155 ------156 FileNotFoundError157 If no listing data can be found at any location.158 """159 if path is not None:160 p = Path(path)161 df = _safe_read_csv(p, nrows=nrows, usecols=usecols)162 if df is not None:163 return df164 raise FileNotFoundError(f"Specified listing file not found: {p}")165166 # Try full extract first167 if _file_available(LISTINGS_CSV_GZ):168 df = _safe_read_csv(169 LISTINGS_CSV_GZ,170 compression="gzip",171 nrows=nrows,172 usecols=usecols,173 low_memory=True,174 )175 if df is not None:176 return df177178 # Fall back to cleaned sample179 if _file_available(CLEANED_SAMPLE_CSV):180 df = _safe_read_csv(CLEANED_SAMPLE_CSV, nrows=nrows, usecols=usecols)181 if df is not None:182 return df183184 raise FileNotFoundError(185 "No listing data found. Place listings.csv.gz in data/raw/ "186 "or ensure cleaned_dataset_sample.csv exists at the project root."187 )188189190def load_calendar(191 *,192 path: Optional[Union[str, Path]] = None,193 nrows: Optional[int] = None,194) -> Optional[pd.DataFrame]:195 """Load Inside Airbnb calendar data (availability & price over time).196197 Returns ``None`` if the file is not available.198 """199 p = Path(path) if path else CALENDAR_CSV_GZ200 compression = "gzip" if p.suffix == ".gz" else None201 df = _safe_read_csv(p, compression=compression, nrows=nrows, low_memory=True)202 if df is not None and "date" in df.columns:203 df["date"] = pd.to_datetime(df["date"], errors="coerce")204 return df205206207def load_reviews(208 *,209 path: Optional[Union[str, Path]] = None,210 nrows: Optional[int] = None,211) -> Optional[pd.DataFrame]:212 """Load Inside Airbnb reviews data.213214 Returns ``None`` if the file is not available.215 """216 p = Path(path) if path else REVIEWS_CSV_GZ217 compression = "gzip" if p.suffix == ".gz" else None218 df = _safe_read_csv(p, compression=compression, nrows=nrows, low_memory=True)219 if df is not None and "date" in df.columns:220 df["date"] = pd.to_datetime(df["date"], errors="coerce")221 return df222223224def load_neighbourhoods(225 *,226 path: Optional[Union[str, Path]] = None,227 prefer_geojson: bool = True,228) -> Optional[pd.DataFrame]:229 """Load neighbourhood boundaries or names.230231 If *prefer_geojson* is True (default), attempts GeoJSON first, then232 CSV. Returns a plain ``DataFrame`` (not GeoDataFrame) unless233 geopandas is installed, in which case it returns a ``GeoDataFrame``.234235 Returns ``None`` if no file is available.236 """237 csv_path = Path(path) if path else NEIGHBOURHOODS_CSV238 geojson_path = NEIGHBOURHOODS_GEOJSON239240 if prefer_geojson and _file_available(geojson_path):241 try:242 import geopandas as gpd # type: ignore[import-untyped]243244 gdf = gpd.read_file(geojson_path)245 logger.info(246 "Loaded %s neighbourhoods from GeoJSON", f"{len(gdf):,}"247 )248 return gdf249 except ImportError:250 logger.warning(251 "geopandas not installed; falling back to CSV neighbourhoods"252 )253 except Exception:254 logger.exception("Failed to read GeoJSON: %s", geojson_path)255256 if _file_available(csv_path):257 return _safe_read_csv(csv_path)258259 logger.info("No neighbourhood data available")260 return None261262263def load_interim(name: str) -> Optional[pd.DataFrame]:264 """Load a parquet or CSV from ``data/interim/`` by stem name."""265 parquet = DATA_INTERIM / f"{name}.parquet"266 csv = DATA_INTERIM / f"{name}.csv"267268 if _file_available(parquet):269 try:270 df = pd.read_parquet(parquet)271 logger.info("Loaded interim parquet: %s (%s rows)", name, f"{len(df):,}")272 return df273 except Exception:274 logger.exception("Failed to read interim parquet: %s", parquet)275276 if _file_available(csv):277 return _safe_read_csv(csv)278279 logger.info("Interim artefact '%s' not found", name)280 return None281282283def load_processed(name: str) -> Optional[pd.DataFrame]:284 """Load a parquet or CSV from ``data/processed/`` by stem name."""285 parquet = DATA_PROCESSED / f"{name}.parquet"286 csv = DATA_PROCESSED / f"{name}.csv"287288 if _file_available(parquet):289 try:290 df = pd.read_parquet(parquet)291 logger.info("Loaded processed parquet: %s (%s rows)", name, f"{len(df):,}")292 return df293 except Exception:294 logger.exception("Failed to read processed parquet: %s", parquet)295296 if _file_available(csv):297 return _safe_read_csv(csv)298299 logger.info("Processed artefact '%s' not found", name)300 return None301302303def load_all_data(304 *,305 nrows: Optional[int] = None,306) -> Dict[str, Optional[pd.DataFrame]]:307 """Master loader -- returns a dict of every available dataset.308309 Keys310 ----311 ``"listings"``312 Always present (raises if not).313 ``"calendar"``, ``"reviews"``, ``"neighbourhoods"``314 ``None`` when the corresponding raw file is missing.315316 Parameters317 ----------318 nrows : int, optional319 Limit rows for quick exploratory loads.320 """321 data: Dict[str, Optional[pd.DataFrame]] = {}322323 data["listings"] = load_listings(nrows=nrows)324 data["calendar"] = load_calendar(nrows=nrows)325 data["reviews"] = load_reviews(nrows=nrows)326 data["neighbourhoods"] = load_neighbourhoods()327328 available = [k for k, v in data.items() if v is not None]329 logger.info("Loaded datasets: %s", ", ".join(available))330 return data331332333# ---------------------------------------------------------------------------334# Validation335# ---------------------------------------------------------------------------336337338def validate_data(339 df: pd.DataFrame,340 *,341 strict: bool = False,342) -> Dict[str, Any]:343 """Validate a listings DataFrame against expected schema.344345 Parameters346 ----------347 df : pd.DataFrame348 The listings data to validate.349 strict : bool350 If True, raise ``ValueError`` on critical failures instead of351 just logging warnings.352353 Returns354 -------355 dict356 ``{"valid": bool, "warnings": [...], "errors": [...]}``.357 """358 warnings_list: List[str] = []359 errors_list: List[str] = []360361 # -- Required columns --------------------------------------------------362 missing_required = [c for c in _REQUIRED_COLUMNS if c not in df.columns]363 if missing_required:364 msg = f"Missing required columns: {missing_required}"365 errors_list.append(msg)366 logger.error(msg)367368 # -- Expected columns (informational) -----------------------------------369 missing_expected = [370 c for c in _EXPECTED_COLUMNS_SAMPLE if c not in df.columns371 ]372 if missing_expected:373 msg = f"Missing expected (non-critical) columns: {missing_expected}"374 warnings_list.append(msg)375 logger.warning(msg)376377 # -- Row count ----------------------------------------------------------378 if len(df) == 0:379 errors_list.append("DataFrame is empty")380 elif len(df) < 100:381 warnings_list.append(382 f"Very small dataset ({len(df)} rows); results may be unreliable"383 )384385 # -- Price range --------------------------------------------------------386 if "price_numeric" in df.columns:387 prices = df["price_numeric"].dropna()388 if len(prices) == 0:389 errors_list.append("All price values are missing")390 else:391 if prices.min() <= 0:392 warnings_list.append(393 f"Prices include non-positive values (min={prices.min():.2f})"394 )395 if prices.max() > 100_000:396 warnings_list.append(397 f"Extreme price outliers present (max={prices.max():,.0f})"398 )399400 # -- Coordinate ranges --------------------------------------------------401 if "lat" in df.columns and "long" in df.columns:402 lat_min, lat_max = QUEBEC_LAT_RANGE403 lon_min, lon_max = QUEBEC_LON_RANGE404405 out_of_bounds = df[406 (df["lat"] < lat_min)407 | (df["lat"] > lat_max)408 | (df["long"] < lon_min)409 | (df["long"] > lon_max)410 ]411 if len(out_of_bounds) > 0:412 pct = 100.0 * len(out_of_bounds) / len(df)413 warnings_list.append(414 f"{len(out_of_bounds)} listings ({pct:.1f}%) have coordinates "415 f"outside Quebec bounding box"416 )417418 # -- Duplicate coordinates (exact) --------------------------------------419 if "lat" in df.columns and "long" in df.columns:420 dup_coords = df.duplicated(subset=["lat", "long"], keep=False).sum()421 if dup_coords > 0:422 pct = 100.0 * dup_coords / len(df)423 warnings_list.append(424 f"{dup_coords} listings ({pct:.1f}%) share exact coordinates "425 f"(possible duplicates or jittered points)"426 )427428 # -- Boolean columns ----------------------------------------------------429 for col in ("is_superhost", "is_guest_favorite", "pets_allowed"):430 if col in df.columns and df[col].dtype not in ("bool", "boolean"):431 warnings_list.append(432 f"Column '{col}' has dtype {df[col].dtype}, expected bool"433 )434435 # -- Rating range -------------------------------------------------------436 if "rating" in df.columns:437 valid_ratings = df["rating"].dropna()438 if len(valid_ratings) > 0:439 if valid_ratings.min() < 0 or valid_ratings.max() > 5:440 warnings_list.append(441 f"Ratings outside 0-5 range: "442 f"min={valid_ratings.min():.2f}, max={valid_ratings.max():.2f}"443 )444445 # -- Summary ------------------------------------------------------------446 is_valid = len(errors_list) == 0447 result = {448 "valid": is_valid,449 "warnings": warnings_list,450 "errors": errors_list,451 "n_rows": len(df),452 "n_cols": len(df.columns),453 "columns": list(df.columns),454 }455456 if is_valid:457 logger.info(458 "Validation passed (%s warnings)", len(warnings_list)459 )460 else:461 logger.error("Validation FAILED with %s error(s)", len(errors_list))462463 if strict and not is_valid:464 raise ValueError(465 "Data validation failed:\n" + "\n".join(errors_list)466 )467468 return result469470471# ---------------------------------------------------------------------------472# Summary statistics473# ---------------------------------------------------------------------------474475476def get_data_summary(df: pd.DataFrame) -> Dict[str, Any]:477 """Return a rich summary-statistics dictionary for a listings DataFrame.478479 Useful for dashboards, logging, and the cleaning report.480 """481 summary: Dict[str, Any] = {482 "n_rows": len(df),483 "n_cols": len(df.columns),484 "columns": list(df.columns),485 "dtypes": {col: str(dt) for col, dt in df.dtypes.items()},486 "missing_counts": df.isnull().sum().to_dict(),487 "missing_pct": (100.0 * df.isnull().mean()).round(2).to_dict(),488 }489490 # Numeric column stats491 numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()492 if numeric_cols:493 desc = df[numeric_cols].describe(percentiles=[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99])494 summary["numeric_stats"] = desc.to_dict()495496 # Price specifics497 if "price_numeric" in df.columns:498 prices = df["price_numeric"].dropna()499 summary["price"] = {500 "count": int(len(prices)),501 "missing": int(df["price_numeric"].isnull().sum()),502 "mean": float(prices.mean()) if len(prices) > 0 else None,503 "median": float(prices.median()) if len(prices) > 0 else None,504 "std": float(prices.std()) if len(prices) > 0 else None,505 "min": float(prices.min()) if len(prices) > 0 else None,506 "max": float(prices.max()) if len(prices) > 0 else None,507 "p1": float(prices.quantile(0.01)) if len(prices) > 0 else None,508 "p5": float(prices.quantile(0.05)) if len(prices) > 0 else None,509 "p95": float(prices.quantile(0.95)) if len(prices) > 0 else None,510 "p99": float(prices.quantile(0.99)) if len(prices) > 0 else None,511 "skewness": float(prices.skew()) if len(prices) > 0 else None,512 "kurtosis": float(prices.kurtosis()) if len(prices) > 0 else None,513 }514515 # Categorical breakdowns516 for cat_col in ("city", "property_type", "price_category", "rating_category", "room_type"):517 if cat_col in df.columns:518 vc = df[cat_col].value_counts(dropna=False)519 summary[f"{cat_col}_counts"] = vc.to_dict()520521 # Boolean flags522 for bool_col in ("is_superhost", "is_guest_favorite", "pets_allowed"):523 if bool_col in df.columns:524 true_count = int(df[bool_col].sum())525 summary[f"{bool_col}_rate"] = round(526 100.0 * true_count / len(df), 2527 ) if len(df) > 0 else 0.0528529 # Geo extent530 if "lat" in df.columns and "long" in df.columns:531 summary["geo_extent"] = {532 "lat_min": float(df["lat"].min()),533 "lat_max": float(df["lat"].max()),534 "lon_min": float(df["long"].min()),535 "lon_max": float(df["long"].max()),536 }537538 # Top cities by count539 if "city" in df.columns:540 top_cities = df["city"].value_counts().head(15)541 summary["top_cities"] = top_cities.to_dict()542543 logger.info(544 "Data summary: %s rows, %s cols, %s numeric features",545 f"{summary['n_rows']:,}",546 summary["n_cols"],547 len(numeric_cols),548 )549 return summary550