add CustomNextPlaceModel

CustomNextPlaceModel.py

@@ -0,0 +1,189 @@
+from typing import Tuple, TypedDict, Optional
+import datetime
+from datetime import datetime, timedelta
+import pandas as pd
+from next_place_ai.classes import DataPreparation, DatasetManager, AzureScore
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+
+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]
+
+
+class CustomNextPlaceModel:
+
+    def __init__(self):
+        self.repo_id = os.getenv('REPO_ID')
+        self.hf_token = os.getenv('HF_TOKEN')
+        self._load_model()
+
+    def _load_model(self):
+        """
+        Load all required models for the prediction pipeline
+        """
+        try:
+            # Model A scoring
+            self.score_a = AzureScore(
+                repo_id=self.repo_id,
+                token=self.hf_token,
+                model_filename='A',
+                scored_labels='A'
+            )
+
+            # Model B scorings
+            self.score_b_1 = AzureScore(
+                repo_id=self.repo_id,
+                token=self.hf_token,
+                model_filename='B_1',
+                scored_labels='B'
+            )
+            self.score_b_2 = AzureScore(
+                repo_id=self.repo_id,
+                token=self.hf_token,
+                model_filename='B_2',
+                scored_labels='B'
+            )
+            self.score_b_3 = AzureScore(
+                repo_id=self.repo_id,
+                token=self.hf_token,
+                model_filename='B_3',
+                scored_labels='B'
+            )
+
+            # Model C scorings
+            self.score_c_models = {
+                '1': AzureScore(repo_id=self.repo_id, token=self.hf_token, model_filename='model_[1]', scored_labels='price'),
+                '2': AzureScore(repo_id=self.repo_id, token=self.hf_token, model_filename='model_[2]', scored_labels='price'),
+                '3_4': AzureScore(repo_id=self.repo_id, token=self.hf_token, model_filename='model_[3, 4]', scored_labels='price'),
+                '5_6': AzureScore(repo_id=self.repo_id, token=self.hf_token, model_filename='model_[5, 6]', scored_labels='price'),
+                '7': AzureScore(repo_id=self.repo_id, token=self.hf_token, model_filename='model_[7]', scored_labels='price'),
+                '8_9': AzureScore(repo_id=self.repo_id, token=self.hf_token, model_filename='model_C_8_9', scored_labels='price')
+            }
+
+            # Time model
+            self.score_t_1 = AzureScore(
+                repo_id=self.repo_id,
+                token=self.hf_token,
+                model_filename='model_T_1',
+                scored_labels='days'
+            )
+
+            # Data preparation module
+            self.data_manager = DatasetManager(repo_id=self.repo_id, token=self.hf_token)
+
+        except Exception as e:
+            raise ValueError(f"Error loading models: {str(e)}")
+
+    def predict(self, validators_data: pd.DataFrame) -> pd.DataFrame:
+        """
+        Main prediction pipeline for processing input data
+
+        Args:
+            validators_data (pd.DataFrame): Input validation dataset
+
+        Returns:
+            pd.DataFrame: Processed prediction results
+        """
+        # Ensure input is a DataFrame and has at least one row
+        if not isinstance(validators_data, pd.DataFrame) or validators_data.empty:
+            raise ValueError("Input must be a non-empty pandas DataFrame")
+
+        # Prepare data preparation instance
+        dp = DataPreparation(validators_data)
+
+        # Prepare initial dataset
+        dp.prepare_data()
+
+        # Predict A scores
+        score_A = self.score_a.predict_proba_dataset(dp.X)
+
+        # Combine datasets
+        combined_dataset = dp.combine_datasets(score_A, dp.X)
+        combined_dataset = combined_dataset.drop(columns=['0'])
+        combined_dataset, _ = dp.create_convolution_features(combined_dataset, combined_dataset.columns.to_list(), 3)
+
+        # Predict B scores for different categories
+        score_B_1 = self.score_b_1.predict_proba_dataset(combined_dataset[combined_dataset['A']==1])
+        score_B_2 = self.score_b_2.predict_proba_dataset(combined_dataset[combined_dataset['A']==2])
+        score_B_3 = self.score_b_3.predict_proba_dataset(combined_dataset[combined_dataset['A']==3])
+
+        # Concatenate B scores
+        df_B = pd.concat([score_B_1, score_B_2, score_B_3], ignore_index=True)
+
+        # Further combine and process dataset
+        combined_dataset = dp.combine_datasets(df_B, dp.X)
+        combined_dataset = combined_dataset.drop(columns=['0'])
+        combined_dataset, _ = dp.create_convolution_features(combined_dataset, combined_dataset.columns.to_list(), 3)
+
+        # Predict C scores for different categories
+        c_scores = {
+            '1': self.score_c_models['1'].predict_dataset(combined_dataset[combined_dataset['B'].isin([1])])
+                  if not combined_dataset[combined_dataset['B'].isin([1])].empty else pd.DataFrame({'price': [0]}),
+            '2': self.score_c_models['2'].predict_dataset(combined_dataset[combined_dataset['B'].isin([2])])
+                  if not combined_dataset[combined_dataset['B'].isin([2])].empty else pd.DataFrame({'price': [0]}),
+            '3_4': self.score_c_models['3_4'].predict_dataset(combined_dataset[combined_dataset['B'].isin([3, 4])])
+                  if not combined_dataset[combined_dataset['B'].isin([3, 4])].empty else pd.DataFrame({'price': [0]}),
+            '5_6': self.score_c_models['5_6'].predict_dataset(combined_dataset[combined_dataset['B'].isin([5, 6])])
+                  if not combined_dataset[combined_dataset['B'].isin([5, 6])].empty else pd.DataFrame({'price': [0]}),
+            '7': self.score_c_models['7'].predict_dataset(combined_dataset[combined_dataset['B'].isin([7])])
+                  if not combined_dataset[combined_dataset['B'].isin([7])].empty else pd.DataFrame({'price': [0]}),
+            '8_9': self.score_c_models['8_9'].predict_dataset(combined_dataset[combined_dataset['B'].isin([8, 9])])
+                  if not combined_dataset[combined_dataset['B'].isin([8, 9])].empty else pd.DataFrame({'price': [0]})
+        }
+        df_C = pd.concat(
+            [c_scores[key][['price']] for key in c_scores
+            if isinstance(c_scores[key], pd.DataFrame) and 'price' in c_scores[key].columns and not c_scores[key].empty],
+            ignore_index=True
+        )
+
+        df_C_ = df_C[df_C['price'] != 0].copy()
+
+        # Combine datasets
+        t_df_ = pd.concat([combined_dataset.reset_index(drop=True), df_C_.reset_index(drop=True)], axis=1)
+
+        # Predict time
+        score_t_1 = self.score_t_1.predict_dataset(t_df_).astype(int)
+
+        # Final result
+        result = pd.concat([df_C_.reset_index(drop=True), score_t_1.reset_index(drop=True)], axis=1)
+
+        return result
+
+    def run_inference(self, input_data: ProcessedSynapse) -> Tuple[float, str]:
+
+        input_data = pd.DataFrame([input_data])
+        result = self.predict(input_data)
+        predicted_sale_price, predicted_days = result['price'].iloc[0], result['days'].iloc[0] # кол-во дней нужно преобразовать в дату в виде строки
+
+        current_days_on_market = input_data.get('days_on_market', 0) or 0
+
+        # Вычисление даты размещения на рынке
+        date_listed = datetime.now() - timedelta(days=current_days_on_market)
+
+        # Вычисление предсказанной даты продажи
+        predicted_sale_date = (date_listed + timedelta(days=predicted_days)).strftime('%Y-%m-%d')
+
+        return predicted_sale_price, predicted_sale_date