social-media-trends-research

社交媒体趋势研究

概述

使用三款免费工具进行程序化趋势研究：

• pytrends：Google Trends 数据（增长速度、搜索量、相关查询）
• yars：无需 API 密钥即可抓取 Reddit 数据
• Perplexity MCP：Twitter/TikTok/Web 趋势（通过 Claude 内置的 MCP）

此技能提供可执行的趋势研究代码。可搭配 content-marketing-social-listening 制定策略，并使用 perplexity-search 进行深度查询。

快速设置

# Install dependencies (one-time)
pip install pytrends requests --break-system-packages

无需 API 密钥。抓取 Reddit 数据使用公开的 .json 端点。

---

工具 1：pytrends（Google Trends）

提供的内容

• 按国家/地区查看实时热门搜索
• 关键词的随时间变化的关注度
• 相关查询（上升趋势 = 增长速度指标）
• 按地区查看关注度
• 相关主题

基本用法

from pytrends.request import TrendReq
import time

# Initialize (no API key needed)
pytrends = TrendReq(hl='en-US', tz=330)  # tz=330 for India (IST)

# Get real-time trending searches
trending = pytrends.trending_searches(pn='india')
print(trending.head(20))

研究所在细分领域的关键词

from pytrends.request import TrendReq
import time

pytrends = TrendReq(hl='en-US', tz=330)

# Define your niche keywords (max 5 per request)
keywords = ['heart health', 'cardiology', 'cholesterol']

# Build payload
pytrends.build_payload(keywords, timeframe='now 7-d', geo='IN')

# Get interest over time
interest = pytrends.interest_over_time()
print(interest)

# CRITICAL: Wait between requests to avoid rate limiting
time.sleep(3)

# Get related queries (THIS IS GOLD - shows rising topics)
related = pytrends.related_queries()
for kw in keywords:
    print(f"\n=== Rising queries for '{kw}' ===")
    rising = related[kw]['rising']
    if rising is not None:
        print(rising.head(10))

查找爆红/突破性话题

from pytrends.request import TrendReq
import time

pytrends = TrendReq(hl='en-US', tz=330)

def find_breakout_topics(keyword, geo=''):
    """Find topics with explosive growth (potential viral content)"""
    pytrends.build_payload([keyword], timeframe='today 3-m', geo=geo)
    time.sleep(3)  # Rate limiting
    
    related = pytrends.related_queries()
    rising = related[keyword]['rising']
    
    if rising is not None:
        # Filter for breakout topics (marked as "Breakout" or very high %)
        breakouts = rising[rising['value'] >= 1000]  # 1000%+ growth
        return breakouts
    return None

# Example usage
breakouts = find_breakout_topics('heart health', geo='IN')
print(breakouts)

pytrends 的速率限制规则

import time

# SAFE: 1 request per 3-5 seconds for casual use
time.sleep(5)

# BULK RESEARCH: 1 request per 60 seconds
time.sleep(60)

# If you get rate limited (429 error): Wait 60-120 seconds, then continue
# If persistent issues: Wait 4-6 hours before resuming

常用时间范围

时间范围	使用场景
'now 1-H'	过去 1 小时（实时峰值）
'now 4-H'	过去 4 小时
'now 1-d'	过去 24 小时
'now 7-d'	过去 7 天（最适合追踪趋势）
'today 1-m'	过去 30 天
'today 3-m'	过去 90 天（增长速度分析）
'today 12-m'	过去一年（季节性模式）

---

工具 2：Reddit（无需 API 密钥——公开 JSON 端点）

提供的内容

• 在 Reddit 上搜索任意关键词
• 获取 subreddit 中的热门、最高赞或上升帖子
• 帖子互动数据（赞数、评论数）
• 无需身份验证

基本用法

import requests
import time

headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}

# Search Reddit for your niche
url = "https://www.reddit.com/search.json?q=heart+health&limit=10&sort=relevance&t=week"
response = requests.get(url, headers=headers, timeout=10)
data = response.json()

# Display results
for child in data.get('data', {}).get('children', []):
    post = child.get('data', {})
    print(f"Title: {post.get('title')}")
    print(f"Subreddit: r/{post.get('subreddit')}")
    print(f"Score: {post.get('score')}")
    print("---")

获取指定 subreddit 中的热门帖子

import requests
import time

headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}

# Define subreddits relevant to your niche
subreddits = ['cardiology', 'health', 'medicine']

for sub in subreddits:
    print(f"\n=== Hot in r/{sub} ===")
    try:
        url = f"https://www.reddit.com/r/{sub}/hot.json?limit=10"
        response = requests.get(url, headers=headers, timeout=10)
        data = response.json()
        
        for child in data.get('data', {}).get('children', [])[:5]:
            post = child.get('data', {})
            print(f"- [{post.get('score')}] {post.get('title')[:60]}...")
    except Exception as e:
        print(f"Error: {e}")
    
    time.sleep(3)  # Rate limiting between requests

