Factorization Machines & DeepFM

import numpy as np

class MatrixFactorization:
    """Basic MF with SGD training."""

    def __init__(self, n_users, n_items, k=10, learning_rate=0.01, reg=0.01):
        self.n_users = n_users
        self.n_items = n_items
        self.k = k
        self.lr = learning_rate
        self.reg = reg

        # Initialize factors
        self.P = np.random.randn(n_users, k) * 0.01  # User factors
        self.Q = np.random.randn(n_items, k) * 0.01  # Item factors
        self.b_u = np.zeros(n_users)  # User biases
        self.b_i = np.zeros(n_items)  # Item biases
        self.mu = 0.0  # Global mean

    def predict(self, user_id, item_id):
        """Predict rating for user-item pair."""
        return self.mu + self.b_u[user_id] + self.b_i[item_id] + np.dot(self.P[user_id], self.Q[item_id])

    def train(self, user_ids, item_ids, ratings, epochs=10):
        """SGD training."""
        self.mu = np.mean(ratings)

        for epoch in range(epochs):
            total_loss = 0
            for u, i, r in zip(user_ids, item_ids, ratings):
                pred = self.predict(u, i)
                error = r - pred

                # Update biases
                self.b_u[u] += self.lr * (error - self.reg * self.b_u[u])
                self.b_i[i] += self.lr * (error - self.reg * self.b_i[i])

                # Update factors
                q_copy = self.Q[i].copy()
                self.P[u] += self.lr * (error * self.Q[i] - self.reg * self.P[u])
                self.Q[i] += self.lr * (error * self.P[u] - self.reg * self.Q[i])

                total_loss += error ** 2

            print(f"Epoch {epoch+1}: MSE = {total_loss / len(user_ids):.4f}")


# Example usage
np.random.seed(42)
mf = MatrixFactorization(n_users=5, n_items=4, k=2)

# Synthetic training data
user_ids = np.array([0, 0, 1, 1, 2, 2, 3, 3, 4, 4])
item_ids = np.array([0, 2, 1, 3, 0, 1, 0, 2, 1, 2])
ratings = np.array([5, 4, 3, 2, 4, 5, 3, 2, 4, 5])

mf.train(user_ids, item_ids, ratings, epochs=50)

class FactorizationMachine:
    """FM for binary classification (CTR prediction)."""

    def __init__(self, n_features, k=8, learning_rate=0.01, reg_w=0.0001, reg_v=0.001):
        self.n_features = n_features
        self.k = k
        self.lr = learning_rate
        self.reg_w = reg_w
        self.reg_v = reg_v

        self.w0 = 0.0  # Global bias
        self.w = np.zeros(n_features)  # Linear weights
        self.V = np.random.randn(n_features, k) * 0.01  # Latent factors

    def sigmoid(self, x):
        return 1.0 / (1.0 + np.exp(-np.clip(x, -500, 500)))

    def predict(self, x):
        """Efficient O(kn) prediction with computational trick."""
        # Linear term
        linear = self.w0 + np.dot(self.w, x)

        # Quadratic term (O(kn) via the trick)
        interaction = 0.0
        for f in range(self.k):
            sum_vx = np.dot(self.V[:, f], x)
            sum_vv_xx = np.dot(self.V[:, f] ** 2, x ** 2)
            interaction += sum_vx ** 2 - sum_vv_xx
        interaction *= 0.5

        logit = linear + interaction
        return self.sigmoid(logit)

    def train(self, X, y, epochs=20):
        """SGD training for binary classification."""
        n_samples = X.shape[0]

        for epoch in range(epochs):
            total_loss = 0
            for idx in range(n_samples):
                x = X[idx]
                target = y[idx]

                pred = self.predict(x)
                error = pred - target

                # Update w0
                self.w0 -= self.lr * error

                # Update w (linear weights)
                self.w -= self.lr * (error * x + 2 * self.reg_w * self.w)

                # Update V (latent factors) - O(kn)
                for f in range(self.k):
                    sum_vx = np.dot(self.V[:, f], x)
                    for i in range(self.n_features):
                        if x[i] != 0:
                            grad = error * x[i] * (sum_vx - self.V[i, f] * x[i])
                            self.V[i, f] -= self.lr * (grad + 2 * self.reg_v * self.V[i, f])

                total_loss += -(target * np.log(pred + 1e-10) + (1 - target) * np.log(1 - pred + 1e-10))

            auc = self.evaluate(X, y)
            print(f"Epoch {epoch+1}: Loss = {total_loss / n_samples:.4f}, AUC = {auc:.4f}")

    def evaluate(self, X, y):
        """Compute AUC."""
        preds = np.array([self.predict(x) for x in X])
        return np.mean((y == 1) & (preds > 0.5)) / (np.mean(y == 1) + 1e-10)


# Synthetic CTR data
np.random.seed(42)
n_samples = 200
n_features = 20

