推荐系统案例：特征平台

# -*- coding: utf-8 -*-"""简化版特征平台:离线加工 + 在线服务 + 缓存加速"""import random  class FeatureDefinition:    """特征定义:描述特征名称、类型与负责人"""    def __init__(self, name, ftype, owner, description=""):        self.name = name        self.ftype = ftype        self.owner = owner        self.description = description  class FeatureRegistry:    """元数据中心:统一注册特征,支持查询与复用"""    def __init__(self):        self.features_map = {}     def register(self, feature):        if feature.name in self.features_map:            raise ValueError("feature %s already exists" % feature.name)        self.features_map[feature.name] = feature     def get(self, name):        return self.features_map.get(name)  class FeatureStore:    """特征存储:key=(user,item),value=特征向量"""    def __init__(self):        self.data_map = {}     def write(self, key, value):        self.data_map[key] = value     def read(self, key):        return self.data_map.get(key)  class OnlineServing:    """在线特征服务:带LRU缓存,自动统计命中率"""    def __init__(self, store, cache_size=128):        self.store = store        self.cache = {}        self.cache_order = []        self.cache_size = cache_size        self.hit_count = 0        self.request_count = 0     def get_features(self, user, item):        key = (user, item)        self.request_count += 1        if key in self.cache:            self.hit_count += 1            return self.cache[key]        value = self.store.read(key)        if value is not None:            self._put_cache(key, value)        return value     def _put_cache(self, key, value):        if key in self.cache:            self.cache_order.remove(key)        self.cache[key] = value        self.cache_order.append(key)        if len(self.cache_order) > self.cache_size:            old_key = self.cache_order.pop(0)            del self.cache[old_key]     def hit_rate(self):        if self.request_count == 0:            return 0.0        return self.hit_count / float(self.request_count)  def offline_compute(logs):    """离线批量计算:把原始行为日志加工成特征向量"""    store = FeatureStore()    for log in logs:        key = (log["user"], log["item"])        hist = store.read(key)        if hist is None:            hist = [0, 0, 0.0]        hist[0] += 1                                  # 点击次数累加        hist[1] += log.get("view_time", 0)             # 累计浏览时长        hist[2] = max(hist[2], log.get("price", 0.0))  # 历史最高价格        store.write(key, hist)    return store  def main():    random.seed(42)    registry = FeatureRegistry()    registry.register(FeatureDefinition("click_cnt", "int", "rec_team"))    registry.register(FeatureDefinition("max_price", "float", "rec_team"))     users = [1001, 1002, 1003]    items = [2001, 2002, 2003]    logs = []    for _ in range(3000):        logs.append({            "user": random.choice(users),            "item": random.choice(items),            "view_time": random.randint(1, 60),            "price": random.uniform(1.0, 10.0),        })     # 离线加工:所有日志变成特征向量    store = offline_compute(logs)     # 在线服务:带缓存的批量特征查询    serving = OnlineServing(store)    requests = [(random.choice(users), random.choice(items)) for _ in range(500)]     sample = serving.get_features(requests[0][0], requests[0][1])    print("sample feature:", sample)     for user, item in requests:        serving.get_features(user, item)     print("cache hit rate: %.2f" % serving.hit_rate())    print("registered features:", len(registry.features_map))  if __name__ == "__main__":    main()