<feat/recommendation> Implement a simple ranking model

recommendation_system/main.py

@@ -17,11 +17,11 @@ def recommend(cls, recipient_id: Text) -> List[Dict]:
         """
         experiment_config: Dict = cls._load_experiment_config(recipient_id)
         user_features: Dict = cls._get_feature(recipient_id)
-        candidates: List[Dict] = []
+        candidates: List[List[Dict]] = []
         for cancidate_model in cls._get_candidate_models(experiment_config):
-            candidates += cancidate_model.get_candidates(user_features)
+            candidates.append(cancidate_model.get_candidates(user_features))
         # TODO: For now, it only support 1 ranking layer. We should suuport multiple ranking layers to sort at some point.
-        result = cls._get_ranking_model(experiment_config).rank(candidates)
+        result = cls._get_ranking_model(experiment_config).rank(user_features, candidates)
         filtered_result = cls._filter(result)
         return filtered_result
 
@@ -35,7 +35,7 @@ def _load_experiment_config(recipient_id):
                 "demo",
                 # 'other candidate model for you guys to implement'
             ],
-            "ranking_model": "demo",
+            "ranking_model": "simple",
             # TODO: should replace base ranking model with your own!
         }
 

recommendation_system/ranking_layer/base.py

@@ -4,5 +4,5 @@
 
 class BaseRankingModel(abc.ABC):
     @abc.abstractmethod
-    def rank(user_features: Dict) -> List[Dict]:
+    def rank(user_features: Dict, candidates: List[List[Dict]]) -> List[Dict]:
         pass

recommendation_system/ranking_layer/simple.py

@@ -0,0 +1,66 @@
+from copy import deepcopy
+from typing import List, Dict, Tuple
+
+from recommendation_system.ranking_layer.base import BaseRankingModel
+
+class SimpleRankingModel(BaseRankingModel):
+    def __init__(self) -> None:
+        super().__init__()
+
+    def rank(self, user_features: Dict, candidates: List[List[Dict]], top_k: int) -> List[Dict]:
+        candidates = self.process_data(candidates=candidates)
+        rankings = self.calc_candidate_score(candidates=candidates)
+
+        new_candidates = []
+        for item, score in rankings:
+            new_item = self.get_item(item)
+            new_item['score'] = score
+            new_candidates.append(new_item)
+
+        return new_candidates[:top_k]
+
+    def get_item(self, item):
+        return deepcopy(self._items_mapping[item])
+
+    def calc_candidate_score(self, candidates):
+        candidate_ranking = {}
+
+        for item in self._items_mapping:
+            candidate_ranking.setdefault(item, [])
+            for ranker in candidates:
+                if item in ranker:
+                    k, score = ranker.get(item)
+                    candidate_ranking[item].append(score/k)
+
+        for item, scores in candidate_ranking.items():
+            candidate_ranking[item] = sum(scores)
+
+        return sorted(candidate_ranking.items(), key=lambda d: d[1], reverse=True)
+
+    def process_data(self, candidates: List[List[Dict]]) -> List[List[Tuple]]:
+        items_mapping = {}
+        new_candidates = []
+
+        for ranker in candidates:
+            new_candidate_per_ranker = {}
+            for i, candidate in enumerate(ranker):
+                items_mapping.setdefault(candidate['title'], {
+                    'title': candidate['title'],
+                    'subtitle': candidate['subtitle'],
+                    'image_url': candidate['image_url'],
+                    'date': candidate['date'],
+                    'url': candidate['url']
+                })
+                new_candidate_per_ranker[candidate['title']] = (i + 1, candidate.get('score', 1)) # (rank, score)
+
+            new_candidates.append(new_candidate_per_ranker)
+
+        self._items_mapping = items_mapping
+
+        return new_candidates
+
+    def build_user_embedding(self, user_features):
+        pass
+
+    def calc_similarity(self, user_features):
+        pass