""" report.py -- Price decomposition and reporting for hedonic pricing analysis. Provides functions to decompose predicted prices into interpretable components (structural, amenity, host, review, location, text, spatial), identify the strongest positive and negative price drivers for individual listings, and orchestrate the full output-generation pipeline (tables, figures, decomposition summaries). """ from __future__ import annotations import logging import time from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import numpy as np import pandas as pd from src import config logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 1. Single-listing price decomposition # --------------------------------------------------------------------------- def explain_listing_price( model: Any, df: pd.DataFrame, listing_idx: int, feature_groups: Dict[str, List[str]], *, n_top: int = 10, ) -> Dict[str, Any]: """Decompose a single listing's predicted price into component contributions. Parameters ---------- model : fitted statsmodels-style regression result Must expose ``.params`` (Series of coefficients) and, when the model includes an intercept, a ``const`` / ``Intercept`` key. df : pd.DataFrame The design matrix used for estimation (same columns as model.params). listing_idx : int Row position (iloc index) of the listing to explain. feature_groups : dict[str, list[str]] Mapping of group name -> list of feature column names. Expected keys: ``"structural"``, ``"amenity"``, ``"host"``, ``"review"``, ``"location"``, ``"text"``, ``"spatial"``. Missing keys are treated as empty. n_top : int Number of top positive / negative drivers to return. Returns ------- dict with keys: actual_price, predicted_price, prediction_error, base_component, property_component, amenity_component, host_component, review_component, location_component, text_component, spatial_component, top_positive_drivers, top_negative_drivers """ row = df.iloc[listing_idx] params = model.params # Determine which features actually exist in both the model and the data model_features = set(params.index) data_features = set(df.columns) common_features = model_features & data_features # --- intercept / base component --- intercept_keys = {"const", "Intercept", "intercept"} intercept_name = intercept_keys & model_features base_component = float(params[intercept_name.pop()]) if intercept_name else 0.0 # --- compute individual feature contributions --- contributions: Dict[str, float] = {} for feat in common_features: val = row.get(feat, 0.0) if pd.isna(val): val = 0.0 contributions[feat] = float(params[feat]) * float(val) # --- aggregate by group --- group_aliases = { "structural": "property_component", "amenity": "amenity_component", "host": "host_component", "review": "review_component", "location": "location_component", "text": "text_component", "spatial": "spatial_component", } component_sums: Dict[str, float] = {} assigned_features: set = set() for group_key, component_name in group_aliases.items(): group_feats = feature_groups.get(group_key, []) total = 0.0 for feat in group_feats: if feat in contributions: total += contributions[feat] assigned_features.add(feat) component_sums[component_name] = total # Features present in the model but not assigned to any group are folded # into property_component (structural catch-all). for feat, contrib in contributions.items(): if feat not in assigned_features: component_sums["property_component"] = ( component_sums.get("property_component", 0.0) + contrib ) # --- predicted price --- predicted_price = base_component + sum(contributions.values()) # --- actual price --- # Try common dependent-variable column names actual_price = np.nan for dep_col in ("log_price", "price_numeric", "price"): if dep_col in df.columns: actual_price = float(row[dep_col]) break prediction_error = actual_price - predicted_price # --- top drivers --- sorted_contribs = sorted(contributions.items(), key=lambda x: x[1], reverse=True) top_positive = [(f, c) for f, c in sorted_contribs if c > 0][:n_top] top_negative = [(f, c) for f, c in sorted_contribs if c < 0][-n_top:] # Sort negatives by magnitude (most negative first) top_negative = sorted(top_negative, key=lambda x: x[1]) return { "actual_price": actual_price, "predicted_price": predicted_price, "prediction_error": prediction_error, "base_component": base_component, **component_sums, "top_positive_drivers": top_positive, "top_negative_drivers": top_negative, } # --------------------------------------------------------------------------- # 2. Full-dataset decomposition # --------------------------------------------------------------------------- def decompose_all_prices( model: Any, df: pd.DataFrame, feature_groups: Dict[str, List[str]], ) -> pd.DataFrame: """Apply :func:`explain_listing_price` to every row and return a DataFrame. Each row corresponds to a listing; columns include all scalar components produced by the single-listing function. The ``top_positive_drivers`` and ``top_negative_drivers`` lists are excluded from the DataFrame (use :func:`explain_listing_price` for individual inspection). Parameters ---------- model : fitted statsmodels result df : pd.DataFrame feature_groups : dict[str, list[str]] Returns ------- pd.DataFrame """ logger.info("Decomposing prices for %d listings ...", len(df)) t0 = time.perf_counter() records: List[Dict[str, float]] = [] scalar_keys = [ "actual_price", "predicted_price", "prediction_error", "base_component", "property_component", "amenity_component", "host_component", "review_component", "location_component", "text_component", "spatial_component", ] for idx in range(len(df)): try: expl = explain_listing_price(model, df, idx, feature_groups) records.append({k: expl.get(k, 0.0) for k in scalar_keys}) except Exception: logger.debug("Decomposition failed for listing idx=%d", idx, exc_info=True) records.append({k: np.nan for k in scalar_keys}) elapsed = time.perf_counter() - t0 logger.info("Price decomposition complete in %.1f s", elapsed) return pd.DataFrame(records, index=df.index) # --------------------------------------------------------------------------- # 3. Decomposition summary statistics # --------------------------------------------------------------------------- def generate_decomposition_summary(decomp_df: pd.DataFrame) -> pd.DataFrame: """Compute summary statistics for each price component. Parameters ---------- decomp_df : pd.DataFrame Output of :func:`decompose_all_prices`. Returns ------- pd.DataFrame Index = component names, columns = mean, median, std, min, max. """ component_cols = [ c for c in decomp_df.columns if c.endswith("_component") or c in ("predicted_price", "actual_price", "prediction_error") ] summary = decomp_df[component_cols].agg(["mean", "median", "std", "min", "max"]).T summary.columns = ["mean", "median", "std", "min", "max"] summary = summary.round(4) logger.info("Decomposition summary:\n%s", summary.to_string()) return summary # --------------------------------------------------------------------------- # 4. Pretty-print a single listing explanation # --------------------------------------------------------------------------- def format_decomposition_report(listing_explanation: Dict[str, Any]) -> str: """Return a human-readable string summarising a listing's price decomposition. Parameters ---------- listing_explanation : dict Output of :func:`explain_listing_price`. Returns ------- str """ e = listing_explanation lines: List[str] = [] sep = "-" * 64 lines.append(sep) lines.append("LISTING PRICE DECOMPOSITION REPORT") lines.append(sep) lines.append(f" Actual price : {e.get('actual_price', np.nan):>12.4f}") lines.append(f" Predicted price : {e.get('predicted_price', np.nan):>12.4f}") lines.append(f" Prediction error : {e.get('prediction_error', np.nan):>12.4f}") lines.append("") lines.append(" COMPONENTS") lines.append(f" Base (intercept) : {e.get('base_component', 0.0):>12.4f}") lines.append(f" Property : {e.get('property_component', 0.0):>12.4f}") lines.append(f" Amenity : {e.get('amenity_component', 0.0):>12.4f}") lines.append(f" Host : {e.get('host_component', 0.0):>12.4f}") lines.append(f" Review : {e.get('review_component', 0.0):>12.4f}") lines.append(f" Location : {e.get('location_component', 0.0):>12.4f}") lines.append(f" Text / NLP : {e.get('text_component', 0.0):>12.4f}") lines.append(f" Spatial : {e.get('spatial_component', 0.0):>12.4f}") lines.append("") # Top positive drivers pos = e.get("top_positive_drivers", []) if pos: lines.append(" TOP POSITIVE DRIVERS") for feat, contrib in pos: lines.append(f" {feat:<40s} +{contrib:>10.4f}") lines.append("") # Top negative drivers neg = e.get("top_negative_drivers", []) if neg: lines.append(" TOP NEGATIVE DRIVERS") for feat, contrib in neg: lines.append(f" {feat:<40s} {contrib:>10.4f}") lines.append("") lines.append(sep) return "\n".join(lines) # --------------------------------------------------------------------------- # 5. Master report / output generation pipeline # --------------------------------------------------------------------------- def generate_full_report( df: pd.DataFrame, models_dict: Dict[str, Any], ml_models_dict: Dict[str, Any], spatial_results: Dict[str, Any], feature_groups: Dict[str, List[str]], output_dir: str | Path = config.OUTPUTS_DIR, ) -> Dict[str, Any]: """Orchestrate full output generation: tables, figures, decomposition. Calls the ``tables`` and ``figures`` modules to produce every artefact required for the research paper, then runs the price-decomposition pipeline and saves the summary. Parameters ---------- df : pd.DataFrame Feature-engineered dataset. models_dict : dict[str, Any] Mapping of model name -> fitted hedonic / quantile result. ml_models_dict : dict[str, Any] Mapping of model name -> fitted ML model or results dict. spatial_results : dict[str, Any] Output from the spatial-models module. feature_groups : dict[str, list[str]] Feature group registry. output_dir : str or Path Root output directory (tables/ and figures/ subdirectories are created). Returns ------- dict with keys: ``decomp_df``, ``decomp_summary``, ``tables_generated``, ``figures_generated``, ``elapsed_seconds``. """ t0 = time.perf_counter() output_dir = Path(output_dir) tables_dir = output_dir / "tables" figures_dir = output_dir / "figures" tables_dir.mkdir(parents=True, exist_ok=True) figures_dir.mkdir(parents=True, exist_ok=True) report_artifacts: Dict[str, Any] = { "tables_generated": [], "figures_generated": [], } # ---- Tables ---- logger.info("Generating tables ...") try: from src import tables as tbl_mod table_funcs = [ ("descriptive_stats", lambda: tbl_mod.generate_descriptive_stats_table(df, tables_dir)), ("correlation_matrix", lambda: tbl_mod.generate_correlation_table(df, tables_dir)), ("hedonic_results", lambda: tbl_mod.generate_hedonic_results_table(models_dict, tables_dir)), ("quantile_results", lambda: tbl_mod.generate_quantile_results_table(models_dict, tables_dir)), ("ml_comparison", lambda: tbl_mod.generate_ml_comparison_table(ml_models_dict, tables_dir)), ("spatial_results", lambda: tbl_mod.generate_spatial_results_table(spatial_results, tables_dir)), ] for name, func in table_funcs: try: func() report_artifacts["tables_generated"].append(name) logger.info(" Table '%s' generated.", name) except Exception: logger.warning(" Table '%s' failed.", name, exc_info=True) except ImportError: logger.warning("tables module not available; skipping table generation.") except Exception: logger.warning("Table generation failed.", exc_info=True) # ---- Figures ---- logger.info("Generating figures ...") try: from src import figures as fig_mod figure_funcs = [ ("price_distribution", lambda: fig_mod.plot_price_distribution(df, figures_dir)), ("correlation_heatmap", lambda: fig_mod.plot_correlation_heatmap(df, figures_dir)), ("coefficient_plot", lambda: fig_mod.plot_coefficient_comparison(models_dict, figures_dir)), ("quantile_plot", lambda: fig_mod.plot_quantile_coefficients(models_dict, figures_dir)), ("spatial_map", lambda: fig_mod.plot_spatial_map(df, spatial_results, figures_dir)), ("residual_diagnostics", lambda: fig_mod.plot_residual_diagnostics(models_dict, df, figures_dir)), ("shap_summary", lambda: fig_mod.plot_shap_summary(ml_models_dict, df, figures_dir)), ] for name, func in figure_funcs: try: func() report_artifacts["figures_generated"].append(name) logger.info(" Figure '%s' generated.", name) except Exception: logger.warning(" Figure '%s' failed.", name, exc_info=True) except ImportError: logger.warning("figures module not available; skipping figure generation.") except Exception: logger.warning("Figure generation failed.", exc_info=True) # ---- Price decomposition ---- logger.info("Running price decomposition ...") decomp_df = pd.DataFrame() decomp_summary = pd.DataFrame() # Pick the best available hedonic model for decomposition preferred_model_keys = [ "ols_semilog", "ols_log", "ols_linear", "robust_huber", "robust_bisquare", ] decomp_model = None decomp_model_name = None for key in preferred_model_keys: if key in models_dict and models_dict[key] is not None: decomp_model = models_dict[key] decomp_model_name = key break if decomp_model is not None: try: logger.info("Using model '%s' for price decomposition.", decomp_model_name) decomp_df = decompose_all_prices(decomp_model, df, feature_groups) decomp_summary = generate_decomposition_summary(decomp_df) # Save artefacts decomp_csv = output_dir / "tables" / "price_decomposition.csv" decomp_df.to_csv(decomp_csv, index=False) logger.info("Decomposition saved to %s", decomp_csv) summary_csv = output_dir / "tables" / "decomposition_summary.csv" decomp_summary.to_csv(summary_csv) logger.info("Summary saved to %s", summary_csv) # Pretty-print a sample listing if len(decomp_df) > 0: sample_idx = 0 sample_expl = explain_listing_price( decomp_model, df, sample_idx, feature_groups, ) logger.info( "Sample decomposition:\n%s", format_decomposition_report(sample_expl), ) except Exception: logger.warning("Price decomposition failed.", exc_info=True) else: logger.warning( "No suitable hedonic model found for decomposition (tried: %s).", ", ".join(preferred_model_keys), ) elapsed = time.perf_counter() - t0 report_artifacts["decomp_df"] = decomp_df report_artifacts["decomp_summary"] = decomp_summary report_artifacts["elapsed_seconds"] = elapsed logger.info("Full report generation complete in %.1f s", elapsed) return report_artifacts