X = np.random.randint(0, 2, (n_samples, n_features))  # One-hot features
y = (X[:, 0] & X[:, 5] | X[:, 10]).astype(int)  # Synthetic label with interactions
y = np.random.binomial(1, 0.3 + 0.4 * y)  # Add noise

fm = FactorizationMachine(n_features=n_features, k=4)
fm.train(X, y, epochs=30)

class SimpleDeepFM:
    """DeepFM with single hidden layer (simplified for demo)."""

    def __init__(self, n_features, k=8, hidden_units=64, learning_rate=0.01):
        self.n_features = n_features
        self.k = k
        self.lr = learning_rate

        # FM component
        self.w0 = 0.0
        self.w = np.zeros(n_features)
        self.V = np.random.randn(n_features, k) * 0.01

        # Deep component
        self.embedding_dim = k  # Use FM embeddings
        embedding_size = n_features * k
        self.W1 = np.random.randn(embedding_size, hidden_units) * 0.01
        self.b1 = np.zeros(hidden_units)
        self.W2 = np.random.randn(hidden_units, 1) * 0.01
        self.b2 = np.zeros(1)

    def relu(self, x):
        return np.maximum(0, x)

    def relu_grad(self, x):
        return (x > 0).astype(float)

    def sigmoid(self, x):
        return 1.0 / (1.0 + np.exp(-np.clip(x, -500, 500)))

    def predict(self, x):
        """Forward pass."""
        # FM prediction
        linear = self.w0 + np.dot(self.w, x)
        interaction = 0.0
        for f in range(self.k):
            sum_vx = np.dot(self.V[:, f], x)
            sum_vv_xx = np.dot(self.V[:, f] ** 2, x ** 2)
            interaction += sum_vx ** 2 - sum_vv_xx
        interaction *= 0.5
        y_fm = linear + interaction

        # Deep prediction
        emb = self.V @ x  # Embeddings (k,)
        emb_flat = emb.flatten()  # Flatten if needed
        hidden = self.relu(np.dot(emb_flat, self.W1) + self.b1)
        y_deep = np.dot(hidden, self.W2) + self.b2

        # Combine
        logit = y_fm + y_deep
        return self.sigmoid(logit)

    def train(self, X, y, epochs=20):
        """Simple SGD training (no proper backprop, just for demo)."""
        for epoch in range(epochs):
            total_loss = 0
            for idx in range(len(X)):
                x = X[idx]
                target = y[idx]
                pred = self.predict(x)
                loss = -(target * np.log(pred + 1e-10) + (1 - target) * np.log(1 - pred + 1e-10))
                total_loss += loss

                # Simple gradient update (not full backprop)
                error = pred - target
                self.w0 -= self.lr * error
                self.w -= self.lr * error * x

            print(f"Epoch {epoch+1}: Loss = {total_loss / len(X):.4f}")


# Test on same data
deepfm = SimpleDeepFM(n_features=20, k=4, hidden_units=16)
deepfm.train(X, y, epochs=20)

"""
Complete example: Predicting click probability on ads.
Features: user age (categorical), ad category (categorical), hour of day (categorical)
"""

import pandas as pd
from sklearn.preprocessing import LabelEncoder

# Synthetic CTR dataset
data = {
    'user_age': ['18-25', '26-35', '18-25', '36-50', '26-35'] * 20,
    'ad_category': ['Tech', 'News', 'Tech', 'Finance', 'News'] * 20,
    'hour': [9, 14, 18, 10, 15] * 20,
    'clicked': [1, 0, 1, 0, 1] * 20
}
df = pd.DataFrame(data)

# One-hot encoding
le_age = LabelEncoder()
le_cat = LabelEncoder()
le_hour = LabelEncoder()

df['user_age_enc'] = le_age.fit_transform(df['user_age'])
df['ad_category_enc'] = le_cat.fit_transform(df['ad_category'])
df['hour_enc'] = le_hour.fit_transform(df['hour'])

# Create feature matrix (one-hot)
n_age = len(df['user_age'].unique())
n_cat = len(df['ad_category'].unique())
n_hour = len(df['hour'].unique())
n_features = n_age + n_cat + n_hour

X = np.zeros((len(df), n_features))
for idx, row in df.iterrows():
    X[idx, row['user_age_enc']] = 1
    X[idx, n_age + row['ad_category_enc']] = 1
    X[idx, n_age + n_cat + row['hour_enc']] = 1

y = df['clicked'].values

# Train FM
print("=== Training Factorization Machine ===")
fm = FactorizationMachine(n_features=n_features, k=4)
fm.train(X, y, epochs=20)

# Make predictions
print("\n=== Sample Predictions ===")
for idx in range(min(5, len(X))):
    pred_prob = fm.predict(X[idx])
    print(f"Sample {idx}: Click probability = {pred_prob:.4f}, Actual = {y[idx]}")