Update modelLoanAPI.py

modelLoanAPI.py

@@ -1,318 +1,254 @@
-# ```python
 from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
+from fastapi.responses import JSONResponse, FileResponse
 import pandas as pd
 import numpy as np
-from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
-from sklearn.preprocessing import LabelEncoder, StandardScaler
+from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
+from sklearn.preprocessing import RobustScaler, LabelEncoder
 from sklearn.metrics import accuracy_score
+from sklearn.model_selection import TimeSeriesSplit, GridSearchCV
 import matplotlib.pyplot as plt
 import json
 import base64
 from io import BytesIO
+from PIL import Image
 import warnings
-import logging
-
-# Set up logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
+import os
+from pydantic import BaseModel
 
 warnings.filterwarnings("ignore")
 
 app = FastAPI()
 
-class WorkerIdRequest(BaseModel):
+class WorkerInput(BaseModel):
     worker_id: int
 
-@app.post("/predict_worker_earnings/")
-async def predict_worker_earnings(request: WorkerIdRequest):
+@app.post("/worker_forecast/")
+async def worker_forecast(input_data: WorkerInput):
+    worker_id = input_data.worker_id
+
+    # Initialize result dictionary
+    results = {
+        'worker_id': worker_id,
+        'metrics': {},
+        'worker_profile': {},
+        'plot': ''
+    }
+
+    # Load dataset
     try:
