Implement fit_partial() for ImplicitALSWrapperModel and LightFMWrapperModel

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - `Debias` mechanism for classification, ranking and auc metrics. New parameter `is_debiased` to `calc_from_confusion_df`, `calc_per_user_from_confusion_df` methods of classification metrics, `calc_from_fitted`, `calc_per_user_from_fitted` methods of auc and rankning (`MAP`) metrics, `calc_from_merged`, `calc_per_user_from_merged` methods of ranking (`NDCG`, `MRR`) metrics. ([#152](https://github.com/MobileTeleSystems/RecTools/pull/152))
 - `nbformat >= 4.2.0` dependency to `[visuals]` extra ([#169](https://github.com/MobileTeleSystems/RecTools/pull/169))
+- Implement `fit_partial()` for `ImplicitALSWrapperModel` and `LightFMWrapperModel` ([#179](https://github.com/MobileTeleSystems/RecTools/pull/179))
 
 ### Fixed
 - `display()` method in `MetricsApp` ([#169](https://github.com/MobileTeleSystems/RecTools/pull/169))

rectools/dataset/dataset.py

@@ -21,6 +21,7 @@
 from scipy import sparse
 
 from rectools import Columns
+from rectools.types import InternalIdsArray
 
 from .features import AbsentIdError, DenseFeatures, Features, SparseFeatures
 from .identifiers import IdMap
@@ -91,6 +92,14 @@ def get_hot_item_features(self) -> tp.Optional[Features]:
             return None
         return self.item_features.take(range(self.n_hot_items))
 
+    def get_hot_users(self) -> InternalIdsArray:
+        """Return internal ids of hot users."""
+        return self.interactions.df[Columns.User].unique()
+
+    def get_hot_items(self) -> InternalIdsArray:
+        """Return internal ids of hot items."""
+        return self.interactions.df[Columns.Item].unique()
+
     @classmethod
     def construct(
         cls,
@@ -138,9 +147,7 @@ def construct(
         Dataset
             Container with all input data, converted to `rectools` structures.
         """
-        for col in (Columns.User, Columns.Item):
-            if col not in interactions_df:
-                raise KeyError(f"Column '{col}' must be present in `interactions_df`")
+        cls._check_columns_present(interactions_df)
         user_id_map = IdMap.from_values(interactions_df[Columns.User].values)
         item_id_map = IdMap.from_values(interactions_df[Columns.Item].values)
         interactions = Interactions.from_raw(interactions_df, user_id_map, item_id_map)
@@ -194,6 +201,12 @@ def _make_features(
         except Exception as e:  # pragma: no cover
             raise RuntimeError(f"An error has occurred while constructing {feature_type} features: {e!r}")
 
+    @staticmethod
+    def _check_columns_present(interactions_df: pd.DataFrame) -> None:
+        for col in (Columns.User, Columns.Item):
+            if col not in interactions_df:
+                raise KeyError(f"Column '{col}' must be present in `interactions_df`")
+
     def get_user_item_matrix(
         self,
         include_weights: bool = True,
@@ -245,3 +258,72 @@ def get_raw_interactions(self, include_weight: bool = True, include_datetime: bo
         pd.DataFrame
         """
         return self.interactions.to_external(self.user_id_map, self.item_id_map, include_weight, include_datetime)
+
+    def construct_new_datasets(
+        self,
+        interactions_df: pd.DataFrame,
+        user_features_df: tp.Optional[pd.DataFrame] = None,
+        cat_user_features: tp.Iterable[str] = (),
+        make_dense_user_features: bool = False,
+        item_features_df: tp.Optional[pd.DataFrame] = None,
+        cat_item_features: tp.Iterable[str] = (),
+        make_dense_item_features: bool = False,
+    ) -> "Dataset":
+        """
+        Create new dataset by merging user_id_map and item_id_map.
+        This function is useful when you want to use fit_partial.
+
+        Parameters
+        ----------
+        interactions_df : pd.DataFrame
+            New interactions table.
+            The same structure as in `construct` method.
+        user_features_df, item_features_df : pd.DataFrame, optional
+            New user (item) explicit features table.
+            The same structure as in `construct` method.
+        cat_user_features, cat_item_features : tp.Iterable[str], default ``()``
+            List of categorical user (item) feature names for
+            `SparseFeatures.from_flatten` method.
+            Used only if `make_dense_user_features` (`make_dense_item_features`)
+            flag is ``False`` and `user_features_df` (`item_features_df`) is not ``None``.
+        make_dense_user_features, make_dense_item_features : bool, default ``False``
+            Create user (item) features as dense or sparse.
+            Used only if `user_features_df` (`item_features_df`) is not ``None``.
+            - if ``False``, `SparseFeatures.from_flatten` method will be used;
+            - if ``True``,  `DenseFeatures.from_dataframe` method will be used.
+
+        Returns
+        -------
+        Dataset
+            New dataset with added data.
+        """
+        self._check_columns_present(interactions_df)
+
+        new_user_id_map = self.user_id_map.add_ids(interactions_df[Columns.User].values, raise_if_already_present=False)
+        new_item_id_map = self.item_id_map.add_ids(interactions_df[Columns.Item].values, raise_if_already_present=False)
+        new_interactions = Interactions.from_raw(interactions_df, new_user_id_map, new_item_id_map)
+
+        new_user_features, new_user_id_map = self._make_features(
+            user_features_df,
+            cat_user_features,
+            make_dense_user_features,
+            new_user_id_map,
+            Columns.User,
+            "user",
+        )
+        new_item_features, new_item_id_map = self._make_features(
+            item_features_df,
+            cat_item_features,
+            make_dense_item_features,
+            new_item_id_map,
+            Columns.Item,
+            "item",
+        )
+
+        return Dataset(
+            new_user_id_map,
+            new_item_id_map,
+            new_interactions,
+            new_user_features,
+            new_item_features,
+        )

rectools/models/base.py

@@ -71,6 +71,26 @@ def fit(self: T, dataset: Dataset, *args: tp.Any, **kwargs: tp.Any) -> T:
     def _fit(self, dataset: Dataset, *args: tp.Any, **kwargs: tp.Any) -> None:
         raise NotImplementedError()
 
+    def fit_partial(self: T, dataset: Dataset, *args: tp.Any, **kwargs: tp.Any) -> T:
+        """
+        Partial fit model.
+
+        Parameters
+        ----------
+        dataset : Dataset
+            Dataset with input data.
+
+        Returns
+        -------
+        self
+        """
+        self._fit_partial(dataset, *args, **kwargs)
+        self.is_fitted = True
+        return self
+
+    def _fit_partial(self, dataset: Dataset, *args: tp.Any, **kwargs: tp.Any) -> None:
+        raise NotImplementedError()
+
     def recommend(
         self,
         users: AnyIds,

rectools/models/implicit_als.py

@@ -90,6 +90,25 @@ def _fit(self, dataset: Dataset) -> None:  # type: ignore
                 self.verbose,
             )
 
+    def _fit_partial(self, dataset: Dataset) -> None:
+        # deepcopy does not copy model.item_factors and model.user_factors.
+        # That causes issues with partial fit.
+        users = dataset.get_hot_users()
+        items = dataset.get_hot_items()
+
+        ui_csr = dataset.get_user_item_matrix(
+            include_weights=True, include_warm_users=True, include_warm_items=True
+        ).astype(np.float32)
+        iu_csr = ui_csr[:, items].T.tocsr(copy=False)
+
+        # TODO: implement partial fit for explicit features
+        if dataset.get_hot_item_features() or dataset.get_hot_user_features():
+            raise NotImplementedError("fit_partial with explicit features is not implemented")
+
+        for _ in range(self.model.iterations):
+            self.model.partial_fit_users(users, ui_csr[users])
+            self.model.partial_fit_items(items, iu_csr)
+
     def _get_users_factors(self, dataset: Dataset) -> Factors:
         return Factors(get_users_vectors(self.model))
 

rectools/models/lightfm.py

@@ -14,6 +14,7 @@
 
 import typing as tp
 from copy import deepcopy
+from typing import Any
 
 import numpy as np
 from lightfm import LightFM
@@ -89,6 +90,23 @@ def _fit(self, dataset: Dataset) -> None:  # type: ignore
             verbose=self.verbose > 0,
         )
 
+    def _fit_partial(self, dataset: Dataset, *args: Any, **kwargs: Any) -> None:  # type: ignore
+        self.model = deepcopy(self._model)
+
+        ui_coo = dataset.get_user_item_matrix(include_weights=True).tocoo(copy=False)
+        user_features = self._prepare_features(dataset.get_hot_user_features(), dataset.n_hot_users)
+        item_features = self._prepare_features(dataset.get_hot_item_features(), dataset.n_hot_items)
+
+        self.model.fit_partial(
+            ui_coo,
+            user_features=user_features,
+            item_features=item_features,
+            sample_weight=ui_coo,
+            epochs=self.n_epochs,
+            num_threads=self.n_threads,
+            verbose=self.verbose > 0,
+        )
+
     @staticmethod
     def _prepare_features(features: tp.Optional[Features], n_hot: int) -> tp.Optional[sparse.csr_matrix]:
         if features is None:

tests/models/test_implicit_als.py

@@ -346,3 +346,63 @@ def test_i2i_with_warm_and_cold_items(self, use_gpu: bool, dataset: Dataset) ->
                 dataset=dataset,
                 k=2,
             )
+
+    def test_fit_partial(self, use_gpu: bool, dataset: Dataset) -> None:
+        base_model = AlternatingLeastSquares(factors=8, num_threads=2, use_gpu=use_gpu, random_state=1)
+        model = ImplicitALSWrapperModel(model=base_model).fit(dataset)
+        data = [
+            [150, 11],
+            [150, 12],
+            [150, 15],
+        ]
+        new_interactions = pd.DataFrame(data, columns=Columns.UserItem)
+        new_interactions[Columns.Weight] = 1
+        new_interactions[Columns.Datetime] = "2021-09-10"
+        new_dataset = dataset.construct_new_datasets(new_interactions)
+        model.fit_partial(new_dataset)
+        actual = model.recommend(
+            users=[150],  # new user
+            dataset=new_dataset,
+            k=2,
+            filter_viewed=False,
+        )
+        expected = pd.DataFrame(
+            {
+                Columns.User: [150, 150],
+                Columns.Item: [12, 11],
+                Columns.Rank: [1, 2],
+            }
+        )
+        pd.testing.assert_frame_equal(actual.drop(columns=Columns.Score), expected)
+        pd.testing.assert_frame_equal(
+            actual.sort_values([Columns.User, Columns.Score], ascending=[True, False]).reset_index(drop=True), actual
+        )
+
+    def test_fit_partial_with_features(self, use_gpu: bool, dataset: Dataset) -> None:
+        user_id_map = IdMap.from_values(["u1", "u2", "u3"])
+        item_id_map = IdMap.from_values(["i1", "i2", "i3"])
+        interactions_df = pd.DataFrame(
+            [
+                ["u1", "i1", 0.1, "2021-09-09"],
+                ["u2", "i1", 0.1, "2021-09-09"],
+                ["u2", "i2", 0.5, "2021-09-05"],
+                ["u2", "i3", 0.2, "2021-09-05"],
+                ["u1", "i3", 0.2, "2021-09-05"],
+                ["u3", "i1", 0.2, "2021-09-05"],
+            ],
+            columns=[Columns.User, Columns.Item, Columns.Weight, Columns.Datetime],
+        )
+        interactions = Interactions.from_raw(interactions_df, user_id_map, item_id_map)
+        user_features_df = pd.DataFrame({"id": ["u1", "u2", "u3"], "f1": [0.3, 0.4, 0.5]})
+        user_features = DenseFeatures.from_dataframe(user_features_df, user_id_map)
+        item_features_df = pd.DataFrame({"id": ["i1", "i1"], "feature": ["f1", "f2"], "value": [2.1, 100]})
+        item_features = SparseFeatures.from_flatten(item_features_df, item_id_map)
+        dataset = Dataset(user_id_map, item_id_map, interactions, user_features, item_features)
+
+        # In case of big number of iterations there are differences between CPU and GPU results
+        base_model = AlternatingLeastSquares(factors=32, num_threads=2, use_gpu=use_gpu)
+        self._init_model_factors_inplace(base_model, dataset)
+
+        model = ImplicitALSWrapperModel(model=base_model, fit_features_together=False).fit(dataset)
+        with pytest.raises(NotImplementedError, match="fit_partial with explicit features is not implemented"):
+            model.fit_partial(dataset)

tests/models/test_lightfm.py

@@ -222,6 +222,38 @@ def test_with_weights(self, interactions_df: pd.DataFrame) -> None:
             actual,
         )
 
+    def test_fit_partial(self, dataset: Dataset) -> None:
+        base_model = DeterministicLightFM(no_components=2, loss="logistic")
+        model = LightFMWrapperModel(model=base_model, epochs=50).fit(dataset)
+        data = [
+            [150, 11],
+            [150, 12],
+            [150, 15],
+        ]
+        new_interactions = pd.DataFrame(data, columns=Columns.UserItem)
+        new_interactions[Columns.Weight] = 1
+        new_interactions[Columns.Datetime] = "2021-09-10"
+        new_dataset = dataset.construct_new_datasets(interactions_df=new_interactions)
+        model.fit_partial(new_dataset)
+        actual = model.recommend(
+            users=np.array([150]),  # new user
+            dataset=new_dataset,
+            k=2,
+            filter_viewed=False,
+        )
+        expected = pd.DataFrame(
+            {
+                Columns.User: [150, 150],
+                Columns.Item: [15, 12],
+                Columns.Rank: [1, 2],
+            }
+        )
+        pd.testing.assert_frame_equal(actual.drop(columns=Columns.Score), expected)
+        pd.testing.assert_frame_equal(
+            actual.sort_values([Columns.User, Columns.Score], ascending=[True, False]).reset_index(drop=True),
+            actual,
+        )
+
     def test_with_warp_kos(self, dataset: Dataset) -> None:
         base_model = DeterministicLightFM(no_components=2, loss="warp-kos")
         try: