111 stream module multi model handling

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "streamsight"
-version = "0.2.10"
+version = "0.2.11"
 description = "A toolkit for offline evaluation of Recommender Systems"
 authors = ["Ng Tze Kean <[email protected]>"]
 readme = "README.md"

streamsight/evaluators/evaluator_stream.py

@@ -242,8 +242,14 @@ def get_data(self, algo_id: UUID) -> InteractionMatrix:
         if not self.has_started:
             raise ValueError(f"call start_stream() before requesting data for algorithm {algo_id}")
 
+        logger.debug(f"Getting data for algorithm {algo_id}")
+
         # check if we need to move to the next window
-        if self.setting.is_sliding_window_setting and self.status_registry.is_all_predicted():
+        if (
+            self.setting.is_sliding_window_setting
+            and self.status_registry.is_all_predicted()
+            and self.status_registry.is_all_same_data_segment()
+        ):
             self.user_item_base._reset_unknown_user_item_base()
             incremental_data = self.setting.next_incremental_data()
             self.user_item_base._update_known_user_item_base(incremental_data)
@@ -303,6 +309,7 @@ def get_unlabeled_data(self, algo_id: UUID) -> Optional[InteractionMatrix]:
         :return: The unlabeled data for prediction
         :rtype: Optional[InteractionMatrix]
         """
+        logger.debug(f"Getting unlabeled data for algorithm {algo_id}")
         status = self.status_registry[algo_id].state
         if status in [AlgorithmStateEnum.READY, AlgorithmStateEnum.PREDICTED]:
             return self._unlabeled_data_cache
@@ -344,6 +351,7 @@ def submit_prediction(self, algo_id: UUID, X_pred: Union[csr_matrix, Interaction
         :type X_pred: csr_matrix
         :raises ValueError: If X_pred is not an InteractionMatrix or csr_matrix
         """
+        logger.debug(f"Submitting prediction for algorithm {algo_id}")
         status = self.status_registry[algo_id].state
 
         if status == AlgorithmStateEnum.READY:
@@ -428,15 +436,15 @@ def _cache_evaluation_data(self) -> None:
         self._run_step += 1
 
         logger.debug(f"Caching evaluation data for step {self._run_step}")
-        
+
         try:
             unlabeled_data, ground_truth_data, _ = self._get_evaluation_data()
         except EOWSetting as e:
             raise e
 
         self._unlabeled_data_cache = unlabeled_data
         self._ground_truth_data_cache = ground_truth_data
-        
+
         logger.debug(f"Data cached for step {self._run_step} complete")
 
     def _evaluate(self, algo_id: UUID, X_pred: csr_matrix) -> None:
