TikTok 比任何其他平台更快地推动产品趋势——产品可以在几天内从未知变成售空。及早发现趋势可以为电子商务品牌和代发货商提供至关重要的优势。本教程使用 Scavio TikTok API 构建趋势检测管道，用于监控与产品相关的主题标签、跟踪浏览速度并在新兴产品达到顶峰之前识别它们。该系统分析主题标签增长率、创作者采用模式和参与率，以生成趋势分数。每次检测运行需要 5-10 个积分 (0.025-0.05 美元)。
前置条件
- 已安装 Python 3.9+
- 请求已安装库
- 来自 scavio.dev 的 Scavio API 密钥
- 要监控的产品类别或利基市场
操作指南
步骤 1: 定义产品监控类别
设置要监控的产品类别和相关搜索词。每个类别都有多个关键词来捕捉同一趋势的不同角度。
Python
categories = {
'skincare': {
'keywords': ['skincare routine', 'skincare haul', 'skincare viral',
'skincare hack', 'skincare tiktok made me buy'],
'hashtags': ['skincareroutine', 'skincarehaul', 'tiktokmademebuyit']
},
'tech_gadgets': {
'keywords': ['tech gadget viral', 'amazon finds tech',
'tiktok gadget review', 'must have tech 2026'],
'hashtags': ['techfinds', 'amazontechfinds', 'gadgetreview']
},
'kitchen': {
'keywords': ['kitchen gadget tiktok', 'cooking hack viral',
'amazon kitchen finds', 'kitchen must have'],
'hashtags': ['kitchenhacks', 'kitchenfinds', 'cookingtiktok']
}
}
total_searches = sum(len(c['keywords']) + len(c['hashtags']) for c in categories.values())
print(f'{len(categories)} categories, {total_searches} total API calls')
print(f'Estimated cost per run: ${total_searches * 0.005:.2f}')步骤 2: 搜索热门产品视频
对于每个关键词，搜索 TikTok 并收集视频数据。跟踪播放、点赞和创建时间以计算速度指标。
Python
import requests, os, time
API_KEY = os.environ['SCAVIO_API_KEY']
TIKTOK_URL = 'https://api.scavio.dev/api/v1/tiktok'
def search_tiktok(keyword: str, count: int = 20) -> list:
resp = requests.post(f'{TIKTOK_URL}/search/videos',
headers={'Authorization': f'Bearer {API_KEY}',
'Content-Type': 'application/json'},
json={'keyword': keyword, 'count': count, 'cursor': 0})
resp.raise_for_status()
videos = resp.json().get('data', {}).get('videos', [])
return [{
'id': v.get('id', ''),
'author': v.get('author', {}).get('uniqueId', ''),
'desc': v.get('desc', ''),
'plays': v.get('stats', {}).get('playCount', 0),
'likes': v.get('stats', {}).get('diggCount', 0),
'comments': v.get('stats', {}).get('commentCount', 0),
'shares': v.get('stats', {}).get('shareCount', 0),
'create_time': v.get('createTime', 0),
'keyword': keyword
} for v in videos]步骤 3: 从视频描述中提取产品名称
解析视频描述以识别特定的产品名称和品牌。视频通常会提及确切的产品名称、品牌或亚马逊 ASIN。
Python
import re
from collections import Counter
def extract_products(videos: list) -> Counter:
"""Extract product mentions from video descriptions."""
product_counter = Counter()
# Common patterns: "Product Name" in caps, @brand mentions, #productname
for v in videos:
desc = v.get('desc', '')
# Hashtag products (e.g., #CeraVe #StanleyCup)
hashtags = re.findall(r'#(\w+)', desc)
for tag in hashtags:
# Filter out generic tags
if len(tag) > 3 and tag.lower() not in {
'fyp', 'foryou', 'viral', 'trending', 'tiktok',
'skincare', 'review', 'haul', 'musthave'}:
product_counter[tag.lower()] += 1
# Brand mentions (capitalized words that look like brand names)
brands = re.findall(r'\b([A-Z][a-zA-Z]+(?:\s[A-Z][a-zA-Z]+)?)\b', desc)
for brand in brands:
if len(brand) > 3 and brand.lower() not in {'this', 'that', 'the', 'with'}:
product_counter[brand.lower()] += 1
return product_counter
# Example:
products = extract_products(search_tiktok('skincare viral'))
for product, count in products.most_common(10):
print(f' {product}: {count} mentions')步骤 4: 计算趋势速度分数
根据提及次数增长的速度对每个产品进行评分。高观看速度加上许多独特的创作者标志着一种趋势产品。
Python
import statistics
from datetime import datetime
def calculate_trend_score(product: str, videos: list) -> dict:
relevant = [v for v in videos if product in v.get('desc', '').lower()]
if len(relevant) < 2:
return {'product': product, 'trend_score': 0, 'reason': 'Too few videos'}
total_plays = sum(v['plays'] for v in relevant)
total_engagement = sum(v['likes'] + v['comments'] + v['shares'] for v in relevant)
unique_creators = len(set(v['author'] for v in relevant))
# Recency: weight recent videos more heavily
now = time.time()
recency_scores = []
for v in relevant:
age_days = (now - v['create_time']) / 86400 if v['create_time'] > 0 else 30
recency_scores.append(max(0, 1 - (age_days / 30))) # 0-1, 1=today
avg_recency = statistics.mean(recency_scores)
# Composite trend score
play_score = min(total_plays / 100000, 40) # up to 40 points
creator_score = min(unique_creators * 5, 30) # up to 30 points
recency_score = avg_recency * 30 # up to 30 points
total_score = round(play_score + creator_score + recency_score, 1)
return {
'product': product,
'trend_score': total_score,
'total_plays': total_plays,
'unique_creators': unique_creators,
'video_count': len(relevant),
'avg_recency': round(avg_recency, 2)
}步骤 5: 运行完整的趋势检测管道
将所有步骤合并到一个管道中，该管道可扫描类别、提取产品、对趋势进行评分并输出排名报告。
Python
def detect_trends(categories: dict) -> list:
all_videos = []
credits_used = 0
for category, config in categories.items():
print(f'Scanning {category}...')
for keyword in config['keywords']:
videos = search_tiktok(keyword, count=20)
all_videos.extend(videos)
credits_used += 1
time.sleep(0.3)
# Extract and score products
product_counts = extract_products(all_videos)
trends = []
for product, count in product_counts.most_common(20):
if count >= 3: # minimum mention threshold
score = calculate_trend_score(product, all_videos)
if score['trend_score'] > 10:
trends.append(score)
trends.sort(key=lambda t: t['trend_score'], reverse=True)
print(f'\nDetected {len(trends)} trending products')
print(f'Credits used: {credits_used} (${credits_used * 0.005:.2f})')
for t in trends[:10]:
emoji_bar = '#' * int(t['trend_score'] / 5)
print(f' [{t["trend_score"]:5.1f}] {t["product"]}: '
f'{t["total_plays"]:,} plays, {t["unique_creators"]} creators '
f'{emoji_bar}')
return trends
trends = detect_trends(categories)Python 示例
Python
import os, requests, time, re
from collections import Counter
API_KEY = os.environ['SCAVIO_API_KEY']
TT = 'https://api.scavio.dev/api/v1/tiktok'
def search(keyword, count=20):
resp = requests.post(f'{TT}/search/videos',
headers={'Authorization': f'Bearer {API_KEY}', 'Content-Type': 'application/json'},
json={'keyword': keyword, 'count': count, 'cursor': 0})
return resp.json().get('data', {}).get('videos', [])
def detect_trends(keywords):
all_videos = []
for kw in keywords:
all_videos.extend(search(kw))
time.sleep(0.3)
products = Counter()
for v in all_videos:
for tag in re.findall(r'#(\w{4,})', v.get('desc', '')):
if tag.lower() not in {'fyp', 'foryou', 'viral', 'trending'}:
products[tag.lower()] += 1
print(f'Scanned {len(all_videos)} videos, found {len(products)} products')
for product, count in products.most_common(10):
print(f' {product}: {count} mentions')
return products
detect_trends(['skincare viral', 'amazon finds 2026', 'tiktok made me buy'])JavaScript 示例
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY;
const TT = 'https://api.scavio.dev/api/v1/tiktok';
async function searchTikTok(keyword) {
const resp = await fetch(`${TT}/search/videos`, {
method: 'POST',
headers: { 'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json' },
body: JSON.stringify({ keyword, count: 20, cursor: 0 })
});
return (await resp.json()).data?.videos || [];
}
async function detectTrends(keywords) {
const products = {};
for (const kw of keywords) {
const videos = await searchTikTok(kw);
videos.forEach(v => {
const tags = (v.desc || '').match(/#(\w{4,})/g) || [];
tags.forEach(t => {
const tag = t.slice(1).toLowerCase();
if (!['fyp', 'foryou', 'viral'].includes(tag)) {
products[tag] = (products[tag] || 0) + 1;
}
});
});
}
Object.entries(products).sort((a, b) => b[1] - a[1]).slice(0, 10)
.forEach(([p, c]) => console.log(` ${p}: ${c} mentions`));
}
detectTrends(['skincare viral', 'amazon finds 2026']);预期输出
JSON
Scanning skincare...
Scanning tech_gadgets...
Scanning kitchen...
Detected 8 trending products
Credits used: 12 ($0.06)
[ 78.5] cerave: 1,234,000 plays, 15 creators ################
[ 65.2] stanleycup: 890,000 plays, 12 creators #############
[ 52.1] dysonairwrap: 567,000 plays, 8 creators ###########
[ 45.8] theordinary: 445,000 plays, 9 creators #########
[ 38.4] airfryer: 334,000 plays, 7 creators ########
[ 31.0] laneige: 234,000 plays, 6 creators ######
[ 24.5] owala: 178,000 plays, 5 creators #####
[ 18.2] hexclad: 123,000 plays, 4 creators ####