Update modelLoanAPI.py

modelLoanAPI.py

@@ -1,50 +1,318 @@
+# ```python
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
 import pandas as pd
-import traceback
+import numpy as np
+from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
+from sklearn.preprocessing import LabelEncoder, StandardScaler
+from sklearn.metrics import accuracy_score
+import matplotlib.pyplot as plt
+import json
+import base64
+from io import BytesIO
+import warnings
+import logging
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+warnings.filterwarnings("ignore")
 
 app = FastAPI()
 
-# Request model
 class WorkerIdRequest(BaseModel):
     worker_id: int
 
-# Load dataset once on startup
-CSV_PATH = "/app/extended_worker_dataset.csv"  # adjust if you use /app/data
-
-try:
-    df = pd.read_csv(CSV_PATH)
-except Exception as e:
-    print("==== ERROR LOADING CSV ====")
-    print(str(e))
-    traceback.print_exc()
-    raise RuntimeError(f"Failed to load dataset from {CSV_PATH}: {str(e)}")
-
 @app.post("/predict_worker_earnings/")
 async def predict_worker_earnings(request: WorkerIdRequest):
     try:
         worker_id = request.worker_id
+        logger.info(f"Processing request for worker_id: {worker_id}")
+        
+        # Initialize result dictionary
+        results = {
+            'worker_id': worker_id,
+            'classification_metrics': {},
+            'worker_profile': {},
+            'plot': ''
+        }
 
-        # Ensure worker exists
-        if worker_id not in df['worker_id'].values:
-            raise HTTPException(status_code=404, detail=f"Worker ID {worker_id} not found")
+        # Load data
+        try:
+            df = pd.read_csv('/app/extended_worker_dataset.csv')
+            logger.info(f"CSV loaded successfully. Columns: {list(df.columns)}")
+        except FileNotFoundError:
+            logger.error("CSV file not found at /app/extended_worker_dataset.csv")
+            raise HTTPException(status_code=500, detail="CSV file not found at /app/data/extended_worker_dataset.csv")
+        except Exception as e:
+            logger.error(f"Error reading CSV file: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error reading CSV file: {str(e)}")
 
-        # Dummy earnings calculation (replace with ML model later)
-        worker_data = df[df['worker_id'] == worker_id].iloc[0]
-        base_salary = worker_data.get("base_salary", 10000)
-        experience_years = worker_data.get("experience_years", 1)
-        predicted_earnings = base_salary + (experience_years * 500)
+        # Verify expected columns
+        expected_columns = ['worker_id', 'state', 'labour_category', 'contracted_wage', 'age', 'gender', 
+                           'migration_status', 'years_of_experience', 'feedback_score', 'job_type', 'timestamp']
+        if not all(col in df.columns for col in expected_columns):
+            missing_cols = [col for col in expected_columns if col not in df.columns]
+            logger.error(f"Missing columns in CSV: {missing_cols}")
+            raise HTTPException(status_code=500, detail=f"Missing columns in CSV: {missing_cols}")
 
