""" Product Recommendation Layer - ML-Based Ranking System Enhances grocery selection with ML probability scoring + price normalization This module adds intelligent ranking on top of ML inference: - Extracts ML probability scores for diet suitability - Computes normalized price scores (lower price = higher score) - Combines both using weighted aggregation for final recommendation strength """ import numpy as np import pandas as pd from pathlib import Path def compute_recommendation_scores(products_df: pd.DataFrame, ml_probabilities: np.ndarray, target_diet_idx: int, diet_weight: float = 0.6, price_weight: float = 0.4) -> pd.DataFrame: """ Compute recommendation scores combining ML suitability and price affordability. Score Formula: final_score = (diet_weight × diet_proba) + (price_weight × price_score) Args: products_df: DataFrame with product information ml_probabilities: Shape (n_products, 5) - probability matrix from model.predict_proba() target_diet_idx: Index of target diet in probability matrix diet_weight: Weight for ML diet suitability (default 0.6 = 60%) price_weight: Weight for price affordability (default 0.4 = 40%) Returns: DataFrame with added 'diet_proba', 'price_score', 'recommendation_score' columns Rationale: - diet_proba (60%): ML model's confidence in diet suitability - price_score (40%): Affordability factor (inverse normalized price) - Weighted combination ensures both factors influence ranking """ # Step 1: Extract diet probability for target diet type # This represents ML model's confidence that product suits the diet diet_proba = ml_probabilities[:, target_diet_idx] # Step 2: Compute normalized price score # Lower price → higher score (inverted and normalized to [0, 1]) price_min = products_df['price'].min() price_max = products_df['price'].max() price_range = price_max - price_min if price_range == 0: price_normalized = np.zeros(len(products_df), dtype=float) else: price_normalized = (products_df['price'] - price_min) / price_range price_score = 1 - price_normalized # Invert: lower price gets higher score # Step 3: Compute weighted final recommendation score # Combines ML suitability with affordability recommendation_score = (diet_weight * diet_proba) + (price_weight * price_score) # Add scores to dataframe result_df = products_df.copy() result_df['diet_proba'] = diet_proba result_df['price_score'] = price_score result_df['recommendation_score'] = recommendation_score return result_df def recommend_products(products_df: pd.DataFrame, budget: float, family_size: int, max_items: int = None) -> tuple: """ Select and rank products based on recommendation scores within budget. Strategy: 1. Sort by recommendation_score (descending) - best recommendations first 2. Iterate through ranked products 3. Select items until budget is exhausted 4. Adjust selection for family size Args: products_df: DataFrame with recommendation_score column budget: Budget constraint (PKR) family_size: Number of family members max_items: Maximum items to select (auto-calculated if None) Returns: (selected_items, total_cost) - tuple with selected products and total spent Why this ranking approach: - ML probability ensures diet compatibility - Price score ensures affordability - Combined score balances both factors - Top-ranked products offer best value for the diet type """ # Calculate target items based on family size if max_items is None: max_items = max(10, min(32, 8 + family_size * 2)) # Higher family size should consume more budget by increasing per-item quantity. quantity_multiplier = max(1, int(round(family_size / 3))) # Larger households should aim for higher budget utilization. target_utilization = min(0.65 + (family_size * 0.03), 0.95) # Sort by recommendation score (descending) - highest recommendations first # This ensures most suitable AND affordable items are selected first ranked_products = products_df.sort_values( by='recommendation_score', ascending=False ) selected_items = [] total_cost = 0.0 # Iterate through ranked products and select within budget for idx, (_, product) in enumerate(ranked_products.iterrows()): price = float(product['price']) quantity = quantity_multiplier item_cost = price * quantity # Check if adding this item stays within budget if total_cost + item_cost <= budget: selected_items.append({ 'rank': idx + 1, # Ranking position 'product': product['product'], 'category': product['category'], 'quantity': quantity, 'price_per_unit': price, 'total_cost': float(item_cost), 'diet_proba': float(product['diet_proba']), # ML confidence 'price_score': float(product['price_score']), # Affordability 'recommendation_score': float(product['recommendation_score']) # Final score }) total_cost += item_cost # Stop if reached target items or budget threshold if len(selected_items) >= max_items or total_cost >= budget * target_utilization: break return selected_items, total_cost def get_recommendation_summary(selected_items: list, budget: float) -> dict: """ Generate summary statistics for recommendation results. Returns: Dictionary with ranking and score statistics """ if not selected_items: return {} recommendation_scores = [item['recommendation_score'] for item in selected_items] diet_probas = [item['diet_proba'] for item in selected_items] price_scores = [item['price_score'] for item in selected_items] return { 'avg_recommendation_score': float(np.mean(recommendation_scores)), 'avg_diet_proba': float(np.mean(diet_probas)), 'avg_price_score': float(np.mean(price_scores)), 'best_score': float(max(recommendation_scores)), 'worst_score': float(min(recommendation_scores)), 'score_std_dev': float(np.std(recommendation_scores)) } # Example usage (for testing): if __name__ == "__main__": import joblib from data_adapter import build_app_dataset_from_pakistan_csv # Load model and data MODEL = joblib.load("final_model_1.pkl") ENCODERS = joblib.load("final_model_1_encoders.pkl") DATASET = build_app_dataset_from_pakistan_csv(Path("Pakistan_Food_Prices_2025.csv")) # Encode features category_encoded = ENCODERS['category'].transform(DATASET['category']) price_min = ENCODERS['price_min'] price_max = ENCODERS['price_max'] price_normalized = (DATASET['price'] - price_min) / (price_max - price_min) # Get ML predictions X = np.column_stack([category_encoded, price_normalized]) probabilities = MODEL.predict_proba(X) # Get target diet index target_diet = "Keto" diet_idx = list(ENCODERS['diet'].classes_).index(target_diet) # Compute recommendation scores scored_products = compute_recommendation_scores(DATASET, probabilities, diet_idx) # Recommend products selected, total_cost = recommend_products(scored_products, budget=5000, family_size=3) print(f"\nRecommended {len(selected)} items for {target_diet} diet:") print(f"Total Cost: {total_cost} PKR\n") for item in selected[:5]: print(f" {item['product']}") print(f" Score: {item['recommendation_score']:.4f} " + f"(Diet: {item['diet_proba']:.2%}, Price: {item['price_score']:.2%})")