""" NLP Feature Extraction for Airbnb Hedonic Pricing Analysis. Extracts linguistic, semantic, and structural features from listing text fields (primarily 'name', but also 'description', 'neighborhood_overview', 'host_about' when available in richer datasets). Supports bilingual (French/English) Airbnb listings common in Quebec markets. Functions --------- extract_basic_text_features : Character-level and token-level text statistics. extract_language_scores : Keyword-based thematic scoring (luxury, cozy, ...). extract_property_info_from_name : Structured parsing of the Airbnb name template. create_text_embeddings : TF-IDF or CountVectorizer with TruncatedSVD reduction. find_price_associated_words : Correlation of TF-IDF terms with (log) price. create_all_nlp_features : Master pipeline combining all of the above. """ from __future__ import annotations import logging import re from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd from scipy import stats from sklearn.decomposition import TruncatedSVD from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Keyword dictionaries (bilingual FR/EN) # --------------------------------------------------------------------------- LANGUAGE_KEYWORDS: Dict[str, List[str]] = { "luxury_language_score": [ "luxe", "luxury", "premium", "elegant", "chic", "spa", "designer", "upscale", "exclusive", "prestige", "haut de gamme", "somptueux", "raffiné", "opulent", "magnifique", ], "cozy_language_score": [ "cozy", "cosy", "chaleureux", "douillet", "confortable", "warm", "charming", "intime", "accueillant", "nid", "mignon", "petit coin", "cocoon", ], "professional_tone_score": [ "professional", "modern", "clean", "bright", "renovated", "new", "stylish", "design", "moderne", "lumineux", "rénové", "neuf", "propre", "contemporain", "minimaliste", ], "family_language_score": [ "family", "famille", "kids", "enfants", "children", "spacious", "grand", "spacieux", "familial", "aire de jeux", "playground", "bébé", "baby", ], "business_language_score": [ "business", "workspace", "bureau", "wifi", "downtown", "centre-ville", "corporate", "professionnel", "travail", "work", "desk", "office", ], "romantic_language_score": [ "romantic", "romantique", "intimate", "couple", "retreat", "getaway", "escapade", "amoureux", "lune de miel", "honeymoon", "chandelle", ], "tourism_language_score": [ "tourist", "touristique", "attractions", "view", "vue", "panoramic", "nature", "montagne", "lac", "ski", "mountain", "lake", "panoramique", "plein air", "randonnée", "hiking", "plage", "beach", "forêt", "forest", ], } # --------------------------------------------------------------------------- # 1. Basic text features # --------------------------------------------------------------------------- def extract_basic_text_features(df: pd.DataFrame) -> pd.DataFrame: """Extract character-level and token-level statistics from listing text. Parameters ---------- df : pd.DataFrame Must contain a ``name`` column. Optionally ``description``, ``neighborhood_overview``, and ``host_about``. Returns ------- pd.DataFrame Copy of *df* with appended feature columns: ``title_length``, ``title_word_count``, ``exclamation_count``, ``uppercase_ratio``, ``has_star_rating``, ``extracted_star_rating``. """ logger.info("Extracting basic text features ...") result = df.copy() name_series: pd.Series = result["name"].fillna("") # Character count result["title_length"] = name_series.str.len() # Word count (split on whitespace) result["title_word_count"] = name_series.str.split().str.len().fillna(0).astype(int) # Exclamation marks result["exclamation_count"] = name_series.str.count("!") # Upper-case ratio (letters only; avoid division by zero) alpha_chars = name_series.str.replace(r"[^A-Za-zÀ-ÿ]", "", regex=True) upper_chars = name_series.str.replace(r"[^A-ZÀ-Ý]", "", regex=True) alpha_len = alpha_chars.str.len().replace(0, np.nan) result["uppercase_ratio"] = (upper_chars.str.len() / alpha_len).fillna(0.0) # Star rating detection result["has_star_rating"] = name_series.str.contains("★", regex=False) # Numeric star rating extraction (e.g. ★4.94 or ★5.0) extracted = name_series.str.extract(r"★(\d+\.?\d*)", expand=False) result["extracted_star_rating"] = pd.to_numeric(extracted, errors="coerce") # --- Optional: description-level features (if column exists) ----------- for col in ("description", "neighborhood_overview", "host_about"): if col in result.columns: text = result[col].fillna("") result[f"{col}_length"] = text.str.len() result[f"{col}_word_count"] = text.str.split().str.len().fillna(0).astype(int) logger.info(" Added length/word_count for '%s'.", col) logger.info("Basic text features extracted: %d rows.", len(result)) return result # --------------------------------------------------------------------------- # 2. Language / keyword scores # --------------------------------------------------------------------------- def _score_keywords(text: str, keywords: List[str]) -> int: """Count how many distinct keywords appear in *text* (case-insensitive).""" text_lower = text.lower() return sum(1 for kw in keywords if kw in text_lower) def extract_language_scores( df: pd.DataFrame, text_columns: Optional[List[str]] = None, ) -> pd.DataFrame: """Score each listing on thematic keyword dimensions. For every theme defined in ``LANGUAGE_KEYWORDS`` the score equals the number of matching keywords found in the concatenated text of all *text_columns* (case-insensitive substring matching). Parameters ---------- df : pd.DataFrame Input dataframe. text_columns : list[str] | None Columns to scan. Defaults to ``['name']`` plus any of ``description``, ``neighborhood_overview``, ``host_about`` that exist in *df*. Returns ------- pd.DataFrame Copy with one new column per theme key in ``LANGUAGE_KEYWORDS``. """ logger.info("Extracting language keyword scores ...") result = df.copy() if text_columns is None: text_columns = ["name"] for optional in ("description", "neighborhood_overview", "host_about"): if optional in result.columns: text_columns.append(optional) # Build a single concatenated text per row combined_text: pd.Series = ( result[text_columns] .fillna("") .apply(lambda row: " ".join(row), axis=1) ) for score_name, keywords in LANGUAGE_KEYWORDS.items(): result[score_name] = combined_text.apply(_score_keywords, keywords=keywords) logger.debug(" %s mean=%.3f", score_name, result[score_name].mean()) logger.info("Language scores computed for %d themes.", len(LANGUAGE_KEYWORDS)) return result # --------------------------------------------------------------------------- # 3. Structured name parsing # --------------------------------------------------------------------------- # Pre-compiled patterns for the Airbnb name template: # "PropertyType in CityName · ★Rating · N bedroom(s) · N bed(s) · N bath(s)" _RE_CITY = re.compile( r"^.+?\s+in\s+(.+?)(?:\s+·|$)", re.IGNORECASE ) _RE_BEDROOMS = re.compile( r"(\d+)\s+bedroom", re.IGNORECASE ) _RE_STUDIO = re.compile( r"\bStudio\b", re.IGNORECASE ) _RE_BEDS = re.compile( r"(\d+)\s+beds?\b", re.IGNORECASE ) _RE_BATHS = re.compile( r"([\d.]+)\s+(?:shared\s+)?(?:private\s+)?bath", re.IGNORECASE ) _RE_HALF_BATH = re.compile( r"\bHalf-bath\b", re.IGNORECASE ) _RE_NEW = re.compile( r"★New\b", re.IGNORECASE ) _RE_PROPERTY_TYPE = re.compile( r"^(.+?)\s+in\s+", re.IGNORECASE ) def extract_property_info_from_name(df: pd.DataFrame) -> pd.DataFrame: """Parse the structured Airbnb ``name`` field into discrete features. Parameters ---------- df : pd.DataFrame Must contain column ``name``. Returns ------- pd.DataFrame Copy with columns: ``parsed_property_type``, ``parsed_city``, ``parsed_bedrooms``, ``parsed_beds``, ``parsed_baths``, ``is_studio``, ``is_new_listing``, ``has_shared_bath``, ``has_private_bath``. """ logger.info("Parsing property info from name field ...") result = df.copy() names: pd.Series = result["name"].fillna("") # --- Parsed property type ------------------------------------------------ result["parsed_property_type"] = names.str.extract( _RE_PROPERTY_TYPE, expand=False ) # --- City ---------------------------------------------------------------- result["parsed_city"] = names.str.extract(_RE_CITY, expand=False) # --- Bedrooms (0 for Studio) -------------------------------------------- bedroom_match = names.str.extract(_RE_BEDROOMS, expand=False) is_studio = names.str.contains(_RE_STUDIO, regex=True) result["parsed_bedrooms"] = pd.to_numeric(bedroom_match, errors="coerce") result.loc[is_studio & result["parsed_bedrooms"].isna(), "parsed_bedrooms"] = 0 result["is_studio"] = is_studio # --- Beds ---------------------------------------------------------------- beds_match = names.str.extract(_RE_BEDS, expand=False) result["parsed_beds"] = pd.to_numeric(beds_match, errors="coerce") # --- Baths (handles "1.5 baths", "Half-bath", "shared bath") ------------ baths_match = names.str.extract(_RE_BATHS, expand=False) result["parsed_baths"] = pd.to_numeric(baths_match, errors="coerce") # Handle "Half-bath" => 0.5 half_bath_mask = names.str.contains(_RE_HALF_BATH, regex=True) result.loc[half_bath_mask & result["parsed_baths"].isna(), "parsed_baths"] = 0.5 # --- Qualifiers ---------------------------------------------------------- result["is_new_listing"] = names.str.contains(_RE_NEW, regex=True) result["has_shared_bath"] = names.str.contains(r"shared\s+bath", case=False, regex=True) result["has_private_bath"] = names.str.contains(r"private\s+bath", case=False, regex=True) logger.info("Name parsing complete: %d rows.", len(result)) return result # --------------------------------------------------------------------------- # 4. Text embeddings (TF-IDF / CountVectorizer + TruncatedSVD) # --------------------------------------------------------------------------- def create_text_embeddings( df: pd.DataFrame, method: str = "tfidf", n_components: int = 10, text_columns: Optional[List[str]] = None, max_features: int = 5000, ngram_range: Tuple[int, int] = (1, 2), random_state: int = 42, ) -> pd.DataFrame: """Create dense low-dimensional text embeddings via TF-IDF/Count + SVD. Parameters ---------- df : pd.DataFrame Input data. method : {'tfidf', 'count'} Vectoriser method. n_components : int Number of SVD components to retain. text_columns : list[str] | None Text columns to concatenate. Defaults to ``['name']`` + optional ``description`` if present. max_features : int Maximum vocabulary size for the vectoriser. ngram_range : tuple[int, int] N-gram range passed to the vectoriser. random_state : int Seed for ``TruncatedSVD``. Returns ------- pd.DataFrame Copy with ``text_embed_0`` ... ``text_embed_{n_components-1}`` appended. """ logger.info("Creating text embeddings (method=%s, n_components=%d) ...", method, n_components) result = df.copy() if text_columns is None: text_columns = ["name"] if "description" in result.columns: text_columns.append("description") # Concatenate and clean corpus: pd.Series = ( result[text_columns] .fillna("") .apply(lambda row: " ".join(row), axis=1) ) corpus = corpus.str.lower() corpus = corpus.str.replace(r"[^\w\sÀ-ÿ]", " ", regex=True) # strip punctuation corpus = corpus.str.replace(r"\s+", " ", regex=True).str.strip() # Replace empty strings to avoid all-zero rows corpus = corpus.replace("", "empty_listing") # Vectorise if method == "count": vectoriser = CountVectorizer( max_features=max_features, ngram_range=ngram_range, stop_words=None, # keep French words ) else: vectoriser = TfidfVectorizer( max_features=max_features, ngram_range=ngram_range, stop_words=None, sublinear_tf=True, ) dtm = vectoriser.fit_transform(corpus) logger.info(" Vocabulary size: %d, document-term matrix shape: %s", len(vectoriser.vocabulary_), dtm.shape) # Ensure n_components does not exceed matrix rank effective_components = min(n_components, dtm.shape[1] - 1, dtm.shape[0] - 1) if effective_components < n_components: logger.warning( "Reducing n_components from %d to %d (matrix rank constraint).", n_components, effective_components, ) if effective_components < 1: logger.warning("Not enough data for SVD; skipping embeddings.") return result svd = TruncatedSVD(n_components=effective_components, random_state=random_state) embeddings = svd.fit_transform(dtm) explained = svd.explained_variance_ratio_.sum() logger.info(" SVD explained variance: %.2f%%", explained * 100) embed_cols = [f"text_embed_{i}" for i in range(effective_components)] embed_df = pd.DataFrame(embeddings, columns=embed_cols, index=result.index) result = pd.concat([result, embed_df], axis=1) # Pad with zeros if we reduced components for i in range(effective_components, n_components): result[f"text_embed_{i}"] = 0.0 logger.info("Text embeddings added: %d components.", n_components) return result # --------------------------------------------------------------------------- # 5. Price-associated words # --------------------------------------------------------------------------- def find_price_associated_words( df: pd.DataFrame, price_col: str = "log_price", top_n: int = 20, text_columns: Optional[List[str]] = None, max_features: int = 3000, ) -> pd.DataFrame: """Identify words most correlated with listing price. Builds a TF-IDF matrix from listing text and computes Pearson correlation of each term column with the price variable. Parameters ---------- df : pd.DataFrame Must contain *price_col* and at least ``name``. price_col : str Column with (log-)price values. Rows with NaN are excluded. top_n : int Number of top positively and negatively correlated words. text_columns : list[str] | None Text columns to concatenate. max_features : int Maximum vocabulary size. Returns ------- pd.DataFrame Columns: ``word``, ``correlation``, ``abs_correlation``, ``direction`` ('positive' | 'negative'). Sorted by absolute correlation descending. """ logger.info("Finding price-associated words (price_col=%s, top_n=%d) ...", price_col, top_n) if text_columns is None: text_columns = ["name"] if "description" in df.columns: text_columns.append("description") # Work on rows with valid price if price_col not in df.columns: # Attempt to create log_price on the fly if "price_numeric" in df.columns: work = df.copy() work["log_price"] = np.log1p(pd.to_numeric(work["price_numeric"], errors="coerce")) price_col = "log_price" else: logger.error("Price column '%s' not found. Cannot compute word correlations.", price_col) return pd.DataFrame(columns=["word", "correlation", "abs_correlation", "direction"]) else: work = df.copy() mask = work[price_col].notna() work = work.loc[mask].reset_index(drop=True) if len(work) < 30: logger.warning("Too few rows with valid price (%d); skipping word-price correlation.", len(work)) return pd.DataFrame(columns=["word", "correlation", "abs_correlation", "direction"]) # Concatenate text corpus = ( work[text_columns] .fillna("") .apply(lambda row: " ".join(row), axis=1) .str.lower() .str.replace(r"[^\w\sÀ-ÿ]", " ", regex=True) .str.replace(r"\s+", " ", regex=True) .str.strip() ) vectoriser = TfidfVectorizer( max_features=max_features, ngram_range=(1, 2), stop_words=None, sublinear_tf=True, min_df=5, ) dtm = vectoriser.fit_transform(corpus) feature_names = vectoriser.get_feature_names_out() price_values = work[price_col].values # Compute correlations correlations = np.zeros(len(feature_names)) dtm_dense = dtm.toarray() for i in range(len(feature_names)): col_vals = dtm_dense[:, i] if col_vals.std() == 0: correlations[i] = 0.0 else: correlations[i], _ = stats.pearsonr(col_vals, price_values) corr_df = pd.DataFrame({ "word": feature_names, "correlation": correlations, }) corr_df["abs_correlation"] = corr_df["correlation"].abs() corr_df["direction"] = np.where(corr_df["correlation"] >= 0, "positive", "negative") # Select top positive and top negative top_positive = corr_df.nlargest(top_n, "correlation") top_negative = corr_df.nsmallest(top_n, "correlation") combined = ( pd.concat([top_positive, top_negative]) .drop_duplicates(subset=["word"]) .sort_values("abs_correlation", ascending=False) .reset_index(drop=True) ) logger.info( "Top correlated word: '%s' (r=%.3f); bottom: '%s' (r=%.3f)", combined.iloc[0]["word"], combined.iloc[0]["correlation"], combined.iloc[-1]["word"], combined.iloc[-1]["correlation"], ) return combined # --------------------------------------------------------------------------- # 6. Master function # --------------------------------------------------------------------------- def create_all_nlp_features( df: pd.DataFrame, embedding_method: str = "tfidf", n_components: int = 10, compute_embeddings: bool = True, compute_price_words: bool = True, price_col: str = "log_price", top_n_words: int = 20, ) -> Tuple[pd.DataFrame, Optional[pd.DataFrame]]: """Run the full NLP feature-extraction pipeline. Parameters ---------- df : pd.DataFrame Raw listing data (must have at least ``name``). embedding_method : {'tfidf', 'count'} Vectorisation method for embeddings. n_components : int Number of SVD dimensions. compute_embeddings : bool Whether to compute text embeddings. compute_price_words : bool Whether to compute price-associated word analysis. price_col : str Price column for word-price correlation. top_n_words : int Number of top words per direction. Returns ------- tuple[pd.DataFrame, pd.DataFrame | None] - Enriched DataFrame with all NLP features. - Price-word correlation DataFrame (or ``None`` if skipped). """ logger.info("=" * 60) logger.info("NLP FEATURE EXTRACTION PIPELINE") logger.info("=" * 60) logger.info("Input shape: %s", df.shape) if "name" not in df.columns: logger.error("Column 'name' not found. Returning input unchanged.") return df.copy(), None # --- Step 1: basic text features ---------------------------------------- enriched = extract_basic_text_features(df) # --- Step 2: language keyword scores ------------------------------------ enriched = extract_language_scores(enriched) # --- Step 3: structured name parsing ------------------------------------ enriched = extract_property_info_from_name(enriched) # --- Step 4: text embeddings -------------------------------------------- if compute_embeddings: enriched = create_text_embeddings( enriched, method=embedding_method, n_components=n_components, ) # --- Step 5: price-associated words ------------------------------------- price_words: Optional[pd.DataFrame] = None if compute_price_words: price_words = find_price_associated_words( enriched, price_col=price_col, top_n=top_n_words, ) # --- Summary ------------------------------------------------------------ new_cols = [c for c in enriched.columns if c not in df.columns] logger.info("NLP pipeline complete. %d new features added.", len(new_cols)) logger.info("New columns: %s", new_cols) logger.info("Output shape: %s", enriched.shape) return enriched, price_words # --------------------------------------------------------------------------- # CLI quick-test # --------------------------------------------------------------------------- if __name__ == "__main__": import sys logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s", stream=sys.stdout, ) sample_path = "cleaned_dataset_sample.csv" logger.info("Loading sample data from '%s' ...", sample_path) data = pd.read_csv(sample_path) logger.info("Loaded %d rows, %d columns.", *data.shape) enriched_df, price_word_df = create_all_nlp_features(data) print("\n--- Enriched DataFrame (first 5 rows, NLP columns) ---") nlp_cols = [ "title_length", "title_word_count", "exclamation_count", "uppercase_ratio", "has_star_rating", "extracted_star_rating", "luxury_language_score", "cozy_language_score", "professional_tone_score", "family_language_score", "business_language_score", "romantic_language_score", "tourism_language_score", "parsed_property_type", "parsed_city", "parsed_bedrooms", "parsed_beds", "parsed_baths", "is_studio", "is_new_listing", "has_shared_bath", "has_private_bath", ] existing_nlp_cols = [c for c in nlp_cols if c in enriched_df.columns] print(enriched_df[existing_nlp_cols].head().to_string()) embed_cols = [c for c in enriched_df.columns if c.startswith("text_embed_")] if embed_cols: print(f"\n--- Text Embedding columns ({len(embed_cols)}) ---") print(enriched_df[embed_cols].describe().to_string()) if price_word_df is not None and not price_word_df.empty: print("\n--- Top Price-Associated Words ---") print(price_word_df.to_string(index=False)) else: print("\n(No price-word correlations computed.)")