Spaces:
Sleeping
Sleeping
| import pandas as pd | |
| import numpy as np | |
| import joblib | |
| import yfinance as yf | |
| from datetime import datetime, timedelta | |
| import os | |
| from functools import lru_cache | |
| import db_logger | |
| # --- KONFIGURASI FILE --- | |
| # Path untuk file model dan data referensi | |
| MODEL_PATH = 'lgbm_model.joblib' | |
| DATA_PATH = 'processed_real_estate_data.csv' | |
| # ============================================================================== | |
| # DATA CONSTANTS (JANGAN DIHAPUS, DIPAKAI UI) | |
| # Daftar kota, tipe properti, dan tahun yang dikenali model | |
| # ============================================================================== | |
| TRAIN_TOWNS = [ | |
| 'Norwalk', 'Hamden', 'Bloomfield', 'Windsor', 'Bristol', 'Trumbull', 'Newington', | |
| 'Stratford', 'Woodbury', 'Guilford', 'North haven', 'Simsbury', 'New canaan', | |
| 'Wolcott', 'Avon', 'Wallingford', 'Torrington', 'New hartford', 'Wethersfield', | |
| 'Seymour', 'Waterbury', 'Greenwich', 'West hartford', 'New milford', 'Bethel', | |
| 'Hartford', 'Norwich', 'Fairfield', 'West haven', 'Danbury', 'South windsor', | |
| 'Wilton', 'East windsor', 'Plainfield', 'Woodbridge', 'East lyme', 'Derby', | |
| 'New britain', 'Southington', 'Westport', 'Middletown', 'Ellington', 'East haven', | |
| 'New haven', 'Stamford', 'Meriden', 'Westbrook', 'Suffield', 'Canton', 'Vernon', | |
| 'Manchester', 'Woodstock', 'Milford', 'Rocky hill', 'Bridgewater', 'Redding', | |
| 'Old lyme', 'Bridgeport', 'Preston', 'Griswold', 'East hartford', 'East granby', | |
| 'Haddam', 'Old saybrook', 'Shelton', 'Branford', 'Southbury', 'Windham', 'Enfield', | |
| 'Hebron', 'New fairfield', 'Deep river', 'Mansfield', 'Darien', 'Winchester', | |
| 'Willington', 'Litchfield', 'Madison', 'Cromwell', 'Monroe', 'Essex', 'Oxford', | |
| 'Naugatuck', 'Glastonbury', 'Tolland', 'Sharon', 'Brookfield', 'Canterbury', | |
| 'Windsor locks', 'Ledyard', 'Prospect', 'Ridgefield', 'Burlington', 'Portland', | |
| 'Groton', 'Colchester', 'Beacon falls', 'Easton', 'North stonington', 'New london', | |
| 'Plainville', 'East hampton', 'East haddam', 'Newtown', 'Salem', 'Cornwall', | |
| 'Sherman', 'Berlin', 'Stonington', 'Farmington', 'Somers', 'Coventry', 'Lebanon', | |
| 'Putnam', 'Granby', 'Washington', 'Lisbon', 'Ansonia', 'Bolton', 'Middlefield', | |
| 'Orange', 'Weston', 'Watertown', 'Plymouth', 'Cheshire', 'Harwinton', 'Killingly', | |
| 'Brooklyn', 'Marlborough', 'Waterford', 'Clinton', 'Montville', 'North branford', | |
| 'Goshen', 'Bethany', 'Thompson', 'Stafford', 'North canaan', 'Kent', 'Hampton', | |
| 'Sterling', 'Voluntown', 'Middlebury', 'Franklin', 'Columbia', 'Barkhamsted', | |
| 'Roxbury', 'Killingworth', 'Salisbury', 'Warren', 'Durham', 'Thomaston', 'Sprague', | |
| 'Pomfret', 'Chester', 'Hartland', 'Bethlehem', 'Colebrook', 'Morris', 'Andover', | |
| 'Ashford', 'Eastford', 'Bozrah', 'Canaan', 'Lyme', 'Union', 'Chaplin', 'Norfolk', | |
| 'Scotland' | |
| ] | |
| TRAIN_PROP_TYPES = [ | |
| 'Two family', 'Three family', 'Condo', 'Single family', 'Commercial', | |
| 'Apartments', 'Vacant land', 'Four family', 'Industrial', 'Public utility' | |
| ] | |
| TRAIN_YEARS = [str(y) for y in range(2001, 2024)] | |
| # Mapping untuk normalisasi input pengguna | |
| town_map = {t.upper(): t for t in TRAIN_TOWNS} | |
| prop_map = {p.upper(): p for p in TRAIN_PROP_TYPES} | |
| # ============================================================================== | |
| # LOAD RESOURCES | |
| # ============================================================================== | |
| model = None | |
| if os.path.exists(MODEL_PATH): | |
| model = joblib.load(MODEL_PATH) | |
| print("✅ Engine: Model Loaded.") | |
| df_lookup = None | |
| if os.path.exists(DATA_PATH): | |
| try: | |
| # Load dataset untuk imputasi data | |
| df_lookup = pd.read_csv(DATA_PATH, usecols=['Town', 'Property_Residential', 'List Year', 'Assessed Value']) | |
| df_lookup['Town'] = df_lookup['Town'].astype(str).str.title().str.strip() | |
| df_lookup['Property_Residential'] = df_lookup['Property_Residential'].astype(str).str.title().str.strip() | |
| df_lookup['List Year'] = df_lookup['List Year'].astype(str) | |
| # Filter nilai outlier kecil untuk data lookup | |
| df_lookup = df_lookup[df_lookup['Assessed Value'] > 1000] | |
| print("✅ Engine: Lookup Data Loaded.") | |
| except: | |
| pass | |
| # ============================================================================== | |
| # CORE LOGIC | |
| # ============================================================================== | |
| def get_exact_category(user_input, mapping_dict, fallback='***unknown***'): | |
| """ | |
| Normalisasi input string berdasarkan dictionary mapping. | |
| """ | |
| if not user_input: return fallback | |
| key = str(user_input).strip().upper() | |
| return mapping_dict.get(key, fallback) | |
| def get_lagged_stock_data(date_str, tickers=("DHI", "LEN", "^SP500-60"), lag_days=7): | |
| """ | |
| Mengambil data saham historis (lagged) untuk memperkaya fitur model. | |
| Menggunakan cache agar efisien. | |
| """ | |
| FALLBACK_VALS = (85.0, 90.0, 250.0) | |
| try: | |
| prediction_date = datetime.strptime(date_str, '%Y-%m-%d') | |
| except: | |
| return FALLBACK_VALS | |
| target_lagged_date = prediction_date - timedelta(days=lag_days) | |
| start_fetch = target_lagged_date - timedelta(days=14) | |
| try: | |
| ticker_list = list(tickers) | |
| df = yf.download(ticker_list, start=start_fetch, end=prediction_date, auto_adjust=False, progress=False)['Close'] | |
| if df.empty: return FALLBACK_VALS | |
| vals = [] | |
| for t in ticker_list: | |
| col_name = t | |
| # Handle MultiIndex column output dari yfinance | |
| if isinstance(df.columns, pd.MultiIndex): | |
| matches = [c for c in df.columns if t in c] | |
| if matches: col_name = matches[0] | |
| if col_name in df.columns or col_name in df: | |
| series = df[col_name] if col_name in df else df[t] | |
| mask = series.index <= target_lagged_date | |
| val = series.loc[mask].iloc[-1] if mask.any() else series.iloc[0] | |
| if isinstance(val, (pd.Series, np.ndarray)): val = val.item() | |
| vals.append(float(val)) | |
| else: | |
| vals.append(FALLBACK_VALS[ticker_list.index(t)]) | |
| return tuple(vals) | |
| except: | |
| return FALLBACK_VALS | |
| def predict_single(assessed_value, town, property_residential, list_year, prediction_date, return_debug=False): | |
| """ | |
| Melakukan prediksi harga jual untuk satu properti. | |
| """ | |
| debug_info = {} | |
| if model is None: | |
| return (0.0, {"error": "Model not loaded"}) if return_debug else 0.0 | |
| if assessed_value <= 0: assessed_value = 1.0 | |
| # Transformasi logaritma pada input numeric sesuai model training | |
| assessed_value_log = np.log1p(assessed_value) | |
| # Ambil fitur eksternal (saham) | |
| dhi, len_stock, sp500 = get_lagged_stock_data(prediction_date) | |
| # Normalisasi kategori | |
| clean_town = get_exact_category(town, town_map, fallback='***unknown***') | |
| clean_prop = get_exact_category(property_residential, prop_map, fallback='Single family') | |
| clean_year = str(list_year) | |
| # Simpan info debug untuk audit | |
| debug_info = { | |
| "Input_Assessed_Log": float(assessed_value_log), | |
| "Input_Town_Norm": clean_town, | |
| "Input_Prop_Norm": clean_prop, | |
| "Input_Year_Norm": clean_year, | |
| "Stock_DHI": dhi, | |
| "Stock_LEN": len_stock, | |
| "Stock_SP500": sp500 | |
| } | |
| # Siapkan DataFrame input untuk model | |
| input_data = pd.DataFrame([[ | |
| assessed_value_log, clean_town, clean_prop, clean_year, dhi, len_stock, sp500 | |
| ]], columns=['Assessed Value log', 'Town', 'Property_Residential', 'List Year', 'DHI_lag7', 'LEN_lag7', '^SP500-60_lag7']) | |
| # Pastikan tipe data categorical sesuai training | |
| input_data['Town'] = pd.Categorical(input_data['Town'], categories=TRAIN_TOWNS) | |
| input_data['Property_Residential'] = pd.Categorical(input_data['Property_Residential'], categories=TRAIN_PROP_TYPES) | |
| input_data['List Year'] = pd.Categorical(input_data['List Year'], categories=TRAIN_YEARS) | |
| try: | |
| # Prediksi (output model adalah log price) | |
| pred_log = model.predict(input_data)[0] | |
| # Inverse transform (expm1) untuk mendapatkan harga asli | |
| prediction = float(np.expm1(pred_log)) | |
| debug_info["Model_Pred_Log"] = float(pred_log) | |
| except Exception as e: | |
| prediction = 0.0 | |
| debug_info["Error"] = str(e) | |
| # Log prediksi ke database | |
| db_logger.log_prediction( | |
| town=clean_town, | |
| prop_type=clean_prop, | |
| year=clean_year, | |
| assessed=assessed_value, | |
| predicted=prediction, | |
| debug_info=debug_info | |
| ) | |
| if return_debug: | |
| return prediction, debug_info | |
| return prediction | |
| def get_suggested_assessed_value(town, prop_type, year): | |
| """ | |
| Mengembalikan nilai 'Assessed Value' yang disarankan berdasarkan data historis. | |
| Menggunakan nilai MODUS (paling sering muncul) sesuai permintaan user. | |
| Jika tidak ada modus yang jelas, gunakan median sebagai fallback. | |
| """ | |
| if df_lookup is None: return 150000.0 | |
| target_town = str(town).strip().title() | |
| target_prop = str(prop_type).strip().title() | |
| target_year = str(year).strip() | |
| # Filter bertahap: Town + Type + Year -> Town + Type -> Town only | |
| mask1 = (df_lookup['Town'] == target_town) & \ | |
| (df_lookup['Property_Residential'] == target_prop) & \ | |
| (df_lookup['List Year'] == target_year) | |
| subset = df_lookup[mask1] | |
| if subset.empty: | |
| subset = df_lookup[(df_lookup['Town'] == target_town) & \ | |
| (df_lookup['Property_Residential'] == target_prop)] | |
| if subset.empty: | |
| subset = df_lookup[df_lookup['Town'] == target_town] | |
| if not subset.empty: | |
| try: | |
| # Coba cari Modus (nominal yang sering muncul) | |
| modes = subset['Assessed Value'].mode() | |
| if not modes.empty: | |
| return float(modes.iloc[0]) # Ambil modus pertama | |
| else: | |
| return float(subset['Assessed Value'].median()) | |
| except: | |
| return float(subset['Assessed Value'].median()) | |
| return 150000.0 | |
| def calculate_key_metrics(df): | |
| """ | |
| Menghitung KPI dashboard sederhana. | |
| """ | |
| total_value = df['Predicted_Sale_Amount'].sum() | |
| avg_price = df['Predicted_Sale_Amount'].mean() | |
| df['Diff'] = df['Predicted_Sale_Amount'] - df['Assessed Value'] | |
| valid_assessed = df[df['Assessed Value'] > 0] | |
| if not valid_assessed.empty: | |
| avg_gain_pct = (valid_assessed['Diff'] / valid_assessed['Assessed Value']).mean() * 100 | |
| else: | |
| avg_gain_pct = 0.0 | |
| return { | |
| "total_count": f"{len(df)} Unit", | |
| "total_val": f"${total_value:,.0f}", | |
| "avg_price": f"${avg_price:,.0f}", | |
| "avg_gain": f"{avg_gain_pct:+.1f}%" | |
| } | |