对核心商户进行重排序

api_test.py

@@ -4,7 +4,7 @@ import json
 url = "http://127.0.0.1:7960/brandcultivation/api/v1/recommend"
 payload = {
     "city_uuid": "00000000000000000000000011445301",
-    "product_code": "700239",
+    "product_code": "700234",
     "cust_code_list": ["445300108802", "445300108803", "445300104219"],
     "cultivacation_id": "10000003",
     "limit_cycle_name": "202606W1(06.01-06.07)"

models/recommend.py

@@ -11,6 +11,26 @@ from core import get_logger
 logger = get_logger("models.recommend")
 
 
+CORE_RERANK_CONFIG = {
+    "existing": {
+        "core_model_weight": 0.75,
+        "core_quality_weight": 0.15,
+        "core_boost": 35,
+        "low_model_threshold": 35,
+        "low_model_weight": 0.85,
+        "low_quality_weight": 0.10,
+        "low_core_boost": 65,
+        "normal_model_weight": 0.90,
+    },
+    "new": {
+        "core_model_weight": 0.55,
+        "core_quality_weight": 0.25,
+        "core_boost": 50,
+        "normal_model_weight": 0.90,
+    },
+}
+
+
 class Recommend:
     def __init__(self, city_uuid):
         self._redis = RedisDatabaseHelper().redis
@@ -71,6 +91,7 @@ class Recommend:
         # 获取推理用的feats_map
         feats_map = generate_feats_map(product_data, cust_data)
         recommend_list = self._gbdtlr_model.get_recommend_list(feats_map, ordered_recall_list)
+        recommend_list = self._rerank_existing_product(recommend_list, cust_code_list)
         # recommend_list = self.filter_recommend_list(recommend_list)
         logger.info(f"GBDT-LR recommend completed: {len(recommend_list)} results")
         return recommend_list
@@ -83,9 +104,236 @@ class Recommend:
         recommend_list = self._item2vec_model.get_recommend_cust_list(product_id, cust_code_list=cust_code_list)
         recommend_list = recommend_list.drop(columns=["sale_qty"])
         recommend_list = recommend_list.to_dict(orient='records')
+        recommend_list = self._rerank_new_product(recommend_list, cust_code_list)
         # recommend_list = self.filter_recommend_list(recommend_list)
         logger.info(f"Item2Vec recommend completed: {len(recommend_list)} results")
         return recommend_list
+
+    def _rerank_existing_product(self, recommend_list, core_cust_list):
+        """Rerank existing-product results with core-customer boosts, then sort by score."""
+        core_set = {str(cust_code) for cust_code in (core_cust_list or [])}
+        if not core_set or not recommend_list:
+            return recommend_list
+
+        quality_score_map = self._build_quality_score_map(core_set)
+        cfg = CORE_RERANK_CONFIG["existing"]
+
+        for item in recommend_list:
+            cust_code = str(item["cust_code"])
+            model_score = float(item.get("recommend_score", 0) or 0)
+            is_core = cust_code in core_set
+            quality_score = quality_score_map.get(cust_code, 60.0)
+
+            if is_core:
+                if model_score >= cfg["low_model_threshold"]:
+                    final_score = (
+                        model_score * cfg["core_model_weight"]
+                        + quality_score * cfg["core_quality_weight"]
+                        + cfg["core_boost"]
+                    )
+                else:
+                    final_score = (
+                        model_score * cfg["low_model_weight"]
+                        + quality_score * cfg["low_quality_weight"]
+                        + cfg["low_core_boost"]
+                    )
+            else:
+                final_score = model_score * cfg["normal_model_weight"]
+
+            item["model_score"] = model_score
+            item["is_core_cust"] = is_core
+            item["core_quality_score"] = quality_score if is_core else None
+            item["recommend_score"] = min(float(final_score), 100.0)
+
+        recommend_list.sort(key=lambda x: x["recommend_score"], reverse=True)
+        logger.info(f"Core boost rerank completed for existing product: core_count={len(core_set)}")
+        return recommend_list
+
+    def _rerank_new_product(self, recommend_list, core_cust_list):
+        """Rerank Item2Vec cold-start results with core-customer boosts, then sort by score."""
+        core_set = {str(cust_code) for cust_code in (core_cust_list or [])}
+        if not core_set or not recommend_list:
+            return recommend_list
+
+        quality_score_map = self._build_quality_score_map(core_set)
+        cfg = CORE_RERANK_CONFIG["new"]
+
+        for item in recommend_list:
+            cust_code = str(item["cust_code"])
+            model_score = float(item.get("recommend_score", 0) or 0)
+            is_core = cust_code in core_set
+            quality_score = quality_score_map.get(cust_code, 60.0)
+
+            if is_core:
+                final_score = (
+                    model_score * cfg["core_model_weight"]
+                    + quality_score * cfg["core_quality_weight"]
+                    + cfg["core_boost"]
+                )
+            else:
+                final_score = model_score * cfg["normal_model_weight"]
+
+            item["item2vec_score"] = model_score
+            item["is_core_cust"] = is_core
+            item["core_quality_score"] = quality_score if is_core else None
+            item["recommend_score"] = min(float(final_score), 100.0)
+
+        recommend_list.sort(key=lambda x: x["recommend_score"], reverse=True)
+        logger.info(f"Core boost rerank completed for new product: core_count={len(core_set)}")
+        return recommend_list
+
+    def _build_quality_score_map(self, cust_list):
+        """Build a 0-100 business-quality score for candidate customers."""
+        if not cust_list:
+            return {}
+
+        unique_cust_list = list(dict.fromkeys(str(cust_code) for cust_code in cust_list))
+        cust_data = self._dao.get_cust_by_ids(self._city_uuid, unique_cust_list)
+        if cust_data.empty:
+            return {cust_code: 60.0 for cust_code in unique_cust_list}
+
+        score_map = {}
+        for _, row in cust_data.iterrows():
+            cust_code = row.get("cust_code")
+            if pd.isna(cust_code):
+                continue
+
+            score_map[str(cust_code)] = self._calculate_core_quality_score(row)
+
+        for cust_code in unique_cust_list:
+            score_map.setdefault(cust_code, 60.0)
+        return score_map
+
+    def _calculate_core_quality_score(self, row):
+        """Calculate a 0-100 quality score using only fields defined in CustConfig."""
+        field_scores = [
+            ("terminal_star_name", {
+                "五星终端": 100,
+                "四星终端": 90,
+                "三星终端": 80,
+                "二星终端": 70,
+                "一星终端": 60,
+                "其他": 50,
+                "无": 40,
+            }, 0.18),
+            ("cooperate_codename", {
+                "好": 90,
+                "较好": 75,
+                "一般": 60,
+            }, 0.14),
+            ("store_appearance_name", {
+                "好": 90,
+                "较好": 75,
+                "一般": 60,
+                "差": 40,
+            }, 0.12),
+            ("is_modern_terminalname", {
+                "是": 85,
+                "否": 55,
+            }, 0.10),
+            ("modern_terminal_name", {
+                "直营终端": 95,
+                "合作终端": 90,
+                "加盟终端": 85,
+                "一般现代终端": 75,
+                "普通终端": 60,
+                "无法识别": 50,
+            }, 0.08),
+            ("cooperate_type_name", {
+                "品牌加盟": 90,
+                "冠名加盟": 85,
+                "无": 55,
+            }, 0.08),
+            ("creditclass_name", {
+                "AAA": 95,
+                "AA": 90,
+                "A": 85,
+                "C": 60,
+                "D": 45,
+            }, 0.10),
+            ("counter_status_name", {
+                "有": 80,
+                "计划中": 65,
+                "无": 50,
+            }, 0.05),
+            ("counter_put_type_name", {
+                "独立陈列": 85,
+                "混杂陈列": 70,
+                "无陈列": 50,
+            }, 0.05),
+            ("back_counter_status_name", {
+                "有": 80,
+                "计划中": 65,
+                "无": 50,
+            }, 0.04),
+            ("back_counter_put_type_name", {
+                "独立陈列": 85,
+                "混杂陈列": 70,
+                "无陈列": 50,
+            }, 0.03),
+            ("back_counter_has_show_name", {
+                "有": 80,
+                "无": 50,
+            }, 0.03),
+        ]
+
+        weighted_score = 0.0
+        total_weight = 0.0
+
+        for field, score_map, weight in field_scores:
+            score = self._score_by_config_value(row, field, score_map)
+            if score is None:
+                continue
+            weighted_score += score * weight
+            total_weight += weight
+
+        for field, weight in [("counter_number", 0.05), ("back_counter_number", 0.05)]:
+            score = self._counter_score(self._get_row_value(row, field))
+            if score is None:
+                continue
+            weighted_score += score * weight
+            total_weight += weight
+
+        if total_weight == 0:
+            return 60.0
+
+        return round(weighted_score / total_weight, 4)
+
+    def _score_by_config_value(self, row, field, score_map):
+        if field not in CustConfig.FEATURE_COLUMNS:
+            return None
+
+        value = self._get_row_value(row, field)
+        if pd.isna(value):
+            return None
+
+        text = str(value)
+        if text not in CustConfig.ONEHOT_CAT.get(field, []):
+            return None
+
+        return float(score_map.get(text, 60.0))
+
+    @staticmethod
+    def _get_row_value(row, field):
+        if field not in row.index:
+            return None
+        return row.get(field)
+
+    @staticmethod
+    def _counter_score(value):
+        if pd.isna(value):
+            return None
+
+        try:
+            number = float(value)
+        except (TypeError, ValueError):
+            return None
+
+        if number <= 0:
+            return 50.0
+        if number >= 4:
+            return 90.0
+        return 50.0 + number * 10.0
         
     def filter_recommend_list(self, recommend_list):
         """过滤掉已经歇业的商铺"""