predict_fastapi.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from prophet import Prophet
import pandas as pd
from datetime import datetime, timedelta
import logging
import traceback
import numpy as np
import uvicorn

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

np.random.seed(42)

# Create FastAPI app
app = FastAPI(
    title="Church Financial Prediction API",
    description="API for predicting church financial data using Prophet",
    version="1.0.0"
)

# Pydantic models for request/response
class DataPoint(BaseModel):
    ds: str
    y: float

class PredictionRequest(BaseModel):
    data: List[DataPoint]

class PredictionResponse(BaseModel):
    ds: str
    month: str
    date_formatted: str
    yhat: float
    yhat_lower: float
    yhat_upper: float

class HealthResponse(BaseModel):
    status: str
    timestamp: str
    version: str

@app.get("/", response_model=HealthResponse)
async def root():
    """Root endpoint with health information"""
    return HealthResponse(
        status="healthy",
        timestamp=datetime.now().isoformat(),
        version="1.0.0"
    )

@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Health check endpoint"""
    return HealthResponse(
        status="healthy",
        timestamp=datetime.now().isoformat(),
        version="1.0.0"
    )

@app.post("/predict", response_model=List[PredictionResponse])
async def predict(request: PredictionRequest):
    """Make predictions for church financial data"""
    try:
        logger.info("Received prediction request")
        
        if not request.data:
            raise HTTPException(status_code=400, detail="No data provided")
            
        logger.info(f"Received data points: {len(request.data)}")
        
        # Convert to DataFrame
        data_list = [{"ds": point.ds, "y": point.y} for point in request.data]
        df = pd.DataFrame(data_list)
        df['ds'] = pd.to_datetime(df['ds'])
        
        # Sort by date to ensure proper order
        df = df.sort_values('ds')
        
        # Handle missing or invalid values
        df = df.dropna()
        df = df[df['y'] >= 0]  # Remove negative values
        
        # Debug: Log the input data
        logger.info(f"Input data summary:")
        logger.info(f"Date range: {df['ds'].min()} to {df['ds'].max()}")
        logger.info(f"Value range: {df['y'].min()} to {df['y'].max()}")
        logger.info(f"Mean value: {df['y'].mean():.2f}")
        logger.info(f"Standard deviation: {df['y'].std():.2f}")
        
        # Check for sufficient data points
        if len(df) < 3:
            raise HTTPException(status_code=400, detail="Need at least 3 data points for prediction")
        
        # Configure Prophet model with adaptive settings based on data
        data_variation = df['y'].std()
        data_mean = df['y'].mean()
        
        # Adjust model parameters based on data characteristics
        if data_variation < 100 or len(df) < 6:
            logger.info("Using enhanced model settings for low variation data")
            model = Prophet(
                yearly_seasonality=True,
                weekly_seasonality=False,
                daily_seasonality=False,
                seasonality_mode='multiplicative',
                changepoint_prior_scale=0.15,
                seasonality_prior_scale=25.0,
                holidays_prior_scale=25.0,
                interval_width=0.8,
                growth='linear'
            )
        else:
            logger.info("Using sophisticated model settings for varied data")
            model = Prophet(
                yearly_seasonality=True,
                weekly_seasonality=False,
                daily_seasonality=False,
                seasonality_mode='multiplicative',
                changepoint_prior_scale=0.1,
                seasonality_prior_scale=20.0,
                holidays_prior_scale=20.0,
                interval_width=0.85,
                growth='linear'
            )
        
        # Add custom seasonality for church donations
        model.add_seasonality(name='monthly', period=30.5, fourier_order=5)
        model.add_seasonality(name='quarterly', period=91.25, fourier_order=3)
        
        # Fit the model
        logger.info("Fitting Prophet model...")
        model.fit(df)
        
        # Create future dates for 2025 predictions
        future_dates = pd.date_range(start='2025-01-01', end='2025-12-31', freq='MS')
        future = pd.DataFrame({'ds': future_dates})
        
        # Make predictions
        forecast = model.predict(future)
        
        # Get predictions for 2025 and ensure proper formatting for chart
        predictions_2025 = forecast[forecast['ds'].dt.year == 2025][['ds', 'yhat', 'yhat_lower', 'yhat_upper']]
        
        # Convert to response format
        result = []
        for _, row in predictions_2025.iterrows():
            date_obj = row['ds']
            month_key = date_obj.strftime('%Y-%m')
            date_formatted = date_obj.strftime('%B 01, %Y')
            
            result.append(PredictionResponse(
                ds=row['ds'].strftime('%Y-%m-%d'),
                month=month_key,
                date_formatted=date_formatted,
                yhat=max(0, round(float(row['yhat']), 2)),
                yhat_lower=max(0, round(float(row['yhat_lower']), 2)),
                yhat_upper=max(0, round(float(row['yhat_upper']), 2))
            ))
        
        logger.info(f"Generated {len(result)} predictions for 2025")
        
        # Log prediction statistics for monitoring
        prediction_values = [r.yhat for r in result]
        prediction_std = np.std(prediction_values)
        prediction_mean = np.mean(prediction_values)
        
        logger.info(f"Prediction statistics - Mean: {prediction_mean:.2f}, Std: {prediction_std:.2f}")
        
        return result
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error in prediction: {str(e)}")
        logger.error(f"Traceback: {traceback.format_exc()}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@app.get("/test")
async def test_prediction():
    """Test endpoint to verify prediction functionality"""
    try:
        # Create sample data for testing
        test_data = [
            DataPoint(ds='2024-01-01', y=25000),
            DataPoint(ds='2024-02-01', y=28000),
            DataPoint(ds='2024-03-01', y=32000),
            DataPoint(ds='2024-04-01', y=45000),
            DataPoint(ds='2024-05-01', y=30000),
            DataPoint(ds='2024-06-01', y=25000),
            DataPoint(ds='2024-07-01', y=22000),
            DataPoint(ds='2024-08-01', y=24000),
            DataPoint(ds='2024-09-01', y=28000),
            DataPoint(ds='2024-10-01', y=32000),
            DataPoint(ds='2024-11-01', y=38000),
            DataPoint(ds='2024-12-01', y=55000)
        ]
        
        # Make prediction request
        request = PredictionRequest(data=test_data)
        predictions = await predict(request)
        
        return {
            'status': 'success',
            'message': 'Prediction test successful',
            'predictions_count': len(predictions),
            'sample_prediction': predictions[0].dict() if predictions else None,
            'prediction_range': {
                'min': min([p.yhat for p in predictions]) if predictions else 0,
                'max': max([p.yhat for p in predictions]) if predictions else 0
            }
        }
        
    except Exception as e:
        return {
            'status': 'error',
            'message': f'Test failed: {str(e)}'
        }

if __name__ == "__main__":
    uvicorn.run(
        "predict_fastapi:app",
        host="0.0.0.0",
        port=5000,
        reload=False,
        workers=1
    )