Refactor dataprocessing

Author: krystianfranus

configs/data_processing/config.yaml

@@ -1,6 +1,4 @@
 defaults:
-  - method: simple
-
   - override hydra/job_logging: custom
   - _self_  # Primary config overrides values of configs from the Defaults List
 
@@ -10,4 +8,7 @@ hydra:
     dir: .
 
 project_name: "MyProject"
-use_remote_storage: false
+tags: ["${now:%Y-%m-%d}"]
+list_size: 5
+split_date: "2019/04/09"
+history_size: 5

configs/data_processing/method/bpr.yaml

@@ -15,8 +15,15 @@
 
 
 def load_raw_data() -> tuple[DataFrame, DataFrame]:
+    """
+    Load raw interaction and impression data from local cache.
+    If data is not available locally, download it from remote storage.
+
+    Returns:
+        tuple[DataFrame, DataFrame]: DataFrames containing interactions and impressions data.
+    """
     data_dir = get_cache_path() / "data-cw10m"
-    # Download data if necessary
+    # Download data if doesn't exist
     if not data_dir.exists():
         log.info("Data are not available locally!")
         log.info(f"Downloading data from remote storage to {data_dir} ...")
@@ -36,28 +43,66 @@ def load_raw_data() -> tuple[DataFrame, DataFrame]:
     return interactions, impressions
 
 
-def process_simple(
+def process_data(
     interactions: DataFrame,
     impressions: DataFrame,
+    list_size: int,
+    split_date: str,
+    history_size: int,
 ) -> tuple[DataFrame, DataFrame, DataFrame, dict]:
-    interactions = _common(interactions, impressions)
-
-    # Split data (train: 2019/01/07-2019/04/13; val: 2019/04/14-2019/04/15)
-    split_date = dt.datetime(2019, 4, 14)
-    train = interactions[interactions["timestamp"] < split_date]
-    val = interactions[interactions["timestamp"] >= split_date]
-
-    # Keep lists with condition 0 < mean_target < 1
-    train_valid_lists = (
-        train.groupby("list_id")
-        .agg({"target": "mean"})
-        .rename(columns={"target": "mean"})
-        .reset_index()
+    """
+    Process and clean raw interaction and impression data, splitting it into training and validation sets.
+
+    Args:
+        interactions (DataFrame): Raw interactions data.
+        impressions (DataFrame): Raw impressions data.
+        list_size (int): Minimum list size to be included in the dataset.
+        split_date (str): Date string to split train and validation sets (format: YYYY/MM/DD).
+        history_size (int): Number of past clicks to store as user history.
+
+    Returns:
+        tuple[DataFrame, DataFrame, DataFrame, dict]:
+            - Processed training dataset.
+            - Processed validation dataset.
+            - User-to-index mapping DataFrame.
+            - Item-to-index mapping DataFrame.
+    """
+    data = _preprocess_data(interactions, impressions)
+
+    # Remove lists shorter than 'list_size'
+    data_valid_lists = (
+        data.groupby("list_id").size().reset_index().rename(columns={0: "list_size"})
     )
-    train_valid_lists = train_valid_lists[
-        (train_valid_lists["mean"] > 0) & (train_valid_lists["mean"] < 1)
-    ]
-    train = train.merge(train_valid_lists, "inner", "list_id")
+    data_valid_lists = data_valid_lists[data_valid_lists["list_size"] >= list_size]
+    data_valid_lists = data_valid_lists.drop(columns="list_size")
+    data = data.merge(data_valid_lists, "inner", "list_id")
+
+    # Split data (train: 2019/01/07-2019/04/08; val: 2019/04/09-2019/04/15)
+    split_date = dt.datetime.strptime(split_date, "%Y/%m/%d")
+    train = data[data["timestamp"] < split_date]
+    val = data[data["timestamp"] >= split_date]
+
+    # Remove overlapping lists between train and val
+    train_lists = train["list_id"].unique()
+    val_lists = val["list_id"].unique()
+    train_lists_expected = np.setdiff1d(train_lists, val_lists)
+    val_lists_expected = np.setdiff1d(val_lists, train_lists)
+    train = train[train["list_id"].isin(train_lists_expected)]
+    val = val[val["list_id"].isin(val_lists_expected)]
+
+    # Remove cold users from val
+    train_users = train["user"].unique()
+    val = val[val["user"].isin(train_users)]
+
+    # Remove lists containing cold items on val
+    train_items = train["item"].unique()
+    val_items = val["item"].unique()
+    cold_items = np.setdiff1d(val_items, train_items)
+    val["cold_item"] = val["item"].isin(cold_items)
+    n_cold_items_per_list = val.groupby("list_id")["cold_item"].sum()
+    valid_lists = n_cold_items_per_list[n_cold_items_per_list == 0].index
+    val = val[val["list_id"].isin(valid_lists)]
+    val = val.drop(columns="cold_item")
 
     # Prepare user/item to idx mappers based on train data
     unique_train_users = train["user"].unique()
@@ -69,153 +114,73 @@ def process_simple(
         {"item": unique_train_items, "item_idx": np.arange(unique_train_items.size)}
     )
 
-    # Map user/item to idx - it removes cold users and items from validation
-    train = train.merge(user_mapper, on="user", how="inner")
-    train = train.merge(item_mapper, on="item", how="inner")
-    val = val.merge(user_mapper, on="user", how="inner")
-    val = val.merge(item_mapper, on="item", how="inner")
-
-    # Keep lists with condition 0 < mean_target < 1
-    val_valid_lists = (
-        val.groupby("list_id")
-        .agg({"target": "mean"})
-        .rename(columns={"target": "mean"})
-        .reset_index()
+    # Create user_history column - list of last n clicked items per user
+    train_clicks = train.sort_values(by=["user", "timestamp"])
+    train_clicks = train_clicks[train_clicks["target"] == 1].reset_index(drop=True)
+
+    def last_clicks(series):
+        history = []
+        result = []
+        for item in series:
+            result.append(history.copy())  # Append the current state of history
+            if len(history) == history_size:
+                history.pop(0)  # Keep only the last n items
+            history.append(item)
+        # Pad with None if history is shorter than threshold
+        return [([None] * (history_size - len(h)) + h) for h in result]
+
+    # Apply function per user
+    train_clicks["user_history"] = train_clicks.groupby("user")["item"].transform(
+        last_clicks
     )
-    val_valid_lists = val_valid_lists[
-        (val_valid_lists["mean"] > 0) & (val_valid_lists["mean"] < 1)
-    ]
-    val = val.merge(val_valid_lists, "inner", "list_id")
+    train = train.merge(
+        train_clicks[["timestamp", "user", "user_history"]],
+        on=["timestamp", "user"],
+        how="left",
+    )
+    tmp = train_clicks.loc[
+        train_clicks.groupby("user")["timestamp"].idxmax()
+    ].reset_index(drop=True)
+    tmp = tmp[["user", "user_history"]]
+    val = val.merge(tmp, "inner", "user")
 
+    # Sort train and val by timestamp
     train = train.sort_values("timestamp").reset_index(drop=True)
     val = val.sort_values("timestamp").reset_index(drop=True)
 
-    # Select valid columns
-    train = train[["timestamp", "list_id", "user_idx", "item_idx", "target"]]
-    train.columns = ["timestamp", "list_id", "user", "item", "target"]
-    val = val[["timestamp", "list_id", "user_idx", "item_idx", "target"]]
-    val.columns = ["timestamp", "list_id", "user", "item", "target"]
-
-    # Mock test_data
-    test = val.copy()  # test set == validation set (should be changed in the future!)
-
-    # Prepare statistics
-    unique_val_users = val["user"].unique()
-    unique_val_items = val["item"].unique()
-    stats = {}
-    stats["train_n_users"] = unique_train_users.size
-    stats["train_n_items"] = unique_train_items.size
-    stats["train_n_lists"] = train["list_id"].nunique()
-    stats["train_n_clicks"] = int(train["target"].sum())
-    stats["train_n_impressions"] = len(train) - stats["train_n_clicks"]
-    stats["train_ctr"] = stats["train_n_clicks"] / stats["train_n_impressions"]
-    stats["val_n_users"] = unique_val_users.size
-    stats["val_n_items"] = unique_val_items.size
-    stats["val_n_lists"] = val["list_id"].nunique()
-    stats["val_n_clicks"] = int(val["target"].sum())
-    stats["val_n_impressions"] = len(val) - stats["val_n_clicks"]
-    stats["val_ctr"] = stats["val_n_clicks"] / stats["val_n_impressions"]
-
-    return train, val, test, stats, user_mapper, item_mapper
-
-
-def process_bpr(
-    interactions: DataFrame,
-    impressions: DataFrame,
-) -> tuple[DataFrame, DataFrame, DataFrame, dict]:
-    interactions = _common(interactions, impressions)
-
-    # Split data
-    split_date = dt.datetime(2019, 4, 14)
-    train = interactions[interactions["timestamp"] < split_date]
-    tmp0 = train.loc[train["target"] == 0, ["user", "item"]]
-    tmp1 = train.loc[train["target"] == 1, ["user", "item"]]
-    train = tmp0.merge(tmp1, "inner", "user", suffixes=("_neg", "_pos"))
-    val = interactions[interactions["timestamp"] >= split_date]
-
-    # Prepare user/item to idx mappers based on train data
-    unique_train_users = train["user"].unique()
-    # unique_users = train["user"].unique()
-    item_neg_set = set(train["item_neg"])
-    item_pos_set = set(train["item_pos"])
-    unique_train_items = pd.Series(list(item_neg_set | item_pos_set)).unique()
-
-    user_mapper = pd.DataFrame(
-        {"user": unique_train_users, "user_idx": np.arange(unique_train_users.size)}
-    )
-    item_mapper = pd.DataFrame(
-        {"item": unique_train_items, "item_idx": np.arange(unique_train_items.size)}
-    )
-
-    # Map user/item to idx and handle column names conflicts
-    train = train.merge(user_mapper, on="user", how="inner")
-    train = train[["user_idx", "item_neg", "item_pos"]].rename(
-        columns={"user_idx": "user"}
-    )
-    train = train.merge(item_mapper, left_on="item_neg", right_on="item", how="inner")
-    train = train[["user", "item_idx", "item_pos"]].rename(
-        columns={"item_idx": "item_neg"}
-    )
-    train = train.merge(item_mapper, left_on="item_pos", right_on="item", how="inner")
-    train = train[["user", "item_neg", "item_idx"]].rename(
-        columns={"item_idx": "item_pos"}
-    )
-
-    val = val.merge(user_mapper, on="user", how="inner")
-    val = val.merge(item_mapper, on="item", how="inner")
-
-    # Keep lists with condition 0 < mean_target < 1
-    val_valid_lists = (
-        val.groupby("list_id")
-        .agg({"target": "mean"})
-        .rename(columns={"target": "mean"})
-        .reset_index()
-    )
-    val_valid_lists = val_valid_lists[
-        (val_valid_lists["mean"] > 0) & (val_valid_lists["mean"] < 1)
-    ]
-    val = val.merge(val_valid_lists, "inner", "list_id")
-
-    val = val[["timestamp", "list_id", "user_idx", "item_idx", "target"]]
-    val = val.rename(columns={"user_idx": "user", "item_idx": "item"})
-
-    # Mock test_data
-    test = val.copy()  # test set == validation set (to change in the future!)
-
-    # Prepare statistics
-    unique_val_users = val["user"].unique()
-    unique_val_items = val["item"].unique()
-    stats = {}
-    stats["train_n_users"] = unique_train_users.size
-    stats["train_n_items"] = unique_train_items.size
-    stats["val_n_users"] = unique_val_users.size
-    stats["val_n_items"] = unique_val_items.size
-    stats["val_n_lists"] = val["list_id"].nunique()
-    stats["val_n_clicks"] = int(val["target"].sum())
-    stats["val_n_impressions"] = len(val) - stats["val_n_clicks"]
-    stats["val_ctr"] = stats["val_n_clicks"] / stats["val_n_impressions"]
-
-    return train, val, test, stats, user_mapper, item_mapper
+    return train, val, user_mapper, item_mapper
 
 
 def save_data(
     task: Task,
     train: DataFrame,
     val: DataFrame,
-    test: DataFrame,
-    stats: dict,
     user_mapper: DataFrame,
     item_mapper: DataFrame,
 ) -> None:
+    """
+    Save processed data and mappings as artifacts in ClearML.
+
+    Args:
+        task (Task): ClearML task instance.
+        train (DataFrame): Processed training dataset.
+        val (DataFrame): Processed validation dataset.
+        user_mapper (DataFrame): User-to-index mapping.
+        item_mapper (DataFrame): Item-to-index mapping.
+    """
     task.upload_artifact("train", train, extension_name=".parquet")
     task.upload_artifact("val", val, extension_name=".parquet")
-    task.upload_artifact("test", test, extension_name=".parquet")
-    task.upload_artifact("stats", stats)
     task.upload_artifact("user_mapper", user_mapper, extension_name=".parquet")
     task.upload_artifact("item_mapper", item_mapper, extension_name=".parquet")
 
 
 def _download_data(data_dir: Path) -> None:
+    """
+    Download interaction and impression data from remote S3 storage.
+
+    Args:
+        data_dir (Path): Local directory path to store downloaded data.
+    """
     s3 = s3fs.S3FileSystem()
     prefix = "s3://kf-north-bucket/data-science-template/data/contentwise/CW10M"
 
@@ -236,10 +201,13 @@ def _download_data(data_dir: Path) -> None:
             df.to_parquet(f"{data_dir}/impressions-direct-link/{Path(p).name}")
 
 
-def _common(
+def _preprocess_data(
     interactions: DataFrame,
     impressions_dl: DataFrame,
 ) -> DataFrame:
+    """
+    Preprocess raw interactions and impressions data for further processing.
+    """
     # Select only movies from item types
     interactions = interactions[interactions["item_type"] == 0]
     # Select only clicks as an interaction type
@@ -250,28 +218,33 @@ def _common(
         unit="ms",
     )
 
+    # Assume that user can have only one interaction at exact timestamp
+    interactions = interactions.drop_duplicates(["utc_ts_milliseconds", "user_id"])
+
     impressions_dl = impressions_dl.explode("recommended_series_list")
     impressions_dl["recommended_series_list"] = pd.to_numeric(
         impressions_dl["recommended_series_list"]
     )
 
     # Join positive interactions (clicks) with negative interactions (impressions)
-    interactions = interactions.merge(impressions_dl, "inner", "recommendation_id")
+    data = interactions.merge(impressions_dl, "inner", "recommendation_id")
 
-    # Create unique id per (recommandation_id, user_id) pairs
-    interactions["list_id"] = pd.factorize(
-        interactions[["recommendation_id", "user_id"]].apply(tuple, axis=1)
+    # Create unique id per (utc_ts_milliseconds, recommandation_id, user_id)
+    data["list_id"] = pd.factorize(
+        data[["utc_ts_milliseconds", "recommendation_id", "user_id"]].apply(
+            tuple, axis=1
+        )
     )[0]
 
     # Mark positive interactions with 1 and negative with 0
-    interactions["target"] = np.where(
-        interactions["series_id"] == interactions["recommended_series_list"],
+    data["target"] = np.where(
+        data["series_id"] == data["recommended_series_list"],
         1,
         0,
     )
-    interactions["target"] = interactions["target"].astype("int32")
+    data["target"] = data["target"].astype("int32")
 
-    interactions = interactions[
+    data = data[
         [
             "utc_ts_milliseconds",
             "list_id",
@@ -280,6 +253,6 @@ def _common(
             "target",
         ]
     ]
-    interactions.columns = ["timestamp", "list_id", "user", "item", "target"]
+    data.columns = ["timestamp", "list_id", "user", "item", "target"]
 
-    return interactions
+    return data