46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
"""基金推荐候选过滤与排序。"""
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from typing import Iterable
|
|
|
|
from common.common_const import MEMORY_INFO_TYPE_OPINION
|
|
from common.suitability import check_suitability
|
|
|
|
|
|
def _opinion_bonuses(memories: Iterable[dict]) -> dict[str, float]:
|
|
bonuses: dict[str, float] = defaultdict(float)
|
|
for memory in memories:
|
|
if memory.get("info_type") != MEMORY_INFO_TYPE_OPINION:
|
|
continue
|
|
product_code = memory.get("product_code")
|
|
if product_code:
|
|
bonuses[str(product_code)] += float(memory.get("score", 0.0) or 0.0)
|
|
return bonuses
|
|
|
|
|
|
def recommend_candidates(
|
|
customer_risk: str,
|
|
candidates: Iterable[dict],
|
|
*,
|
|
memories: Iterable[dict] = (),
|
|
) -> list[dict]:
|
|
"""硬过滤不适当产品，再按业绩和主观观点排序。"""
|
|
bonuses = _opinion_bonuses(memories)
|
|
result = []
|
|
for candidate in candidates:
|
|
suitability = check_suitability(customer_risk, candidate.get("risk_level", ""))
|
|
if not suitability.ok:
|
|
continue
|
|
item = dict(candidate)
|
|
base_score = float(item.get("performance_score", 0.0) or 0.0)
|
|
item["recommendation_score"] = base_score + bonuses.get(
|
|
str(item.get("product_code")), 0.0
|
|
)
|
|
result.append(item)
|
|
return sorted(
|
|
result,
|
|
key=lambda item: item["recommendation_score"],
|
|
reverse=True,
|
|
)
|