-        worker_id = request.worker_id
-        logger.info(f"Processing request for worker_id: {worker_id}")
+        df = pd.read_csv('extended_worker_dataset_random_reduced.csv')
+    except FileNotFoundError:
+        raise HTTPException(status_code=500, detail="Dataset file not found")
+
+    # Filter for one worker_id
+    df = df[df['worker_id'] == worker_id].copy()
+    if df.empty:
+        raise HTTPException(status_code=404, detail=f"No data found for worker_id {worker_id}")
+
+    # Data preprocessing
+    df['timestamp'] = pd.to_datetime(df['timestamp'], dayfirst=True, errors='coerce')
+    df['has_job'] = (df['job_type'] != "No Job").astype(int)
+
+    # Wage capping
+    wage_cap = df[df['contracted_wage'] > 0]['contracted_wage'].quantile(0.98)
+    df['contracted_wage'] = df['contracted_wage'].clip(lower=500, upper=wage_cap)
+
+    # Encode categorical
+    le_job = LabelEncoder()
+    df['job_type_encoded'] = le_job.fit_transform(df['job_type'])
+    le_labour = LabelEncoder()
+    df['labour_category_encoded'] = le_labour.fit_transform(df['labour_category'])
+
+    # Lagged and rolling features
+    df['prev_wage'] = df['contracted_wage'].shift(1).fillna(0)
+    df['prev_wage2'] = df['contracted_wage'].shift(2).fillna(0)
+    df['prev_wage3'] = df['contracted_wage'].shift(3).fillna(0)
+    df['rolling_mean_3'] = df['contracted_wage'].rolling(3, min_periods=1).mean().shift(1).fillna(0)
+    df['rolling_std_3'] = df['contracted_wage'].rolling(3, min_periods=1).std().shift(1).fillna(0)
+    df['rolling_mean_7'] = df['contracted_wage'].rolling(7, min_periods=1).mean().shift(1).fillna(0)
+
+    # Train/test split
+    split_point = int(len(df) * 0.8)
+    train_df, test_df = df.iloc[:split_point].copy(), df.iloc[split_point:].copy()
+
+    # Scaling
+    scaler = RobustScaler()
+    train_df[['job_type_scaled', 'years_exp_scaled', 'prev_wage_scaled', 'prev_wage2_scaled', 'prev_wage3_scaled',
+              'rolling_mean_3_scaled', 'rolling_std_3_scaled', 'rolling_mean_7_scaled', 'labour_category_scaled']] = scaler.fit_transform(
+        train_df[['job_type_encoded', 'years_of_experience', 'prev_wage', 'prev_wage2', 'prev_wage3',
+                  'rolling_mean_3', 'rolling_std_3', 'rolling_mean_7', 'labour_category_encoded']]
+    )
+    train_df['job_exp_interaction'] = train_df['job_type_scaled'] * train_df['years_exp_scaled']
+
+    # Date features
+    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)
+
+    # Classification model
+    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']
+    classifier = RandomForestClassifier(n_estimators=500, max_depth=12, min_samples_split=5, random_state=42)
+    classifier.fit(X_train_class, y_train_class)
+
+    # Regression model (only when has_job=1)
+    train_df_reg = train_df[train_df['has_job'] == 1].copy()
+    X_train_reg = train_df_reg[['dayofweek', 'month', 'year', 'dayofyear',
+                                'is_weekend', 'job_type_scaled', 'feedback_score',
+                                'years_exp_scaled', 'job_exp_interaction', 'prev_wage_scaled',
+                                'prev_wage2_scaled', 'prev_wage3_scaled', 'rolling_mean_3_scaled',
+                                'rolling_std_3_scaled', 'rolling_mean_7_scaled', 'labour_category_scaled']]
+    y_train_reg = train_df_reg['contracted_wage']
+
+    # Hyperparameter tuning
+    tscv = TimeSeriesSplit(n_splits=5)
+    param_grid = {
+        'n_estimators': [200, 300, 400],
+        'learning_rate': [0.01, 0.05],
+        'max_depth': [3, 4, 5],
+        'min_samples_split': [3, 4],
+        'min_samples_leaf': [2, 3]
+    }
+    grid_search = GridSearchCV(GradientBoostingRegressor(random_state=42),
+                               param_grid, cv=tscv, scoring='neg_mean_absolute_error', n_jobs=-1)
+    grid_search.fit(X_train_reg, y_train_reg)
+    best_reg = grid_search.best_estimator_
+    best_params = grid_search.best_params_
+
+    # Quantile regressors
+    reg_lower = GradientBoostingRegressor(loss='quantile', alpha=0.025, **best_params, random_state=42)
+    reg_upper = GradientBoostingRegressor(loss='quantile', alpha=0.975, **best_params, random_state=42)
+    reg_lower.fit(X_train_reg, y_train_reg)
+    reg_upper.fit(X_train_reg, y_train_reg)
+
+    # Future dataframe
+    future_df = test_df[['timestamp', 'job_type', 'job_type_encoded', 'feedback_score', 'years_of_experience',
+                         'prev_wage', 'prev_wage2', 'prev_wage3', 'rolling_mean_3', 'rolling_std_3', 'rolling_mean_7',
+                         'labour_category_encoded']].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)
+    future_df[['job_type_scaled', 'years_exp_scaled', 'prev_wage_scaled', 'prev_wage2_scaled', 'prev_wage3_scaled',
+               'rolling_mean_3_scaled', 'rolling_std_3_scaled', 'rolling_mean_7_scaled', 'labour_category_scaled']] = scaler.transform(
+        future_df[['job_type_encoded', 'years_of_experience', 'prev_wage', 'prev_wage2', 'prev_wage3',
+                   'rolling_mean_3', 'rolling_std_3', 'rolling_mean_7', 'labour_category_encoded']]
+    )
+    future_df['job_exp_interaction'] = future_df['job_type_scaled'] * future_df['years_exp_scaled']
+
+    # Predictions
+    future_df['has_job_predicted'] = classifier.predict(
+        future_df[['dayofweek', 'month', 'year', 'dayofyear', 'is_weekend', 'job_type_encoded',
+                   'feedback_score', 'years_of_experience']]
+    )
+    future_df['yhat'] = best_reg.predict(
+        future_df[['dayofweek', 'month', 'year', 'dayofyear', 'is_weekend', 'job_type_scaled', 'feedback_score',
+                   'years_exp_scaled', 'job_exp_interaction', 'prev_wage_scaled', 'prev_wage2_scaled', 'prev_wage3_scaled',
+                   'rolling_mean_3_scaled', 'rolling_std_3_scaled', 'rolling_mean_7_scaled', 'labour_category_scaled']]
+    )
+    future_df['yhat_lower'] = reg_lower.predict(
+        future_df[['dayofweek', 'month', 'year', 'dayofyear', 'is_weekend', 'job_type_scaled', 'feedback_score',
+                   'years_exp_scaled', 'job_exp_interaction', 'prev_wage_scaled', 'prev_wage2_scaled', 'prev_wage3_scaled',
+                   'rolling_mean_3_scaled', 'rolling_std_3_scaled', 'rolling_mean_7_scaled', 'labour_category_scaled']]
+    )
+    future_df['yhat_upper'] = reg_upper.predict(
+        future_df[['dayofweek', 'month', 'year', 'dayofyear', 'is_weekend', 'job_type_scaled', 'feedback_score',
+                   'years_exp_scaled', 'job_exp_interaction', 'prev_wage_scaled', 'prev_wage2_scaled', 'prev_wage3_scaled',
+                   'rolling_mean_3_scaled', 'rolling_std_3_scaled', 'rolling_mean_7_scaled', 'labour_category_scaled']]
+    )
+
+    # Apply job 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)
+    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'])
+    final_forecast_df['yhat_lower'] = np.maximum(final_forecast_df['yhat_lower'], 0)
+
+    # Evaluation
+    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', how='left'
+    )
+    comparison_df = comparison_df.set_index(final_forecast_df.index)  # Align indices
+    valid_comparison_df = comparison_df[comparison_df['y'] > 0].copy()
+
+    if not valid_comparison_df.empty:
+        valid_y = valid_comparison_df['y'].values
+        valid_yhat = valid_comparison_df['yhat'].values
+        weights = valid_comparison_df['y'].values / valid_comparison_df['y'].mean()
         
