趋势情报平台

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

# Run the CLI instantly
uvx trend-pulse trending

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

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

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())

# 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)

# macOS
brew install uv

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

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

{
  "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"
      }
    }
  }
}

pip install "trend-pulse[mcp]"

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

# 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

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})

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

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()

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

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

# 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

{
  "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": {}
}

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

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']}")