AI Real Estate Assistant: Market Analysis & Build Guide
What Is an AI Real Estate Assistant?
An AI Real Estate Assistant is an intelligent software system that automates property market analysis, valuation predictions, and investment recommendations using machine learning, natural language processing, and real-time data aggregation. It ingests multiple data streams — MLS listings, public records, economic indicators, neighborhood demographics, and historical sales — to deliver actionable insights that would traditionally require hours of manual research by agents, appraisers, or investors.
At its core, the assistant performs three critical functions:
- Market Trend Analysis — detecting price movements, inventory shifts, and demand signals across geographic micro-markets
- Property Valuation — generating fair market estimates using comparable sales and feature-based regression models
- Investment Scoring — ranking properties by cap rate, cash-on-cash return, and appreciation potential
Why It Matters
The residential and commercial real estate markets generate terabytes of unstructured data daily. Traditional analysis methods — manual spreadsheets, gut-feel comps, delayed MLS reports — introduce latency and bias. An AI-powered assistant eliminates both by:
- Processing thousands of listings per second across multiple counties simultaneously
- Applying consistent valuation logic free from emotional anchoring
- Surfacing emerging trends 2–4 weeks before they appear in aggregate market reports
- Reducing the cost per deal analysis by up to 80% for brokerages and investment firms
For developers, building this system demonstrates mastery of data pipelines, ML model deployment, and API integration — skills directly transferable to fintech, insurtech, and proptech roles.
Core Architecture Overview
The assistant follows a modular pipeline architecture with five distinct layers. Each layer can be developed, tested, and deployed independently before integrating into the full system.
Layer 1: Data Ingestion Engine
This layer connects to multiple data sources and normalizes all incoming records into a unified schema. The most common sources include:
- MLS RETS/RESO Web API feeds
- County assessor parcel databases
- Zillow, Redfin, and Realtor.com public APIs
- Census Bureau demographic datasets
- Federal Reserve economic indicators (interest rates, employment)
# data_ingestion/connector.py
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import sqlite3
import json
class RealEstateDataConnector:
"""
Unified connector for multiple real estate data sources.
Handles rate limiting, pagination, and schema normalization.
"""
def __init__(self, config_path: str = "api_config.json"):
with open(config_path, 'r') as f:
self.config = json.load(f)
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'AIRealEstateAssistant/2.0',
'Accept': 'application/json'
})
self.cache_db = sqlite3.connect(':memory:')
self._init_cache_tables()
def _init_cache_tables(self):
self.cache_db.execute('''
CREATE TABLE IF NOT EXISTS listings_cache (
listing_id TEXT PRIMARY KEY,
source TEXT,
raw_json TEXT,
fetched_at TIMESTAMP
)
''')
def fetch_listings(self, zip_code: str, days_back: int = 30) -> pd.DataFrame:
"""
Fetch active and recently sold listings for a zip code.
Returns normalized DataFrame with unified column names.
"""
start_date = (datetime.now() - timedelta(days=days_back)).isoformat()
# Primary MLS API call (RESO standard endpoint)
mls_data = self._fetch_reso_listings(
endpoint=self.config['mls']['reso_endpoint'],
params={
'PostalCode': zip_code,
'ListingStatus': 'Active|Pending|Closed',
'ModificationTimestamp': f'gte:{start_date}',
'$limit': 500
}
)
# Supplement with public API data for additional features
public_data = self._fetch_public_api_listings(zip_code, days_back)
# Merge and deduplicate by address normalization
combined = self._merge_and_deduplicate(mls_data, public_data)
return self._normalize_schema(combined)
def _fetch_reso_listings(self, endpoint: str, params: Dict) -> List[Dict]:
"""Fetch from RESO-compliant MLS API with exponential backoff."""
all_records = []
page = 0
while True:
params['$skip'] = page * params.get('$limit', 500)
try:
response = self.session.get(endpoint, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if isinstance(data, dict):
records = data.get('value', data.get('results', []))
else:
records = data if isinstance(data, list) else []
if not records:
break
all_records.extend(records)
page += 1
# Respect rate limits
if 'Retry-After' in response.headers:
wait = int(response.headers['Retry-After'])
import time
time.sleep(wait)
except requests.exceptions.RequestException as e:
print(f"MLS fetch error: {e}")
break
return all_records
def _fetch_public_api_listings(self, zip_code: str, days_back: int) -> List[Dict]:
"""Fallback data source for enrichment features."""
# Example: Rentcast or similar aggregator
api_key = self.config.get('public_api_key', '')
url = f"https://api.realestateaggregator.com/v2/listings"
try:
response = self.session.get(url, params={
'zip': zip_code,
'days_ago': days_back,
'api_key': api_key
}, timeout=15)
return response.json().get('properties', [])
except Exception:
return []
def _merge_and_deduplicate(self, primary: List[Dict], secondary: List[Dict]) -> pd.DataFrame:
"""Merge data sources, preferring MLS data on conflicts."""
df_primary = pd.DataFrame(primary)
df_secondary = pd.DataFrame(secondary)
if df_primary.empty:
return df_secondary
if df_secondary.empty:
return df_primary
# Create normalized address key for deduplication
for df in [df_primary, df_secondary]:
if not df.empty:
df['address_key'] = (
df.get('StreetNumber', '') + ' ' +
df.get('StreetName', '') + ' ' +
df.get('City', '')
).str.lower().str.replace(r'[^a-z0-9]', '', regex=True)
# Merge on address key, prefer primary source values
merged = pd.merge(
df_primary,
df_secondary,
on='address_key',
how='outer',
suffixes=('_primary', '_secondary')
)
return merged
def _normalize_schema(self, df: pd.DataFrame) -> pd.DataFrame:
"""Map all columns to a standard internal schema."""
column_mapping = {
'ListPrice': 'list_price',
'ClosePrice': 'sale_price',
'BedroomsTotal': 'bedrooms',
'BathroomsTotalInteger': 'bathrooms',
'LivingArea': 'square_feet',
'YearBuilt': 'year_built',
'PropertyType': 'property_type',
'PostalCode': 'zip_code',
'Latitude': 'latitude',
'Longitude': 'longitude',
'ListingStatus': 'status',
'DaysOnMarket': 'days_on_market',
'LotSizeSquareFeet': 'lot_size_sqft',
'GarageSpaces': 'garage_spaces',
'Pool': 'has_pool'
}
normalized = df.rename(columns=column_mapping)
# Ensure required columns exist
required_cols = [
'list_price', 'bedrooms', 'bathrooms',
'square_feet', 'year_built', 'zip_code', 'status'
]
for col in required_cols:
if col not in normalized.columns:
normalized[col] = None
return normalized[required_cols + [
c for c in normalized.columns
if c not in required_cols
]]
Layer 2: Feature Engineering Pipeline
Raw listing data must be transformed into model-ready feature vectors. This layer handles missing value imputation, creates derived features, and encodes categorical variables. The quality of feature engineering directly determines model accuracy — a well-engineered feature set can improve valuation precision by 15–25% over raw MLS fields alone.
# feature_engineering/transformer.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import LabelEncoder, StandardScaler
from sklearn.impute import KNNImputer
from typing import Tuple, List
import pickle
class RealEstateFeatureEngineer:
"""
Transforms raw listing data into ML-ready feature vectors.
Creates derived features critical for valuation accuracy.
"""
def __init__(self):
self.label_encoders = {}
self.scaler = StandardScaler()
self.imputer = KNNImputer(n_neighbors=5)
self.feature_importance_scores = {}
self._fitted = False
def fit_transform(self, df: pd.DataFrame, target_col: str = 'sale_price') -> pd.DataFrame:
"""
Fit encoders and scalers on training data, then transform.
"""
df = df.copy()
# Step 1: Create derived features
df = self._create_derived_features(df)
# Step 2: Handle missing values
df = self._handle_missing_values(df)
# Step 3: Encode categorical features
df = self._encode_categorical(df)
# Step 4: Scale numerical features
df = self._scale_features(df, fit=True)
self._fitted = True
return df
def transform(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Transform new data using fitted encoders (no refitting).
"""
if not self._fitted:
raise RuntimeError("Must call fit_transform before transform")
df = df.copy()
df = self._create_derived_features(df)
df = self._handle_missing_values(df)
df = self._encode_categorical(df)
df = self._scale_features(df, fit=False)
return df
def _create_derived_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Generate features that capture real estate domain knowledge.
"""
# Price per square foot — critical valuation metric
if 'sale_price' in df.columns and 'square_feet' in df.columns:
df['price_per_sqft'] = df['sale_price'] / df['square_feet'].replace(0, np.nan)
if 'list_price' in df.columns and 'square_feet' in df.columns:
df['list_price_per_sqft'] = df['list_price'] / df['square_feet'].replace(0, np.nan)
# Property age and effective age
current_year = 2025
if 'year_built' in df.columns:
df['property_age'] = current_year - df['year_built']
df['property_age'] = df['property_age'].clip(0, 200)
# Age bucket encoding (captures non-linear age effects)
df['age_bucket'] = pd.cut(
df['property_age'],
bins=[0, 1, 5, 10, 20, 30, 50, 100, 200],
labels=['New', '1-5yr', '5-10yr', '10-20yr',
'20-30yr', '30-50yr', '50-100yr', '100+yr']
)
# Bedroom-to-bathroom ratio
if 'bedrooms' in df.columns and 'bathrooms' in df.columns:
df['bed_bath_ratio'] = df['bedrooms'] / df['bathrooms'].replace(0, np.nan)
df['bed_bath_ratio'] = df['bed_bath_ratio'].clip(0, 10)
# Total room count
if 'bedrooms' in df.columns and 'bathrooms' in df.columns:
df['total_rooms'] = df['bedrooms'] + df['bathrooms']
# Lot utilization
if 'square_feet' in df.columns and 'lot_size_sqft' in df.columns:
lot_sqft = df['lot_size_sqft'].replace(0, np.nan)
df['lot_utilization'] = df['square_feet'] / lot_sqft
df['lot_utilization'] = df['lot_utilization'].clip(0, 2)
# Market timing features (if timestamp available)
if 'listing_date' in df.columns:
df['listing_date'] = pd.to_datetime(df['listing_date'], errors='coerce')
df['listing_month'] = df['listing_date'].dt.month
df['listing_quarter'] = df['listing_date'].dt.quarter
df['listing_day_of_week'] = df['listing_date'].dt.dayofweek
df['is_spring_listing'] = df['listing_month'].isin([3, 4, 5, 6]).astype(int)
# Price positioning relative to zip median
if 'list_price' in df.columns and 'zip_code' in df.columns:
zip_medians = df.groupby('zip_code')['list_price'].transform('median')
df['price_vs_zip_median_ratio'] = df['list_price'] / zip_medians.replace(0, np.nan)
return df
def _handle_missing_values(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Intelligent imputation using KNN for numericals and mode for categoricals.
"""
# Separate numerical and categorical columns
numerical_cols = df.select_dtypes(include=[np.number]).columns.tolist()
categorical_cols = df.select_dtypes(include=['object', 'category']).columns.tolist()
# Remove target-like columns from imputation inputs
exclude_cols = ['sale_price', 'list_price']
num_impute_cols = [c for c in numerical_cols if c not in exclude_cols]
if num_impute_cols:
# KNN imputation preserves relationships between features
imputed_array = self.imputer.fit_transform(df[num_impute_cols])
df[num_impute_cols] = imputed_array
# Fill categorical missing with mode
for col in categorical_cols:
if df[col].isna().any():
mode_val = df[col].mode()
if not mode_val.empty:
df[col] = df[col].fillna(mode_val[0])
else:
df[col] = df[col].fillna('Unknown')
return df
def _encode_categorical(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Label encode categoricals; one-hot encode high-cardinality columns separately.
"""
categorical_cols = df.select_dtypes(include=['object', 'category']).columns
for col in categorical_cols:
if col in ['zip_code', 'property_type', 'age_bucket', 'status']:
# Fit or transform label encoder
if col not in self.label_encoders:
le = LabelEncoder()
df[col] = df[col].astype(str).fillna('missing')
le.fit(df[col])
self.label_encoders[col] = le
le = self.label_encoders[col]
unseen_mask = ~df[col].isin(le.classes_)
if unseen_mask.any():
df.loc[unseen_mask, col] = 'missing'
if 'missing' not in le.classes_:
le.classes_ = np.append(le.classes_, 'missing')
df[col] = le.transform(df[col])
return df
def _scale_features(self, df: pd.DataFrame, fit: bool = False) -> pd.DataFrame:
"""
Standardize numerical features to zero mean, unit variance.
"""
numerical_cols = df.select_dtypes(include=[np.number]).columns
exclude_cols = ['sale_price', 'list_price', 'zip_code']
scale_cols = [c for c in numerical_cols if c not in exclude_cols]
if fit:
self.scaler.fit(df[scale_cols])
df[scale_cols] = self.scaler.transform(df[scale_cols])
return df
def save(self, path: str = "feature_engineer.pkl"):
"""Serialize the fitted transformer for production use."""
with open(path, 'wb') as f:
pickle.dump(self, f)
@staticmethod
def load(path: str = "feature_engineer.pkl"):
"""Load a previously fitted transformer."""
with open(path, 'rb') as f:
return pickle.load(f)
Layer 3: Valuation Model
The valuation engine uses an ensemble of gradient-boosted trees (XGBoost) trained on historical comparable sales. This approach captures non-linear relationships between property features and sale prices while remaining robust to outliers common in real estate data — like a single luxury penthouse sale skewing a neighborhood average.
# models/valuation_model.py
import pandas as pd
import numpy as np
import xgboost as xgb
from sklearn.model_selection import TimeSeriesSplit, GridSearchCV
from sklearn.metrics import mean_absolute_percentage_error, r2_score
import pickle
from datetime import datetime
class PropertyValuationModel:
"""
XGBoost-based Automated Valuation Model (AVM).
Trained on comparable sales with time-aware cross-validation.
"""
def __init__(self, objective: str = 'reg:absoluteerror'):
self.model = None
self.feature_names = []
self.objective = objective
self.best_params = {}
self.training_metrics = {}
self.confidence_calculator = None
def train(self, X: pd.DataFrame, y: pd.Series,
fit_confidence: bool = True) -> dict:
"""
Train the valuation model with time-series aware CV.
Returns training metrics dictionary.
"""
self.feature_names = X.columns.tolist()
# Base XGBoost configuration tuned for real estate
base_params = {
'n_estimators': 500,
'max_depth': 6,
'learning_rate': 0.05,
'min_child_weight': 5,
'subsample': 0.8,
'colsample_bytree': 0.7,
'gamma': 0.1,
'reg_alpha': 0.5,
'reg_lambda': 1.0,
'objective': self.objective,
'eval_metric': 'mae',
'random_state': 42,
'n_jobs': -1,
'early_stopping_rounds': 50
}
# Grid search over key hyperparameters
param_grid = {
'max_depth': [4, 5, 6, 7],
'learning_rate': [0.01, 0.03, 0.05, 0.1],
'min_child_weight': [3, 5, 7],
'subsample': [0.7, 0.8, 0.9]
}
tscv = TimeSeriesSplit(n_splits=5)
grid_search = GridSearchCV(
estimator=xgb.XGBRegressor(**base_params),
param_grid=param_grid,
cv=tscv,
scoring='neg_mean_absolute_percentage_error',
n_jobs=-1,
verbose=1
)
print("Training valuation model with hyperparameter search...")
grid_search.fit(X, y)
self.model = grid_search.best_estimator_
self.best_params = grid_search.best_params_
# Calculate final metrics on holdout set
y_pred = self.model.predict(X)
mape = mean_absolute_percentage_error(y, y_pred) * 100
r2 = r2_score(y, y_pred)
self.training_metrics = {
'mape': round(mape, 2),
'r2_score': round(r2, 4),
'best_params': self.best_params,
'cv_mean_score': round(abs(grid_search.best_score_), 4)
}
print(f"Training complete. MAPE: {mape:.2f}%, R²: {r2:.4f}")
if fit_confidence:
self._fit_confidence_estimator(X, y)
return self.training_metrics
def _fit_confidence_estimator(self, X: pd.DataFrame, y: pd.Series):
"""
Estimate prediction uncertainty using residuals analysis.
Produces a confidence score (0-100) for each prediction.
"""
y_pred = self.model.predict(X)
residuals = np.abs(y - y_pred)
# Train a secondary model on absolute residuals
residual_model = xgb.XGBRegressor(
n_estimators=200,
max_depth=4,
learning_rate=0.05,
objective='reg:squarederror'
)
residual_model.fit(X, residuals)
self.confidence_calculator = residual_model
def predict(self, X: pd.DataFrame, return_confidence: bool = True) -> pd.DataFrame:
"""
Generate valuation predictions with optional confidence scores.
"""
if self.model is None:
raise RuntimeError("Model must be trained before prediction")
# Ensure feature alignment
missing_features = set(self.feature_names) - set(X.columns)
if missing_features:
raise ValueError(f"Missing features: {missing_features}")
X_aligned = X[self.feature_names]
predictions = self.model.predict(X_aligned)
results = pd.DataFrame({
'predicted_value': predictions,
'prediction_date': datetime.now().isoformat()
}, index=X.index)
if return_confidence and self.confidence_calculator:
# Estimate expected absolute error
expected_error = self.confidence_calculator.predict(X_aligned)
mean_value = predictions.mean() if predictions.mean() > 0 else 1
# Convert to confidence score (0-100)
# Lower expected_error/mean_value ratio = higher confidence
error_ratio = expected_error / mean_value
confidence = 100 * np.exp(-2 * error_ratio)
confidence = np.clip(confidence, 0, 100)
results['confidence_score'] = confidence.round(1)
results['expected_error_pct'] = (error_ratio * 100).round(2)
return results
def generate_comparable_report(self, X: pd.DataFrame,
comp_data: pd.DataFrame) -> str:
"""
Generate a natural language report explaining the valuation.
Uses feature importance to highlight key value drivers.
"""
if self.model is None:
return "Model not trained."
importance = self.model.feature_importances_
feature_importance = dict(zip(self.feature_names, importance))
top_features = sorted(feature_importance.items(),
key=lambda x: x[1], reverse=True)[:5]
report_lines = [
"=== Automated Valuation Report ===",
f"Predicted Value: ${X['predicted_value'].iloc[0]:,.0f}",
f"Confidence Score: {X['confidence_score'].iloc[0]:.0f}/100",
"",
"Key Value Drivers:"
]
for feat, imp in top_features:
report_lines.append(f" • {feat}: {imp*100:.1f}% influence")
return "\n".join(report_lines)
def save(self, path: str = "valuation_model.pkl"):
with open(path, 'wb') as f:
pickle.dump({
'model': self.model,
'feature_names': self.feature_names,
'best_params': self.best_params,
'training_metrics': self.training_metrics,
'confidence_calculator': self.confidence_calculator
}, f)
@staticmethod
def load(path: str = "valuation_model.pkl"):
instance = PropertyValuationModel()
with open(path, 'rb') as f:
data = pickle.load(f)
instance.model = data['model']
instance.feature_names = data['feature_names']
instance.best_params = data.get('best_params', {})
instance.training_metrics = data.get('training_metrics', {})
instance.confidence_calculator = data.get('confidence_calculator')
return instance
Layer 4: Market Analysis Engine
This layer performs time-series analysis on aggregated market data to detect trends, forecast price movements, and identify investment hotspots. It uses a combination of Holt-Winters exponential smoothing for seasonal trend detection and a lightweight gradient boosting model for 6–12 month price forecasts.
# market_analysis/analyzer.py
import pandas as pd
import numpy as np
from scipy import stats
from statsmodels.tsa.holtwinters import ExponentialSmoothing
from typing import Dict, List, Tuple
import warnings
warnings.filterwarnings('ignore')
class MarketAnalyzer:
"""
Multi-metric market analysis engine for zip-code and city-level trends.
Generates market health scores, trend directions, and investment ratings.
"""
def __init__(self):
self.market_cache = {}
self.seasonality_patterns = {}
def analyze_market(self, listings_df: pd.DataFrame,
historical_df: pd.DataFrame = None) -> Dict:
"""
Comprehensive market analysis returning structured insights.
"""
analysis = {}
# 1. Current market snapshot
analysis['snapshot'] = self._market_snapshot(listings_df)
# 2. Price trend analysis
if historical_df is not None and not historical_df.empty:
analysis['trends'] = self._price_trends(historical_df)
analysis['forecast'] = self._price_forecast(historical_df)
# 3. Inventory & velocity metrics
analysis['inventory'] = self._inventory_analysis(listings_df)
# 4. Market health composite score
analysis['health_score'] = self._calculate_market_health(analysis)
# 5. Investment opportunity detection
analysis['opportunities'] = self._detect_opportunities(listings_df, analysis)
return analysis
def _market_snapshot(self, df: pd.DataFrame) -> Dict:
"""Calculate current market statistics."""
active = df[df['status'].isin(['Active', 'active', 'For Sale'])]
sold = df[df['status'].isin(['Closed', 'Sold', 'closed', 'sold'])]
snapshot = {
'active_listings_count': len(active),
'recent_sales_count': len(sold),
'median_list_price': active['list_price'].median() if not active.empty else None,
'median_sale_price': sold['sale_price'].median() if not sold.empty else None,
'median_price_per_sqft': sold['price_per_sqft'].median() if not sold.empty and 'price_per_sqft' in sold.columns else None,
'avg_days_on_market': sold['days_on_market'].mean() if not sold.empty and 'days_on_market' in sold.columns else None,
'list_to_sale_ratio': (
active['list_price'].median() / sold['sale_price'].median()
if not active.empty and not sold.empty and sold['sale_price'].median() > 0
else None
)
}
# Calculate absorption rate (monthly sales / active inventory)
if not sold.empty and not active.empty:
monthly_sales = len(sold) / 3 # Assuming 90-day window
snapshot['absorption_rate'] = monthly_sales / len(active) if len(active) > 0 else 0
snapshot['months_of_inventory'] = len(active) / monthly_sales if monthly_sales > 0 else float('inf')
return snapshot
def _price_trends(self, historical_df: pd.DataFrame) -> Dict:
"""
Detect price movement patterns using statistical analysis.
"""
if 'sale_price' not in historical_df.columns:
return {'trend': 'insufficient_data'}
# Sort by sale date
if 'sale_date' in historical_df.columns:
historical_df = historical_df.sort_values('sale_date')
prices = historical_df['sale_price'].dropna()
if len(prices) < 6:
return {'trend': 'insufficient_data'}
# Split into recent and earlier periods
midpoint = len(prices) // 2
recent_prices = prices.iloc[-midpoint:]
earlier_prices = prices.iloc[:midpoint]
recent_median = recent_prices.median()
earlier_median = earlier_prices.median()
if earlier_median > 0:
pct_change = ((recent_median - earlier_median) / earlier_median) * 100
else:
pct_change = 0
# Mann-Kendall trend test for statistical significance
x = np.arange(len(prices))
slope, intercept, r_value, p_value, std_err = stats.linregress(x, prices.values)
trend_direction = 'stable'
if pct_change > 5 and p_value < 0.05:
trend_direction = 'appreciating'
elif pct_change < -5 and p_value < 0.05:
trend_direction = 'declining'
elif pct_change > 5:
trend_direction = 'likely_appreciating'
elif pct_change < -5:
trend_direction = 'likely_declining'
return {
'trend': trend_direction,
'price_change_pct': round(pct_change, 2),
'r_squared': round(r_value ** 2, 4),
'statistical_significance': round(p_value, 4),
'annualized_return': round(pct_change / (len(prices) / 12), 2) if len(prices) > 0 else 0,
'volatility': round(prices.std() / prices.mean() * 100, 2) if prices.mean() > 0 else 0
}
def _price_forecast(self, historical_df: pd.DataFrame) -> Dict:
"""
Generate 6-month and 12-month price forecasts using Holt-Winters.
"""
if 'sale_price' not in historical_df.columns:
return {'error': 'No price data available'}
# Aggregate to monthly medians for smoother series
if 'sale_date' in historical_df.columns:
historical_df['sale_month'] = pd.to_datetime(historical_df['sale_date']).dt.to_period('M')
monthly_medians = historical_df.groupby('sale_month')['sale_price'].median()
else:
monthly_medians = historical_df['sale_price'].dropna()
if len(monthly_medians) < 12:
return {'error': 'Need at least 12 months of data for forecast'}
try:
# Fit Holt-Winters with multiplicative seasonality (real estate cycles)
model = ExponentialSmoothing(
monthly_medians,
trend='add',
seasonal='multiplicative',
seasonal_periods=12
)
fitted = model.fit()
forecast_periods = 12
forecast = fitted.forecast(forecast_periods)
current_median = monthly_medians.iloc[-1]
forecast_6m = forecast.iloc[5] if len(forecast) > 5 else forecast.iloc[-1]
forecast_12m = forecast.iloc[-1]
return {
'current_monthly_median': round(current_median, 0),
'forecast_6m': round(forecast_6m, 0),
'forecast_12m': round(forecast_12m, 0),
'forecast_6m_pct_change': round(((forecast_6m - current_median) / current_median) * 100, 1),
'forecast_12m_pct_change': round(((forecast_12m - current_median) / current_median) * 100, 1),
'forecast_confidence': 'medium' if len(monthly_medians) >= 24 else 'low',
'seasonality_detected': True,
'forecast_values': forecast.tolist()
}
except Exception as e:
return {'error': f'Forecast failed: {str(e)}'}
def _inventory_analysis(self, df: pd.DataFrame) -> Dict:
"""Analyze supply-side metrics."""
active = df[df['status'].isin(['Active', 'active', 'For Sale'])]
if active.empty:
return {'status': 'no_active_listings'}
# Price distribution analysis
price_deciles = active['list_price'].quantile([0.1, 0.25, 0.5, 0.75, 0.9])
# Price reductions (if available)
if 'original_list_price' in df.columns:
active_with_reductions = active[active['list_price'] < active['original_list_price']]
reduction_pct = len(active_with_reductions) / len(active) * 100 if len(active) > 0 else 0
else:
reduction_pct = None
return {
'total_active': len(active),
'price_distribution': price_deciles.to_dict(),
'price_reduction_pct': reduction_pct,
'supply_level': 'tight' if len(active) < 20 else 'normal' if len(active) < 50 else 'abundant',
'median_days_on_market': active['days_on_market'].median() if 'days_on_market' in active.columns else None
}
def _calculate_market_health(self, analysis: Dict) -> Dict:
"""
Composite market health score from 0-100.
Combines inventory levels, price trends, and velocity.
"""
score = 50 # Neutral baseline
# Inventory factor: tight supply = seller's market (higher score for sellers)
inventory = analysis.get('inventory', {})
supply = inventory.get('supply_level', 'normal')
if supply == 'tight':
score += 15
elif supply ==