|
|
from typing import Tuple, TypedDict, Optional |
|
|
import datetime |
|
|
|
|
|
|
|
|
class ProcessedSynapse(TypedDict): |
|
|
id: Optional[str] |
|
|
nextplace_id: Optional[str] |
|
|
property_id: Optional[str] |
|
|
listing_id: Optional[str] |
|
|
address: Optional[str] |
|
|
city: Optional[str] |
|
|
state: Optional[str] |
|
|
zip_code: Optional[str] |
|
|
price: Optional[float] |
|
|
beds: Optional[int] |
|
|
baths: Optional[float] |
|
|
sqft: Optional[int] |
|
|
lot_size: Optional[int] |
|
|
year_built: Optional[int] |
|
|
days_on_market: Optional[int] |
|
|
latitude: Optional[float] |
|
|
longitude: Optional[float] |
|
|
property_type: Optional[str] |
|
|
last_sale_date: Optional[str] |
|
|
hoa_dues: Optional[float] |
|
|
query_date: Optional[str] |
|
|
market: Optional[str] |
|
|
|
|
|
|
|
|
class StatisticalBaseModel: |
|
|
|
|
|
def __init__(self): |
|
|
self._load_model() |
|
|
|
|
|
def _load_model(self): |
|
|
""" |
|
|
Perform any actions needed to load the model. |
|
|
EX: Establish API connections, download an ML model for inference, etc... |
|
|
""" |
|
|
print("Loading model...") |
|
|
|
|
|
print("Model loaded.") |
|
|
|
|
|
def _get_average_for_market(self, market: str) -> int: |
|
|
""" |
|
|
Get the average days on market for a house in a given market |
|
|
:param market: the housing market |
|
|
:return: the average days on market |
|
|
""" |
|
|
|
|
|
|
|
|
if market == 'San Francisco': |
|
|
return 23 |
|
|
elif market == 'Los Angeles': |
|
|
return 68 |
|
|
elif market == 'Seattle': |
|
|
return 27 |
|
|
elif market == 'Austin': |
|
|
return 78 |
|
|
elif market == 'Houston': |
|
|
return 73 |
|
|
elif market == 'Chicago': |
|
|
return 25 |
|
|
elif market == 'New York': |
|
|
return 20 |
|
|
elif market == 'Denver': |
|
|
return 24 |
|
|
return 34 |
|
|
|
|
|
|
|
|
def _sale_date_predictor(self, input_data: ProcessedSynapse): |
|
|
""" |
|
|
Calculate the expected sale date based on the national average |
|
|
:param days_on_market: number of days this house has been on the market |
|
|
:return: the predicted sale date, based on the national average of 34 days |
|
|
""" |
|
|
if 'days_on_market' not in input_data: |
|
|
return datetime.date.today() + datetime.timedelta(days=1) |
|
|
|
|
|
if 'market' not in input_data: |
|
|
average = 34 |
|
|
|
|
|
else: |
|
|
average = self._get_average_for_market(input_data['market']) |
|
|
|
|
|
days_on_market = input_data['days_on_market'] |
|
|
if days_on_market < average: |
|
|
days_until_sale = average - days_on_market |
|
|
sale_date = datetime.date.today() + datetime.timedelta(days=days_until_sale) |
|
|
return sale_date |
|
|
else: |
|
|
return datetime.date.today() + datetime.timedelta(days=1) |
|
|
|
|
|
def _get_price_multiplier(self, market: str) -> float: |
|
|
""" |
|
|
Calculate the price multiplier based on the market |
|
|
:param market: the marked the house is in |
|
|
:return: the multiplier for the predicted price |
|
|
""" |
|
|
|
|
|
|
|
|
if market == 'San Francisco': |
|
|
return 1.18 |
|
|
elif market == 'Los Angeles': |
|
|
return 1.2 |
|
|
elif market == 'Seattle': |
|
|
return 1.13 |
|
|
elif market == 'Austin': |
|
|
return 1.11 |
|
|
elif market == 'Houston': |
|
|
return 1.15 |
|
|
elif market == 'Chicago': |
|
|
return 1.12 |
|
|
elif market == 'New York': |
|
|
return 1.05 |
|
|
elif market == 'Denver': |
|
|
return 1.11 |
|
|
return 1.0 |
|
|
|
|
|
def run_inference(self, input_data: ProcessedSynapse) -> Tuple[float, str]: |
|
|
""" |
|
|
Predict the sale price and sale date for the house represented by `input_data` |
|
|
:param input_data: a formatted Synapse from the validator, representing a currently listed house |
|
|
:return: the predicted sale price and predicted sale date for this home |
|
|
""" |
|
|
listing_price = float(input_data['price']) if 'price' in input_data else 1.0 |
|
|
sale_multiplier = self._get_price_multiplier(input_data['market']) if 'market' in input_data else 1.0 |
|
|
predicted_sale_price = listing_price * sale_multiplier |
|
|
predicted_sale_date = self._sale_date_predictor(input_data) |
|
|
predicted_sale_date = predicted_sale_date.strftime("%Y-%m-%d") |
|
|
return predicted_sale_price, predicted_sale_date |