Trend Pulse

趋势脉冲

   

代理趋势智能平台——37个来源、插件系统、矢量搜索、生命周期预测、6个代理内容工厂、web仪表板和29个工具MCP服务器。

用作 Python库, CLI工具, MCP服务器 适用于Claude Code/AI代理，或独立代理 网络仪表盘.

单线MCP设置（零安装）：

{ "mcpServers": { "trend-pulse": { "command": "uvx", "args": ["--from", "trend-pulse[mcp]", "trend-pulse-server"], "type": "stdio" } } }

粘贴到 .mcp.json 你完了。需要 紫外线 (brew install uv 或 curl -LsSf https://astral.sh/uv/install.sh | sh).

来源

内置资源（20）

所有内置源都是免费的，需要 零身份验证:

插件源（17）

插件源代码位于 src/trend_pulse/plugins/sources/ 并在启动时自动发现：

安装

使用uvx进行零安装（推荐）

uvx 直接运行Python包——无需安装，无需venv，无需设置：

# Run the CLI instantly
uvx trend-pulse trending

# Run the MCP server
uvx --from "trend-pulse[mcp]" trend-pulse-server

uvx Python相当于 npx它附带了 紫外线 --安装uv curl -LsSf https://astral.sh/uv/install.sh | sh

pip 安装

pip install trend-pulse                  # core (httpx + aiosqlite)
pip install "trend-pulse[mcp]"           # MCP server
pip install "trend-pulse[dashboard]"     # Streamlit + FastAPI
pip install "trend-pulse[llm]"           # Claude API for hybrid scoring
pip install "trend-pulse[all]"           # everything

快速开始

命令行界面

# What's trending right now? (all 37 sources, merged ranking)
trend-pulse trending

# Taiwan trends from Google + Hacker News
trend-pulse trending --sources google_trends,hackernews --geo TW

# Fetch + save snapshot to history DB
trend-pulse trending --save --count 10

# Take a full snapshot (all sources, auto-saves)
trend-pulse snapshot

# Query historical trends for a keyword
trend-pulse history "Claude" --days 7

# Search across sources
trend-pulse search "AI agent"

# List available sources (built-in + plugins)
trend-pulse sources

python

import asyncio
from trend_pulse.aggregator import TrendAggregator

async def main():
    agg = TrendAggregator()

    # All sources, merged ranking
    result = await agg.trending(geo="TW", count=10)
    for item in result["merged_top"]:
        print(f"[{item['source']}] {item['keyword']} ({item.get('traffic', '')})")

    # With snapshot saving + velocity enrichment
    result = await agg.trending(count=5, save=True)
    for item in result["merged_top"]:
        print(f"{item['keyword']} — {item['direction']} (velocity: {item['velocity']})")

    # Query history
    history = await agg.history("Claude", days=7)
    for record in history["records"]:
        print(f"  {record['timestamp']}: score={record['score']}")

    # Search
    result = await agg.search("Claude AI")
    for item in result["merged_top"][:5]:
        print(f"{item['keyword']} - {item['score']:.0f}")

asyncio.run(main())

单一来源

import asyncio
from trend_pulse.sources import HackerNewsSource

async def main():
    hn = HackerNewsSource()
    items = await hn.fetch_trending(count=5)
    for item in items:
        print(f"{item.keyword} ({item.traffic})")

    # HN also supports search
    results = await hn.search("Python")
    for item in results[:3]:
        print(f"  {item.keyword}")

asyncio.run(main())

第1-3阶段情报API

# Lifecycle prediction
from trend_pulse.core.intelligence.lifecycle import predict_lifecycle, LifecycleStage
stage = predict_lifecycle(current_score=75, history=[{"score": s} for s in [20, 35, 50, 65]])
# -> LifecycleStage.EMERGING

# Trend clustering
from trend_pulse.core.intelligence.clusters import cluster_trends
clusters = await cluster_trends(items, threshold=0.25)

# 6-agent content workflow
from trend_pulse.core.agents.workflow import run_content_workflow
state = await run_content_workflow(
    trends=items,
    platforms=["threads", "x"],
    brand_voice="casual",
    topic="AI tools",
)
content = state["final_content"]  # {"threads": "...", "x": "..."}

# Hybrid scoring (heuristic + optional Claude API)
from trend_pulse.core.scoring.hybrid import score_content
result = await score_content("Your post content", "threads")
print(result.total, result.grade, result.mode)  # 78.5, B+, heuristic

# Vector similarity search
from trend_pulse.core.vector.simple import SimpleVectorStore
store = SimpleVectorStore()
await store.upsert(items)
similar = await store.search_similar("artificial intelligence", k=5)

MCP服务器（用于克劳德代码/AI代理）

步骤1:安装uv（如果你没有）

# macOS
brew install uv

# Linux / WSL
curl -LsSf https://astral.sh/uv/install.sh | sh

步骤2：添加到 .mcp.json

创建或编辑 .mcp.json 在项目根目录中（或 ~/.claude/.mcp.json 全球）：

{
  "mcpServers": {
    "trend-pulse": {
      "command": "uvx",
      "args": ["--from", "trend-pulse[mcp]", "trend-pulse-server"],
      "type": "stdio"
    }
  }
}

就是这样。不 pip install，没有venv，没有Python版本管理。uvx在第一次运行时自动下载并缓存包。

步骤3（可选）：启用浏览器渲染

这 render_page 该工具使用Cloudflare浏览器渲染来获取JS繁重的页面。如果您想要此功能，请添加您的Cloudflare凭据：

{
  "mcpServers": {
    "trend-pulse": {
      "command": "uvx",
      "args": ["--from", "trend-pulse[mcp]", "trend-pulse-server"],
      "type": "stdio",
      "env": {
        "CF_ACCOUNT_ID": "your-cloudflare-account-id",
        "CF_API_TOKEN": "your-cloudflare-api-token"
      }
    }
  }
}

从以下地址获取这些 Cloudflare仪表板 → 工人和页面→ 概述。如果你不需要它，跳过这一步——所有其他28个工具都可以在没有任何凭据的情况下工作。

替代方案：pip安装

如果您更喜欢传统安装而不是uvx：

pip install "trend-pulse[mcp]"

{
  "mcpServers": {
    "trend-pulse": {
      "command": "trend-pulse-server",
      "type": "stdio"
    }
  }
}

可用工具（29）

趋势数据（5）：

情报（5）：

内容指南（5）：

所有内容指南工具返回 结构化指南 --法学硕士做所有的判断和创造性工作。

代理内容（8）：

操作（5）：

浏览器（1，可选-需要Cloudflare凭据）：

仪表板和REST API

# Start Streamlit dashboard
streamlit run src/trend_pulse/dashboard/app.py

# Start FastAPI REST API
uvicorn trend_pulse.dashboard.api:app --port 8000

# Docker Compose (all services)
docker compose up

Docker撰写服务：

通知

from trend_pulse.notifications.channels import DiscordWebhook
notifier = DiscordWebhook(webhook_url="https://discord.com/api/webhooks/...")
await notifier.send("Trending now", {"keyword": "Claude AI", "score": 95})

CLI 参考

trend-pulse trending [--sources SRC] [--geo CODE] [--count N] [--save]
trend-pulse search QUERY [--sources SRC] [--geo CODE]
trend-pulse history KEYWORD [--days N] [--source SRC]
trend-pulse snapshot [--sources SRC] [--geo CODE] [--count N]
trend-pulse sources

--sources：以逗号分隔的源ID。

内置： google_trends, hackernews, mastodon, bluesky, wikipedia, github, pypi, google_news, lobsters, devto, npm, reddit, coingecko, dockerhub, stackoverflow, arxiv, producthunt, lemmy, dcard, ptt

插件： weibo, youtube_trending, threads, x_trending, tiktok_trending, line_today, mobile01, bahamut, ettoday, yahoo_tw, udn, coinmarketcap, dexscreener, pinterest, linkedin_trending, indie_hackers, xiaohongshu

--geo：ISO国家代码（例如。， TW, US, JP, DE).

