|
|
|
|
| import numpy as np
|
| import pandas as pd
|
|
|
| import inference_core
|
|
|
| from fe_avm_bdo26 import (
|
| map_kecamatan,
|
| transform_area_group,
|
| transform_area_features,
|
| transform_structure_features,
|
| transform_interaction_features,
|
| fit_desirability_features,
|
| transform_desirability_features,
|
| )
|
|
|
| from gemini_utils import gemini_client, GEMINI_ENABLED
|
|
|
|
|
|
|
|
|
|
|
| DIMENSIONS = [
|
| "size_fit",
|
| "room_functionality_fit",
|
| "family_fit",
|
| "school_fit",
|
| "rentability_fit",
|
| "lifestyle_fit",
|
| "prestige_fit",
|
| "access_convenience_fit",
|
| "mixed_use_fit",
|
| "affordability_fit",
|
| "rumah_fit",
|
| "apartemen_fit",
|
| ]
|
|
|
| DIMENSION_LABELS = {
|
| "size_fit": "larger typical space",
|
| "room_functionality_fit": "more practical room layout",
|
| "family_fit": "stronger family-oriented environment",
|
| "school_fit": "better school access",
|
| "rentability_fit": "better rental appeal",
|
| "lifestyle_fit": "stronger lifestyle access",
|
| "prestige_fit": "more premium area profile",
|
| "access_convenience_fit": "better day-to-day convenience",
|
| "mixed_use_fit": "more mixed-use flexibility",
|
| "affordability_fit": "better affordability",
|
| "rumah_fit": "better fit for house-seeking buyers",
|
| "apartemen_fit": "better fit for apartment-seeking buyers",
|
| }
|
|
|
| BUYER_ORIENTATION_LIBRARY = {
|
| "family_oriented": {
|
| "size_fit": 9,
|
| "room_functionality_fit": 8,
|
| "family_fit": 10,
|
| "school_fit": 10,
|
| "rentability_fit": 1,
|
| "lifestyle_fit": 3,
|
| "prestige_fit": 2,
|
| "access_convenience_fit": 7,
|
| "mixed_use_fit": 0,
|
| "affordability_fit": 7,
|
| "rumah_fit": 10,
|
| "apartemen_fit": 1,
|
| },
|
| "investment_oriented": {
|
| "size_fit": 6,
|
| "room_functionality_fit": 7,
|
| "family_fit": 1,
|
| "school_fit": 1,
|
| "rentability_fit": 10,
|
| "lifestyle_fit": 6,
|
| "prestige_fit": 5,
|
| "access_convenience_fit": 7,
|
| "mixed_use_fit": 8,
|
| "affordability_fit": 8,
|
| "rumah_fit": 4,
|
| "apartemen_fit": 4,
|
| },
|
| "lifestyle_oriented": {
|
| "size_fit": 4,
|
| "room_functionality_fit": 6,
|
| "family_fit": 1,
|
| "school_fit": 1,
|
| "rentability_fit": 2,
|
| "lifestyle_fit": 10,
|
| "prestige_fit": 8,
|
| "access_convenience_fit": 8,
|
| "mixed_use_fit": 4,
|
| "affordability_fit": 4,
|
| "rumah_fit": 3,
|
| "apartemen_fit": 9,
|
| },
|
| "value_oriented": {
|
| "size_fit": 7,
|
| "room_functionality_fit": 7,
|
| "family_fit": 5,
|
| "school_fit": 4,
|
| "rentability_fit": 4,
|
| "lifestyle_fit": 3,
|
| "prestige_fit": 1,
|
| "access_convenience_fit": 6,
|
| "mixed_use_fit": 2,
|
| "affordability_fit": 10,
|
| "rumah_fit": 5,
|
| "apartemen_fit": 6,
|
| },
|
| }
|
|
|
| ORIENTATION_TO_CORE_PROFILE = {
|
| "family_oriented": "family",
|
| "investment_oriented": "investment",
|
| "lifestyle_oriented": "lifestyle",
|
| "value_oriented": "affordability",
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
| def norm01(series):
|
| s = pd.to_numeric(series, errors="coerce")
|
| mn = s.min()
|
| mx = s.max()
|
|
|
| if pd.isna(mn) or pd.isna(mx) or mx == mn:
|
| return pd.Series(0.0, index=s.index)
|
|
|
| return (s - mn) / (mx - mn)
|
|
|
|
|
| def _safe_numeric(df, col, fill_value=0.0):
|
| if col not in df.columns:
|
| return pd.Series(fill_value, index=df.index, dtype=float)
|
| return pd.to_numeric(df[col], errors="coerce").fillna(fill_value)
|
|
|
|
|
| def _clean_orientation_name(name):
|
| if name is None:
|
| return None
|
| name = str(name).strip().lower()
|
| if name in BUYER_ORIENTATION_LIBRARY:
|
| return name
|
| alias_map = {
|
| "family": "family_oriented",
|
| "investment": "investment_oriented",
|
| "lifestyle": "lifestyle_oriented",
|
| "affordability": "value_oriented",
|
| "value": "value_oriented",
|
| }
|
| return alias_map.get(name, name)
|
|
|
|
|
| def _orientation_label(name):
|
| if not name:
|
| return "general"
|
| return str(name).replace("_", " ").title()
|
|
|
|
|
| def _core_profile_from_orientation(name):
|
| clean = _clean_orientation_name(name)
|
| return ORIENTATION_TO_CORE_PROFILE.get(clean, None)
|
|
|
|
|
| def _build_listing_key(df):
|
| if "url" in df.columns:
|
| return "url"
|
| if "listing_summary" in df.columns:
|
| return "listing_summary"
|
| return None
|
|
|
|
|
|
|
|
|
|
|
|
|
| def prepare_listing_features_for_simulation(raw_df):
|
| """
|
| Rebuild engineered features for simulation layer from raw listing data.
|
| """
|
| df = raw_df.copy()
|
|
|
| df = map_kecamatan(df)
|
| df = transform_area_group(df)
|
|
|
| df = transform_area_features(df)
|
| df = transform_structure_features(df)
|
| df = transform_interaction_features(df)
|
|
|
| kec_map_sim, loc_map_sim = fit_desirability_features(df, trusted_cutoff=10)
|
| df = transform_desirability_features(df, kec_map_sim, loc_map_sim)
|
|
|
| return df, kec_map_sim, loc_map_sim
|
|
|
|
|
| def build_location_affordability_table(listing_df, loc_map_sim):
|
| """
|
| Weighted land price / m² reference by area.
|
| """
|
| land_ref = (
|
| listing_df.groupby("location", as_index=False)
|
| .agg(typical_land_area_m2=("land_area_m2", "median"))
|
| )
|
|
|
| aff = loc_map_sim.merge(land_ref, on="location", how="left").copy()
|
|
|
| aff["weighted_land_price_per_m2"] = np.where(
|
| aff["typical_land_area_m2"].fillna(0) > 0,
|
| aff["loc_weighted_price"] / aff["typical_land_area_m2"],
|
| np.nan,
|
| )
|
|
|
| return aff
|
|
|
|
|
| def build_listing_dimension_table(listing_sim_df, location_afford_df):
|
| """
|
| Listing-level 12-dimension table for Buyer Fit Simulation.
|
| """
|
| df = listing_sim_df.copy()
|
|
|
| df = df.merge(
|
| location_afford_df[["location", "weighted_land_price_per_m2"]],
|
| on="location",
|
| how="left",
|
| )
|
|
|
| df["is_rumah"] = (df["building_type"] == "Rumah").astype(float)
|
| df["is_apartemen"] = (df["building_type"] == "Apartemen").astype(float)
|
| df["is_ruko"] = (df["building_type"] == "Ruko").astype(float)
|
|
|
| bath_bed_ratio = df["bath_bed_ratio"].fillna(0)
|
| df["bath_bed_ratio_closeness"] = 1 - np.minimum(np.abs(bath_bed_ratio - 1), 1)
|
|
|
| df["size_fit"] = (
|
| 0.6 * norm01(np.log1p(_safe_numeric(df, "building_area_m2"))) +
|
| 0.4 * norm01(np.log1p(_safe_numeric(df, "land_area_m2")))
|
| )
|
|
|
| df["room_functionality_fit"] = (
|
| 0.4 * norm01(_safe_numeric(df, "bedroom_total")) +
|
| 0.4 * norm01(_safe_numeric(df, "bathroom_total")) +
|
| 0.2 * norm01(_safe_numeric(df, "carport_total"))
|
| )
|
|
|
| df["family_fit"] = (
|
| 0.5 * norm01(_safe_numeric(df, "family_access_score")) +
|
| 0.3 * norm01(_safe_numeric(df, "family_density_score")) +
|
| 0.2 * norm01(_safe_numeric(df, "family_access_zone_score"))
|
| )
|
|
|
| df["school_fit"] = (
|
| 0.6 * norm01(_safe_numeric(df, "school_access_score")) +
|
| 0.4 * norm01(_safe_numeric(df, "fav_school_score"))
|
| )
|
|
|
| df["rentability_fit"] = (
|
| 0.35 * norm01(_safe_numeric(df, "univ_access_score")) +
|
| 0.30 * df["bath_bed_ratio_closeness"] +
|
| 0.20 * norm01(_safe_numeric(df, "bedroom_total")) +
|
| 0.15 * _safe_numeric(df, "investment_property_flag")
|
| )
|
|
|
| df["lifestyle_fit"] = (
|
| 0.65 * norm01(_safe_numeric(df, "lifestyle_score")) +
|
| 0.35 * norm01(_safe_numeric(df, "mall_access_score"))
|
| )
|
|
|
| df["prestige_fit"] = norm01(_safe_numeric(df, "location_premium_score"))
|
| df["access_convenience_fit"] = norm01(_safe_numeric(df, "access_quality_score"))
|
|
|
| df["mixed_use_fit"] = (
|
| 0.5 * df["is_ruko"] +
|
| 0.25 * norm01(_safe_numeric(df, "lifestyle_score")) +
|
| 0.25 * norm01(_safe_numeric(df, "access_quality_score"))
|
| )
|
|
|
| df["affordability_fit"] = 1 - norm01(_safe_numeric(df, "weighted_land_price_per_m2"))
|
| df["rumah_fit"] = df["is_rumah"]
|
| df["apartemen_fit"] = df["is_apartemen"]
|
|
|
| keep_cols = [
|
| "location",
|
| "building_type",
|
| "price_million_idr",
|
| "building_area_m2",
|
| "land_area_m2",
|
| "bedroom_total",
|
| "bathroom_total",
|
| "carport_total",
|
| "listing_summary",
|
| "url",
|
| ]
|
| keep_cols = [c for c in keep_cols if c in df.columns]
|
|
|
| return df[keep_cols + DIMENSIONS].copy()
|
|
|
|
|
| def build_area_simulation_table(listing_sim_df, loc_map_sim, location_afford_df):
|
| """
|
| Area-level 12-dimension table for price-shock simulation.
|
| """
|
| area_df = (
|
| listing_sim_df.groupby("location", as_index=False)
|
| .agg(
|
| kecamatan_mapped=("kecamatan_mapped", "first"),
|
| typical_price=("price_million_idr", "median"),
|
| typical_building_area=("building_area_m2", "median"),
|
| typical_land_area=("land_area_m2", "median"),
|
| bedroom_total=("bedroom_total", "median"),
|
| bathroom_total=("bathroom_total", "median"),
|
| carport_total=("carport_total", "median"),
|
| bath_bed_ratio=("bath_bed_ratio", "median"),
|
| family_access_score=("family_access_score", "mean"),
|
| family_density_score=("family_density_score", "mean"),
|
| family_access_zone_score=("family_access_zone_score", "mean"),
|
| school_access_score=("school_access_score", "mean"),
|
| fav_school_score=("fav_school_score", "mean"),
|
| univ_access_score=("univ_access_score", "mean"),
|
| lifestyle_score=("lifestyle_score", "mean"),
|
| mall_access_score=("mall_access_score", "mean"),
|
| access_quality_score=("access_quality_score", "mean"),
|
| investment_property_flag=("investment_property_flag", "mean"),
|
| rumah_fit=("building_type", lambda s: (s == "Rumah").mean()),
|
| apartemen_fit=("building_type", lambda s: (s == "Apartemen").mean()),
|
| ruko_share=("building_type", lambda s: (s == "Ruko").mean()),
|
| )
|
| )
|
|
|
| area_df = area_df.merge(
|
| loc_map_sim[["location", "loc_weighted_price", "location_premium_score"]],
|
| on="location",
|
| how="left",
|
| )
|
|
|
| area_df = area_df.merge(
|
| location_afford_df[["location", "weighted_land_price_per_m2"]],
|
| on="location",
|
| how="left",
|
| )
|
|
|
| area_df["size_fit"] = (
|
| 0.6 * norm01(np.log1p(_safe_numeric(area_df, "typical_building_area"))) +
|
| 0.4 * norm01(np.log1p(_safe_numeric(area_df, "typical_land_area")))
|
| )
|
|
|
| area_df["room_functionality_fit"] = (
|
| 0.4 * norm01(_safe_numeric(area_df, "bedroom_total")) +
|
| 0.4 * norm01(_safe_numeric(area_df, "bathroom_total")) +
|
| 0.2 * norm01(_safe_numeric(area_df, "carport_total"))
|
| )
|
|
|
| area_df["family_fit"] = (
|
| 0.5 * norm01(_safe_numeric(area_df, "family_access_score")) +
|
| 0.3 * norm01(_safe_numeric(area_df, "family_density_score")) +
|
| 0.2 * norm01(_safe_numeric(area_df, "family_access_zone_score"))
|
| )
|
|
|
| area_df["school_fit"] = (
|
| 0.6 * norm01(_safe_numeric(area_df, "school_access_score")) +
|
| 0.4 * norm01(_safe_numeric(area_df, "fav_school_score"))
|
| )
|
|
|
| bath_bed_ratio_closeness = 1 - np.minimum(
|
| np.abs(_safe_numeric(area_df, "bath_bed_ratio") - 1),
|
| 1,
|
| )
|
|
|
| area_df["rentability_fit"] = (
|
| 0.35 * norm01(_safe_numeric(area_df, "univ_access_score")) +
|
| 0.30 * bath_bed_ratio_closeness +
|
| 0.20 * norm01(_safe_numeric(area_df, "bedroom_total")) +
|
| 0.15 * _safe_numeric(area_df, "investment_property_flag")
|
| )
|
|
|
| area_df["lifestyle_fit"] = (
|
| 0.65 * norm01(_safe_numeric(area_df, "lifestyle_score")) +
|
| 0.35 * norm01(_safe_numeric(area_df, "mall_access_score"))
|
| )
|
|
|
| area_df["prestige_fit"] = norm01(_safe_numeric(area_df, "location_premium_score"))
|
| area_df["access_convenience_fit"] = norm01(_safe_numeric(area_df, "access_quality_score"))
|
|
|
| area_df["mixed_use_fit"] = (
|
| 0.5 * _safe_numeric(area_df, "ruko_share") +
|
| 0.25 * norm01(_safe_numeric(area_df, "lifestyle_score")) +
|
| 0.25 * norm01(_safe_numeric(area_df, "access_quality_score"))
|
| )
|
|
|
| area_df["affordability_fit"] = 1 - norm01(_safe_numeric(area_df, "weighted_land_price_per_m2"))
|
|
|
| keep_cols = [
|
| "location",
|
| "kecamatan_mapped",
|
| "typical_price",
|
| "loc_weighted_price",
|
| "weighted_land_price_per_m2",
|
| ] + DIMENSIONS
|
|
|
| return area_df[keep_cols].copy()
|
|
|
|
|
| def prepare_simulation_base(raw_listing_df=None):
|
| """
|
| One-stop prep for both simulation modules.
|
| """
|
| if raw_listing_df is None:
|
| raw_listing_df = inference_core.listing_df.copy()
|
|
|
| listing_sim_df, kec_map_sim, loc_map_sim = prepare_listing_features_for_simulation(raw_listing_df)
|
| location_afford_df = build_location_affordability_table(listing_sim_df, loc_map_sim)
|
| listing_dimension_df = build_listing_dimension_table(listing_sim_df, location_afford_df)
|
| area_sim_df = build_area_simulation_table(listing_sim_df, loc_map_sim, location_afford_df)
|
|
|
| return {
|
| "raw_listing_df": raw_listing_df,
|
| "listing_sim_df": listing_sim_df,
|
| "kec_map_sim": kec_map_sim,
|
| "loc_map_sim": loc_map_sim,
|
| "location_afford_df": location_afford_df,
|
| "listing_dimension_df": listing_dimension_df,
|
| "area_sim_df": area_sim_df,
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
| def compute_clarity_score(ratings_dict):
|
| s = pd.Series(ratings_dict, dtype=float).clip(lower=0)
|
| s = s.reindex(DIMENSIONS).fillna(0.0)
|
|
|
| clarity = s.std() / 3.0
|
| return float(np.clip(clarity, 0, 1))
|
|
|
|
|
| def ratings_to_weights(ratings_dict):
|
| s = pd.Series(ratings_dict, dtype=float).clip(lower=0)
|
| s = s.reindex(DIMENSIONS).fillna(0.0)
|
|
|
| total = s.sum()
|
| if total == 0:
|
| return pd.Series(1 / len(s), index=s.index)
|
|
|
| return s / total
|
|
|
|
|
| def get_orientation_template(name):
|
| clean = _clean_orientation_name(name)
|
| if clean not in BUYER_ORIENTATION_LIBRARY:
|
| raise ValueError(f"Unknown buyer orientation: {name}")
|
| return BUYER_ORIENTATION_LIBRARY[clean].copy()
|
|
|
|
|
| def build_final_client_weights(client_ratings, buyer_orientation):
|
| clean_orientation = _clean_orientation_name(buyer_orientation)
|
| fallback_ratings = BUYER_ORIENTATION_LIBRARY[clean_orientation]
|
|
|
| client_w = ratings_to_weights(client_ratings)
|
| fallback_w = ratings_to_weights(fallback_ratings)
|
|
|
| clarity = compute_clarity_score(client_ratings)
|
|
|
| final_w = clarity * client_w + (1 - clarity) * fallback_w
|
| final_w = final_w / final_w.sum()
|
|
|
| return final_w, clarity
|
|
|
|
|
| def build_single_client_input(
|
| location,
|
| buyer_orientation,
|
| budget_million_idr=None,
|
| consider_similar_areas=False,
|
| custom_ratings=None,
|
| ):
|
| """
|
| Creates a single-client row for deployment use.
|
| """
|
| clean_orientation = _clean_orientation_name(buyer_orientation)
|
| if clean_orientation not in BUYER_ORIENTATION_LIBRARY:
|
| raise ValueError(f"Unknown buyer orientation: {buyer_orientation}")
|
|
|
| base_ratings = BUYER_ORIENTATION_LIBRARY[clean_orientation].copy()
|
|
|
| if custom_ratings:
|
| for k, v in custom_ratings.items():
|
| if k in DIMENSIONS and v is not None:
|
| base_ratings[k] = float(v)
|
|
|
| row = {
|
| "client_id": "client_1",
|
| "location": location,
|
| "buyer_orientation": clean_orientation,
|
| "budget_million_idr": budget_million_idr,
|
| "strict_location": not bool(consider_similar_areas),
|
| "fallback_to_kecamatan": bool(consider_similar_areas),
|
| "fallback_to_alternatives": bool(consider_similar_areas),
|
| }
|
| row.update(base_ratings)
|
|
|
| return row
|
|
|
|
|
| def get_candidate_listings_for_client(
|
| client_row,
|
| listing_dim_df,
|
| min_candidates=12,
|
| use_core_recs=True,
|
| ):
|
| target_location = client_row["location"]
|
| strict_location = bool(client_row.get("strict_location", True))
|
| fallback_to_kecamatan = bool(client_row.get("fallback_to_kecamatan", True))
|
| fallback_to_alternatives = bool(client_row.get("fallback_to_alternatives", False))
|
|
|
| exact_df = listing_dim_df[
|
| listing_dim_df["location"] == target_location
|
| ].copy()
|
|
|
| if strict_location:
|
| return exact_df
|
|
|
| candidate_df = exact_df.copy()
|
|
|
| if len(candidate_df) < min_candidates and fallback_to_kecamatan:
|
| target_kec = inference_core.location_to_kec.get(target_location)
|
|
|
| if target_kec is not None:
|
| same_kec_locations = [
|
| loc for loc, kec in inference_core.location_to_kec.items()
|
| if kec == target_kec
|
| ]
|
|
|
| kec_df = listing_dim_df[
|
| listing_dim_df["location"].isin(same_kec_locations)
|
| ].copy()
|
|
|
| candidate_df = pd.concat([candidate_df, kec_df], ignore_index=True)
|
|
|
| if len(candidate_df) < min_candidates and fallback_to_alternatives and use_core_recs:
|
| try:
|
| user_info = {
|
| "location": client_row["location"],
|
| "budget_million_idr": client_row.get("budget_million_idr"),
|
| "profile": _core_profile_from_orientation(client_row.get("buyer_orientation")),
|
| }
|
|
|
| recs = inference_core.recommend_locations(user_info, top_n=5)
|
|
|
| if not recs.empty:
|
| alt_areas = recs["Alternative Area"].tolist()
|
| alt_df = listing_dim_df[
|
| listing_dim_df["location"].isin(alt_areas)
|
| ].copy()
|
| candidate_df = pd.concat([candidate_df, alt_df], ignore_index=True)
|
|
|
| except Exception as e:
|
| print(f"Alternative-area fallback failed: {e}")
|
|
|
| listing_key = _build_listing_key(candidate_df)
|
| if listing_key:
|
| candidate_df = candidate_df.drop_duplicates(subset=listing_key)
|
| else:
|
| candidate_df = candidate_df.drop_duplicates()
|
|
|
| return candidate_df
|
|
|
|
|
| def score_candidate_listings_for_client(
|
| client_row,
|
| candidate_df,
|
| dimensions=DIMENSIONS,
|
| ):
|
| if candidate_df.empty:
|
| return pd.DataFrame()
|
|
|
| df = candidate_df.copy()
|
|
|
| client_ratings = {dim: client_row[dim] for dim in dimensions}
|
| buyer_orientation = client_row["buyer_orientation"]
|
|
|
| final_w, clarity = build_final_client_weights(
|
| client_ratings=client_ratings,
|
| buyer_orientation=buyer_orientation,
|
| )
|
|
|
| df["base_match_score"] = 0.0
|
| for dim in dimensions:
|
| df["base_match_score"] += final_w[dim] * df[dim]
|
|
|
| budget = client_row.get("budget_million_idr", None)
|
| if budget is not None and "price_million_idr" in df.columns:
|
| over_ratio = (df["price_million_idr"] - budget) / max(budget, 1)
|
|
|
| df["budget_penalty"] = np.where(
|
| df["price_million_idr"] <= budget,
|
| 1.0,
|
| np.clip(1 - over_ratio, 0.2, 1.0),
|
| )
|
| else:
|
| df["budget_penalty"] = 1.0
|
|
|
| df["match_score"] = df["base_match_score"] * df["budget_penalty"]
|
|
|
| df["client_id"] = client_row["client_id"]
|
| df["buyer_orientation"] = buyer_orientation
|
| df["clarity_score"] = clarity
|
|
|
| return df.sort_values("match_score", ascending=False)
|
|
|
|
|
| def _make_listing_reason(row, buyer_orientation):
|
| orientation = _clean_orientation_name(buyer_orientation)
|
|
|
| if orientation == "family_oriented":
|
| return "Strong school and family fit with practical household layout."
|
| if orientation == "investment_oriented":
|
| return "Good rentability profile with relatively efficient buyer economics."
|
| if orientation == "lifestyle_oriented":
|
| return "Better lifestyle access with a stronger convenience profile."
|
| if orientation == "value_oriented":
|
| return "More cost-efficient option while preserving overall fit."
|
|
|
| return "Balanced fit across buyer priorities."
|
|
|
|
|
| def format_buyer_fit_results(scored_df, client_row, top_n=5):
|
| if scored_df.empty:
|
| return {
|
| "client_summary": {
|
| "Preferred Area": client_row.get("location"),
|
| "Buyer Orientation": _orientation_label(client_row.get("buyer_orientation")),
|
| "Budget (IDR Mn)": client_row.get("budget_million_idr"),
|
| "Consider Similar Areas": bool(client_row.get("fallback_to_alternatives", False)),
|
| "Clarity Score": None,
|
| },
|
| "summary": f"No suitable listings were found for {client_row.get('location')}.",
|
| "listing_table": pd.DataFrame(),
|
| "details_table": pd.DataFrame(),
|
| }
|
|
|
| top_df = scored_df.head(top_n).copy()
|
|
|
| top_df["Why it Fits"] = top_df.apply(
|
| lambda row: _make_listing_reason(row, client_row.get("buyer_orientation")),
|
| axis=1,
|
| )
|
|
|
| display_cols = []
|
| rename_map = {}
|
|
|
| for old, new in [
|
| ("location", "Location"),
|
| ("building_type", "Type"),
|
| ("price_million_idr", "Price"),
|
| ("building_area_m2", "Size (m²)"),
|
| ("bedroom_total", "Bed"),
|
| ("bathroom_total", "Bath"),
|
| ("match_score", "Match Score"),
|
| ("Why it Fits", "Why it Fits"),
|
| ]:
|
| if old in top_df.columns:
|
| display_cols.append(old)
|
| rename_map[old] = new
|
|
|
| if "url" in top_df.columns:
|
| display_cols.append("url")
|
|
|
| listing_table = top_df[display_cols].rename(columns=rename_map).copy()
|
|
|
| if "Match Score" in listing_table.columns:
|
| listing_table["Match Score"] = listing_table["Match Score"].round(3)
|
| if "Price" in listing_table.columns:
|
| listing_table["Price"] = pd.to_numeric(listing_table["Price"], errors="coerce").round(1)
|
| if "Size (m²)" in listing_table.columns:
|
| listing_table["Size (m²)"] = pd.to_numeric(listing_table["Size (m²)"], errors="coerce").round(1)
|
|
|
| clarity_score = float(top_df["clarity_score"].iloc[0]) if "clarity_score" in top_df.columns else None
|
|
|
| consider_similar = bool(client_row.get("fallback_to_alternatives", False))
|
| target_area = client_row.get("location")
|
| orientation_label = _orientation_label(client_row.get("buyer_orientation"))
|
|
|
| if consider_similar:
|
| summary = (
|
| f"For a {orientation_label.lower()} buyer targeting {target_area}, "
|
| f"these listings provide the strongest fit after allowing similar-area substitutes."
|
| )
|
| else:
|
| summary = (
|
| f"For a {orientation_label.lower()} buyer targeting {target_area}, "
|
| f"these are the strongest in-area listing matches."
|
| )
|
|
|
| client_summary = {
|
| "Preferred Area": target_area,
|
| "Buyer Orientation": orientation_label,
|
| "Budget (IDR Mn)": client_row.get("budget_million_idr"),
|
| "Consider Similar Areas": consider_similar,
|
| "Clarity Score": round(clarity_score, 2) if clarity_score is not None else None,
|
| }
|
|
|
| details_cols = [
|
| c for c in [
|
| "location",
|
| "building_type",
|
| "price_million_idr",
|
| "building_area_m2",
|
| "land_area_m2",
|
| "bedroom_total",
|
| "bathroom_total",
|
| "carport_total",
|
| "base_match_score",
|
| "budget_penalty",
|
| "match_score",
|
| "listing_summary",
|
| "url",
|
| ] if c in top_df.columns
|
| ]
|
| details_table = top_df[details_cols].copy()
|
|
|
| return {
|
| "client_summary": client_summary,
|
| "summary": summary,
|
| "listing_table": listing_table.reset_index(drop=True),
|
| "details_table": details_table.reset_index(drop=True),
|
| }
|
|
|
|
|
| def run_buyer_fit_simulation(
|
| location,
|
| buyer_orientation,
|
| budget_million_idr=None,
|
| consider_similar_areas=False,
|
| custom_ratings=None,
|
| simulation_base=None,
|
| top_n=5,
|
| min_candidates=12,
|
| ):
|
| """
|
| Deployment wrapper for single-client buyer fit simulation.
|
| """
|
| if simulation_base is None:
|
| simulation_base = prepare_simulation_base()
|
|
|
| listing_dimension_df = simulation_base["listing_dimension_df"]
|
|
|
| client_row = build_single_client_input(
|
| location=location,
|
| buyer_orientation=buyer_orientation,
|
| budget_million_idr=budget_million_idr,
|
| consider_similar_areas=consider_similar_areas,
|
| custom_ratings=custom_ratings,
|
| )
|
|
|
| candidate_df = get_candidate_listings_for_client(
|
| client_row=client_row,
|
| listing_dim_df=listing_dimension_df,
|
| min_candidates=min_candidates,
|
| use_core_recs=True,
|
| )
|
|
|
| scored_df = score_candidate_listings_for_client(
|
| client_row=client_row,
|
| candidate_df=candidate_df,
|
| )
|
|
|
| formatted = format_buyer_fit_results(
|
| scored_df=scored_df,
|
| client_row=client_row,
|
| top_n=top_n,
|
| )
|
|
|
| return {
|
| "client_input": client_row,
|
| "candidate_count": len(candidate_df),
|
| "scored_candidates": scored_df,
|
| **formatted,
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
| def get_substitute_area_pool(target_area, buyer_orientation=None, top_n=12):
|
| user_info = {
|
| "location": target_area,
|
| "profile": _core_profile_from_orientation(buyer_orientation),
|
| }
|
|
|
| recs = inference_core.recommend_locations(user_info, top_n=top_n)
|
|
|
| pool = [target_area]
|
| similarity_map = {target_area: 1.0}
|
|
|
| if not recs.empty:
|
| for _, row in recs.iterrows():
|
| area = row["Alternative Area"]
|
| pool.append(area)
|
| similarity_map[area] = row["Similarity Score"]
|
|
|
| pool = list(dict.fromkeys(pool))
|
| return pool, similarity_map
|
|
|
|
|
| def simulate_target_area_price_shock(
|
| area_sim_df,
|
| target_area,
|
| buyer_orientation=None,
|
| shock_pct=0.10,
|
| top_n_pool=12,
|
| similarity_weight=0.60,
|
| orientation_weight=0.40,
|
| ):
|
| """
|
| Target-relative substitute simulation.
|
| If orientation is None, defaults to value-oriented for a general affordability-sensitive view.
|
| """
|
| df = area_sim_df.copy()
|
|
|
| clean_orientation = _clean_orientation_name(buyer_orientation)
|
| if clean_orientation is None:
|
| clean_orientation = "value_oriented"
|
|
|
| pool, similarity_map = get_substitute_area_pool(
|
| target_area=target_area,
|
| buyer_orientation=clean_orientation,
|
| top_n=top_n_pool,
|
| )
|
|
|
| df = df[df["location"].isin(pool)].copy()
|
|
|
| if df.empty:
|
| return pd.DataFrame()
|
|
|
| df["similarity_to_target"] = df["location"].map(similarity_map).fillna(0.0)
|
|
|
| df["weighted_land_price_per_m2_after_shock"] = df["weighted_land_price_per_m2"].copy()
|
|
|
| mask = df["location"] == target_area
|
| df.loc[mask, "weighted_land_price_per_m2_after_shock"] = (
|
| df.loc[mask, "weighted_land_price_per_m2_after_shock"] * (1 + shock_pct)
|
| )
|
|
|
| df["affordability_fit_before"] = 1 - norm01(df["weighted_land_price_per_m2"])
|
| df["affordability_fit_after"] = 1 - norm01(df["weighted_land_price_per_m2_after_shock"])
|
|
|
| orientation_ratings = BUYER_ORIENTATION_LIBRARY[clean_orientation]
|
| weights = ratings_to_weights(orientation_ratings)
|
|
|
| df["orientation_score_before"] = 0.0
|
| for dim in DIMENSIONS:
|
| df["orientation_score_before"] += weights[dim] * df[dim]
|
|
|
| df["orientation_score_after"] = (
|
| df["orientation_score_before"]
|
| - weights["affordability_fit"] * df["affordability_fit_before"]
|
| + weights["affordability_fit"] * df["affordability_fit_after"]
|
| )
|
|
|
| df["substitute_score_before"] = (
|
| similarity_weight * df["similarity_to_target"] +
|
| orientation_weight * df["orientation_score_before"]
|
| )
|
|
|
| df["substitute_score_after"] = (
|
| similarity_weight * df["similarity_to_target"] +
|
| orientation_weight * df["orientation_score_after"]
|
| )
|
|
|
| for col in [
|
| "similarity_to_target",
|
| "orientation_score_before",
|
| "orientation_score_after",
|
| "substitute_score_before",
|
| "substitute_score_after",
|
| ]:
|
| df[col] = pd.to_numeric(df[col], errors="coerce")
|
|
|
| df = df.dropna(
|
| subset=["substitute_score_before", "substitute_score_after"]
|
| ).copy()
|
|
|
| df["rank_before"] = df["substitute_score_before"].rank(ascending=False, method="dense")
|
| df["rank_after"] = df["substitute_score_after"].rank(ascending=False, method="dense")
|
| df["rank_change"] = df["rank_before"] - df["rank_after"]
|
| df["score_change"] = df["substitute_score_after"] - df["substitute_score_before"]
|
|
|
| out = df.copy()
|
| out["Area"] = out["location"]
|
| out["Role"] = np.where(out["location"] == target_area, "Target", "Alternative")
|
| out["Buyer Fit Before"] = out["orientation_score_before"].round(3)
|
| out["Buyer Fit After"] = out["orientation_score_after"].round(3)
|
| out["Similarity"] = out["similarity_to_target"].round(2)
|
| out["Rank Before"] = out["rank_before"].round().astype("Int64")
|
| out["Rank After"] = out["rank_after"].round().astype("Int64")
|
| out["Rank Change"] = out["rank_change"].round().astype("Int64")
|
| out["Price / m² Before"] = out["weighted_land_price_per_m2"].round(2)
|
| out["Price / m² After Shock"] = out["weighted_land_price_per_m2_after_shock"].round(2)
|
| out["Substitute Strength Before"] = out["substitute_score_before"].round(3)
|
| out["Substitute Strength After"] = out["substitute_score_after"].round(3)
|
|
|
| target_row = out[out["Role"] == "Target"].copy()
|
| alt_rows = out[out["Role"] == "Alternative"].copy()
|
|
|
| alt_rows = alt_rows.sort_values(
|
| by=["Buyer Fit After", "Substitute Strength After", "Similarity"],
|
| ascending=[False, False, False],
|
| )
|
|
|
| out = pd.concat([target_row, alt_rows], ignore_index=True)
|
|
|
| return out
|
|
|
|
|
| def summarize_target_area_price_shock(
|
| shock_result,
|
| target_area,
|
| shock_pct,
|
| buyer_orientation=None,
|
| top_n=3,
|
| ):
|
| if shock_result.empty:
|
| return f"No strong substitutes were found for {target_area}."
|
|
|
| alt_only = shock_result[shock_result["Role"] == "Alternative"].copy()
|
|
|
| if alt_only.empty:
|
| return f"No strong substitutes were found for {target_area}."
|
|
|
| gainers = alt_only.head(top_n)["Area"].tolist()
|
|
|
| orientation_label = _orientation_label(
|
| _clean_orientation_name(buyer_orientation) or "value_oriented"
|
| ).lower()
|
|
|
| return (
|
| f"{target_area} increased in price by {shock_pct:.0%}. "
|
| f"Under this scenario, the areas most likely to become stronger substitutes "
|
| f"for {orientation_label} buyers are {', '.join(gainers)}."
|
| )
|
|
|
|
|
| def format_price_shock_results(
|
| shock_result,
|
| target_area,
|
| shock_pct,
|
| buyer_orientation=None,
|
| top_n=5,
|
| ):
|
| if shock_result.empty:
|
| return {
|
| "case_summary": f"No substitute signal was found for {target_area}.",
|
| "target_impact": {},
|
| "substitute_table": pd.DataFrame(),
|
| "details_table": pd.DataFrame(),
|
| }
|
|
|
| target_row = shock_result[shock_result["Role"] == "Target"].copy()
|
| alt_rows = shock_result[shock_result["Role"] == "Alternative"].copy()
|
|
|
| case_summary = summarize_target_area_price_shock(
|
| shock_result=shock_result,
|
| target_area=target_area,
|
| shock_pct=shock_pct,
|
| buyer_orientation=buyer_orientation,
|
| top_n=min(top_n, 3),
|
| )
|
|
|
| target_impact = {}
|
| if not target_row.empty:
|
| tr = target_row.iloc[0]
|
| target_impact = {
|
| "Target Area": target_area,
|
| "Shock Applied": f"{shock_pct:.0%}",
|
| "Buyer Orientation": _orientation_label(
|
| _clean_orientation_name(buyer_orientation) or "value_oriented"
|
| ),
|
| "Buyer Fit Before": round(float(tr["Buyer Fit Before"]), 3),
|
| "Buyer Fit After": round(float(tr["Buyer Fit After"]), 3),
|
| "Rank Before": int(tr["Rank Before"]) if pd.notna(tr["Rank Before"]) else None,
|
| "Rank After": int(tr["Rank After"]) if pd.notna(tr["Rank After"]) else None,
|
| "Rank Change": int(tr["Rank Change"]) if pd.notna(tr["Rank Change"]) else None,
|
| "Price / m² Before": round(float(tr["Price / m² Before"]), 2),
|
| "Price / m² After Shock": round(float(tr["Price / m² After Shock"]), 2),
|
| }
|
|
|
| substitute_cols = [
|
| "Area",
|
| "Buyer Fit After",
|
| "Buyer Fit Before",
|
| "Similarity",
|
| "Rank Change",
|
| ]
|
| substitute_cols = [c for c in substitute_cols if c in alt_rows.columns]
|
|
|
| substitute_table = (
|
| alt_rows[substitute_cols]
|
| .head(top_n)
|
| .reset_index(drop=True)
|
| .copy()
|
| )
|
|
|
| details_cols = [
|
| "Area",
|
| "Role",
|
| "Buyer Fit Before",
|
| "Buyer Fit After",
|
| "Similarity",
|
| "Rank Before",
|
| "Rank After",
|
| "Rank Change",
|
| "Price / m² Before",
|
| "Price / m² After Shock",
|
| "Substitute Strength Before",
|
| "Substitute Strength After",
|
| ]
|
| details_cols = [c for c in details_cols if c in shock_result.columns]
|
| details_table = shock_result[details_cols].copy().reset_index(drop=True)
|
|
|
| return {
|
| "case_summary": case_summary,
|
| "target_impact": target_impact,
|
| "substitute_table": substitute_table,
|
| "details_table": details_table,
|
| }
|
|
|
|
|
| def extract_area_tradeoff_drivers(area_sim_df, areas, top_k=3):
|
| """
|
| For each area, identify the strongest fit dimensions to help explain
|
| why it may become a stronger substitute.
|
| """
|
| if area_sim_df is None or area_sim_df.empty:
|
| return {}
|
|
|
| df = area_sim_df.copy()
|
| df = df[df["location"].isin(areas)].copy()
|
|
|
| if df.empty:
|
| return {}
|
|
|
| driver_map = {}
|
|
|
| for _, row in df.iterrows():
|
| area = row["location"]
|
|
|
| dim_scores = {}
|
| for dim in DIMENSIONS:
|
| dim_scores[dim] = float(row.get(dim, 0.0))
|
|
|
| top_dims = sorted(
|
| dim_scores.items(),
|
| key=lambda x: x[1],
|
| reverse=True
|
| )[:top_k]
|
|
|
| driver_map[area] = [
|
| {
|
| "dimension": dim,
|
| "label": DIMENSION_LABELS.get(dim, dim),
|
| "score": round(score, 3)
|
| }
|
| for dim, score in top_dims
|
| ]
|
|
|
| return driver_map
|
|
|
| def build_shock_reasoning_payload(shock_result, area_sim_df, target_area, top_n=3):
|
| """
|
| Build a compact comparison payload for Gemini so it explains using
|
| deterministic outputs, not invented logic.
|
| """
|
| if shock_result is None or shock_result.empty:
|
| return {}
|
|
|
| top_subs = (
|
| shock_result[shock_result["Role"] == "Alternative"]
|
| .head(top_n)
|
| .copy()
|
| )
|
|
|
| if top_subs.empty:
|
| return {}
|
|
|
| areas = [target_area] + top_subs["Area"].tolist()
|
| driver_map = extract_area_tradeoff_drivers(area_sim_df, areas, top_k=3)
|
|
|
| payload = {
|
| "target_area": target_area,
|
| "target_drivers": driver_map.get(target_area, []),
|
| "substitutes": []
|
| }
|
|
|
| for _, row in top_subs.iterrows():
|
| area = row["Area"]
|
|
|
| payload["substitutes"].append({
|
| "area": area,
|
| "buyer_fit_before": round(float(row.get("Buyer Fit Before", 0)), 3),
|
| "buyer_fit_after": round(float(row.get("Buyer Fit After", 0)), 3),
|
| "similarity": round(float(row.get("Similarity", 0)), 3),
|
| "rank_change": (
|
| int(row["Rank Change"])
|
| if pd.notna(row.get("Rank Change"))
|
| else None
|
| ),
|
| "drivers": driver_map.get(area, [])
|
| })
|
|
|
| return payload
|
|
|
| def gemini_price_shock_summary(result, area_sim_df, client=None):
|
| """
|
| Optional LLM explanation for area price-shock simulation.
|
| Uses deterministic outputs + extracted drivers to explain why
|
| areas became stronger substitutes.
|
| """
|
| fallback = result.get("case_summary", "")
|
| active_client = client if client is not None else gemini_client
|
|
|
| if active_client is None:
|
| return fallback
|
|
|
| try:
|
| target_impact = result.get("target_impact", {})
|
| substitute_table = result.get("substitute_table", pd.DataFrame())
|
|
|
| if substitute_table is None or substitute_table.empty:
|
| return fallback
|
|
|
| target_area = target_impact.get("Target Area")
|
| reasoning_payload = build_shock_reasoning_payload(
|
| shock_result=result.get("shock_result"),
|
| area_sim_df=area_sim_df,
|
| target_area=target_area,
|
| top_n=3
|
| )
|
|
|
| prompt = f"""
|
| Explain a deterministic Bandung housing price-shock simulation.
|
|
|
| Target impact:
|
| {target_impact}
|
|
|
| Top substitute areas:
|
| {substitute_table.head(3).to_dict(orient="records")}
|
|
|
| Reasoning payload:
|
| {reasoning_payload}
|
|
|
| Rules:
|
| - Explain WHY these areas became stronger substitutes
|
| - Use only the supplied metrics and driver labels, DO NOT repeat raw metrics.
|
| - Mention tradeoffs briefly
|
| - Do not invent new data, scores, or claims
|
| - Similarity means how close the substitute remains to the original target profile
|
| - Buyer Fit reflects post-shock suitability for that buyer orientation
|
| - Rank Change means upward movement is stronger substitute positioning
|
| - Max 4 sentences
|
| - Under 90 words
|
| - Plain text only, CONCRETE comparison wording.
|
| """
|
|
|
| response = active_client.models.generate_content(
|
| model="gemini-2.5-flash-lite",
|
| contents=prompt,
|
| config={
|
| "max_output_tokens": 120,
|
| "temperature": 0.3,
|
| "top_p": 0.95
|
| }
|
| )
|
|
|
| if hasattr(response, "text") and response.text:
|
| text = response.text.strip()
|
| else:
|
| text = str(response).strip()
|
|
|
| if text and len(text.split()) >= 10:
|
| return text
|
|
|
| return fallback
|
|
|
| except Exception as e:
|
| print("Gemini price-shock summary failed:", e)
|
| return fallback
|
|
|
|
|
| def run_area_price_shock_simulation(
|
| target_area,
|
| shock_pct=0.10,
|
| buyer_orientation=None,
|
| simulation_base=None,
|
| top_n=5,
|
| top_n_pool=12,
|
| similarity_weight=0.60,
|
| orientation_weight=0.40,
|
| use_llm_summary=False,
|
| client=None,
|
| ):
|
| """
|
| Deployment wrapper for area price-shock simulation.
|
| """
|
| if simulation_base is None:
|
| simulation_base = prepare_simulation_base()
|
|
|
| area_sim_df = simulation_base["area_sim_df"]
|
|
|
| shock_result = simulate_target_area_price_shock(
|
| area_sim_df=area_sim_df,
|
| target_area=target_area,
|
| buyer_orientation=buyer_orientation,
|
| shock_pct=shock_pct,
|
| top_n_pool=top_n_pool,
|
| similarity_weight=similarity_weight,
|
| orientation_weight=orientation_weight,
|
| )
|
|
|
| formatted = format_price_shock_results(
|
| shock_result=shock_result,
|
| target_area=target_area,
|
| shock_pct=shock_pct,
|
| buyer_orientation=buyer_orientation,
|
| top_n=top_n,
|
| )
|
|
|
| llm_summary = formatted["case_summary"]
|
| if use_llm_summary:
|
| llm_summary = gemini_price_shock_summary(
|
| result={
|
| "shock_result": shock_result,
|
| **formatted
|
| },
|
| area_sim_df=area_sim_df,
|
| client=client
|
| )
|
|
|
| return {
|
| "shock_result": shock_result,
|
| **formatted,
|
| "summary": llm_summary,
|
| } |
