Fixed drift report generation -Evidently

AWS_SETUP_GUIDE.md

@@ -191,4 +191,3 @@ For a production deployment, consider adding:
 - **VPC Endpoints** - Reduce S3 data transfer costs
 - **Secrets Manager** - Store database credentials securely
 
-See the original AWS_SETUP_GUIDE_FULL.md for complete production setup instructions.

src/app/api.py

@@ -1,7 +1,4 @@
-"""
-FastAPI application for Flight Price Prediction.
-Provides REST API endpoints for predictions, monitoring, and model management.
-"""
+"""FastAPI application for Flight Price Prediction."""
 
 import os
 import sys
@@ -61,8 +58,6 @@
 
 
 class FlightFeatures(BaseModel):
-    """Input features for flight price prediction."""
-
     airline: str = Field(..., description="Airline name")
     source_city: str = Field(..., description="Source city")
     destination_city: str = Field(..., description="Destination city")
@@ -91,8 +86,6 @@ class Config:
 
 
 class PredictionResponse(BaseModel):
-    """Response model for predictions."""
-
     predicted_price: float
     model_name: str
     model_version: str
@@ -101,8 +94,6 @@ class PredictionResponse(BaseModel):
 
 
 class ModelInfoResponse(BaseModel):
-    """Response model for model information."""
-
     model_name: str
     model_version: Optional[str]
     model_alias: Optional[str]
@@ -111,16 +102,7 @@ class ModelInfoResponse(BaseModel):
 
 
 def load_model_from_mlflow(alias: str = None, version: str = None):
-    """
-    Load model from MLflow registry.
-
-    Args:
-        alias: Model alias (e.g., 'production', 'staging')
-        version: Specific model version
-
-    Returns:
-        Loaded model and metadata
-    """
+    """Load model from MLflow registry by alias or version."""
     mlflow_uri = os.getenv('MLFLOW_TRACKING_URI', 'http://localhost:5000')
     mlflow.set_tracking_uri(mlflow_uri)
 
@@ -190,7 +172,7 @@ def load_model_local():
 
 
 def initialize_model():
-    """Initialize model on startup."""
+    """Load model on startup."""
     global current_model, current_model_info
 
     try:
@@ -218,15 +200,13 @@ def initialize_model():
 
 @app.on_event("startup")
 async def startup_event():
-    """Run initialization on startup."""
     logger.info("Starting Flight Price Prediction API...")
     initialize_model()
     logger.info("API startup complete")
 
 
 @app.get("/health", status_code=status.HTTP_200_OK)
 async def health_check():
-    """Health check endpoint."""
     REQUEST_COUNT.labels(method='GET', endpoint='/health', status='200').inc()
 
     return {
@@ -238,13 +218,11 @@ async def health_check():
 
 @app.get("/metrics")
 async def metrics():
-    """Prometheus metrics endpoint."""
     return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)
 
 
 @app.get("/model_info", response_model=ModelInfoResponse)
 async def get_model_info():
-    """Get current model information."""
     REQUEST_COUNT.labels(method='GET', endpoint='/model_info', status='200').inc()
 
     if not current_model:
@@ -255,13 +233,7 @@ async def get_model_info():
 
 @app.post("/reload")
 async def reload_model(alias: str = None, version: str = None):
-    """
-    Reload model without restarting the service (zero-downtime update).
-
-    Args:
-        alias: Model alias to load (e.g., 'production')
-        version: Specific version to load
-    """
+    """Reload model without restarting the service."""
     global current_model, current_model_info
 
     try:
@@ -299,15 +271,7 @@ async def reload_model(alias: str = None, version: str = None):
 
 @app.post("/predict", response_model=PredictionResponse)
 async def predict(features: FlightFeatures):
-    """
-    Predict flight price based on input features.
-
-    Args:
-        features: Flight features for prediction
-
-    Returns:
-        Predicted price and metadata
-    """
+    """Predict flight price from input features."""
     start_time = time.time()
 
     if not current_model:
@@ -316,20 +280,10 @@ async def predict(features: FlightFeatures):
         raise HTTPException(status_code=503, detail="Model not loaded")
 
     try:
-        # Convert features to format expected by model
-        # This is simplified - you'll need to match your preprocessing pipeline
         feature_dict = features.dict(by_alias=True)
 
-        # For now, we'll assume the model expects preprocessed features
-        # In production, you'd apply the same preprocessing as training
-        # TODO: Integrate with data preprocessing pipeline
-
-        # Make prediction (placeholder - needs actual feature engineering)
-        # predicted_price = current_model.predict(processed_features)[0]
-
-        # Temporary: return a mock prediction
-        # Replace this with actual model inference
-        predicted_price = 5000.0  # Placeholder
+        # TODO: integrate actual preprocessing pipeline
+        predicted_price = 5000.0
 
         # Calculate latency
         latency_ms = (time.time() - start_time) * 1000
@@ -373,7 +327,6 @@ async def predict(features: FlightFeatures):
 
 @app.get("/")
 async def root():