使用捆绑的 Reddit 抓取工具

声明档/reddit_scraper.py 中包含一个辅助类：

from scripts.reddit_scraper import SimpleRedditScraper

scraper = SimpleRedditScraper()

# Search
results = scraper.search("heart health tips", limit=20)
for post in results['posts']:
    print(f"[{post['score']}] r/{post['subreddit']}: {post['title']}")

# Get subreddit hot posts
results = scraper.get_subreddit("health", sort="hot", limit=10)
for post in results['posts']:
    print(f"[{post['score']}] {post['title']}")

Reddit 的速率限制规则

import time

# SAFE: 1 request per 2-3 seconds
time.sleep(3)

# If you get 429 errors: Wait 5-10 minutes
# Never do more than 60 requests per hour

---

工具 3：Perplexity MCP（Twitter/TikTok/Web）

对于无法直接抓取的平台，请使用 Claude 内置的 Perplexity MCP。

趋势研究查询模板

Twitter/X 趋势：

"What are the most discussed [YOUR NICHE] topics on Twitter/X this week? 
Include specific examples of viral tweets and their engagement."

TikTok 趋势（可在印度使用）：

"What [YOUR NICHE] content is trending on TikTok right now? 
Include hashtags, view counts, and content formats that are working."

YouTube 趋势：

"What [YOUR NICHE] videos are getting the most views on YouTube this week? 
Include channel names, view counts, and video topics."

LinkedIn 专业领域：

"What [YOUR NICHE] topics are professionals discussing on LinkedIn this week? 
Include examples of high-engagement posts."

通用爆红内容：

"What [YOUR NICHE] content has gone viral across social media in the past 7 days? 
Include platform, format, and why it resonated."

将 Perplexity 与 perplexity-search 技能配合使用

如果已安装 perplexity-search 技能：

python scripts/perplexity_search.py \
  "What cardiology topics are trending on Twitter and TikTok this week? Include specific viral posts and hashtags." \
  --model sonar-pro

---

综合研究工作流

完整的趋势研究函数

from pytrends.request import TrendReq
import requests
import time
import json
from datetime import datetime

class TrendResearcher:
    def __init__(self):
        self.pytrends = TrendReq(hl='en-US', tz=330)
        self.reddit_headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
    
    def _reddit_request(self, url):
        """Make a Reddit API request."""
        try:
            response = requests.get(url, headers=self.reddit_headers, timeout=10)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            return {'error': str(e)}
    
    def research_niche(self, keywords, subreddits=None, geo='IN'):
        """
        Complete trend research for a niche.
        
        Args:
            keywords: List of keywords (max 5)
            subreddits: List of subreddit names to monitor
            geo: Geographic region code
        
        Returns:
            Dictionary with all research data
        """
        results = {
            'timestamp': datetime.now().isoformat(),
            'keywords': keywords,
            'google_trends': {},
            'reddit': {},
            'recommendations': []
        }
        
        # 1. Google Trends - Interest Over Time
        print("📊 Fetching Google Trends data...")
        try:
            self.pytrends.build_payload(keywords[:5], timeframe='now 7-d', geo=geo)
            results['google_trends']['interest'] = self.pytrends.interest_over_time().to_dict()
            time.sleep(5)
            
            # Related queries (rising topics)
            related = self.pytrends.related_queries()
            results['google_trends']['rising_queries'] = {}
            for kw in keywords[:5]:
                rising = related[kw]['rising']
                if rising is not None:
                    results['google_trends']['rising_queries'][kw] = rising.head(10).to_dict()
            time.sleep(5)
        except Exception as e:
            results['google_trends']['error'] = str(e)
        
        # 2. Reddit Research
        print("👽 Fetching Reddit discussions...")
        if subreddits:
            for sub in subreddits[:5]:
                try:
                    url = f"https://www.reddit.com/r/{sub}/hot.json?limit=10"
                    data = self._reddit_request(url)
                    posts = []
                    for child in data.get('data', {}).get('children', [])[:5]:
                        post = child.get('data', {})
                        posts.append({
                            'title': post.get('title', ''),
                            'score': post.get('score', 0),
                            'comments': post.get('num_comments', 0)
                        })
                    results['reddit'][sub] = posts
                    time.sleep(3)
                except Exception as e:
                    results['reddit'][sub] = {'error': str(e)}
        
        # 3. Keyword search on Reddit
        print("🔍 Searching Reddit for keywords...")
        for kw in keywords[:3]:
            try:
                url = f"https://www.reddit.com/search.json?q={kw}&limit=10&sort=relevance&t=week"
                data = self._reddit_request(url)
                posts = []
                for child in data.get('data', {}).get('children', [])[:5]:
                    post = child.get('data', {})
                    posts.append({
                        'title': post.get('title', ''),
                        'subreddit': post.get('subreddit', ''),
                        'score': post.get('score', 0),
                        'comments': post.get('num_comments', 0)
                    })
                results['reddit'][f'search_{kw}'] = posts
                time.sleep(3)
            except Exception as e:
                results['reddit'][f'search_{kw}'] = {'error': str(e)}
        
        # 4. Generate recommendations
        results['recommendations'] = self._generate_recommendations(results)
        
        return results
    
    def _generate_recommendations(self, data):
        """Generate content recommendations from research data"""
        recommendations = []
        
        # From rising queries
        rising = data.get('google_trends', {}).get('rising_queries', {})
        for kw, queries in rising.items():
            if isinstance(queries, dict) and 'query' in queries:
                for query in list(queries['query'].values())[:3]:
                    recommendations.append({
                        'source': 'Google Trends',
                        'topic': query,
                        'reason': f"Rising search term related to '{kw}'"
                    })
        
        # From Reddit hot posts
        for sub, posts in data.get('reddit', {}).items():
            if isinstance(posts, list):
                for post in posts[:2]:
                    if post.get('score', 0) > 50:
                        recommendations.append({
                            'source': f'Reddit r/{sub}',
                            'topic': post.get('title', ''),
                            'reason': f"High engagement ({post.get('score')} upvotes)"
                        })
        
        return recommendations