@@ -456,7 +464,7 @@ def _evaluate(self, algo_id: UUID, X_pred: csr_matrix) -> None:
         """
         X_true = self._ground_truth_data_cache.get_users_n_first_interaction(self.metric_k)
         X_true = X_true.binary_values
-        
+
         X_pred = self._prediction_shape_handler(X_true.shape, X_pred)
         algorithm_name = self.status_registry.get_algorithm_identifier(algo_id)
 
@@ -469,5 +477,5 @@ def _evaluate(self, algo_id: UUID, X_pred: csr_matrix) -> None:
                 metric:Metric = metric_cls(timestamp_limit=self._current_timestamp)
             metric.calculate(X_true, X_pred)
             self._acc.add(metric=metric, algorithm_name=algorithm_name)
-        
+
         logger.debug(f"Prediction evaluated for algorithm {algo_id} complete")

streamsight/matrix/interaction_matrix.py

@@ -460,6 +460,12 @@ def __sub__(self, im: "InteractionMatrix") -> "InteractionMatrix":
     def __repr__(self):
         return repr(self._df)
 
+    def __eq__(self, value: object) -> bool:
+        if not isinstance(value, InteractionMatrix):
+            logger.debug(f"Comparing {type(value)} with InteractionMatrix is not supported")
+            return False
+        return self._df.equals(value._df)
+
     @overload
     def items_in(self, I: Set[int], inplace=False) -> "InteractionMatrix": ...
     @overload

streamsight/registries/registry.py

@@ -299,6 +299,12 @@ def update(self, algo_id: UUID, state: AlgorithmStateEnum, data_segment: Optiona
     def is_all_predicted(self) -> bool:
         return self.status_counts[AlgorithmStateEnum.PREDICTED] == len(self.registered)
 
+    def is_all_same_data_segment(self) -> bool:
+        data_segments = set()
+        for key in self:
+            data_segments.add(self[key].data_segment)
+        return len(data_segments) == 1
+
     def all_algo_states(self) -> Dict[str, AlgorithmStateEnum]:
         states = {}
         for key in self:

test/evaluator/test_full_run.py

@@ -1,7 +1,7 @@
 import pytest
 from streamsight.datasets import TestDataset
 from streamsight.settings import SlidingWindowSetting, SingleTimePointSetting
-from streamsight.evaluators import EvaluatorBuilder, EvaluatorStreamerBuilder
+from streamsight.evaluators import EvaluatorPipelineBuilder, EvaluatorStreamerBuilder
 
 @pytest.fixture()
 def sliding_window():
@@ -25,28 +25,46 @@ def single_time_point():
     return setting
 
 class TestFullRun:
-    def test_sliding_window(self, sliding_window):
-        b = EvaluatorBuilder()
+    def test_sliding_window_without_unknown_user_item(self, sliding_window):
+        b = EvaluatorPipelineBuilder(True,True)
+        b.add_setting(sliding_window)
         b.add_algorithm("ItemKNNIncremental", {"K": 1})
         b.add_metric("PrecisionK")
         b.add_metric("RecallK")
+        evaluator = b.build()
+        evaluator.run()
+
+    def test_sliding_window_without_unknown_user(self, sliding_window):
+        b = EvaluatorPipelineBuilder(True,False)
+        b.add_setting(sliding_window)
+        b.add_algorithm("ItemKNNIncremental", {"K": 1})
+        b.add_metric("PrecisionK")
+        b.add_metric("RecallK")
+        evaluator = b.build()
+        evaluator.run()
+
+    def test_sliding_window_with_unknowns(self, sliding_window):
+        b = EvaluatorPipelineBuilder(False,False)
         b.add_setting(sliding_window)
+        b.add_algorithm("ItemKNNIncremental", {"K": 1})
+        b.add_metric("PrecisionK")
+        b.add_metric("RecallK")
         evaluator = b.build()
         evaluator.run()
 
     def test_single_time_point(self, single_time_point):
-        b = EvaluatorBuilder()
+        b = EvaluatorPipelineBuilder()
+        b.add_setting(single_time_point)
         b.add_algorithm("ItemKNNIncremental", {"K": 1})
         b.add_metric("PrecisionK")
         b.add_metric("RecallK")
-        b.add_setting(single_time_point)
         evaluator = b.build()
         evaluator.run()
 
     def test_stream(self, sliding_window):
         b = EvaluatorStreamerBuilder()
-        b.add_metric("PrecisionK")
         b.add_setting(sliding_window)
+        b.add_metric("PrecisionK")
         evaluator = b.build()
 
         from streamsight.algorithms import ItemKNNIncremental

test/evaluator/test_stream.py

@@ -0,0 +1,65 @@
+import pytest
+from streamsight.settings.sliding_window_setting import SlidingWindowSetting
+from test.conftest import BACKGROUND_T, WINDOW_SIZE, SEED, N_SEQ_DATA, SEED
+from streamsight.evaluators import EvaluatorStreamerBuilder
+from streamsight.algorithms import ItemKNNIncremental
+
+
+@pytest.fixture()
+def setting(test_dataset):
+    data = test_dataset.load()
+    setting_obj = SlidingWindowSetting(background_t=BACKGROUND_T,
+                                window_size=WINDOW_SIZE,
+                                n_seq_data=N_SEQ_DATA,
+                                seed=SEED)
+    setting_obj.split(data)
+    return setting_obj
+
+@pytest.fixture()
+def k():
+    return 10
+
+class TestStreamer():
+    def test_algorithm_in_different_data_segment_handling(self, setting, k):
+
+        builder = EvaluatorStreamerBuilder()
+        builder.add_setting(setting)
+        builder.set_metric_K(k)
+        builder.add_metric("PrecisionK")
+        evaluator = builder.build()
+
+        algo = ItemKNNIncremental(K=10)
+        algo_id = evaluator.register_algorithm(algo)
+        print(algo_id)
+
+        from streamsight.algorithms import ItemKNNStatic
+        external_model = ItemKNNIncremental(K=10)
+        external_model_id = evaluator.register_algorithm(external_model)
+        print(external_model_id)
+
+        evaluator.start_stream()
+
+        # first iteration
+        data = evaluator.get_data(algo_id)
+        algo.fit(data)
+        unlabeled_data = evaluator.get_unlabeled_data(algo_id)
+        prediction = algo.predict(unlabeled_data)
+        evaluator.submit_prediction(algo_id, prediction)
+        data = evaluator.get_data(external_model_id)
+        external_model.fit(data)
+        unlabeled_data = evaluator.get_unlabeled_data(external_model_id)
+        prediction = external_model.predict(unlabeled_data)
+        evaluator.submit_prediction(external_model_id, prediction)
+
+        # second iteration
+        print("Second iteration")
+        data = evaluator.get_data(algo_id)
+        algo.fit(data)
+        unlabeled_data = evaluator.get_unlabeled_data(algo_id)
+        prediction = algo.predict(unlabeled_data)
+        evaluator.submit_prediction(algo_id, prediction)
+
+
+        to_validate_data = evaluator.get_data(external_model_id)
+
+        assert(to_validate_data == data)