analyze_topic_trend
Analyzes topic trends by tracking popularity changes, detecting viral spikes, predicting future hotspots, and examining lifecycle patterns to understand topic evolution over time.
Instructions
统一话题趋势分析工具 - 整合多种趋势分析模式
重要：日期范围处理 当用户使用"本周"、"最近7天"等自然语言时，请先调用 resolve_date_range 工具获取精确日期：
调用 resolve_date_range("本周") → 获取 {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"}
将返回的 date_range 传入本工具
Args: topic: 话题关键词（必需） analysis_type: 分析类型，可选值： - "trend": 热度趋势分析（追踪话题的热度变化） - "lifecycle": 生命周期分析（从出现到消失的完整周期） - "viral": 异常热度检测（识别突然爆火的话题） - "predict": 话题预测（预测未来可能的热点） date_range: 日期范围（trend和lifecycle模式），可选 - 格式: {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"} - 获取方式: 调用 resolve_date_range 工具解析自然语言日期 - 默认: 不指定时默认分析最近7天 granularity: 时间粒度（trend模式），默认"day"（仅支持 day，因为底层数据按天聚合） threshold: 热度突增倍数阈值（viral模式），默认3.0 time_window: 检测时间窗口小时数（viral模式），默认24 lookahead_hours: 预测未来小时数（predict模式），默认6 confidence_threshold: 置信度阈值（predict模式），默认0.7
Returns: JSON格式的趋势分析结果
Examples: 用户："分析AI本周的趋势" 推荐调用流程： 1. resolve_date_range("本周") → {"date_range": {"start": "2025-11-18", "end": "2025-11-26"}} 2. analyze_topic_trend(topic="AI", date_range={"start": "2025-11-18", "end": "2025-11-26"})
用户："看看特斯拉最近30天的热度"
推荐调用流程：
1. resolve_date_range("最近30天") → {"date_range": {"start": "2025-10-28", "end": "2025-11-26"}}
2. analyze_topic_trend(topic="特斯拉", analysis_type="lifecycle", date_range=...)Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| topic | Yes | ||
| analysis_type | No | trend | |
| date_range | No | ||
| granularity | No | day | |
| threshold | No | ||
| time_window | No | ||
| lookahead_hours | No | ||
| confidence_threshold | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- mcp_server/tools/analytics.py:156-243 (handler)The handler 'analyze_topic_trend_unified' in 'AnalyticsTools' class which orchestrates the different topic trend analysis modes.
def analyze_topic_trend_unified( self, topic: str, analysis_type: str = "trend", date_range: Optional[Dict[str, str]] = None, granularity: str = "day", threshold: float = 3.0, time_window: int = 24, lookahead_hours: int = 6, confidence_threshold: float = 0.7 ) -> Dict: """ 统一话题趋势分析工具 - 整合多种趋势分析模式 Args: topic: 话题关键词（必需） analysis_type: 分析类型，可选值： - "trend": 热度趋势分析（追踪话题的热度变化） - "lifecycle": 生命周期分析（从出现到消失的完整周期） - "viral": 异常热度检测（识别突然爆火的话题） - "predict": 话题预测（预测未来可能的热点） date_range: 日期范围（trend和lifecycle模式），可选 - **格式**: {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"} - **默认**: 不指定时默认分析最近7天 granularity: 时间粒度（trend模式），默认"day"（hour/day） threshold: 热度突增倍数阈值（viral模式），默认3.0 time_window: 检测时间窗口小时数（viral模式），默认24 lookahead_hours: 预测未来小时数（predict模式），默认6 confidence_threshold: 置信度阈值（predict模式），默认0.7 Returns: 趋势分析结果字典 Examples (假设今天是 2025-11-17): - 用户："分析AI最近7天的趋势" → analyze_topic_trend_unified(topic="人工智能", analysis_type="trend", date_range={"start": "2025-11-11", "end": "2025-11-17"}) - 用户："看看特斯拉本月的热度" → analyze_topic_trend_unified(topic="特斯拉", analysis_type="lifecycle", date_range={"start": "2025-11-01", "end": "2025-11-17"}) - analyze_topic_trend_unified(topic="比特币", analysis_type="viral", threshold=3.0) - analyze_topic_trend_unified(topic="ChatGPT", analysis_type="predict", lookahead_hours=6) """ try: # 参数验证 topic = validate_keyword(topic) if analysis_type not in ["trend", "lifecycle", "viral", "predict"]: raise InvalidParameterError( f"无效的分析类型: {analysis_type}", suggestion="支持的类型: trend, lifecycle, viral, predict" ) # 根据分析类型调用相应方法 if analysis_type == "trend": return self.get_topic_trend_analysis( topic=topic, date_range=date_range, granularity=granularity ) elif analysis_type == "lifecycle": return self.analyze_topic_lifecycle( topic=topic, date_range=date_range ) elif analysis_type == "viral": # viral模式不需要topic参数，使用通用检测 return self.detect_viral_topics( threshold=threshold, time_window=time_window ) else: # predict # predict模式不需要topic参数，使用通用预测 return self.predict_trending_topics( lookahead_hours=lookahead_hours, confidence_threshold=confidence_threshold ) except MCPError as e: return { "success": False, "error": e.to_dict() } except Exception as e: return { "success": False, "error": { "code": "INTERNAL_ERROR", "message": str(e) } } - mcp_server/tools/analytics.py:244-400 (handler)The actual implementation logic for 'get_topic_trend_analysis' which performs the day-by-day trend tracking and statistics calculation.
def get_topic_trend_analysis( self, topic: str, date_range: Optional[Dict[str, str]] = None, granularity: str = "day" ) -> Dict: """ 热度趋势分析 - 追踪特定话题的热度变化趋势 Args: topic: 话题关键词 date_range: 日期范围（可选） - **格式**: {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"} - **默认**: 不指定时默认分析最近7天 granularity: 时间粒度，仅支持 day（天） Returns: 趋势分析结果字典 Examples: 用户询问示例： - "帮我分析一下'人工智能'这个话题最近一周的热度趋势" - "查看'比特币'过去一周的热度变化" - "看看'iPhone'最近7天的趋势如何" - "分析'特斯拉'最近一个月的热度趋势" - "查看'ChatGPT'2024年12月的趋势变化" 代码调用示例： >>> tools = AnalyticsTools() >>> # 分析7天趋势（假设今天是 2025-11-17） >>> result = tools.get_topic_trend_analysis( ... topic="人工智能", ... date_range={"start": "2025-11-11", "end": "2025-11-17"}, ... granularity="day" ... ) >>> # 分析历史月份趋势 >>> result = tools.get_topic_trend_analysis( ... topic="特斯拉", ... date_range={"start": "2024-12-01", "end": "2024-12-31"}, ... granularity="day" ... ) >>> print(result['trend_data']) """ try: # 验证参数 topic = validate_keyword(topic) # 验证粒度参数（只支持day） if granularity != "day": from ..utils.errors import InvalidParameterError raise InvalidParameterError( f"不支持的粒度参数: {granularity}", suggestion="当前仅支持 'day' 粒度，因为底层数据按天聚合" ) # 处理日期范围（不指定时默认最近7天） if date_range: from ..utils.validators import validate_date_range date_range_tuple = validate_date_range(date_range) start_date, end_date = date_range_tuple else: # 默认最近7天 end_date = datetime.now() start_date = end_date - timedelta(days=6) # 收集趋势数据 trend_data = [] current_date = start_date while current_date <= end_date: try: all_titles, _, _ = self.data_service.parser.read_all_titles_for_date( date=current_date ) # 统计该时间点的话题出现次数 count = 0 matched_titles = [] for _, titles in all_titles.items(): for title in titles.keys(): if topic.lower() in title.lower(): count += 1 matched_titles.append(title) trend_data.append({ "date": current_date.strftime("%Y-%m-%d"), "count": count, "sample_titles": matched_titles[:3] # 只保留前3个样本 }) except DataNotFoundError: trend_data.append({ "date": current_date.strftime("%Y-%m-%d"), "count": 0, "sample_titles": [] }) # 按天增加时间 current_date += timedelta(days=1) # 计算趋势指标 counts = [item["count"] for item in trend_data] total_days = (end_date - start_date).days + 1 if len(counts) >= 2: # 计算涨跌幅度 first_non_zero = next((c for c in counts if c > 0), 0) last_count = counts[-1] if first_non_zero > 0: change_rate = ((last_count - first_non_zero) / first_non_zero) * 100 else: change_rate = 0 # 找到峰值时间 max_count = max(counts) peak_index = counts.index(max_count) peak_time = trend_data[peak_index]["date"] else: change_rate = 0 peak_time = None max_count = 0 return { "success": True, "topic": topic, "date_range": { "start": start_date.strftime("%Y-%m-%d"), "end": end_date.strftime("%Y-%m-%d"), "total_days": total_days }, "granularity": granularity, "trend_data": trend_data, "statistics": { "total_mentions": sum(counts), "average_mentions": round(sum(counts) / len(counts), 2) if counts else 0, "peak_count": max_count, "peak_time": peak_time, "change_rate": round(change_rate, 2) }, "trend_direction": "上升" if change_rate > 10 else "下降" if change_rate < -10 else "稳定" } except MCPError as e: return { "success": False, "error": e.to_dict() } except Exception as e: return { "success": False, "error": { "code": "INTERNAL_ERROR", "message": str(e) } } - mcp_server/server.py:228-286 (registration)The MCP tool function 'analyze_topic_trend' in 'server.py' that serves as the entry point and registers the tool.
async def analyze_topic_trend( topic: str, analysis_type: str = "trend", date_range: Optional[Dict[str, str]] = None, granularity: str = "day", threshold: float = 3.0, time_window: int = 24, lookahead_hours: int = 6, confidence_threshold: float = 0.7 ) -> str: """ 统一话题趋势分析工具 - 整合多种趋势分析模式 **重要：日期范围处理** 当用户使用"本周"、"最近7天"等自然语言时，请先调用 resolve_date_range 工具获取精确日期： 1. 调用 resolve_date_range("本周") → 获取 {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"} 2. 将返回的 date_range 传入本工具 Args: topic: 话题关键词（必需） analysis_type: 分析类型，可选值： - "trend": 热度趋势分析（追踪话题的热度变化） - "lifecycle": 生命周期分析（从出现到消失的完整周期） - "viral": 异常热度检测（识别突然爆火的话题） - "predict": 话题预测（预测未来可能的热点） date_range: 日期范围（trend和lifecycle模式），可选 - **格式**: {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"} - **获取方式**: 调用 resolve_date_range 工具解析自然语言日期 - **默认**: 不指定时默认分析最近7天 granularity: 时间粒度（trend模式），默认"day"（仅支持 day，因为底层数据按天聚合） threshold: 热度突增倍数阈值（viral模式），默认3.0 time_window: 检测时间窗口小时数（viral模式），默认24 lookahead_hours: 预测未来小时数（predict模式），默认6 confidence_threshold: 置信度阈值（predict模式），默认0.7 Returns: JSON格式的趋势分析结果 Examples: 用户："分析AI本周的趋势" 推荐调用流程： 1. resolve_date_range("本周") → {"date_range": {"start": "2025-11-18", "end": "2025-11-26"}} 2. analyze_topic_trend(topic="AI", date_range={"start": "2025-11-18", "end": "2025-11-26"}) 用户："看看特斯拉最近30天的热度" 推荐调用流程： 1. resolve_date_range("最近30天") → {"date_range": {"start": "2025-10-28", "end": "2025-11-26"}} 2. analyze_topic_trend(topic="特斯拉", analysis_type="lifecycle", date_range=...) """ tools = _get_tools() result = tools['analytics'].analyze_topic_trend_unified( topic=topic, analysis_type=analysis_type, date_range=date_range, granularity=granularity, threshold=threshold, time_window=time_window, lookahead_hours=lookahead_hours, confidence_threshold=confidence_threshold