| | 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 |