• 谷歌趋势/谷歌新闻：按国家筛选
• 维基百科：选择语言版本
• GitHub：被视为语言过滤器（例如。， python)
• 其他来源：忽略（全球数据）

--save：将结果保存到本地SQLite数据库(~/.trend-pulse/history.db)用于速度跟踪。

插件系统

将文件放入 src/trend_pulse/plugins/sources/ --无需注册。这 PluginRegistry 自动发现导出的所有模块 register() 功能。

from trend_pulse.plugins.base import PluginSource
from trend_pulse.sources.base import TrendItem

class MyPluginSource(PluginSource):
    name = "my_plugin"
    description = "My custom plugin source"
    category = "global"   # global, tw, dev, crypto, social, professional
    frequency = "daily"

    async def fetch_trending(self, geo="", count=20) -> list[TrendItem]:
        return [TrendItem(keyword="...", score=80.0, source=self.name)]

def register():
    return MyPluginSource()

插件类别： global, tw, dev, crypto, social, professional

为了遗产 TrendSource 接口（内置源代码），请参阅 docs/custom-sources.md 和 示例/custom_rss_source.py.

速度和方向

当历史数据可用时，每个趋势项包括：

{
  "keyword": "Claude AI",
  "score": 92,
  "direction": "rising",
  "velocity": 15.3,
  "previous_score": 45.0,
  "source": "hackernews"
}

