EASEを使ってみる - Re:ゼロから始めるML生活

この記事は（1人で）基礎から学ぶ推薦システム Advent Calendar 2022の13日目の記事です。

今回は、EASEというアルゴリズムを試してみようと思います。

※見様見真似で書いてみたものの、スコアが低すぎてなんかおかしいので多分後で実装し直します。

Embarrassingly Shallow Autoencoders for Sparse Data (EASE)
- 超ざっくり説明
実装
結果の比較
サンプルコード
参考文献
感想

Embarrassingly Shallow Autoencoders for Sparse Data (EASE)

もともとの論文はこちらです。

arxiv.org

実装がこちら。

github.com

超ざっくり説明

ざっくりもでるの説明としては

S: score
X: 評価値行列
B: アイテム-アイテム weight matrix

としたとき、

$\displaystyle{ S_{uj} = X_{u, \cdot} \cdot B_{\cdot, j} }$

このようなモデルを考えたとき、Bの行列を学習することで推定を可能にします。

このとき、下記のような式について最小化するように学習していきます。

$\displaystyle{ min_{B} || X - XB ||^{2}_{F} + \lambda \cdot || B ||^{2}_{F} }$

※Bの対角成分はすべて0となっています。

その他、詳しくは論文読んでもらったり、こちらのブログ記事がわかりやすいので、そちらも合わせて参照いただければと思います。

理屈は置いておいて、特徴的なのは実装が非常にシンプル点です。もともとの実装(movielensが想定されてる)ですら、モデルの定義がこれくらいなので、非常にシンプルなことがわかります。

class EASE:
    def __init__(self):
        self.user_enc = LabelEncoder()
        self.item_enc = LabelEncoder()

    def _get_users_and_items(self, df):
        users = self.user_enc.fit_transform(df.loc[:, 'user_id'])
        items = self.item_enc.fit_transform(df.loc[:, 'item_id'])
        return users, items

    def fit(self, df, lambda_: float = 0.5, implicit=True):
        """
        df: pandas.DataFrame with columns user_id, item_id and (rating)
        lambda_: l2-regularization term
        implicit: if True, ratings are ignored and taken as 1, else normalized ratings are used
        """
        users, items = self._get_users_and_items(df)
        values = (
            np.ones(df.shape[0])
            if implicit
            else df['rating'].to_numpy() / df['rating'].max()
        )

        X = csr_matrix((values, (users, items)))
        self.X = X

        G = X.T.dot(X).toarray()
        diagIndices = np.diag_indices(G.shape[0])
        G[diagIndices] += lambda_
        P = np.linalg.inv(G)
        B = P / (-np.diag(P))
        B[diagIndices] = 0

        self.B = B
        self.pred = X.dot(B)

    def predict(self, train, users, items, k):
        items = self.item_enc.transform(items)
        dd = train.loc[train.user_id.isin(users)]
        dd['ci'] = self.item_enc.transform(dd.item_id)
        dd['cu'] = self.user_enc.transform(dd.user_id)
        g = dd.groupby('cu')
        with Pool(cpu_count()) as p:
            user_preds = p.starmap(
                self.predict_for_user,
                [(user, group, self.pred[user, :], items, k) for user, group in g],
            )
        df = pd.concat(user_preds)
        df['item_id'] = self.item_enc.inverse_transform(df['item_id'])
        df['user_id'] = self.user_enc.inverse_transform(df['user_id'])
        return df

    @staticmethod
    def predict_for_user(user, group, pred, items, k):
        watched = set(group['ci'])
        candidates = [item for item in items if item not in watched]
        pred = np.take(pred, candidates)
        res = np.argpartition(pred, -k)[-k:]
        r = pd.DataFrame(
            {
                "user_id": [user] * len(res),
                "item_id": np.take(candidates, res),
                "score": np.take(pred, res),
            }
        ).sort_values('score', ascending=False)
        return r

実装

念の為自分でも実装して実行してみます。

class EASE:
    def __init__(self, user_df, item_df):
        self.user_enc = LabelEncoder()
        self.item_enc = LabelEncoder()

        self.users = self.user_enc.fit_transform(user_df.loc[:, "user_id"])
        self.items = self.item_enc.fit_transform(item_df.loc[:, "item_id"])

    def _get_users_and_items(self, df):
        users = self.user_enc.transform(df.loc[:, "user_id"])
        items = self.item_enc.transform(df.loc[:, "item_id"])

        return users, items

    def train(self, df, lambda_: float = 0.5, implicit=True):
        """
        df: pandas.DataFrame with columns user_id, item_id and (rating)
        lambda_: l2-regularization term
        implicit: if True, ratings are ignored and taken as 1, else normalized ratings are used
        """
        self.train_df = df
        users, items = self._get_users_and_items(df)
        values = (
            np.ones(df.shape[0])
            if implicit
            else df["rating"].to_numpy() / df["rating"].max()
        )

        X = csr_matrix((values, (users, items)))
        self.X = X

        G = X.T.dot(X).toarray()
        diagIndices = np.diag_indices(G.shape[0])
        G[diagIndices] += lambda_
        P = np.linalg.inv(G)
        B = P / (-np.diag(P))
        B[diagIndices] = 0

        self.B = B
        self.pred = X.dot(B)

    def predict(self, test_df:pd.DataFrame):
        train = self.train_df
        users = test_df["user_id"]
        items = test_df["item_id"]
        k = np.unique(self.items).size

        users = self.user_enc.transform(users)
        items = self.item_enc.transform(items)
        dd = train.loc[train.user_id.isin(users)]
        dd['ci'] = self.item_enc.transform(dd.item_id)
        dd['cu'] = self.user_enc.transform(dd.user_id)
        g = dd.groupby('cu')
        with Pool(cpu_count()) as p:
            user_preds = p.starmap(
                self.predict_for_user,
                [(user, group, self.pred[user, :], items, k) for user, group in g],
            )
        df = pd.concat(user_preds)
        df['item_id'] = self.item_enc.inverse_transform(df['item_id'])
        df['user_id'] = self.user_enc.inverse_transform(df['user_id'])
        result = test_df.merge(df, on=["user_id", "item_id"], how="left")
        return result["score"]

    @staticmethod
    def predict_for_user(user, group, pred, items, k):
        watched = set(group['ci'])
        candidates = [item for item in items if item not in watched]
        pred = np.take(pred, candidates)
        res = np.argpartition(pred, -k)[-k:]
        r = pd.DataFrame(
            {
                "user_id": [user] * len(res),
                "item_id": np.take(candidates, res),
                "score": np.take(pred, res),
            }
        ).sort_values('score', ascending=False)
        return r