[Fix] Saving all past behaviors during merging (#320)

* save consumed using pickle

* fix saving all past behaviors during merging

* ignore ruff `RUF012`

Author: massquantity

libreco/data/consumed.py

@@ -29,22 +29,17 @@ def _remove_duplicates(user_consumed, item_consumed):
     return user_dedup, item_dedup
 
 
-def update_consumed(new_data_info, data_info, merge_behavior):
+def update_consumed(
+    user_indices, item_indices, n_users, n_items, old_info, merge_behavior
+):
+    user_consumed, item_consumed = interaction_consumed(user_indices, item_indices)
     if merge_behavior:
-        new_data_info.user_consumed = _merge_dedup(
-            new_data_info.user_consumed, new_data_info.n_users, data_info.user_consumed
-        )
-        new_data_info.item_consumed = _merge_dedup(
-            new_data_info.item_consumed, new_data_info.n_items, data_info.item_consumed
-        )
+        user_consumed = _merge_dedup(user_consumed, n_users, old_info.user_consumed)
+        item_consumed = _merge_dedup(item_consumed, n_items, old_info.item_consumed)
     else:
-        new_data_info.user_consumed = _fill_empty(
-            new_data_info.user_consumed, new_data_info.n_users, data_info.user_consumed
-        )
-        new_data_info.item_consumed = _fill_empty(
-            new_data_info.item_consumed, new_data_info.n_items, data_info.item_consumed
-        )
-    return new_data_info
+        user_consumed = _fill_empty(user_consumed, n_users, old_info.user_consumed)
+        item_consumed = _fill_empty(item_consumed, n_items, old_info.item_consumed)
+    return user_consumed, item_consumed
 
 
 def _merge_dedup(new_consumed, num, old_consumed):

libreco/data/data_info.py

@@ -1,15 +1,15 @@
 """Classes for Storing Various Data Information."""
 import inspect
 import json
-import os
+import pickle
 from collections import namedtuple
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any, Dict, Iterable, List
 
 import numpy as np
 import pandas as pd
 
-from .consumed import interaction_consumed
 from ..feature.update import (
     get_row_id_masks,
     update_new_dense_feats,
@@ -69,10 +69,10 @@ class DataInfo:
         Unique sparse features for all items in train data.
     item_dense_unique : numpy.ndarray or None, default: None
         Unique dense features for all items in train data.
-    user_indices : numpy.ndarray or None, default: None
-        Mapped inner user indices from train data.
-    item_indices : numpy.ndarray or None, default: None
-        Mapped inner item indices from train data.
+    user_consumed : dict of {int : list} or None, default: None
+        All consumed items by each user.
+    item_consumed : dict of {int : list} or None, default: None
+        All consumed users by each item.
     user_unique_vals : numpy.ndarray or None, default: None
         All the unique users in train data.
     item_unique_vals : numpy.ndarray or None, default: None
@@ -110,8 +110,8 @@ def __init__(
         user_dense_unique=None,
         item_sparse_unique=None,
         item_dense_unique=None,
-        user_indices=None,
-        item_indices=None,
+        user_consumed=None,
+        item_consumed=None,
         user_unique_vals=None,
         item_unique_vals=None,
         sparse_unique_vals=None,
@@ -126,9 +126,8 @@ def __init__(
         self.user_dense_unique = user_dense_unique
         self.item_sparse_unique = item_sparse_unique
         self.item_dense_unique = item_dense_unique
-        self.user_consumed, self.item_consumed = interaction_consumed(
-            user_indices, item_indices
-        )
+        self.user_consumed = user_consumed
+        self.item_consumed = item_consumed
         self.user_unique_vals = user_unique_vals
         self.item_unique_vals = item_unique_vals
         self.sparse_unique_vals = sparse_unique_vals
@@ -440,27 +439,30 @@ def save(self, path, model_name):
         model_name : str
             Name of the saved file.
         """
-        if not os.path.isdir(path):
+        path = Path(path)
+        if not path.is_dir():
             print(f"file folder {path} doesn't exists, creating a new one...")
-            os.makedirs(path)
+            path.mkdir()
         if self.col_name_mapping is not None:
-            name_mapping_path = os.path.join(
-                path, f"{model_name}_data_info_name_mapping.json"
-            )
-            with open(name_mapping_path, "w") as f:
+            with open(path / f"{model_name}_data_info_name_mapping.json", "w") as f:
                 json.dump(
                     self.all_args["col_name_mapping"],
                     f,
                     separators=(",", ":"),
                     indent=4,
                 )
+        if self.user_consumed is not None:
+            with open(path / f"{model_name}_user_consumed.pkl", "wb") as f:
+                pickle.dump(self.user_consumed, f, protocol=pickle.HIGHEST_PROTOCOL)
+        if self.item_consumed is not None:
+            with open(path / f"{model_name}_item_consumed.pkl", "wb") as f:
+                pickle.dump(self.item_consumed, f, protocol=pickle.HIGHEST_PROTOCOL)
 
-        other_path = os.path.join(path, f"{model_name}_data_info")
         hparams = dict()
         arg_names = inspect.signature(self.__init__).parameters.keys()
         for arg in arg_names:
             if (
-                arg == "col_name_mapping"
+                arg in ("col_name_mapping", "user_consumed", "item_consumed")
                 or arg not in self.all_args
                 or self.all_args[arg] is None
             ):
@@ -478,7 +480,7 @@ def save(self, path, model_name):
             else:
                 hparams[arg] = self.all_args[arg]
 
-        np.savez_compressed(other_path, **hparams)
+        np.savez_compressed(path / f"{model_name}_data_info", **hparams)
 
     @classmethod
     def load(cls, path, model_name):
@@ -491,19 +493,26 @@ def load(cls, path, model_name):
         model_name : str
             Name of the saved file.
         """
-        if not os.path.exists(path):
+        path = Path(path)
+        if not path.exists():
             raise OSError(f"file folder {path} doesn't exists...")
 
         hparams = dict()
-        name_mapping_path = os.path.join(
-            path, f"{model_name}_data_info_name_mapping.json"
-        )
-        if os.path.exists(name_mapping_path):
+        name_mapping_path = path / f"{model_name}_data_info_name_mapping.json"
+        if name_mapping_path.exists():
             with open(name_mapping_path, "r") as f:
                 hparams["col_name_mapping"] = json.load(f)
 
-        other_path = os.path.join(path, f"{model_name}_data_info.npz")
-        info = np.load(other_path, allow_pickle=True)
+        user_consumed_path = path / f"{model_name}_user_consumed.pkl"
+        if user_consumed_path.exists():
+            with open(user_consumed_path, "rb") as f:
+                hparams["user_consumed"] = pickle.load(f)
+        item_consumed_path = path / f"{model_name}_item_consumed.pkl"
+        if item_consumed_path.exists():
+            with open(item_consumed_path, "rb") as f:
+                hparams["item_consumed"] = pickle.load(f)
+
+        info = np.load(path / f"{model_name}_data_info.npz", allow_pickle=True)
         info = dict(info.items())
         for arg in info:
             if arg == "interaction_data":
@@ -556,6 +565,7 @@ def store_old_info(data_info):
             # multi_sparse case, second to last cols are redundant.
             # Used in `rebuild_tf_model`, `rebuild_torch_model`
             sparse_len.append(-1)
+
     return OldInfo(
         data_info.n_users,
         data_info.n_items,

libreco/data/dataset.py

@@ -4,7 +4,7 @@
 
 import numpy as np
 
-from .consumed import update_consumed
+from .consumed import interaction_consumed, update_consumed
 from .data_info import DataInfo, store_old_info
 from .transformed import TransformedSet
 from ..feature.column_mapping import col_name2index
@@ -246,10 +246,11 @@ def build_trainset(cls, train_data, shuffle=False, seed=42):
             is_train=True,
             is_ordered=True,
         )
+        user_consumed, item_consumed = interaction_consumed(user_indices, item_indices)
         data_info = DataInfo(
             interaction_data=train_data[["user", "item", "label"]],
-            user_indices=user_indices,
-            item_indices=item_indices,
+            user_consumed=user_consumed,
+            item_consumed=item_consumed,
             user_unique_vals=cls.user_unique_vals,
             item_unique_vals=cls.item_unique_vals,
         )
@@ -303,14 +304,22 @@ def merge_trainset(
             is_train=True,
             is_ordered=False,
         )
+        user_consumed, item_consumed = update_consumed(
+            user_indices,
+            item_indices,
+            len(cls.user_unique_vals),
+            len(cls.item_unique_vals),
+            data_info,
+            merge_behavior,
+        )
+
         new_data_info = DataInfo(
             interaction_data=train_data[["user", "item", "label"]],
-            user_indices=user_indices,
-            item_indices=item_indices,
+            user_consumed=user_consumed,
+            item_consumed=item_consumed,
             user_unique_vals=cls.user_unique_vals,
             item_unique_vals=cls.item_unique_vals,
         )
-        new_data_info = update_consumed(new_data_info, data_info, merge_behavior)
         new_data_info.old_info = store_old_info(data_info)
         cls.train_called = True
         return merge_transformed, new_data_info
@@ -511,15 +520,16 @@ def build_trainset(
             col_name_mapping["multi_sparse"] = multi_sparse_col_map(multi_sparse_col)
 
         interaction_data = train_data[["user", "item", "label"]]
+        user_consumed, item_consumed = interaction_consumed(user_indices, item_indices)
         data_info = DataInfo(
             col_name_mapping,
             interaction_data,
             user_sparse_unique,
             user_dense_unique,
             item_sparse_unique,
             item_dense_unique,
-            user_indices,
-            item_indices,
+            user_consumed,
+            item_consumed,
             cls.user_unique_vals,
             cls.item_unique_vals,
             cls.sparse_unique_vals,
@@ -632,15 +642,24 @@ def merge_trainset(
         )
 
         interaction_data = train_data[["user", "item", "label"]]
+        user_consumed, item_consumed = update_consumed(
+            user_indices,
+            item_indices,
+            len(cls.user_unique_vals),
+            len(cls.item_unique_vals),
+            data_info,
+            merge_behavior,
+        )
+
         new_data_info = DataInfo(
             data_info.col_name_mapping,
             interaction_data,
             user_sparse_unique,
             user_dense_unique,
             item_sparse_unique,
             item_dense_unique,
-            user_indices,
-            item_indices,
+            user_consumed,
+            item_consumed,
             cls.user_unique_vals,
             cls.item_unique_vals,
             cls.sparse_unique_vals,
@@ -649,7 +668,6 @@ def merge_trainset(
             cls.multi_sparse_unique_vals,
             multi_sparse_info,
         )
-        new_data_info = update_consumed(new_data_info, data_info, merge_behavior)
         new_data_info.old_info = store_old_info(data_info)
         cls.train_called = True
         return merge_transformed, new_data_info

pyproject.toml

@@ -90,7 +90,7 @@ filterwarnings = [
 line-length = 88
 target-version = "py38"
 show-source = true
-ignore = ["E501"]
+ignore = ["E501", "RUF012"]
 select = [
     # pyflakes
     "F",

tests/retrain/test_tfmodel_retrain_feat.py

@@ -95,8 +95,8 @@ def test_tfmodel_retrain_feat():
     tf.compat.v1.reset_default_graph()
     new_data_info = DataInfo.load(SAVE_PATH, model_name="din_model")
 
-    # use second half data as second training part
-    second_half_data = all_data[(len(all_data) // 2) :]
+    # use first half of second half data as second training part
+    second_half_data = all_data[(len(all_data) // 2) : (len(all_data) * 3 // 4)]
     train_data_orig, eval_data_orig = split_by_ratio_chrono(
         second_half_data, test_size=0.2
     )
@@ -158,4 +158,60 @@ def test_tfmodel_retrain_feat():
 
     assert new_eval_result["roc_auc"] != eval_result["roc_auc"]
 
+    new_data_info.save(path=SAVE_PATH, model_name="din_model")
+    new_model.save(
+        path=SAVE_PATH, model_name="din_model", manual=True, inference_only=False
+    )
+
+    # ========================== load and retrain 2 =============================
+    tf.compat.v1.reset_default_graph()
+    new_data_info = DataInfo.load(SAVE_PATH, model_name="din_model")
+
+    # use second half of second half data as second training part
+    third_half_data = all_data[(len(all_data) * 3 // 4) :]
+    train_data_orig, eval_data_orig = split_by_ratio_chrono(
+        third_half_data, test_size=0.2
+    )
+    train_data, new_data_info = DatasetFeat.merge_trainset(
+        train_data_orig, new_data_info, merge_behavior=True
+    )
+    eval_data = DatasetFeat.merge_evalset(eval_data_orig, new_data_info)
+    print(new_data_info)
+
+    new_model = DIN(
+        "ranking",
+        new_data_info,
+        loss_type="focal",  # change loss
+        embed_size=16,
+        n_epochs=1,
+        lr=1e-4,
+        lr_decay=False,
+        reg=None,
+        batch_size=2048,
+        hidden_units=(32, 16),
+        recent_num=10,
+        use_tf_attention=True,
+    )
+    new_model.rebuild_model(path=SAVE_PATH, model_name="din_model", full_assign=True)
+    new_model.fit(
+        train_data,
+        neg_sampling=True,
+        verbose=2,
+        shuffle=True,
+        eval_data=eval_data,
+        metrics=[
+            "loss",
+            "balanced_accuracy",
+            "roc_auc",
+            "pr_auc",
+            "precision",
+            "recall",
+            "map",
+            "ndcg",
+        ],
+        eval_user_num=20,
+    )
+    ptest_preds(new_model, "ranking", second_half_data, with_feats=False)
+    ptest_recommends(new_model, new_data_info, second_half_data, with_feats=False)
+
     remove_path(SAVE_PATH)