-    """Root endpoint with API information."""
     return {
         "message": "Flight Price Prediction API",
         "version": os.getenv('API_VERSION', '1.0.0'),

src/database/models.py

@@ -1,7 +1,4 @@
-"""
-Database models for storing predictions and monitoring.
-Uses SQLAlchemy ORM with PostgreSQL.
-"""
+"""Database models for predictions."""
 
 from sqlalchemy import Column, Integer, Float, String, DateTime, JSON, Index, create_engine
 from sqlalchemy.ext.declarative import declarative_base
@@ -13,30 +10,20 @@
 
 
 class Prediction(Base):
-    """Model for storing flight price predictions."""
-
     __tablename__ = 'predictions'
 
     id = Column(Integer, primary_key=True, autoincrement=True)
     timestamp = Column(DateTime, default=datetime.utcnow, nullable=False, index=True)
 
-    # Input features (stored as JSON for flexibility)
     features = Column(JSON, nullable=False)
 
-    # Prediction outputs
     predicted_price = Column(Float, nullable=False)
 
-    # Model metadata
     model_name = Column(String(100), nullable=True, index=True)
     model_version = Column(String(50), nullable=True, index=True)
-
-    # Performance tracking
     latency_ms = Column(Float, nullable=True)
-
-    # Optional: actual price (for feedback loop)
     actual_price = Column(Float, nullable=True)
 
-    # Create composite index for common queries
     __table_args__ = (
         Index('idx_timestamp_model', 'timestamp', 'model_version'),
     )
@@ -46,22 +33,19 @@ def __repr__(self):
 
 
 def get_database_url():
-    """Get database URL from environment or use default."""
     return os.getenv(
         'DATABASE_URL',
         'postgresql://postgres:postgres@localhost:5432/flightprice'
     )
 
 
 def create_tables():
-    """Create all database tables."""
     engine = create_engine(get_database_url())
     Base.metadata.create_all(engine)
     print("Database tables created successfully")
 
 
 def get_session():
-    """Get database session."""
     engine = create_engine(get_database_url())
     Session = sessionmaker(bind=engine)
     return Session()

src/ml/data.py

@@ -1,7 +1,4 @@
-"""
-Data preparation module for Flight Price Prediction.
-Handles data loading, preprocessing, feature engineering, and train/test split.
-"""
+"""Data preparation for Flight Price Prediction."""
 
 import os
 import pandas as pd
@@ -18,22 +15,17 @@
 
 
 class FlightDataProcessor:
-    """Process flight data for model training and inference."""
-
     def __init__(self, config_path: str = "configs/base.yaml"):
-        """Initialize data processor with configuration."""
         self.config = self._load_config(config_path)
         self.scaler = None
         self.feature_names = None
         self.label_encoders = {}  # Store encoders for each categorical column
 
     def _load_config(self, config_path: str) -> Dict[str, Any]:
-        """Load configuration from YAML file."""
         with open(config_path, 'r') as f:
             return yaml.safe_load(f)
 
     def load_data(self, data_path: str) -> pd.DataFrame:
-        """Load flight data from CSV file."""
         logger.info(f"Loading data from {data_path}")
 
         if data_path.endswith('.csv'):
@@ -52,7 +44,6 @@ def load_data(self, data_path: str) -> pd.DataFrame:
         return df
 
     def engineer_features(self, df: pd.DataFrame) -> pd.DataFrame:
-        """Create additional features from existing ones."""
         logger.info("Engineering features...")
 
         df = df.copy()
@@ -83,7 +74,6 @@ def engineer_features(self, df: pd.DataFrame) -> pd.DataFrame:
         return df
 
     def _parse_duration(self, duration_str: str) -> float:
-        """Parse duration string to hours."""
         if pd.isna(duration_str):
             return 0.0
 
@@ -99,16 +89,7 @@ def _parse_duration(self, duration_str: str) -> float:
             return 0.0
 
     def preprocess(self, df: pd.DataFrame, fit: bool = True) -> Tuple[np.ndarray, pd.DataFrame]:
-        """
-        Preprocess the data: handle missing values, encode categoricals, scale numericals.
-
-        Args:
-            df: Input dataframe
-            fit: Whether to fit the scaler (True for training, False for inference)
-
-        Returns:
-            Tuple of (features array, target series)
-        """
+        """Handle missing values, encode categoricals, scale numericals."""
         logger.info("Preprocessing data...")
 
         df = df.copy()
@@ -177,12 +158,7 @@ def preprocess(self, df: pd.DataFrame, fit: bool = True) -> Tuple[np.ndarray, pd
         logger.info(f"Preprocessing complete. Feature shape: {features.shape}")
         return features, target
 
-    def split_data(
-        self,
-        features: np.ndarray,
-        target: pd.Series
-    ) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
-        """Split data into train and test sets."""
+    def split_data(self, features: np.ndarray, target: pd.Series) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
         test_size = self.config['data']['test_size']
         random_state = self.config['data']['random_state']
 
@@ -201,8 +177,7 @@ def split_data(
 
 
 def prepare_datasets():
-    """Main function to prepare datasets for training."""
-    logger.info("Starting data preparation pipeline...")
+    logger.info("Starting data preparation...")
 
     # Initialize processor
     processor = FlightDataProcessor()
@@ -224,11 +199,12 @@ def prepare_datasets():
     output_dir = Path("data/processed")
     output_dir.mkdir(parents=True, exist_ok=True)
 
-    train_df = pd.DataFrame(X_train)
+    # Use feature names from processor to preserve column names
+    train_df = pd.DataFrame(X_train, columns=processor.feature_names)
     train_df['price'] = y_train.values
     train_df.to_parquet(output_dir / "train.parquet", index=False)
 
-    test_df = pd.DataFrame(X_test)
+    test_df = pd.DataFrame(X_test, columns=processor.feature_names)
     test_df['price'] = y_test.values
     test_df.to_parquet(output_dir / "test.parquet", index=False)