-        return {
-            "worker_id": worker_id,
-            "predicted_earnings": predicted_earnings
-        }
+        # Filter for one worker_id
+        df = df[df['worker_id'] == worker_id].copy()
+        if df.empty:
+            logger.warning(f"No data found for worker_id {worker_id}")
+            raise HTTPException(status_code=404, detail=f"No data found for worker_id {worker_id}")
+
+        logger.info(f"Filtered data for worker_id {worker_id}: {len(df)} rows")
+
+        # Data preprocessing
+        try:
+            df['timestamp'] = pd.to_datetime(df['timestamp'])
+        except Exception as e:
+            logger.error(f"Error converting timestamp: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error converting timestamp: {str(e)}")
+
+        df['has_job'] = (df['job_type'] != "No Job").astype(int)
+
+        try:
+            wage_cap = df[df['contracted_wage'] > 0]['contracted_wage'].quantile(0.90)
+            if np.isnan(wage_cap) or wage_cap <= 500:
+                logger.error("Invalid wage cap calculated")
+                raise ValueError("Invalid wage cap calculated")
+            df['contracted_wage'] = df['contracted_wage'].clip(lower=500, upper=wage_cap)
+        except Exception as e:
+            logger.error(f"Error processing wage data: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error processing wage data: {str(e)}")
+
+        # Encode job_type
+        le = LabelEncoder()
+        try:
+            df['job_type_encoded'] = le.fit_transform(df['job_type'])
+        except Exception as e:
+            logger.error(f"Error encoding job_type: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error encoding job_type: {str(e)}")
+
+        # Split data
+        if len(df) < 2:
+            logger.warning("Insufficient data points for training and testing")
+            raise HTTPException(status_code=400, detail="Insufficient data points for training and testing")
+        split_point = int(len(df) * 0.8)
+        train_df = df.iloc[:split_point].copy()
+        test_df = df.iloc[split_point:].copy()
+
+        # Scale features
+        scaler = StandardScaler()
+        try:
+            train_df[['job_type_scaled', 'years_exp_scaled']] = scaler.fit_transform(
+                train_df[['job_type_encoded', 'years_of_experience']]
+            )
+            train_df['job_exp_interaction'] = train_df['job_type_scaled'] * train_df['years_exp_scaled']
+        except Exception as e:
+            logger.error(f"Error scaling features: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error scaling features: {str(e)}")
+
+        for subset in [train_df, test_df]:
+            subset['dayofweek'] = subset['timestamp'].dt.dayofweek
+            subset['month'] = subset['timestamp'].dt.month
+            subset['year'] = subset['timestamp'].dt.year
+            subset['dayofyear'] = subset['timestamp'].dt.dayofyear
+            subset['is_weekend'] = subset['dayofweek'].isin([5, 6]).astype(int)
+
+        # Train classifier
+        X_train_class = train_df[['dayofweek', 'month', 'year', 'dayofyear',
+                                 'is_weekend', 'job_type_encoded', 'feedback_score',
+                                 'years_of_experience']]
+        y_train_class = train_df['has_job']
+
+        try:
+            classifier = RandomForestClassifier(
+                n_estimators=500, max_depth=12, min_samples_split=5, random_state=42
+            )
+            classifier.fit(X_train_class, y_train_class)
+        except Exception as e:
+            logger.error(f"Error training classifier: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error training classifier: {str(e)}")
+
+        # Train regressor
+        train_df_reg = train_df[train_df['has_job'] == 1].copy()
+        if train_df_reg.empty:
+            logger.warning("No data available for regression (all has_job == 0)")
+            raise HTTPException(status_code=404, detail="No data available for regression (all has_job == 0)")
+
+        X_train_reg = train_df_reg[['dayofweek', 'month', 'year', 'dayofyear',
+                                   'is_weekend', 'job_type_scaled', 'feedback_score',
+                                   'years_exp_scaled', 'job_exp_interaction']]
+        y_train_reg = train_df_reg['contracted_wage']
+
+        try:
+            regressor = RandomForestRegressor(
+                n_estimators=300, max_depth=10, min_samples_split=4, random_state=42
+            )
+            regressor.fit(X_train_reg, y_train_reg)
+        except Exception as e:
+            logger.error(f"Error training regressor: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error training regressor: {str(e)}")
+
+        # Prepare future dataframe
+        future_df = test_df[['timestamp', 'job_type', 'job_type_encoded',
+                            'feedback_score', 'years_of_experience']].rename(columns={'timestamp': 'ds'})
+
+        future_df['dayofweek'] = future_df['ds'].dt.dayofweek
+        future_df['month'] = future_df['ds'].dt.month
+        future_df['year'] = future_df['ds'].dt.year
+        future_df['dayofyear'] = future_df['ds'].dt.dayofyear
+        future_df['is_weekend'] = future_df['dayofweek'].isin([5, 6]).astype(int)
 
