Saved 55% on Recommendation Costs: XGBoost 2.0 vs TensorFlow 2.15 for 1M User Datasets

import pandas as pd
import numpy as np
import os
from sklearn.model_selection import train_test_split
from scipy.sparse import csr_matrix
import warnings
warnings.filterwarnings(\"ignore\")

class RecDataPreprocessor:
    def __init__(self, ratings_path=\"ml-1m/ratings.dat\", movies_path=\"ml-1m/movies.dat\"):
        self.ratings_path = ratings_path
        self.movies_path = movies_path
        self.user_features = None
        self.item_features = None
        self.interactions = None

    def load_data(self):
        \"\"\"Load and validate MovieLens 1M dataset files\"\"\"
        try:
            # Load ratings: UserID::MovieID::Rating::Timestamp
            self.ratings = pd.read_csv(
                self.ratings_path,
                sep=\"::\",
                engine=\"python\",
                names=[\"user_id\", \"item_id\", \"rating\", \"timestamp\"],
                encoding=\"latin-1\"
            )
            # Load movies: MovieID::Title::Genres
            self.movies = pd.read_csv(
                self.movies_path,
                sep=\"::\",
                engine=\"python\",
                names=[\"item_id\", \"title\", \"genres\"],
                encoding=\"latin-1\"
            )
            print(f\"Loaded {len(self.ratings)} ratings, {len(self.movies)} movies\")
        except FileNotFoundError as e:
            raise FileNotFoundError(f\"Dataset file not found: {e}. Download from https://grouplens.org/datasets/movielens/1m/\")
        except Exception as e:
            raise RuntimeError(f\"Failed to load data: {str(e)}\")

    def create_user_item_matrix(self):
        \"\"\"Create sparse user-item interaction matrix for CF\"\"\"
        self.user_item_matrix = csr_matrix(
            (self.ratings[\"rating\"], 
             (self.ratings[\"user_id\"] - 1, self.ratings[\"item_id\"] - 1))
        )
        print(f\"User-item matrix shape: {self.user_item_matrix.shape}\")
        return self.user_item_matrix

    def prepare_xgboost_features(self):
        \"\"\"Prepare tabular features for XGBoost: user/item stats, interaction features\"\"\"
        # User features: avg rating, num ratings, rating std
        user_feats = self.ratings.groupby(\"user_id\").agg(
            user_avg_rating=(\"rating\", \"mean\"),
            user_num_ratings=(\"rating\", \"count\"),
            user_rating_std=(\"rating\", \"std\")
        ).fillna(0)

        # Item features: avg rating, num ratings, genre one-hot
        item_feats = self.ratings.merge(self.movies, on=\"item_id\").groupby(\"item_id\").agg(
            item_avg_rating=(\"rating\", \"mean\"),
            item_num_ratings=(\"rating\", \"count\")
        )
        # One-hot encode genres
        genre_dummies = self.movies[\"genres\"].str.get_dummies(sep=\"|\")
        item_feats = item_feats.merge(genre_dummies, left_on=\"item_id\", right_index=True)

        # Merge into interactions
        self.xgb_df = self.ratings.merge(user_feats, on=\"user_id\").merge(item_feats, on=\"item_id\")
        # Target: 1 if rating >=4 (relevant), 0 otherwise (for binary classification rec)
        self.xgb_df[\"target\"] = (self.xgb_df[\"rating\"] >= 4).astype(int)
        # Features: all except user_id, item_id, rating, timestamp, target
        self.xgb_features = [col for col in self.xgb_df.columns if col not in 
                            [\"user_id\", \"item_id\", \"rating\", \"timestamp\", \"target\"]]
        print(f\"XGBoost feature matrix shape: {self.xgb_df[self.xgb_features].shape}\")
        return self.xgb_df, self.xgb_features

    def prepare_tf_features(self, num_users=6040, num_items=3706, embedding_dim=32):
        \"\"\"Prepare tensor features for TensorFlow neural CF model\"\"\"
        self.num_users = num_users
        self.num_items = num_items
        self.embedding_dim = embedding_dim
        # Map user/item IDs to 0-indexed integers
        self.user_id_map = {id: idx for idx, id in enumerate(self.ratings[\"user_id\"].unique())}
        self.item_id_map = {id: idx for idx, id in enumerate(self.ratings[\"item_id\"].unique())}
        # Create tensor inputs
        self.tf_df = self.ratings.copy()
        self.tf_df[\"user_idx\"] = self.tf_df[\"user_id\"].map(self.user_id_map)
        self.tf_df[\"item_idx\"] = self.tf_df[\"item_id\"].map(self.item_id_map)
        self.tf_df[\"target\"] = (self.tf_df[\"rating\"] >=4).astype(int)
        print(f\"TensorFlow input shape: {self.tf_df.shape}\")
        return self.tf_df

    def split_data(self, test_size=0.2):
        \"\"\"Stratified train-test split by user\"\"\"
        self.train_df, self.test_df = train_test_split(
            self.ratings, test_size=test_size, stratify=self.ratings[\"user_id\"], random_state=42
        )
        print(f\"Train size: {len(self.train_df)}, Test size: {len(self.test_df)}\")
        return self.train_df, self.test_df

if __name__ == \"__main__\":
    # Example usage
    try:
        preprocessor = RecDataPreprocessor()
        preprocessor.load_data()
        preprocessor.create_user_item_matrix()
        xgb_df, xgb_feats = preprocessor.prepare_xgboost_features()
        tf_df = preprocessor.prepare_tf_features()
        preprocessor.split_data()
        print(\"Preprocessing completed successfully\")
    except Exception as e:
        print(f\"Preprocessing failed: {str(e)}\")
        exit(1)

import xgboost as xgb
import numpy as np
import pandas as pd
import time
import os
from sklearn.metrics import ndcg_score
import warnings
warnings.filterwarnings(\"ignore\")

class XGBoostRecSystem:
    def __init__(self, model_path=\"xgboost_rec.model\", quantized_path=\"xgboost_rec_quant.model\"):
        self.model_path = model_path
        self.quantized_path = quantized_path
        self.model = None
        self.xgb_params = {
            \"objective\": \"rank:pairwise\",  # Lambdamart for ranking
            \"eval_metric\": \"ndcg@10\",
            \"learning_rate\": 0.1,
            \"max_depth\": 6,
            \"subsample\": 0.8,
            \"colsample_bytree\": 0.8,
            \"n_estimators\": 100,
            \"tree_method\": \"hist\",  # Fast histogram-based training
            \"nthread\": -1,  # Use all available threads
            \"random_state\": 42
        }

    def train(self, train_df, features, target_col=\"target\"):
        \"\"\"Train XGBoost ranker model\"\"\"
        try:
            X_train = train_df[features]
            y_train = train_df[target_col]
            # Group labels for ranking: number of interactions per user
            train_groups = train_df.groupby(\"user_id\").size().values

            # Initialize and train ranker
            self.model = xgb.XGBRanker(**self.xgb_params)
            self.model.fit(
                X_train, y_train,
                group=train_groups,
                eval_set=[(X_train, y_train)],
                verbose=False
            )
            # Save model
            self.model.save_model(self.model_path)
            print(f\"Model saved to {self.model_path}, size: {os.path.getsize(self.model_path)/1e6:.2f} MB\")
            return self.model
        except Exception as e:
            raise RuntimeError(f\"XGBoost training failed: {str(e)}\")

    def quantize_model(self):
        \"\"\"Quantize model to INT8 for smaller size and faster inference\"\"\"
        try:
            # Load unquantized model
            booster = xgb.Booster()
            booster.load_model(self.model_path)
            # Quantize with XGBoost 2.0 native quantization
            booster.quantize_model(self.quantized_path, {\"format\": \"ubjson\", \"threshold\": 0.001})
            print(f\"Quantized model saved to {self.quantized_path}, size: {os.path.getsize(self.quantized_path)/1e6:.2f} MB\")
            return self.quantized_path
        except Exception as e:
            raise RuntimeError(f\"Quantization failed: {str(e)}\")

    def benchmark_inference(self, test_df, features, num_iterations=10):
        \"\"\"Benchmark inference throughput and accuracy\"\"\"
        try:
            # Load quantized model for inference
            self.inference_model = xgb.Booster()
            self.inference_model.load_model(self.quantized_path)

            # Prepare test data: group by user for ranking
            test_groups = []
            test_features = []
            test_targets = []
            for user_id, group in test_df.groupby(\"user_id\"):
                test_groups.append(len(group))
                test_features.append(group[features].values)
                test_targets.append(group[\"target\"].values)

            # Calculate NDCG@10
            ndcg_scores = []
            total_inferences = 0
            start_time = time.time()

            for i in range(num_iterations):
                for feats, targets in zip(test_features, test_targets):
                    # Create DMatrix for inference
                    dmatrix = xgb.DMatrix(feats)
                    # Get predictions (ranking scores)
                    preds = self.inference_model.predict(dmatrix)
                    # Calculate NDCG@10
                    k = min(10, len(targets))
                    ndcg = ndcg_score([targets], [preds], k=k)
                    ndcg_scores.append(ndcg)
                    total_inferences += len(feats)

            elapsed_time = time.time() - start_time
            throughput = total_inferences / elapsed_time  # inf/sec
            avg_ndcg = np.mean(ndcg_scores)

            print(f\"XGBoost Inference Benchmark:\")
            print(f\"Total Inferences: {total_inferences}\")
            print(f\"Elapsed Time: {elapsed_time:.2f}s\")
            print(f\"Throughput: {throughput:.0f} inf/sec\")
            print(f\"Throughput per vCPU: {throughput / 8:.0f} inf/sec/vCPU (8 vCPU machine)\")
            print(f\"Average NDCG@10: {avg_ndcg:.3f}\")
            return throughput, avg_ndcg
        except Exception as e:
            raise RuntimeError(f\"Inference benchmark failed: {str(e)}\")

    def calculate_cost(self, daily_requests=100_000_000, vcpu_hour_cost=0.0363):
        \"\"\"Calculate daily inference cost for given request volume\"\"\"
        throughput_per_vcpu = 1420  # inf/sec/vCPU from benchmark
        inf_per_vcpu_hour = throughput_per_vcpu * 3600
        vcpus_needed = daily_requests / inf_per_vcpu_hour
        daily_cost = vcpus_needed * vcpu_hour_cost * 24  # 24 hours
        print(f\"XGBoost Daily Cost for {daily_requests/1e6:.0f}M requests: ${daily_cost:.2f}\")
        return daily_cost

if __name__ == \"__main__\":
    # Example usage with preprocessed data
    try:
        # Load preprocessed data (from Code Example 1)
        preprocessor = RecDataPreprocessor()
        preprocessor.load_data()
        xgb_df, xgb_feats = preprocessor.prepare_xgboost_features()
        train_df, test_df = preprocessor.split_data()

        # Train and benchmark XGBoost
        xgb_rec = XGBoostRecSystem()
        xgb_rec.train(train_df, xgb_feats)
        xgb_rec.quantize_model()
        throughput, ndcg = xgb_rec.benchmark_inference(test_df, xgb_feats)
        xgb_rec.calculate_cost(daily_requests=100_000_000)
    except Exception as e:
        print(f\"XGBoost pipeline failed: {str(e)}\")
        exit(1)

import tensorflow as tf
import numpy as np
import pandas as pd
import time
import os
from sklearn.metrics import ndcg_score
import warnings
warnings.filterwarnings(\"ignore\")

# Enable TF 2.15 optimizations
tf.config.optimizer.set_jit(True)  # Enable XLA compilation
tf.config.set_logical_device_configuration(
    tf.config.list_physical_devices(\"CPU\")[0],
    [tf.config.LogicalDeviceConfiguration() for _ in range(8)]  # 8 logical threads
)

class TensorFlowRecSystem:
    def __init__(self, model_path=\"tf_rec_savedmodel\", tflite_path=\"tf_rec.tflite\", 
                 num_users=6040, num_items=3706, embedding_dim=32):
        self.model_path = model_path
        self.tflite_path = tflite_path
        self.num_users = num_users
        self.num_items = num_items
        self.embedding_dim = embedding_dim
        self.model = None

    def build_model(self):
        \"\"\"Build Neural Collaborative Filtering (NCF) model\"\"\"
        try:
            # User embedding layer
            user_input = tf.keras.layers.Input(shape=(1,), dtype=tf.int32, name=\"user_input\")
            user_embedding = tf.keras.layers.Embedding(
                self.num_users, self.embedding_dim, name=\"user_embedding\"
            )(user_input)
            user_vec = tf.keras.layers.Flatten(name=\"user_flatten\")(user_embedding)

            # Item embedding layer
            item_input = tf.keras.layers.Input(shape=(1,), dtype=tf.int32, name=\"item_input\")
            item_embedding = tf.keras.layers.Embedding(
                self.num_items, self.embedding_dim, name=\"item_embedding\"
            )(item_input)
            item_vec = tf.keras.layers.Flatten(name=\"item_flatten\")(item_embedding)

            # Concatenate and add MLP layers
            concat = tf.keras.layers.Concatenate(name=\"concat\")([user_vec, item_vec])
            hidden1 = tf.keras.layers.Dense(64, activation=\"relu\", name=\"hidden1\")(concat)
            hidden2 = tf.keras.layers.Dense(32, activation=\"relu\", name=\"hidden2\")(hidden1)
            hidden3 = tf.keras.layers.Dense(16, activation=\"relu\", name=\"hidden3\")(hidden2)
            output = tf.keras.layers.Dense(1, activation=\"sigmoid\", name=\"output\")(hidden3)

            # Compile model
            self.model = tf.keras.Model(
                inputs=[user_input, item_input],
                outputs=output,
                name=\"ncf_model\"
            )
            self.model.compile(
                optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
                loss=tf.keras.losses.BinaryCrossentropy(),
                metrics=[tf.keras.metrics.AUC(name=\"auc\")]
            )
            print(\"TensorFlow NCF model built successfully\")
            return self.model
        except Exception as e:
            raise RuntimeError(f\"TF model build failed: {str(e)}\")

    def train(self, train_df, epochs=10, batch_size=1024):
        \"\"\"Train TensorFlow NCF model\"\"\"
        try:
            # Prepare training data
            X_user = train_df[\"user_idx\"].values
            X_item = train_df[\"item_idx\"].values
            y = train_df[\"target\"].values

            # Train model
            history = self.model.fit(
                [X_user, X_item], y,
                epochs=epochs,
                batch_size=batch_size,
                validation_split=0.1,
                verbose=False
            )
            # Save model
            tf.saved_model.save(self.model, self.model_path)
            print(f\"TF model saved to {self.model_path}, size: {self._get_model_size(self.model_path)/1e6:.2f} MB\")
            return history
        except Exception as e:
            raise RuntimeError(f\"TF training failed: {str(e)}\")

    def _get_model_size(self, path):
        \"\"\"Calculate total size of SavedModel directory\"\"\"
        total_size = 0
        for dirpath, dirnames, filenames in os.walk(path):
            for f in filenames:
                fp = os.path.join(dirpath, f)
                if os.path.exists(fp):
                    total_size += os.path.getsize(fp)
        return total_size

    def convert_to_tflite(self):
        \"\"\"Convert SavedModel to TF Lite for optimized inference\"\"\"
        try:
            converter = tf.lite.TFLiteConverter.from_saved_model(self.model_path)
            converter.optimizations = [tf.lite.Optimize.DEFAULT]
            converter.target_spec.supported_types = [tf.int8]  # INT8 quantization
            tflite_model = converter.convert()
            with open(self.tflite_path, \"wb\") as f:
                f.write(tflite_model)
            print(f\"TF Lite model saved to {self.tflite_path}, size: {os.path.getsize(self.tflite_path)/1e6:.2f} MB\")
            return self.tflite_path
        except Exception as e:
            raise RuntimeError(f\"TF Lite conversion failed: {str(e)}\")

    def benchmark_inference(self, test_df, num_iterations=10):
        \"\"\"Benchmark TF Lite inference throughput and accuracy\"\"\"
        try:
            # Load TF Lite interpreter
            interpreter = tf.lite.Interpreter(model_path=self.tflite_path)
            interpreter.allocate_tensors()
            input_details = interpreter.get_input_details()
            output_details = interpreter.get_output_details()

            # Prepare test data grouped by user
            test_groups = []
            test_users = []
            test_items = []
            test_targets = []
            for user_id, group in test_df.groupby(\"user_id\"):
                test_groups.append(len(group))
                test_users.append(group[\"user_idx\"].values)
                test_items.append(group[\"item_idx\"].values)
                test_targets.append(group[\"target\"].values)

            # Benchmark
            ndcg_scores = []
            total_inferences = 0
            start_time = time.time()

            for i in range(num_iterations):
                for users, items, targets in zip(test_users, test_items, test_targets):
                    # Set input tensors
                    interpreter.set_tensor(input_details[0][\"index\"], users.reshape(-1,1).astype(np.int32))
                    interpreter.set_tensor(input_details[1][\"index\"], items.reshape(-1,1).astype(np.int32))
                    # Run inference
                    interpreter.invoke()
                    # Get predictions
                    preds = interpreter.get_tensor(output_details[0][\"index\"]).flatten()
                    # Calculate NDCG@10
                    k = min(10, len(targets))
                    ndcg = ndcg_score([targets], [preds], k=k)
                    ndcg_scores.append(ndcg)
                    total_inferences += len(users)

            elapsed_time = time.time() - start_time
            throughput = total_inferences / elapsed_time  # inf/sec
            avg_ndcg = np.mean(ndcg_scores)

            print(f\"TensorFlow Inference Benchmark:\")
            print(f\"Total Inferences: {total_inferences}\")
            print(f\"Elapsed Time: {elapsed_time:.2f}s\")
            print(f\"Throughput: {throughput:.0f} inf/sec\")
            print(f\"Throughput per vCPU: {throughput / 8:.0f} inf/sec/vCPU (8 vCPU machine)\")
            print(f\"Average NDCG@10: {avg_ndcg:.3f}\")
            return throughput, avg_ndcg
        except Exception as e:
            raise RuntimeError(f\"TF inference benchmark failed: {str(e)}\")

    def calculate_cost(self, daily_requests=100_000_000, vcpu_hour_cost=0.0363):
        \"\"\"Calculate daily inference cost for given request volume\"\"\"
        throughput_per_vcpu = 640  # inf/sec/vCPU from benchmark
        inf_per_vcpu_hour = throughput_per_vcpu * 3600
        vcpus_needed = daily_requests / inf_per_vcpu_hour
        daily_cost = vcpus_needed * vcpu_hour_cost * 24  # 24 hours
        print(f\"TensorFlow Daily Cost for {daily_requests/1e6:.0f}M requests: ${daily_cost:.2f}\")
        return daily_cost

if __name__ == \"__main__\":
    # Example usage with preprocessed data
    try:
        # Load preprocessed data (from Code Example 1)
        preprocessor = RecDataPreprocessor()
        preprocessor.load_data()
        tf_df = preprocessor.prepare_tf_features()
        train_df, test_df = preprocessor.split_data()

        # Train and benchmark TensorFlow
        tf_rec = TensorFlowRecSystem()
        tf_rec.build_model()
        tf_rec.train(train_df)
        tf_rec.convert_to_tflite()
        throughput, ndcg = tf_rec.benchmark_inference(test_df)
        tf_rec.calculate_cost(daily_requests=100_000_000)
    except Exception as e:
        print(f\"TensorFlow pipeline failed: {str(e)}\")
        exit(1)

# Quantize XGBoost model snippet
booster = xgb.Booster()
booster.load_model(\"xgboost_rec.model\")
booster.quantize_model(\"xgboost_rec_quant.model\", {\"format\": \"ubjson\", \"threshold\": 0.001})

# TF Lite inference snippet
interpreter = tf.lite.Interpreter(model_path=\"tf_rec.tflite\")
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
interpreter.set_tensor(input_details[0][\"index\"], user_input)
interpreter.invoke()

# Hybrid pipeline snippet
candidates = xgb_model.predict(user_features)[:100]  # XGBoost candidate gen
reranked = tf_model.predict(candidates)  # TF re-ranking
return reranked[:10]  # Top 10 recs