# Usage Example
if __name__ == "__main__":
    researcher = TrendResearcher()
    
    results = researcher.research_niche(
        keywords=['heart health', 'cardiology', 'cholesterol'],
        subreddits=['cardiology', 'health', 'medicine'],
        geo='IN'
    )
    
    # Save results
    with open('trend_research.json', 'w') as f:
        json.dump(results, f, indent=2, default=str)
    
    # Print recommendations
    print("\n🎯 CONTENT RECOMMENDATIONS:")
    for rec in results['recommendations']:
        print(f"- [{rec['source']}] {rec['topic']}")
        print(f"  Why: {rec['reason']}")

---

命令速查

每日趋势检查（5 分钟）

from pytrends.request import TrendReq
import requests
import time

# Quick Google Trends check
pytrends = TrendReq(hl='en-US', tz=330)
pytrends.build_payload(['your keyword'], timeframe='now 1-d')
print(pytrends.related_queries()['your keyword']['rising'])

time.sleep(5)

# Quick Reddit check  
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
url = "https://www.reddit.com/search.json?q=your+keyword&limit=10&t=day"
response = requests.get(url, headers=headers, timeout=10)
data = response.json()
for child in data.get('data', {}).get('children', [])[:5]:
    post = child.get('data', {})
    print(f"[{post.get('score')}] {post.get('title')}")

每周深入研究

# Use the TrendResearcher class above with:
# - 5 core keywords
# - 5 relevant subreddits
# - 90-day timeframe for velocity analysis

# Then use Perplexity MCP for:
# - Twitter trends in your niche
# - TikTok viral content
# - YouTube trending videos
# - LinkedIn discussions

---

与写作技能集成

研究完成后，将发现交给写作技能：

1. Run trend research (this skill)
2. Identify top 3-5 opportunities
3. Use content-marketing-social-listening for strategy
4. Use cardiology-content-repurposer or similar for content creation
5. Use authentic-voice for final polish

---

故障排除

pytrends 问题

错误	解决方案
429 Too Many Requests	等待 60 秒，然后延长休眠时间
结果为空	检查关键词是否有搜索量
连接错误	检查网络，5 分钟后重试

Reddit 问题

错误	解决方案
429 Rate Limited	等待 10 分钟
未找到 subreddit	检查 subreddit 名称拼写
结果为空	subreddit 可能是私有的或已被隔离
连接超时	延长超时时间，检查网络

---

最佳实践

1. 始终进行速率限制：在请求之间休眠
2. 分批研究：每周进行深入研究，不要持续轮询
3. 保存结果：在本地缓存研究数据
4. 交叉验证：在多个平台上验证趋势
5. 快速行动：爆红窗口很短（24-72 小时）

---

平台覆盖范围汇总

平台	工具	费用	风险
Google Trends	pytrends	免费	极低
Reddit	requests（公开 JSON）	免费	低
Twitter/X	Perplexity MCP	免费*	无
TikTok	Perplexity MCP	免费*	无
YouTube	Perplexity MCP	免费*	无
LinkedIn	Perplexity MCP	免费*	无

*使用 Claude 内置的 MCP，或在使用 perplexity-search 技能时使用 OpenRouter 额度

---

内置资源

• scripts/trend_research.py：用于完整趋势研究的主命令行工具
• scripts/reddit_scraper.py：简单的 Reddit 抓取器类（无需 API 密钥）