-    except HTTPException:
-        raise  # let FastAPI handle cleanly
+        try:
+            future_df[['job_type_scaled', 'years_exp_scaled']] = scaler.transform(
+                future_df[['job_type_encoded', 'years_of_experience']]
+            )
+            future_df['job_exp_interaction'] = future_df['job_type_scaled'] * future_df['years_exp_scaled']
+        except Exception as e:
+            logger.error(f"Error transforming future dataframe: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error transforming future dataframe: {str(e)}")
+
+        # Predict job/no-job
+        try:
+            future_df['has_job_predicted'] = classifier.predict(
+                future_df[['dayofweek', 'month', 'year', 'dayofyear',
+                          'is_weekend', 'job_type_encoded', 'feedback_score',
+                          'years_of_experience']]
+            )
+        except Exception as e:
+            logger.error(f"Error predicting has_job: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error predicting has_job: {str(e)}")
+
+        # Evaluate classifier accuracy
+        test_df['has_job'] = (test_df['job_type'] != "No Job").astype(int)
+        try:
+            acc = accuracy_score(test_df['has_job'], future_df['has_job_predicted'])
+            results['classification_metrics']['accuracy'] = round(acc * 100, 2)
+        except Exception as e:
+            logger.error(f"Error calculating accuracy: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error calculating accuracy: {str(e)}")
+
+        # Predict wages
+        try:
+            future_df['yhat'] = regressor.predict(
+                future_df[['dayofweek', 'month', 'year', 'dayofyear',
+                          'is_weekend', 'job_type_scaled', 'feedback_score',
+                          'years_exp_scaled', 'job_exp_interaction']]
+            )
+        except Exception as e:
+            logger.error(f"Error predicting wages: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error predicting wages: {str(e)}")
+
+        # Apply job prediction mask
+        final_forecast_df = future_df.copy()
+        final_forecast_df['yhat'] = np.where(final_forecast_df['has_job_predicted'] == 0, 0, final_forecast_df['yhat'])
+        final_forecast_df['yhat'] = np.minimum(final_forecast_df['yhat'], wage_cap)
+
+        # Uncertainty intervals
+        try:
+            predictions = regressor.predict(X_train_reg)
+            std_dev = np.std([tree.predict(X_train_reg) for tree in regressor.estimators_], axis=0)
+            future_df['yhat_lower'] = np.maximum(final_forecast_df['yhat'] - 1.96 * std_dev.mean(), 0)
+            future_df['yhat_upper'] = final_forecast_df['yhat'] + 1.96 * std_dev.mean()
+            final_forecast_df['yhat_lower'] = np.where(final_forecast_df['has_job_predicted'] == 0, 0, future_df['yhat_lower'])
+            final_forecast_df['yhat_upper'] = np.where(final_forecast_df['has_job_predicted'] == 0, 0, future_df['yhat_upper'])
+        except Exception as e:
+            logger.error(f"Error calculating uncertainty intervals: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error calculating uncertainty intervals: {str(e)}")
+
+        # Evaluation
+        try:
+            comparison_df = pd.merge(
+                test_df[['timestamp', 'contracted_wage']].rename(columns={'timestamp': 'ds', 'contracted_wage': 'y'}),
+                final_forecast_df[['ds', 'yhat', 'yhat_lower', 'yhat_upper']], on='ds'
+            )
+
+            valid_comparison_df = comparison_df[comparison_df['y'] > 0]
+            if not valid_comparison_df.empty:
+                weights = valid_comparison_df['y'] / valid_comparison_df['y'].mean()
+                mae = np.average([abs(a - p) for a, p in zip(valid_comparison_df['y'], valid_comparison_df['yhat'])], weights=weights)
+                mape = np.average([abs((a - p) / a) * 100 for a, p in zip(valid_comparison_df['y'], valid_comparison_df['yhat'])], weights=weights)
+            else:
+                mae = np.nan
+                mape = np.nan
+
+            results['classification_metrics']['mae'] = round(mae, 2) if not np.isnan(mae) else None
+            results['classification_metrics']['mape'] = round(mape, 2) if not np.isnan(mape) else None
+        except Exception as e:
+            logger.error(f"Error evaluating predictions: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error evaluating predictions: {str(e)}")
+
+        # Plot results
+        try:
+            plt.figure(figsize=(12, 6))
+            plt.plot(comparison_df['ds'], comparison_df['y'], 'o-', label='Actual Values', markersize=4)
+            plt.plot(comparison_df['ds'], comparison_df['yhat'], '-', label='Forecasted Values')
+            plt.fill_between(comparison_df['ds'], comparison_df['yhat_lower'], comparison_df['yhat_upper'],
+                            color='gray', alpha=0.2, label='Uncertainty Interval')
+            plt.title('Actual vs. Forecasted Daily Earnings (Last 20% of Dataset)')
+            plt.xlabel('Date')
+            plt.ylabel('Contracted Wage')
+            plt.legend()
+            plt.grid(True)
+            plt.xticks(rotation=45)
+            plt.tight_layout()
+
+            buffer = BytesIO()
+            plt.savefig(buffer, format='png')
+            buffer.seek(0)
+            plot_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
+            results['plot'] = f'data:image/png;base64,{plot_base64}'
+            plt.close()
+        except Exception as e:
+            logger.error(f"Error generating plot: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error generating plot: {str(e)}")
+
+        # Worker Profile for Microfinance
+        try:
+            worker_data = df.copy()
+
+            avg_daily_earning = worker_data[worker_data['contracted_wage'] > 0]['contracted_wage'].mean()
+            avg_monthly_earning = avg_daily_earning * 30 if not np.isnan(avg_daily_earning) else 0
+
+            job_distribution = worker_data['job_type'].value_counts(normalize=True) * 100
+
+            avg_feedback = worker_data['feedback_score'].mean()
+
+            workholic_index = job_distribution.drop(labels=['No Job'], errors='ignore').sum() / 100
+
+            if avg_daily_earning > 0:
+                earning_stability = worker_data[worker_data['contracted_wage'] > 0]['contracted_wage'].std() / avg_daily_earning
+            else:
+                earning_stability = np.nan
+
+            results['worker_profile'] = {
+                'average_daily_earning': round(avg_daily_earning, 2) if not np.isnan(avg_daily_earning) else None,
+                'estimated_monthly_earning': round(avg_monthly_earning, 2) if not np.isnan(avg_monthly_earning) else None,
+                'job_distribution': job_distribution.round(2).to_dict(),
+                'average_feedback_score': round(avg_feedback, 2) if not np.isnan(avg_feedback) else None,
+                'workholic_index': round(workholic_index, 2) if not np.isnan(workholic_index) else None,
+                'earning_stability': round(earning_stability, 2) if not np.isnan(earning_stability) else None
+            }
+        except Exception as e:
+            logger.error(f"Error generating worker profile: {str(e)}")
+            raise HTTPException(status_code=500, detail=f"Error generating worker profile: {str(e)}")
+
+        def convert_to_serializable(obj):
+            if isinstance(obj, np.floating):
+                return float(obj)
+            if isinstance(obj, np.integer):
+                return int(obj)
+            if isinstance(obj, np.ndarray):
+                return obj.tolist()
+            return obj
+
+        logger.info("Request processed successfully")
+        return json.loads(json.dumps(results, default=convert_to_serializable))
 
     except Exception as e:
-        print("==== SERVER ERROR ====")
-        print(str(e))
-        traceback.print_exc()
+        logger.error(f"Error processing request: {str(e)}")
         raise HTTPException(status_code=500, detail=f"Error processing request: {str(e)}")
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)