-        # Initialize result dictionary
-        results = {
-            'worker_id': worker_id,
-            'classification_metrics': {},
-            'worker_profile': {},
-            'plot': ''
-        }
-
-        # 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)}")
-
-        # 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}")
-
-        # 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)
-
-        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:
-        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)
+        mae = np.average(np.abs(valid_y - valid_yhat), weights=weights, axis=0)
+        mape = np.average(np.abs((valid_y - valid_yhat) / valid_y) * 100, weights=weights, axis=0)
+    else:
+        mae, mape = np.nan, np.nan
+
+    results['metrics']['mae'] = round(mae, 2) if not np.isnan(mae) else None
+    results['metrics']['mape'] = round(mape, 2) if not np.isnan(mape) else None
+
+    # Plot results
+    plt.figure(figsize=(12, 6))
+    plt.plot(final_forecast_df['ds'], final_forecast_df['yhat'], '-', label='Forecasted', color='blue')
+    plt.fill_between(final_forecast_df['ds'], final_forecast_df['yhat_lower'], final_forecast_df['yhat_upper'],
+                     color='gray', alpha=0.2, label='Uncertainty')
+    plt.title('Forecasted Daily Earnings (Last 20%)')
+    plt.xlabel('Date'); plt.ylabel('Contracted Wage')
+    plt.legend(); plt.grid(True); plt.xticks(rotation=45); plt.tight_layout()
+
+    # Save plot as PNG → JPG for compression
+    buf_png = BytesIO()
+    plt.savefig(buf_png, format="png", dpi=80, bbox_inches="tight")
+    plt.close()
+    buf_png.seek(0)
+    img = Image.open(buf_png).convert("RGB")
+    buf_jpg = BytesIO()
+    img.save(buf_jpg, format="JPEG", quality=70, optimize=True)
+    buf_jpg.seek(0)
+    plot_base64 = base64.b64encode(buf_jpg.getvalue()).decode("utf-8")
+    results['plot'] = f"data:image/jpeg;base64,{plot_base64}"
+
+    # Save plot to file
+    plot_filename = f"worker_{worker_id}_forecast.jpg"
+    with open(plot_filename, "wb") as f:
+        f.write(base64.b64decode(plot_base64))
+
+    # Worker profile
+    worker_data = df.copy()
+    avg_daily = worker_data[worker_data['contracted_wage'] > 0]['contracted_wage'].mean()
+    avg_monthly = avg_daily * 30 if not np.isnan(avg_daily) else 0
+    job_dist = worker_data['job_type'].value_counts(normalize=True) * 100
+    avg_feedback = worker_data['feedback_score'].mean()
+    work_index = job_dist.drop(labels=['No Job'], errors='ignore').sum() / 100
+    earn_stability = worker_data[worker_data['contracted_wage'] > 0]['contracted_wage'].std() / avg_daily if avg_daily > 0 else np.nan
+
+    results['worker_profile'] = {
+        'average_daily_earning': round(avg_daily, 2) if not np.isnan(avg_daily) else None,
+        'estimated_monthly_earning': round(avg_monthly, 2) if not np.isnan(avg_monthly) else None,
+        'job_distribution': job_dist.round(2).to_dict(),
+        'average_feedback_score': round(avg_feedback, 2) if not np.isnan(avg_feedback) else None,
+        'work_index': round(work_index, 2) if not np.isnan(work_index) else None,
+        'earning_stability': round(earn_stability, 2) if not np.isnan(earn_stability) else None
+    }
+
+    def convert_to_serializable(obj):
+        if isinstance(obj, (np.floating, np.float32, np.float64)): return float(obj)
+        if isinstance(obj, (np.integer, np.int32, np.int64)): return int(obj)
+        if isinstance(obj, np.ndarray): return obj.tolist()
+        return obj
+
+    # Return JSON response with results
+    return JSONResponse(content=json.loads(json.dumps(results, default=convert_to_serializable)))
+
+@app.get("/worker_forecast/plot/{worker_id}")
+async def get_forecast_plot(worker_id: int):
+    plot_filename = f"worker_{worker_id}_forecast.jpg"
+    if os.path.exists(plot_filename):
+        return FileResponse(plot_filename, media_type="image/jpeg", filename=f"worker_{worker_id}_forecast.jpg")
+    else:
+        raise HTTPException(status_code=404, detail=f"Plot for worker_id {worker_id} not found")