From 4e564a017c18d5d5c5da1c5872ceda253d8d8e8f Mon Sep 17 00:00:00 2001 From: gana36 Date: Sat, 13 Dec 2025 17:46:34 -0500 Subject: [PATCH] Fixed drift report generation -Evidently --- AWS_SETUP_GUIDE.md | 1 - src/app/api.py | 61 ++++----------------------- src/database/models.py | 18 +------- src/ml/data.py | 38 ++++------------- src/monitoring/drift_detection.py | 68 ++++++++++++++----------------- 5 files changed, 46 insertions(+), 140 deletions(-) diff --git a/AWS_SETUP_GUIDE.md b/AWS_SETUP_GUIDE.md index 27abb0a..35fb1c0 100644 --- a/AWS_SETUP_GUIDE.md +++ b/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. diff --git a/src/app/api.py b/src/app/api.py index e4d9369..732430d 100644 --- a/src/app/api.py +++ b/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,7 +200,6 @@ 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") @@ -226,7 +207,6 @@ async def startup_event(): @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'), diff --git a/src/database/models.py b/src/database/models.py index eb7acb6..a969c2e 100644 --- a/src/database/models.py +++ b/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,7 +33,6 @@ 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' @@ -54,14 +40,12 @@ def get_database_url(): 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() diff --git a/src/ml/data.py b/src/ml/data.py index 6d6663c..7fcf0e3 100644 --- a/src/ml/data.py +++ b/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) diff --git a/src/monitoring/drift_detection.py b/src/monitoring/drift_detection.py index c3fe0d4..0c9459e 100644 --- a/src/monitoring/drift_detection.py +++ b/src/monitoring/drift_detection.py @@ -1,7 +1,4 @@ -""" -Data drift detection using Evidently. -Monitors prediction data for distribution shifts and data quality issues. -""" +"""Data drift detection using Evidently.""" import pandas as pd import numpy as np @@ -17,35 +14,39 @@ class DriftDetector: - """Monitor data drift using Evidently.""" - def __init__(self, reference_data_path: str = "data/processed/reference.parquet"): - """ - Initialize drift detector. - - Args: - reference_data_path: Path to reference dataset - """ self.reference_data = pd.read_parquet(reference_data_path) logger.info(f"Loaded reference data: {len(self.reference_data)} records") - def detect_drift( - self, - current_data: pd.DataFrame, - report_path: str = None - ) -> Dict[str, Any]: - """ - Detect drift between reference and current data. - - Args: - current_data: Current production data - report_path: Path to save HTML report - - Returns: - Drift detection results - """ + def detect_drift(self, current_data: pd.DataFrame, report_path: str = None) -> Dict[str, Any]: + """Compare current data against reference and generate drift report.""" logger.info(f"Detecting drift on {len(current_data)} current records...") + # Find common columns between reference and current data + reference_cols = set(self.reference_data.columns) + current_cols = set(current_data.columns) + common_cols = list(reference_cols.intersection(current_cols)) + + if not common_cols: + raise ValueError("No common columns between reference and current data") + + # Filter to common columns only + reference_aligned = self.reference_data[common_cols].copy() + current_aligned = current_data[common_cols].copy() + + logger.info(f"Comparing {len(common_cols)} common columns: {common_cols}") + + # Ensure consistent dtypes + for col in common_cols: + if reference_aligned[col].dtype != current_aligned[col].dtype: + logger.warning(f"Column {col}: converting types to match") + try: + current_aligned[col] = current_aligned[col].astype(reference_aligned[col].dtype) + except (ValueError, TypeError): + # If conversion fails, convert both to string + reference_aligned[col] = reference_aligned[col].astype(str) + current_aligned[col] = current_aligned[col].astype(str) + # Create Evidently report report = Report(metrics=[ DataDriftPreset(), @@ -54,8 +55,8 @@ def detect_drift( # Run report report.run( - reference_data=self.reference_data, - current_data=current_data + reference_data=reference_aligned, + current_data=current_aligned ) # Save HTML report @@ -68,8 +69,6 @@ def detect_drift( report.save_html(str(report_path)) logger.info(f"Drift report saved to {report_path}") - # Extract key metrics - # Note: Evidently API may vary by version results = { "report_path": str(report_path), "timestamp": datetime.now().isoformat(), @@ -81,12 +80,7 @@ def detect_drift( def monitor_production_drift(hours: int = 24): - """ - Monitor drift in production predictions. - - Args: - hours: Number of hours of recent data to analyze - """ + """Check drift on recent predictions from database.""" from src.database.models import get_session, Prediction logger.info(f"Monitoring drift for last {hours} hours...")