File size: 4,341 Bytes
9bc522a |
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 |
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import pandas as pd
import os
from typing import List, Dict, Any
# Response models
class FeatureExample(BaseModel):
category: str
region: str
inventory_level: float
units_ordered: int
demand_forecast: float
class FeaturesResponse(BaseModel):
available_categories: List[str]
available_regions: List[str]
sample_data: List[FeatureExample]
statistics: Dict[str, Any]
class SampleDataResponse(BaseModel):
total_records: int
sample_records: List[Dict[str, Any]]
# Create router
router = APIRouter()
def load_sample_data():
"""Load sample data from CSV file"""
try:
data_path = "data/sample_data.csv"
if os.path.exists(data_path):
df = pd.read_csv(data_path)
return df
else:
# Return mock data if file doesn't exist
return pd.DataFrame({
'Category': ['Electronics', 'Clothing', 'Home & Garden', 'Sports'],
'Region': ['North', 'South', 'East', 'West'],
'Inventory Level': [150.5, 89.2, 205.7, 125.8],
'Units Ordered': [25, 15, 40, 30],
'Demand Forecast': [200.0, 120.5, 300.2, 180.4]
})
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error loading data: {str(e)}")
@router.get("/get-features", response_model=FeaturesResponse)
async def get_features():
"""
Get available feature values and sample data for prediction.
Returns categories, regions, and example data for users.
"""
try:
df = load_sample_data()
# Get unique values
categories = df['Category'].unique().tolist() if 'Category' in df.columns else []
regions = df['Region'].unique().tolist() if 'Region' in df.columns else []
# Create sample examples (first 3 records)
sample_data = []
for _, row in df.head(3).iterrows():
sample_data.append(FeatureExample(
category=row.get('Category', 'Electronics'),
region=row.get('Region', 'North'),
inventory_level=float(row.get('Inventory Level', 100.0)),
units_ordered=int(row.get('Units Ordered', 20)),
demand_forecast=float(row.get('Demand Forecast', 150.0))
))
# Calculate basic statistics
statistics = {}
if 'Inventory Level' in df.columns:
statistics['inventory_level'] = {
'min': float(df['Inventory Level'].min()),
'max': float(df['Inventory Level'].max()),
'mean': float(df['Inventory Level'].mean())
}
if 'Units Ordered' in df.columns:
statistics['units_ordered'] = {
'min': int(df['Units Ordered'].min()),
'max': int(df['Units Ordered'].max()),
'mean': float(df['Units Ordered'].mean())
}
if 'Demand Forecast' in df.columns:
statistics['demand_forecast'] = {
'min': float(df['Demand Forecast'].min()),
'max': float(df['Demand Forecast'].max()),
'mean': float(df['Demand Forecast'].mean())
}
return FeaturesResponse(
available_categories=categories,
available_regions=regions,
sample_data=sample_data,
statistics=statistics
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error getting features: {str(e)}")
@router.get("/sample-data", response_model=SampleDataResponse)
async def get_sample_data(limit: int = 5):
"""
Get raw sample data from the dataset.
Args:
limit: Number of records to return (default: 5, max: 20)
"""
try:
if limit > 20:
limit = 20
df = load_sample_data()
# Convert to dict records
sample_records = df.head(limit).to_dict('records')
return SampleDataResponse(
total_records=len(df),
sample_records=sample_records
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error getting sample data: {str(e)}")
|