Spaces:
Sleeping
Sleeping
File size: 7,671 Bytes
b2c7817 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """
Data processing utilities for AgriPredict Analysis Service
"""
import pandas as pd
import numpy as np
from datetime import datetime
from typing import List, Dict, Any
from utils.logger import setup_logger
from utils.config import settings
logger = setup_logger(__name__)
class DataProcessor:
"""Handles data processing and validation for forecasting"""
def __init__(self):
self.logger = logger
def process_historical_data(self, historical_data: List[Dict[str, Any]]) -> pd.DataFrame:
"""
Process and validate historical demand data
Args:
historical_data: List of demand data points
Returns:
Processed pandas DataFrame
"""
try:
self.logger.info(f"Processing {len(historical_data)} historical data points")
# Handle Pydantic model instances - convert to dict if needed
processed_data = []
for i, item in enumerate(historical_data):
if hasattr(item, 'model_dump'): # Pydantic v2
processed_data.append(item.model_dump())
self.logger.info(f"Item {i}: Converted Pydantic v2 model")
elif hasattr(item, 'dict'): # Pydantic v1
processed_data.append(item.dict())
self.logger.info(f"Item {i}: Converted Pydantic v1 model")
else:
processed_data.append(item)
self.logger.info(f"Item {i}: Already dict - {type(item)}")
self.logger.info(f"Processed data sample: {processed_data[0] if processed_data else 'None'}")
# Convert to DataFrame
df = pd.DataFrame(processed_data)
self.logger.info(f"DataFrame columns: {list(df.columns)}")
self.logger.info(f"DataFrame shape: {df.shape}")
# Validate required columns
required_columns = ['date', 'quantity', 'price']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
self.logger.error(f"Missing columns: {missing_columns}")
raise ValueError(f"Missing required columns: {missing_columns}")
# Convert date column
df['date'] = pd.to_datetime(df['date'])
# Validate data types and ranges
df['quantity'] = pd.to_numeric(df['quantity'], errors='coerce')
df['price'] = pd.to_numeric(df['price'], errors='coerce')
# Remove invalid data
df = df.dropna(subset=['quantity', 'price'])
df = df[df['quantity'] > 0]
df = df[df['price'] > 0]
# Sort by date
df = df.sort_values('date').reset_index(drop=True)
# Remove duplicates based on date
df = df.drop_duplicates(subset=['date'], keep='last')
# Limit data points if too many
if len(df) > settings.MAX_DATA_POINTS:
self.logger.warning(f"Limiting data from {len(df)} to {settings.MAX_DATA_POINTS} points")
df = df.tail(settings.MAX_DATA_POINTS)
self.logger.info(f"Successfully processed {len(df)} data points")
return df
except Exception as e:
self.logger.error(f"Data processing failed: {str(e)}")
raise
def validate_data_quality(self, df: pd.DataFrame) -> Dict[str, Any]:
"""
Validate data quality and return metrics
Args:
df: Processed DataFrame
Returns:
Dictionary with quality metrics
"""
try:
quality_metrics = {
'total_points': len(df),
'date_range': {
'start': df['date'].min().isoformat() if len(df) > 0 else None,
'end': df['date'].max().isoformat() if len(df) > 0 else None
},
'missing_values': {
'quantity': df['quantity'].isnull().sum(),
'price': df['price'].isnull().sum()
},
'outliers': {
'quantity': self._detect_outliers(df['quantity']),
'price': self._detect_outliers(df['price'])
},
'data_completeness': self._calculate_completeness(df)
}
return quality_metrics
except Exception as e:
self.logger.error(f"Quality validation failed: {str(e)}")
return {}
def _detect_outliers(self, series: pd.Series) -> int:
"""Detect outliers using IQR method"""
try:
Q1 = series.quantile(0.25)
Q3 = series.quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers = ((series < lower_bound) | (series > upper_bound)).sum()
return int(outliers)
except:
return 0
def _calculate_completeness(self, df: pd.DataFrame) -> float:
"""Calculate data completeness percentage"""
try:
total_cells = len(df) * 2 # quantity and price columns
missing_cells = df[['quantity', 'price']].isnull().sum().sum()
completeness = ((total_cells - missing_cells) / total_cells) * 100
return round(completeness, 2)
except:
return 0.0
def prepare_features_for_ml(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Prepare features for machine learning models
Args:
df: Processed DataFrame
Returns:
DataFrame with engineered features
"""
try:
# Create feature engineering
feature_df = df.copy()
# Date-based features
feature_df['day_of_week'] = feature_df['date'].dt.dayofweek
feature_df['month'] = feature_df['date'].dt.month
feature_df['day_of_month'] = feature_df['date'].dt.day
feature_df['quarter'] = feature_df['date'].dt.quarter
# Lag features
for lag in [1, 7, 14, 30]:
if len(feature_df) > lag:
feature_df[f'price_lag_{lag}'] = feature_df['price'].shift(lag)
feature_df[f'quantity_lag_{lag}'] = feature_df['quantity'].shift(lag)
# Rolling statistics
for window in [7, 14, 30]:
if len(feature_df) > window:
feature_df[f'price_rolling_mean_{window}'] = feature_df['price'].rolling(window).mean()
feature_df[f'price_rolling_std_{window}'] = feature_df['price'].rolling(window).std()
feature_df[f'quantity_rolling_mean_{window}'] = feature_df['quantity'].rolling(window).mean()
# Price change features
feature_df['price_change'] = feature_df['price'].pct_change()
feature_df['price_change_7d'] = feature_df['price'].pct_change(7)
# Volume-weighted features
feature_df['value'] = feature_df['quantity'] * feature_df['price']
# Drop rows with NaN values created by lag features
feature_df = feature_df.dropna()
self.logger.info(f"Created {len(feature_df.columns) - len(df.columns)} additional features")
return feature_df
except Exception as e:
self.logger.error(f"Feature engineering failed: {str(e)}")
return df
|