智能推荐：小红书AI种草

#!/usr/bin/env python3"""小红书AI种草演示：KOL智能匹配 + 热点追踪 + 多触点归因""" import numpy as npfrom typing import List, Dict, Tuple  class KOLMatcher:    """KOL/KOC智能匹配器"""        def __init__(self, kol_data: List[dict]):        """        kol_data: 每个KOL包含粉丝数、互动率、粉丝画像、内容标签、报价等信息        """        self.kols = kol_data        self.embeddings = self._build_embeddings()        def _build_embeddings(self) -> np.ndarray:        """构建KOL特征向量（简化版）"""        embeddings = []        for kol in self.kols:            feat = [                np.log10(kol.get('followers', 1000)),  # 粉丝数（对数缩放）                kol.get('engagement_rate', 0.03),      # 互动率                kol.get('match_score', 0.5),            # 人群匹配度                kol.get('content_quality', 0.6),        # 内容质量分                kol.get('authenticity', 0.7),            # 真实度（反作弊）            ]            embeddings.append(feat)        return np.array(embeddings)        def find_best_kols(self, brand_profile: dict, top_n: int = 10,                        budget: float = None) -> List[Tuple[dict, float]]:        """找到最匹配的KOL                综合考虑匹配度、性价比、预期效果        """        scores = []        for i, kol in enumerate(self.kols):            # 综合评分：匹配度 + 互动质量 + 真实度 - 价格因素            match_score = kol.get('match_score', 0.5)            quality = kol.get('engagement_rate', 0.03) / 0.05  # 归一化            authenticity = kol.get('authenticity', 0.7)            price = kol.get('price', 1000)                        # 性价比 = 预期效果 / 价格            expected_effect = match_score * quality * authenticity * np.log10(kol.get('followers', 1000))            cost_effectiveness = expected_effect / (price / 1000)                        # 综合得分            total_score = 0.4 * match_score + 0.3 * quality + 0.2 * authenticity + 0.1 * cost_effectiveness            scores.append((kol, total_score))                # 按得分排序        scores.sort(key=lambda x: x[1], reverse=True)                if budget:            # 预算约束下的贪心选择            selected = []            remaining = budget            for kol, score in scores:                price = kol.get('price', 1000)                if price <= remaining:                    selected.append((kol, score))                    remaining -= price            return selected                return scores[:top_n]  class HotSpotTracker:    """热点话题追踪器"""        def __init__(self):        self.topics = {}  # topic -> {heat, growth_rate, category, trend}        def add_topic_data(self, topic: str, current_heat: float,                         history_heats: List[float]):        """添加话题数据，计算热度趋势"""        # 计算增长率        if len(history_heats) >= 2:            growth_rate = (current_heat - history_heats[-1]) / (history_heats[-1] + 1e-8)        else:            growth_rate = 0.0                # 趋势预测（简单线性外推）        if len(history_heats) >= 3:            recent = history_heats[-3:] + [current_heat]            x = np.arange(len(recent))            slope, _ = np.polyfit(x, recent, 1)            trend = 'rising' if slope > 0 else 'declining'        else:            trend = 'stable'                self.topics[topic] = {            'heat': current_heat,            'growth_rate': growth_rate,            'trend': trend,            'history': history_heats + [current_heat]        }        def get_hot_topics(self, min_heat: float = 1000,                        sort_by: str = 'heat') -> List[Tuple[str, dict]]:        """获取热门话题列表"""        hot = [(t, d) for t, d in self.topics.items() if d['heat'] > min_heat]        if sort_by == 'heat':            hot.sort(key=lambda x: x[1]['heat'], reverse=True)        elif sort_by == 'growth':            hot.sort(key=lambda x: x[1]['growth_rate'], reverse=True)        return hot        def predict_peak_time(self, topic: str) -> float:        """预测热点峰值时间（简化版）"""        if topic not in self.topics:            return -1        data = self.topics[topic]['history']        if len(data) < 3:            return -1        # 简化：如果增长率开始下降，说明接近峰值        growth_rates = []        for i in range(1, len(data)):            g = (data[i] - data[i-1]) / (data[i-1] + 1e-8)            growth_rates.append(g)                if len(growth_rates) >= 2 and growth_rates[-1] < growth_rates[-2]:            # 增长率下降，估计还需2-3个周期到峰值            return len(data) + 2        else:            return -1  # 无法判断  class AttributionModel:    """多触点归因模型"""        def __init__(self):        pass        def last_click_attribution(self, touchpoints: List[dict]) -> Dict[str, float]:        """末次点击归因：最后一个触点全功"""        if not touchpoints:            return {}        last = touchpoints[-1]        return {last['channel']: 1.0}        def first_click_attribution(self, touchpoints: List[dict]) -> Dict[str, float]:        """首次点击归因：第一个触点全功"""        if not touchpoints:            return {}        first = touchpoints[0]        return {first['channel']: 1.0}        def linear_attribution(self, touchpoints: List[dict]) -> Dict[str, float]:        """线性归因：各触点平均分配功劳"""        if not touchpoints:            return {}        n = len(touchpoints)        credit = 1.0 / n        result = {}        for tp in touchpoints:            ch = tp['channel']            result[ch] = result.get(ch, 0) + credit        return result        def time_decay_attribution(self, touchpoints: List[dict],                                 half_life: float = 3.0) -> Dict[str, float]:        """时间衰减归因：越接近转化的触点权重越高"""        if not touchpoints:            return {}        n = len(touchpoints)        weights = []        for i in range(n):            # 距离转化的位置：最后一个距离为0            distance = n - 1 - i            weight = 0.5 ** (distance / half_life)  # 指数衰减            weights.append(weight)                total_weight = sum(weights)        result = {}        for i, tp in enumerate(touchpoints):            ch = tp['channel']            credit = weights[i] / total_weight            result[ch] = result.get(ch, 0) + credit        return result        def markov_attribution(self, touchpoints_list: List[List[dict]],                           conversion_paths: List[List[dict]]) -> Dict[str, float]:        """简化版马尔可夫归因：计算各触点的移除效应                移除某个触点后转化率下降越多，说明该触点越重要        """        # 计算基准转化率        n_total = len(touchpoints_list)        n_conv = len(conversion_paths)        base_cvr = n_conv / n_total if n_total > 0 else 0                # 获取所有渠道        all_channels = set()        for path in touchpoints_list:            for tp in path:                all_channels.add(tp['channel'])                attribution = {}        for ch in all_channels:            # 移除该渠道后，有多少转化路径会中断            removed_conversions = 0            for path in conversion_paths:                channels_in_path = set(tp['channel'] for tp in path)                if ch in channels_in_path:                    # 简化：如果路径中只有这个渠道，移除后转化消失                    if len(channels_in_path) == 1:                        removed_conversions += 1                        # 移除效应 = （基准转化率 - 移除后转化率） / 基准转化率            new_cvr = (n_conv - removed_conversions) / n_total if n_total > 0 else 0            effect = (base_cvr - new_cvr) / base_cvr if base_cvr > 0 else 0            attribution[ch] = effect                # 归一化        total_effect = sum(attribution.values())        if total_effect > 0:            for ch in attribution:                attribution[ch] /= total_effect                return attribution  def demo_ai_zhongcao():    """AI种草综合演示"""    print("=" * 60)    print("小红书AI种草系统演示")    print("=" * 60)        # 1. KOL智能匹配演示    print("\n--- 1. KOL智能匹配 ---")    # 模拟10个KOL数据    np.random.seed(42)    kol_data = []    for i in range(10):        kol_data.append({            'id': f'KOL_{i+1}',            'name': f'博主{i+1}',            'followers': int(np.random.uniform(5000, 500000)),            'engagement_rate': np.random.uniform(0.01, 0.08),            'match_score': np.random.uniform(0.3, 0.95),            'content_quality': np.random.uniform(0.4, 0.9),            'authenticity': np.random.uniform(0.5, 0.95),            'price': int(np.random.uniform(500, 10000))        })        matcher = KOLMatcher(kol_data)    brand_profile = {'category': '护肤', 'target_age': '25-35'}    best_kols = matcher.find_best_kols(brand_profile, top_n=5)        print("Top 5 匹配KOL:")    for kol, score in best_kols:        print(f"  {kol['name']} (粉丝{kol['followers']:,}, 互动率{kol['engagement_rate']:.1%}, "              f"匹配度{kol['match_score']:.2f}, 报价￥{kol['price']:,}) -> 综合得分{score:.3f}")        # 2. 热点追踪演示    print("\n--- 2. 热点话题追踪 ---")    tracker = HotSpotTracker()    topics_data = [        ('早C晚A', 8500, [1000, 2000, 4000, 6000]),        ('以油养肤', 3200, [500, 800, 1500, 2500]),        ('精简护肤', 5600, [3000, 3500, 4200, 5000]),        ('敏皮修护', 2100, [4000, 3500, 2800, 2400]),  # 下降趋势        ('抗老精华', 7800, [5000, 5500, 6500, 7200]),    ]        for topic, heat, history in topics_data:        tracker.add_topic_data(topic, heat, history)        hot_topics = tracker.get_hot_topics(min_heat=1000, sort_by='growth')    print("热门话题（按增长率排序）:")    for topic, data in hot_topics:        print(f"  {topic}: 热度{data['heat']:,}, 增长{data['growth_rate']:.1%}, 趋势{data['trend']}")        # 3. 归因分析演示    print("\n--- 3. 种草转化归因分析 ---")    model = AttributionModel()        # 一个用户的转化路径    user_path = [        {'channel': '信息流广告', 'time': 'day 1'},        {'channel': 'KOL笔记', 'time': 'day 3'},        {'channel': '搜索', 'time': 'day 5'},        {'channel': 'KOC笔记', 'time': 'day 7'},    ]        print("用户转化路径：" + " → ".join(tp['channel'] for tp in user_path))    print()        # 不同归因模型对比    print("不同归因模型的功劳分配：")        last = model.last_click_attribution(user_path)    print(f"  末次归因: {last}")        first = model.first_click_attribution(user_path)    print(f"  首次归因: {first}")        linear = model.linear_attribution(user_path)    print(f"  线性归因: {{k: {round(v,3)} for k,v in linear.items()}}")        decay = model.time_decay_attribution(user_path, half_life=2)    print(f"  时间衰减: {{k: {round(v,3)} for k,v in decay.items()}}")        print("\n观察：不同归因模型对各渠道的功劳分配差异很大，选择合适的归因模型很重要")  if __name__ == "__main__":    demo_ai_zhongcao()