## 历史数据库

快照存储在SQLite中 `~/.trend-pulse/history.db` （用覆盖 `TREND_PULSE_DB` 有人）。

Save snapshots over time

trend-pulse trending --save --count 5

... wait some time ...

trend-pulse trending --save --count 5

Query history

trend-pulse history "Claude" --days 7 trend-pulse history "React" --days 30 --source npm

## 输出格式

所有命令都返回具有统一结构的JSON：

{ "timestamp": "2026-03-12T05:04:12Z", "geo": "TW", "sources_ok": ["google_trends", "hackernews", "reddit"], "sources_error": {"wikipedia": "429 rate limited"}, "merged_top": [ { "keyword": "Temporal: fixing time in JavaScript", "score": 100, "source": "hackernews", "url": "https://...", "traffic": "579 points", "category": "tech", "direction": "rising", "velocity": 12.5, "previous_score": 60.0, "metadata": {} } ], "by_source": {} }

每个项目都有一个标准化 `score` （0–100）用于跨源比较。

## 速率限制

|来源|限制|注释|
|--------|-------|-------|
|谷歌趋势RSS |无限制| RSS提要，无速率限制|
|黑客新闻|无限| Firebase+Algolia，非常慷慨|
|Mastodon |需要300次/5分钟|每个实例|
|蓝天|3000请求/5分钟|公共API|
|维基百科| 100个要求/秒|非常慷慨|
|GitHub |合理| HTML抓取，不要滥用|
|PyPI统计| 1个要求/天/终点|数据每天更新|
|谷歌新闻|无限| RSS订阅源|
|Lobste.rs |无限| JSON API|
|dev.to|无限制|公共API|
|npm|无限制|公共API|
|Reddit | 60请求/分钟|需要User-Agent标头|
|CoinGecko | 10–30请求/分钟|公共API|
|Docker Hub | 100 req/5分钟|公共API|
|堆栈溢出|300请求/天|不带API密钥|
|ArXiv |无限| RSS/API|
|产品搜寻|合理|公共API|
|莱米|无限|公共API（莱米.world）|
|Dcard |合理|公共API|
|PTT |合理|网络抓取，不要滥用|
|微博|合理|网络抓取|
|YouTube趋势|合理|公共RSS/抓取|
|线程|合理|网络抓取|
|X/Twitter|合理|可选的不记名代币提高了限额|
|TikTok趋势|合理|网络刮刮|
|LINE Today TW |合理|网络抓取|
|Mobile01 |合理|网络抓取|
|Bahamut |合理|网络抓取|
|ETtoday |合理|网络抓取|
|雅虎TW |合理|网络抓取|
|UDN |合理|网络抓取|
|CoinMarketCap |合理|公共页面抓取|
|DexScreener|无限制|公共API|
|Pinterest |合理|网络抓取|
|LinkedIn趋势|合理|网络抓取|
|独立黑客|合理|网络抓取|
|小红书|合理|网络刮刮|

## 内容指南工具

MCP工具为创建针对Meta的7项排名专利进行优化的病毒式内容提供了结构化的指南。 **法学硕士做所有的判断和创造性工作** --工具只提供框架和标准。

### 线程算法惩罚预检查

