1. 项目概述：告别爬虫，拥抱稳定的趋势数据API
如果你曾经尝试过用Python抓取Google Trends、新闻热度或者社交媒体趋势数据，那你一定对“429 Too Many Requests”这个错误代码再熟悉不过了。半夜三更，数据管道突然中断，你不得不爬起来写一堆time.sleep()的补丁，或者更糟，开始研究昂贵的代理IP轮换方案。更让人头疼的是，像pytrends这样的经典库，因为其“爬虫”的本质，已经被官方归档，随时可能彻底失效。这种数据获取方式不仅脆弱、不稳定，还充满了法律和合规风险。
今天要聊的news-volume-mcp（以及其背后的trendsmcp生态），就是为了彻底解决这个问题而生的。它不是一个爬虫库，而是一个托管的趋势数据API服务。简单来说，你把抓取、解析、维护数据基础设施的脏活累活都交给了trendsmcp的团队，你只需要一个API Key，就能通过简单的REST调用，稳定、合规地获取到跨13个平台的标准化趋势数据。无论是做SEO关键词研究、市场进入分析、投资信号捕捉，还是内容策略制定，它都能提供一套“开箱即用”的数据解决方案。最吸引人的是，它提供了一个每月100次请求的免费额度，让你可以零成本地体验和验证。
2. 核心设计思路：为什么“托管API”是更好的选择
在深入代码之前，我们有必要先搞清楚trendsmcp这套方案的设计哲学。它解决的不仅仅是“怎么拿到数据”的问题，更是“如何长期、稳定、规模化地拿到数据”的问题。
2.1 与传统爬虫方案的彻底决裂
传统的DIY爬虫方案存在几个无法根治的顽疾：
- 对抗性与不稳定性：平台方（如Google）会不断更新反爬机制（验证码、行为分析、频率限制）。你的爬虫代码需要持续维护，是一场永无止境的“军备竞赛”。
pytrends的归档就是明证——它已经无法适应平台的变化。 - 数据质量与一致性差：自行爬取的数据往往是非结构化的HTML，解析规则复杂且易变。不同时间点爬取的数据可能因为页面布局微调而格式不一致，给后续分析带来巨大麻烦。
- 规模化成本高昂：要实现稳定、高频的数据获取，必须搭建代理IP池、设计复杂的请求调度器、处理各种异常。这不仅仅是代码工作，更是持续的运维和资金投入。
- 法律与合规风险：许多网站的服务条款明确禁止未经授权的自动化抓取。用爬虫为商业项目提供数据支撑，始终存在潜在风险。
trendsmcp的托管API模式，正是针对这些痛点设计的。它相当于一个“数据即服务”（DaaS）层。平台方（trendsmcp）负责与所有数据源进行合规对接和数据采集，进行清洗、标准化和存储，然后通过一个统一的、稳定的API接口提供给开发者。作为用户，你从“数据基础设施的建造和维护者”，变成了“数据服务的消费者”。
2.2 统一化与标准化的价值
trendsmcp支持13个数据源，从Google搜索、YouTube到Reddit、亚马逊，甚至Steam和npm。它的一个巨大优势是数据标准化。
注意：所有平台返回的“趋势值”都被归一化到0-100的同一尺度。这意味着你可以直接对比“气候变化”在Google搜索上的热度（比如85）和在TikTok上的话题度（比如45），而无需担心不同平台内部指标的巨大差异。这为跨平台趋势分析提供了前所未有的便利。
这种设计使得复杂的多源数据融合分析变得非常简单。你可以轻松地计算一个品牌在“新闻声量”、“社交媒体讨论度”和“搜索引擎关注度”上的综合指数，这是自行爬取难以实现的。
2.3 面向AI工作流的原生集成（MCP）
trendsmcp中的“MCP”指的是Model Context Protocol，这是Anthropic为Claude等AI模型设计的一种工具调用协议。这意味着news-volume-mcp包不仅可以作为普通的Python库使用，更可以无缝集成到Claude Desktop、Cursor IDE、VS Code Copilot等支持MCP的AI宿主环境中。
当作为MCP服务器运行时，AI助手（如Claude）可以直接调用“获取趋势”、“计算增长”、“查看热门”等工具函数，并将实时数据作为上下文融入对话或代码生成中。例如，你可以让AI助手“分析一下最近三个月‘人工智能监管’这个话题在新闻和学术圈（通过Wikipedia）的热度增长情况”，AI会直接调用相应的API获取数据并生成分析报告。这极大地扩展了数据驱动型AI应用的可能性。
3. 从零开始：环境配置与核心API使用详解
理论说得再多，不如上手一试。我们来看看如何从零开始，把news-volume-mcp用起来。
3.1 获取通行证：API Key申请与配置
一切始于一个API Key。访问 trendsmcp.ai ，用邮箱注册即可。免费套餐每月100次请求，对于个人探索和小型项目起步完全足够。拿到Key之后，你有两种主要的使用方式。
方式一：作为纯Python客户端（最通用）这是最直接的方式，适合大多数数据分析脚本、后台服务或Jupyter Notebook分析。
# 首先安装包 pip install news-volume-mcp在你的Python脚本或环境中，建议将API Key设置为环境变量，避免硬编码在代码中，提高安全性。
# 在终端中设置（临时） export TRENDSMCP_API_KEY='your_api_key_here' # 或者在 .env 文件中（推荐） TRENDSMCP_API_KEY=your_api_key_here# 在Python代码中使用 import os from news_volume_mcp import TrendsMcpClient, SOURCE # 从环境变量读取Key api_key = os.environ.get("TRENDSMCP_API_KEY") if not api_key: raise ValueError("请设置 TRENDSMCP_API_KEY 环境变量") # 初始化客户端 client = TrendsMcpClient(api_key=api_key)方式二：作为MCP服务器集成到AI工作流如果你想在Claude或Cursor里直接让AI调用趋势数据，需要配置MCP。
- 安装MCP命令行工具（如果尚未安装）：通常Claude Desktop会自带。
- 找到MCP配置文件：
- Claude Desktop: 配置文件通常位于
~/Library/Application Support/Claude/claude_desktop_config.json(Mac) 或%APPDATA%\Claude\claude_desktop_config.json(Windows)。 - Cursor: 在Cursor设置中搜索MCP，或编辑全局配置文件。
- Claude Desktop: 配置文件通常位于
- 编辑配置文件：在
mcpServers部分添加trendsmcp服务器。
{ "mcpServers": { "trends": { "command": "npx", "args": ["-y", "trendsmcp"], "env": { "TRENDS_API_KEY": "你的API_KEY" // 注意这里是 TRENDS_API_KEY } } } }重要提示：在MCP配置中，环境变量名是
TRENDS_API_KEY，而在Python客户端代码中，我们通常从TRENDSMCP_API_KEY读取。这是两个不同的约定，务必不要混淆。配置完成后，重启你的AI应用，你就可以在对话中直接使用趋势查询工具了。
3.2 三大核心功能实战演练
客户端初始化后，你就可以调用其三大核心方法了。我们结合具体场景来看。
场景一：追踪关键词的长期趋势（get_trends）假设你是一个内容创作者，想研究“可持续时尚”这个话题过去几年的热度变化，来决定是否要深度投入。
from news_volume_mcp import TrendsMcpClient, SOURCE import pandas as pd import matplotlib.pyplot as plt client = TrendsMcpClient(api_key=api_key) # 获取5年内的每周趋势数据（默认） # SOURCE 是一个方便的常量，默认为 'news volume'，你也可以用字符串如 'google search' series = client.get_trends(source=SOURCE, keyword="sustainable fashion") print(f"共获取到 {len(series)} 个数据点") print(series[0]) # 查看第一个数据点 # 输出类似：TrendsDataPoint(date='2026-03-28', value=72, keyword='sustainable fashion', source='news volume') # 转换为Pandas DataFrame进行后续分析（无缝支持） df = pd.DataFrame([s.__dict__ for s in series]) df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) # 简单绘制趋势图 plt.figure(figsize=(12, 6)) plt.plot(df.index, df['value'], label='News Volume Index') plt.title('"Sustainable Fashion" News Volume Trend (5 Years, Weekly)') plt.xlabel('Date') plt.ylabel('Normalized Index (0-100)') plt.grid(True, alpha=0.3) plt.legend() plt.show() # 如果你想看最近30天的日度数据，更精细地观察短期波动 daily_series = client.get_trends( source=SOURCE, keyword="sustainable fashion", data_mode="daily" # 切换为日粒度 ) print(f"最近30天日度数据点: {len(daily_series)}")场景二：量化增长表现（get_growth）市场团队想知道上一季度公司品牌词在搜索引擎上的关注度增长了多少，用于评估营销活动效果。
from news_volume_mcp import TrendsMcpClient client = TrendsMcpClient(api_key=api_key) brand_keyword = "YourBrandName" growth_report = client.get_growth( source="google search", # 指定数据源为谷歌搜索 keyword=brand_keyword, percent_growth=["3M", "6M", "1Y"] # 计算环比3个月、6个月和同比1年的增长率 ) for result in growth_report.results: print(f"过去{result.period}，关键词'{brand_keyword}'在Google搜索的热度{result.direction}了{abs(result.growth):.1f}%。") # 示例输出： # 过去3M，关键词'YourBrandName'在Google搜索的热度increased了15.2%。 # 过去6M，关键词'YourBrandName'在Google搜索的热度increased了32.8%。 # 过去1Y，关键词'YourBrandName'在Google搜索的热度increased了58.1%。percent_growth参数支持丰富的预设周期，从7D（7天）到5Y（5年），还有MTD（本月至今）、QTD（本季度至今）等，非常灵活。
场景三：发现实时热点（get_top_trends）新媒体运营需要快速捕捉当天全网热点，寻找内容创作灵感。
from news_volume_mcp import TrendsMcpClient client = TrendsMcpClient(api_key=api_key) # 获取所有平台的热门趋势（默认） all_trending = client.get_top_trends(limit=20) # 限制返回20条 print("全网实时热点TOP 20:") for idx, item in enumerate(all_trending.data, start=1): print(f"{idx:2d}. {item[1]}") # item[0]是排名，item[1]是话题 # 只查看YouTube趋势 youtube_trending = client.get_top_trends(type="YouTube", limit=10) print("\nYouTube热门搜索TOP 10:") for rank, topic in youtube_trending.data: print(f"{rank}. {topic}") # 只查看Reddit热门帖子 reddit_trending = client.get_top_trends(type="Reddit Hot Posts", limit=10) print("\nReddit热门帖子TOP 10:") for rank, topic in reddit_trending.data: print(f"{rank}. {topic}")3.3 异步并发：提升数据获取效率
当你需要同时查询多个关键词或多个平台的数据时，同步请求会变成串行等待，效率低下。news-volume-mcp原生支持异步客户端AsyncTrendsMcpClient，利用asyncio可以轻松实现并发查询。
import asyncio from news_volume_mcp import AsyncTrendsMcpClient async def compare_keywords(): client = AsyncTrendsMcpClient(api_key=api_key) keywords = ["machine learning", "deep learning", "reinforcement learning"] # 并发获取三个关键词在新闻中的趋势 tasks = [client.get_trends(source=SOURCE, keyword=kw) for kw in keywords] results = await asyncio.gather(*tasks) for kw, series in zip(keywords, results): latest_value = series[-1].value if series else None print(f"关键词 '{kw}' 最新新闻热度指数: {latest_value}") async def cross_platform_analysis(): client = AsyncTrendsMcpClient(api_key=api_key) keyword = "electric vehicle" # 并发查询同一关键词在三个不同平台的表现 google_task = client.get_trends(source="google search", keyword=keyword) youtube_task = client.get_trends(source="youtube", keyword=keyword) reddit_task = client.get_trends(source="reddit", keyword=keyword) google_series, youtube_series, reddit_series = await asyncio.gather( google_task, youtube_task, reddit_task ) # 计算各平台最新热度 data = { "Google Search": google_series[-1].value, "YouTube": youtube_series[-1].value, "Reddit": reddit_series[-1].value, } print(f"关键词 '{keyword}' 跨平台热度对比:") for platform, value in data.items(): print(f" {platform}: {value}") # 运行异步函数 asyncio.run(compare_keywords()) asyncio.run(cross_platform_analysis())4. 高级应用与集成方案
掌握了基础用法后，我们可以看看如何将它融入到更复杂的生产环境和现代技术栈中。
4.1 与数据分析栈（Pandas, Plotly）无缝结合
返回的数据点列表可以轻松转化为Pandas DataFrame，这是进行进一步统计分析、可视化或机器学习的基础。
import pandas as pd import plotly.express as px from news_volume_mcp import TrendsMcpClient client = TrendsMcpClient(api_key=api_key) # 假设我们获取多个相关关键词的数据 keywords = ["Python", "JavaScript", "Rust"] all_data = [] for kw in keywords: series = client.get_trends(source="google search", keyword=kw, data_mode="daily") df_temp = pd.DataFrame([s.__dict__ for s in series]) df_temp['keyword'] = kw all_data.append(df_temp) # 合并数据 df_combined = pd.concat(all_data, ignore_index=True) df_combined['date'] = pd.to_datetime(df_combined['date']) # 使用Plotly制作交互式图表 fig = px.line(df_combined, x='date', y='value', color='keyword', title='编程语言Google搜索热度对比 (日度)', labels={'value': '标准化热度 (0-100)', 'date': '日期'}) fig.update_layout(hovermode='x unified') fig.show() # 进行简单的描述性统计 pivot_df = df_combined.pivot(index='date', columns='keyword', values='value') print(pivot_df.describe())4.2 嵌入AI应用框架（LangChain, LlamaIndex）
对于构建AI驱动的数据分析应用，trendsmcp的数据可以很方便地作为“工具”或“上下文”接入。
在LangChain中使用：你可以将TrendsMcpClient包装成一个LangChain Tool，让大模型自主决定何时调用、查询什么。
from langchain.agents import Tool from langchain.agents import initialize_agent from langchain.llms import OpenAI # 或其他LLM from news_volume_mcp import TrendsMcpClient client = TrendsMcpClient(api_key=api_key) def get_trend_data(keyword: str, source: str = "news volume") -> str: """根据关键词和来源获取趋势数据。""" try: series = client.get_trends(source=source, keyword=keyword) # 取最近5个点作为摘要 recent = series[-5:] summary = "\n".join([f"{s.date}: {s.value}" for s in recent]) return f"关键词 '{keyword}' 在 {source} 上的近期趋势:\n{summary}" except Exception as e: return f"查询失败: {str(e)}" trend_tool = Tool( name="Trends Lookup", func=get_trend_data, description="Useful for getting historical trend volume data for a keyword from various sources like news, google search, youtube, etc." ) llm = OpenAI(temperature=0) # 初始化你的LLM agent = initialize_agent([trend_tool], llm, agent="zero-shot-react-description", verbose=True) # 现在你可以用自然语言让Agent查询趋势了 agent.run("最近一个月，'Web3'在新闻里的热度变化怎么样？")在LlamaIndex中使用：你可以将趋势数据序列化为Document，存入向量索引，实现基于趋势信息的语义检索。
from llama_index import Document, VectorStoreIndex from news_volume_mcp import TrendsMcpClient client = TrendsMcpClient(api_key=api_key) def create_trend_document(keyword, source): series = client.get_trends(source=source, keyword=keyword) # 将数据转换为文本描述 text_content = f"Trend report for '{keyword}' on {source}.\n" text_content += f"Overall trend over {len(series)} weeks.\n" text_content += f"Latest value: {series[-1].value} on {series[-1].date}.\n" text_content += f"Peak value: {max(s.value for s in series)}.\n" # 可以添加更多分析... return Document(text=text_content, metadata={"keyword": keyword, "source": source}) # 为多个关键词创建文档 docs = [] for kw in ["AI safety", "quantum computing", "climate tech"]: doc = create_trend_document(kw, SOURCE) docs.append(doc) # 构建索引 index = VectorStoreIndex.from_documents(docs) # 现在你可以进行语义查询了 query_engine = index.as_query_engine() response = query_engine.query("哪个技术话题最近在新闻中增长最快？") print(response)4.3 构建自动化数据管道与监控告警
对于生产环境，你可能需要定时获取数据、存入数据库，并在出现异常波动时触发告警。
import schedule import time import sqlite3 from datetime import datetime from news_volume_mcp import TrendsMcpClient, TrendsMcpError client = TrendsMcpClient(api_key=api_key) KEYWORDS_TO_MONITOR = ["我们的产品", "主要竞品A", "主要竞品B", "行业通用词"] def job(): """定时任务：获取数据并存储""" print(f"[{datetime.now()}] 开始执行趋势数据抓取任务...") conn = sqlite3.connect('trends_monitor.db') cursor = conn.cursor() # 确保表存在 cursor.execute(''' CREATE TABLE IF NOT EXISTS trend_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, fetch_time TIMESTAMP, keyword TEXT, source TEXT, date DATE, value INTEGER ) ''') for keyword in KEYWORDS_TO_MONITOR: try: # 获取最近一天的日度数据（用于监控） series = client.get_trends(source="google search", keyword=keyword, data_mode="daily") if series: latest_point = series[-1] cursor.execute( "INSERT INTO trend_history (fetch_time, keyword, source, date, value) VALUES (?, ?, ?, ?, ?)", (datetime.now(), keyword, "google search", latest_point.date, latest_point.value) ) print(f" 已保存 '{keyword}' 的最新数据: {latest_point.date} = {latest_point.value}") # 简单的告警逻辑：如果最新值相比前一天暴跌超过50% if len(series) >= 2: prev_value = series[-2].value curr_value = latest_point.value if prev_value > 0 and (curr_value / prev_value) < 0.5: send_alert(f"警报！关键词 '{keyword}' 的Google搜索热度24小时内暴跌超过50% ({prev_value} -> {curr_value})") except TrendsMcpError as e: print(f" 查询关键词 '{keyword}' 时出错: {e.message}") except Exception as e: print(f" 处理关键词 '{keyword}' 时发生未知错误: {e}") conn.commit() conn.close() print(f"[{datetime.now()}] 任务完成。\n") def send_alert(message): """模拟发送告警（可集成邮件、Slack、钉钉等）""" print(f"[ALERT] {message}") # 实际项目中，这里调用发送邮件或消息的API # 每天上午9点执行 schedule.every().day.at("09:00").do(job) print("趋势监控服务已启动，等待定时任务...") while True: schedule.run_pending() time.sleep(60)5. 避坑指南与最佳实践
在实际使用中，我总结了一些经验和需要注意的地方，能帮你节省不少时间，避免踩坑。
5.1 免费额度的精打细算与升级策略
免费套餐每月100次请求，听起来不少，但如果不加规划，很容易在开发测试阶段就快速耗尽。一次get_trends调用算一次请求，一次get_growth也算一次，get_top_trends同样。
实操心得：在开发和调试阶段，务必使用缓存。对于相对静态的历史趋势数据（比如过去5年的周度数据），一旦获取后，可以在本地缓存24小时或更久，避免重复调用。你可以使用
functools.lru_cache装饰器或直接写入本地文件/数据库。对于get_top_trends这种实时性要求高的，缓存时间可以设短一些（比如10分钟）。
from functools import lru_cache from datetime import datetime, timedelta import pickle import os class CachedTrendsClient: def __init__(self, api_key, cache_dir="./trends_cache"): self.client = TrendsMcpClient(api_key=api_key) self.cache_dir = cache_dir os.makedirs(cache_dir, exist_ok=True) def get_trends_cached(self, source, keyword, data_mode=None, expire_hours=24): """带文件缓存的趋势获取""" cache_key = f"{source}_{keyword}_{data_mode}.pkl" cache_path = os.path.join(self.cache_dir, cache_key) # 检查缓存是否存在且未过期 if os.path.exists(cache_path): file_mtime = datetime.fromtimestamp(os.path.getmtime(cache_path)) if datetime.now() - file_mtime < timedelta(hours=expire_hours): with open(cache_path, 'rb') as f: print(f"从缓存加载: {cache_key}") return pickle.load(f) # 缓存不存在或已过期，调用API print(f"调用API获取: {cache_key}") data = self.client.get_trends(source=source, keyword=keyword, data_mode=data_mode) # 保存到缓存 with open(cache_path, 'wb') as f: pickle.dump(data, f) return data # 使用缓存客户端 cached_client = CachedTrendsClient(api_key=api_key) # 第一次调用会请求API data1 = cached_client.get_trends_cached(source=SOURCE, keyword="test") # 24小时内的第二次调用会直接读缓存，节省额度 data2 = cached_client.get_trends_cached(source=SOURCE, keyword="test")当你的项目从原型进入生产，需要更多请求量时，再去trendsmcp.ai官网查看付费套餐。通常按需付费或月度订阅的模式，比你自己维护代理IP池和服务器要划算且省心得多。
5.2 错误处理与重试机制
虽然托管API比爬虫稳定得多，但网络波动、服务端临时问题仍有可能发生。健壮的生产代码必须有完善的错误处理。
import time from news_volume_mcp import TrendsMcpClient, TrendsMcpError def robust_trends_fetch(client, source, keyword, max_retries=3, backoff_factor=2): """带指数退避重试的稳健数据获取函数""" for attempt in range(max_retries): try: return client.get_trends(source=source, keyword=keyword) except TrendsMcpError as e: # 如果是配额不足，重试没用，直接抛出 if e.code == "rate_limited" or e.status == 429: raise Exception(f"API配额不足: {e.message}") # 如果是服务器错误（5xx），可以重试 elif 500 <= e.status < 600: wait_time = backoff_factor ** attempt print(f"服务器错误 {e.status}，第{attempt+1}次重试，等待{wait_time}秒...") time.sleep(wait_time) else: # 其他客户端错误（4xx），如参数错误，重试也没用 raise except Exception as e: # 网络超时等通用异常 wait_time = backoff_factor ** attempt print(f"请求异常: {str(e)}，第{attempt+1}次重试，等待{wait_time}秒...") time.sleep(wait_time) raise Exception(f"在{max_retries}次重试后仍未能获取数据") # 使用示例 try: data = robust_trends_fetch(client, SOURCE, "important keyword") except Exception as e: print(f"数据获取最终失败: {e}") # 这里可以触发降级逻辑，例如使用上一次成功获取的缓存数据5.3 数据解读的注意事项
trendsmcp返回的是标准化指数，不是绝对数量。一个关键词在Google搜索上的“85”和另一个关键词的“85”，代表的相对热度在同一平台内是可比的，但不能直接理解为有85次搜索。它的核心价值在于观察同一关键词随时间的变化趋势，以及不同关键词在同一时间点的相对排名。
常见误区：不要试图用这个指数来精确估算实际的搜索量或文章数。它的设计初衷是趋势分析，而非精确计量。例如，指数从50增长到100，意味着热度翻倍，但并不意味着实际数量翻倍（因为是指数化处理后的结果）。
对于跨平台比较，也要谨慎。虽然指数都归一化到0-100，但不同平台的用户基数、行为模式差异巨大。新闻热度“60”和YouTube搜索热度“60”的绝对体量可能完全不同。跨平台比较更适合看趋势方向（上升/下降）和波动幅度，而不是绝对数值的高低。
5.4 与pytrends的迁移建议
如果你有现有的pytrends代码，迁移到trendsmcp在思路上是直接的，但API不同。主要区别在于：
- 无需登录和构建payload：
trendsmcp不需要模拟Cookies或构建复杂的请求参数。 - 更清晰的错误处理：
TrendsMcpError替代了各种网络和解析异常。 - 数据格式更规范：返回的是结构化的
TrendsDataPoint对象列表或GrowthResult对象，而不是原始的字典或列表。 - 异步原生支持：直接使用
AsyncTrendsMcpClient即可，无需自己包装线程池。
迁移时，重点重构数据获取和错误处理部分，业务逻辑（如数据分析、可视化）通常可以复用。
6. 真实场景下的应用案例拆解
最后，我们通过几个具体的场景，看看如何将trendsmcp的数据能力转化为实际的业务洞察。
6.1 SEO内容策略：寻找蓝海关键词
假设你运营一个科技博客，想找一些有潜力但竞争还不算白热化的关键词。
from news_volume_mcp import TrendsMcpClient import pandas as pd client = TrendsMcpClient(api_key=api_key) def find_rising_topics(seed_topics, min_volume_threshold=30, min_growth_threshold=20): """ 从种子话题中寻找正在快速上升的潜力话题。 """ promising_topics = [] for topic in seed_topics: try: # 1. 检查当前热度，过滤掉太冷门或太热门的 series = client.get_trends(source="google search", keyword=topic, data_mode="daily") if len(series) < 7: continue avg_last_week = sum([s.value for s in series[-7:]]) / 7 if avg_last_week < min_volume_threshold: print(f"话题 '{topic}' 近期平均热度({avg_last_week:.1f})过低，跳过。") continue # 2. 计算短期增长势头（对比30天前） growth = client.get_growth( source="google search", keyword=topic, percent_growth=["30D"] ) if growth.results: growth_pct = growth.results[0].growth if growth_pct >= min_growth_threshold: promising_topics.append({ 'topic': topic, 'current_volume': avg_last_week, '30d_growth_pct': growth_pct, 'trend': 'rising' }) print(f"✅ 发现潜力话题: '{topic}'，热度{avg_last_week:.1f}，30天增长{growth_pct:.1f}%") else: print(f"话题 '{topic}' 增长缓慢({growth_pct:.1f}%)，跳过。") except Exception as e: print(f"处理话题 '{topic}' 时出错: {e}") continue # 按增长率和热度综合排序 df_results = pd.DataFrame(promising_topics) if not df_results.empty: # 一个简单的评分：增长率权重更高 df_results['score'] = df_results['30d_growth_pct'] * 0.7 + df_results['current_volume'] * 0.3 df_results = df_results.sort_values('score', ascending=False) print("\n推荐内容主题（按潜力排序）:") print(df_results[['topic', 'current_volume', '30d_growth_pct', 'score']].to_string(index=False)) return df_results # 测试一些种子话题 seed_list = ["Web3 gaming", "AI agent", "low-code platform", "quantum machine learning", "green hydrogen"] find_rising_topics(seed_list)6.2 投资研究：捕捉另类数据信号
对冲基金或投资者可以用社交媒体和新闻数据作为传统金融数据的补充。
import asyncio from news_volume_mcp import AsyncTrendsMcpClient from datetime import datetime, timedelta async def monitor_sentiment_and_buzz(company_name, ticker): """ 监控公司相关的新闻声量和情感趋势。 """ client = AsyncTrendsMcpClient(api_key=api_key) # 并发获取新闻声量和情感数据 volume_task = client.get_trends(source="news volume", keyword=company_name, data_mode="daily") sentiment_task = client.get_trends(source="news sentiment", keyword=company_name, data_mode="daily") volume_series, sentiment_series = await asyncio.gather(volume_task, sentiment_task) # 分析最近7天 recent_volume = volume_series[-7:] if len(volume_series) >= 7 else volume_series recent_sentiment = sentiment_series[-7:] if len(sentiment_series) >= 7 else sentiment_series avg_volume = sum([v.value for v in recent_volume]) / len(recent_volume) if recent_volume else 0 avg_sentiment = sum([s.value for s in recent_sentiment]) / len(recent_sentiment) if recent_sentiment else 50 # 情感指数50为中性 # 计算变化 if len(volume_series) >= 2: volume_change_pct = ((volume_series[-1].value / volume_series[-2].value) - 1) * 100 else: volume_change_pct = 0 # 生成简易信号 signal = "NEUTRAL" if avg_volume > 60 and volume_change_pct > 10 and avg_sentiment > 60: signal = "STRONG POSITIVE BUZZ" elif avg_volume > 60 and avg_sentiment < 40: signal = "HIGH NEGATIVE BUZZ" elif volume_change_pct > 20: signal = "VOLUME SPIKE" elif avg_sentiment < 30: signal = "VERY NEGATIVE SENTIMENT" report = { 'company': company_name, 'ticker': ticker, 'date': datetime.now().strftime('%Y-%m-%d'), 'avg_news_volume_7d': round(avg_volume, 1), 'avg_sentiment_7d': round(avg_sentiment, 1), # >50 正面, <50 负面 'volume_day_over_day_pct': round(volume_change_pct, 1), 'signal': signal, 'interpretation': f"过去一周，{company_name}的新闻声量指数平均为{avg_volume:.1f}，情感指数平均为{avg_sentiment:.1f}。" } return report # 监控一组公司 async def main(): companies = [("Apple", "AAPL"), ("Tesla", "TSLA"), ("Netflix", "NFLX")] tasks = [monitor_sentiment_and_buzz(name, ticker) for name, ticker in companies] results = await asyncio.gather(*tasks) for report in results: print(f"\n=== {report['company']} ({report['ticker']}) ===") print(f" 新闻声量(7日均): {report['avg_news_volume_7d']}") print(f" 情感指数(7日均): {report['avg_sentiment_7d']} (50为中性)") print(f" 声量日环比: {report['volume_day_over_day_pct']:+}%") print(f" 信号: {report['signal']}") print(f" 解读: {report['interpretation']}") asyncio.run(main())6.3 产品市场契合度验证
在考虑开发一款面向开发者的新工具时，可以验证相关技术话题的热度。
from news_volume_mcp import TrendsMcpClient import matplotlib.pyplot as plt client = TrendsMcpClient(api_key=api_key) def validate_tech_trend(primary_keyword, related_keywords): """ 验证一个技术趋势及其生态的热度。 """ print(f"分析技术趋势: {primary_keyword}") print("=" * 40) # 1. 主线技术趋势 primary_series = client.get_trends(source="google search", keyword=primary_keyword) primary_growth = client.get_growth(source="google search", keyword=primary_keyword, percent_growth=["1Y", "3M"]) print(f"1. 主线技术 '{primary_keyword}':") print(f" - 当前热度指数: {primary_series[-1].value if primary_series else 'N/A'}") for g in primary_growth.results: print(f" - 过去{g.period}增长率: {g.growth:.1f}% ({g.direction})") # 2. 相关技术/生态热度 print(f"\n2. 相关生态热度:") eco_data = [] for kw in related_keywords: try: series = client.get_trends(source="google search", keyword=kw) if series: current_val = series[-1].value # 简单计算3个月增长 if len(series) > 13: # 大约13周为3个月 three_months_ago = series[-13].value growth_3m = ((current_val / three_months_ago) - 1) * 100 if three_months_ago > 0 else 0 else: growth_3m = 0 eco_data.append((kw, current_val, growth_3m)) print(f" - {kw}: 热度{current_val}, 近3月增长{growth_3m:+.1f}%") except Exception as e: print(f" - {kw}: 查询失败 - {e}") # 3. 可视化 if eco_data: keywords, values, _ = zip(*eco_data) plt.figure(figsize=(10, 6)) bars = plt.barh(keywords, values, color='skyblue') plt.xlabel('Google搜索热度指数') plt.title(f"'{primary_keyword}' 相关技术生态热度对比") # 在条形末端添加数值 for bar, val in zip(bars, values): plt.text(val + 1, bar.get_y() + bar.get_height()/2, f'{int(val)}', va='center') plt.tight_layout() plt.show() # 4. 综合判断 print(f"\n3. 初步判断:") yearly_growth = next((g.growth for g in primary_growth.results if g.period == '1Y'), 0) if yearly_growth > 30 and any(growth > 20 for _, _, growth in eco_data): print(f" ✅ 趋势强劲。'{primary_keyword}'及其生态年增长超过30%，且有相关技术快速增长，市场关注度在上升。") elif yearly_growth > 10: print(f" ⚠️ 温和增长。'{primary_keyword}'保持增长，但生态热度不均，需进一步细分市场。") else: print(f" ❌ 增长乏力。'{primary_keyword}'年增长低于10%，可能已过峰值或尚未形成主流。") # 示例：验证“Rust”生态 validate_tech_trend( primary_keyword="Rust programming", related_keywords=["WebAssembly", "Wasm", "Actix", "Tokio", "Rust vs Go"] )经过这几个月的实际使用，我的体会是，trendsmcp这类托管API服务真正将数据分析师和开发者从“数据获取”的泥潭中解放了出来。它提供的稳定性和数据一致性，是自行爬取难以企及的。虽然需要付出一定的费用（超出免费额度后），但相比于雇佣工程师维护爬虫集群、处理代理IP和反爬虫的隐性成本，这往往是一笔更划算的交易。对于需要快速验证想法、构建数据原型，或者缺乏爬虫工程能力的小团队来说，它是一个非常高效的起点。最后一个小技巧：在规划数据获取任务时，尽量将多个关键词的查询安排在异步函数中并发执行，这不仅能提升效率，对于按次计费的套餐来说，也能在单位时间内最大化数据获取的吞吐量。
