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) @lru_cache(maxsize=128) 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}%" }