评分前，将内容与4进行核对 **官方处罚** 图案从 [线程创建者页面](https://creators.instagram.com/threads)。任何违规行为都会阻止发布：

|惩罚|什么线索惩罚|行动|
|---------|----------------------|--------|
| `no_clickbait` |Hook承诺了一些身体无法提供的东西|将钩子与内容对齐|
| `no_engagement_bait` |明确要求点赞/转发/关注|替换为自然CTA|
| `no_contest_violation` |竞赛/赠品需要参与才能参加|删除或取消|
| `original_content` |从IG/FB交叉发布，没有原始角度|用原始视角重写|

### 基于专利的评分（5个维度）

|尺寸|重量|基于|
|-----------|--------|----------|
|钩力|25%|EdgeRank重量+仙女座|
|参与度触发器|25%|故事查看器元组+亲爱的Algo|
|对话持久性|20%|线程72小时窗口|
|速度潜力| 15%|仙女座实时|
|格式分数|15%|多模态索引|

分数： **S** (90+), **A.** (80+), **B** (70+), **C** (55+), **D** (\=70，通话耐久性>=55。

### 混合评分

`score_content_hybrid` （MCP）和 `score_content` （Python API）默认运行启发式评分。如果 `ANTHROPIC_API_KEY` 设置和 `pip install "trend-pulse[llm]"` 安装了一个Claude API判断层来补充启发式得分。这 `mode` 结果中的字段指示运行的路径(`heuristic` 或 `llm`).

### 审查清单（15项）

|严重性|计数|示例|
|----------|-------|---------|
| **关键的** |7 |字符限制、总分、对话持久性、4种算法惩罚|
| **警告** |5 |挂钩效果、CTA呈现、媒体增强、音调真实性、主题标签|
| **信息** |3|问题存在、格式可读性、回复策略|

判决： **通过** =所有关键检查均通过， **失败** =任何关键检查都失败。

### Threads创作者见解

关键信号来自 [Threads创建者官方页面](https://creators.instagram.com/threads):

- **发布频率**：每周2-5次（较高=每篇帖子的浏览量更多）
- **回复**：约占Threads视图的50%——积极回复评论
- **媒体**：文本+媒体明显优于纯文本
- **幽默**：正式记录为在Threads上表现良好
- **主题标签**：带有表情符号的多词标签增加了覆盖范围
- **15种内容类型**：文本、图像、视频、旋转木马、投票、GIF、链接附件、文本附件、剧透媒体、剧透文本、重影帖子、引用帖子、回复控制、主题标签、ALT_TEXT

### 工作流程：MCP指导LLM

1. get_content_brief() → LLM gets writing guide with hook examples & strategies
2. LLM creates content → Original text based on brief
3. get_scoring_guide() → LLM runs penalty pre-checks, then self-scores 5 dimensions
4. LLM revises → Iterate until score >= 70
5. get_review_checklist() → LLM checks 15 items (7 critical / 5 warning / 3 info)
6. get_platform_specs() → LLM adapts for each platform

— or — run_content_workflow() → 6-agent factory runs steps 1–6 autonomously

### Python（内容指南工具）

from trend_pulse.content.briefing import ( get_content_brief, get_scoring_guide, get_review_checklist, get_reel_guide, ) from trend_pulse.content.adapter import get_platform_specs

Get writing brief — hook examples, patent strategies, scoring dimensions

brief = get_content_brief("AI tools", "debate", "threads", lang="en") print(f"Topic: {brief['topic']}, Char limit: {brief['char_limit']}") print(f"Hook examples: {len(brief['hook_examples'])}") print(f"Patent strategies: {len(brief['patent_strategies'])}")

Get scoring guide — penalty pre-checks + 5-dimension evaluation framework

guide = get_scoring_guide("en") print(f"Penalty pre-checks: {len(guide['penalty_precheck']['penalties'])}") for name, dim in guide["dimensions"].items(): print(f"{name}: weight={dim['weight']}, {dim['description']}") print(f"Grades: {list(guide['grade_thresholds'].keys())}")

Get review checklist — 15 structured quality checks (7 critical)

checklist = get_review_checklist("threads", "en") for item in checklist["checklist"]: print(f"[{item['severity']}] {item['check']}") print(f"Quality gate: overall >= {checklist['quality_gate']['min_overall']}")

Get platform specs — char limits, algo priority, best times

specs = get_platform_specs("", "en") # all platforms for name, spec in specs.items(): print(f"{name}: {spec['max_chars']} chars, best at {spec['best_times']}")

Get reel guide — scene structure for video scripts

guide = get_reel_guide("educational", 30, "en") for scene in guide["scene_structure"]: print(f"[{scene['type']}] {scene['duration_seconds']}s — {scene['purpose']}")

### 内容类型

`opinion`, `story`, `debate`, `howto`, `list`, `question`, `news`, `meme`

### 平台限制

|平台|字符限制|
|----------|-----------|
|线程|500|
|Instagram | 2200|
|脸书| 63206|
|X（推特）|280|
|尝试|2200|
|LinkedIn | 3000|
|YouTube短片| 5000|
|小红书|1000|

## 安全

- `render_page` 包括一个SSRF防护：方案白名单+私有IP块。请求 `file://`, `127.x`, `10.x`, `192.168.x`，并且拒绝类似的范围。
- 混合评分的LLM裁判使用快速注射隔离来防止趋势内容影响评分指令。

## 需求

- Python 3.10+
- `httpx` （HTTP客户端）
- `aiosqlite` （历史存储）
- 可选： `mcp[cli]` 对于MCP服务器(`pip install "trend-pulse[mcp]"`)
- 可选： `streamlit`, `fastapi`, `uvicorn` 用于仪表板(`pip install "trend-pulse[dashboard]"`)
- 可选： `anthropic` 用于混合LLM评分(`pip install "trend-pulse[llm]"`)
- 可选： `trendspyg`, `pytrends` 增强谷歌趋势

## 许可证

麻省理工学院

## 学分

由...建造 [克劳德世界](https://claude-world.com) --面向开发人员的Claude Code社区。