DEV Community

Vhub Systems
Vhub Systems

Posted on

How to Track TikTok Shop Trending Products Before Your Competitors Find Them

#python #webdev #ecommerce #tutorial

How to Track TikTok Shop Trending Products Before Your Competitors Find Them

TikTok Shop's "Movers & Shakers" section shows products gaining velocity — items that are trending up before they hit mainstream. The data changes every 24 hours. Here's how to monitor it systematically to catch winning products early.

Why TikTok Shop Trending Data Matters

  • Products trending on TikTok Shop often appear on Amazon 2-4 weeks later
  • Early sellers get the review advantage (first to market = first reviews)
  • "Viral" windows are short — you have 48-72 hours before the market floods
  • The data is public but hard to monitor manually across categories

What Data Is Available

TikTok Shop's trending section shows:

  • Product name + category
  • Velocity (% increase in sales/searches in 24h)
  • Price range
  • Commission rate
  • Number of videos made with this product
  • Estimated revenue bracket

Method 1: TikTok Shop Direct Scraping

TikTok Shop uses a React app with an internal API. The trending data comes from a predictable endpoint:

from curl_cffi import requests as cf_requests
import json, time, random
from datetime import datetime

def get_tiktok_shop_trending(category: str = "", country: str = "US") -> list:
    """
    Fetch trending products from TikTok Shop.

    Categories: Electronics, Fashion, Beauty, Home, Sports, etc.
    """
    session = cf_requests.Session()

    # Warm up session on TikTok Shop homepage
    session.get("https://shop.tiktok.com/", impersonate="chrome124",
                headers={"Accept-Language": "en-US,en;q=0.9"})
    time.sleep(2)

    # TikTok Shop trending API endpoint
    # This is the internal API used by the website
    api_url = "https://shop.tiktok.com/api/v1/trending/products"

    headers = {
        "Referer": "https://shop.tiktok.com/view/trending",
        "Accept": "application/json",
        "Content-Type": "application/json",
        "x-tt-store-region": country,
    }

    params = {
        "page": 1,
        "page_size": 50,
        "sort_by": "trending",
        "category": category,
        "region": country,
    }

    response = session.get(api_url, impersonate="chrome124", 
                           headers=headers, params=params)

    if response.status_code != 200:
        # Fallback to Movers & Shakers page scraping
        return scrape_tiktok_movers_shakers(session, country)

    data = response.json()
    return parse_tiktok_products(data)

def scrape_tiktok_movers_shakers(session, country: str = "US") -> list:
    """
    Fallback: scrape the Movers & Shakers page directly.
    """
    url = "https://shop.tiktok.com/view/trending"

    response = session.get(url, impersonate="chrome124",
                           headers={"Accept-Language": "en-US,en;q=0.9"})

    if response.status_code != 200:
        return []

    # Extract NEXT_DATA or __INITIAL_STATE__
    import re

    # Try Next.js data
    match = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', 
                      response.text, re.DOTALL)
    if match:
        try:
            next_data = json.loads(match.group(1))
            # Navigate to trending products in the state tree
            props = next_data.get('props', {}).get('pageProps', {})
            products = (
                props.get('trending_products') or
                props.get('initialState', {}).get('trending', {}).get('products', [])
            )
            if products:
                return products
        except json.JSONDecodeError:
            pass

    return []

def parse_tiktok_products(data: dict) -> list:
    products = []
    for item in data.get('data', data.get('products', [])):
        products.append({
            'id': item.get('product_id', ''),
            'name': item.get('product_name', item.get('title', '')),
            'price': item.get('price', {}).get('min_price', '') or item.get('price', ''),
            'category': item.get('category', ''),
            'velocity_score': item.get('trending_score', item.get('velocity', 0)),
            'commission_rate': item.get('commission_rate', ''),
            'video_count': item.get('video_count', 0),
            'sale_count_24h': item.get('sale_count_24h', 0),
        })
    return products
Enter fullscreen mode Exit fullscreen mode

Method 2: TikTok Creator Marketplace API

If you have a TikTok Business account, the Creator Marketplace API gives structured access:

import requests

class TikTokShopAPI:
    def __init__(self, app_key: str, app_secret: str, access_token: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.base_url = "https://open-api.tiktokglobalshop.com"

    def get_trending_products(self, category_id: str = None, limit: int = 50) -> list:
        """
        Official TikTok Shop API endpoint for product search.
        Requires: TikTok for Business account + approved app
        """
        endpoint = "/product/202309/products/search"

        payload = {
            "page_size": limit,
            "search_status": 2,  # Active products only
            "sort_field": "CREATE_TIME",
            "sort_order": "DESC",
        }
        if category_id:
            payload["category_id"] = category_id

        headers = {
            "x-tts-access-token": self.access_token,
            "Content-Type": "application/json",
        }

        r = requests.post(
            f"{self.base_url}{endpoint}",
            json=payload,
            headers=headers
        )

        return r.json().get('data', {}).get('products', [])
Enter fullscreen mode Exit fullscreen mode

Method 3: Trend Intelligence via Video Data

TikTok product trends show up in video performance before they appear in shop data. Track videos mentioning products:

def get_product_video_velocity(product_keyword: str) -> dict:
    """
    Track how many TikTok videos mention a product keyword in the last 24h vs 7 days.
    Uses TikTok's unofficial search API.
    """
    session = cf_requests.Session()

    search_url = "https://www.tiktok.com/api/search/general/full/"

    params = {
        "keyword": product_keyword,
        "count": 30,
        "offset": 0,
        "from_page": "search",
    }

    headers = {
        "Referer": f"https://www.tiktok.com/search?q={product_keyword}",
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    }

    response = session.get(search_url, impersonate="chrome124", 
                           params=params, headers=headers)

    if response.status_code != 200:
        return {}

    data = response.json()
    videos = data.get('data', {}).get('videos', [])

    # Analyze upload dates to measure velocity
    from datetime import datetime, timedelta
    now = datetime.now()

    last_24h = sum(1 for v in videos 
                   if datetime.fromtimestamp(v.get('createTime', 0)) > now - timedelta(hours=24))
    last_7d = len(videos)

    # Calculate daily average
    daily_avg_7d = last_7d / 7
    velocity = last_24h / daily_avg_7d if daily_avg_7d > 0 else 0

    return {
        'keyword': product_keyword,
        'videos_24h': last_24h,
        'videos_7d': last_7d,
        'velocity_ratio': round(velocity, 2),
        'trending': velocity > 1.5,  # 50% above average = trending
        'total_views_estimate': sum(v.get('stats', {}).get('playCount', 0) for v in videos),
    }

# Usage
products_to_track = [
    "phone grip holder",
    "portable blender",
    "led strip lights",
    "magnetic wallet",
    "silicone cup",
]

for product in products_to_track:
    data = get_product_video_velocity(product)
    if data.get('trending'):
        print(f"🔥 TRENDING: {product}")
        print(f"   Videos today: {data['videos_24h']} (7d avg: {data['videos_7d']/7:.1f}/day)")
        print(f"   Velocity: {data['velocity_ratio']}x normal rate")
Enter fullscreen mode Exit fullscreen mode

Automated Daily Monitoring Pipeline 

import json
from datetime import datetime

def daily_trend_monitor(keywords: list, output_file: str = "trends.json"):
    """
    Run every morning to catch yesterday's trending products.
    Set up as a cron job: 0 8 * * * python3 daily_trend_monitor.py
    """
    results = {
        'date': datetime.now().strftime('%Y-%m-%d'),
        'trending': [],
        'watching': [],
    }

    # Check each keyword
    for keyword in keywords:
        velocity_data = get_product_video_velocity(keyword)

        if velocity_data.get('velocity_ratio', 0) > 2.0:
            results['trending'].append(velocity_data)
            print(f"⚡ HIGH VELOCITY: {keyword} ({velocity_data['velocity_ratio']}x)")
        elif velocity_data.get('velocity_ratio', 0) > 1.3:
            results['watching'].append(velocity_data)
            print(f"👀 WATCH: {keyword} ({velocity_data['velocity_ratio']}x)")

    # Sort by velocity
    results['trending'].sort(key=lambda x: x.get('velocity_ratio', 0), reverse=True)

    # Save results
    with open(output_file, 'w') as f:
        json.dump(results, f, indent=2)

    print(f"\nSummary: {len(results['trending'])} trending, {len(results['watching'])} watching")
    print(f"Saved to {output_file}")

    return results

# Example keyword list for e-commerce tracking
KEYWORDS = [
    "portable charger", "wireless earbuds", "neck massager", "posture corrector",
    "teeth whitening", "hair growth serum", "acne patches", "under eye patches",
    "cable organizer", "monitor stand", "desk mat", "ring light",
    "insulated tumbler", "air fryer accessories", "food scale",
]

daily_trend_monitor(KEYWORDS)
Enter fullscreen mode Exit fullscreen mode

Cross-Reference with Amazon

TikTok Shop trends often predict Amazon bestsellers 2-4 weeks out:

def check_amazon_opportunity(product_name: str) -> dict:
    """
    Quick check: is this TikTok trending product already saturated on Amazon?
    """
    from curl_cffi import requests as cf_requests
    import re

    session = cf_requests.Session()
    search_url = f"https://www.amazon.com/s?k={product_name.replace(' ', '+')}"

    r = session.get(search_url, impersonate="chrome124",
                    headers={"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"})

    # Count results to gauge saturation
    result_count_match = re.search(r'(\d[\d,]+)\s+results', r.text)
    result_count = int(result_count_match.group(1).replace(',', '')) if result_count_match else 0

    # Check if there's a dedicated bestseller badge for this niche
    has_bestseller = '#1 Best Seller' in r.text

    return {
        'product': product_name,
        'amazon_results': result_count,
        'saturation': 'high' if result_count > 10000 else 'medium' if result_count > 1000 else 'low',
        'has_dominant_seller': has_bestseller,
        'opportunity': result_count < 5000 and not has_bestseller,
    }

# For each TikTok trending product, check Amazon saturation
for product in trending_products:
    amazon_data = check_amazon_opportunity(product['name'])
    if amazon_data['opportunity']:
        print(f"✅ OPPORTUNITY: {product['name']}")
        print(f"   TikTok velocity: {product.get('velocity_score')}x")
        print(f"   Amazon results: {amazon_data['amazon_results']:,}")
Enter fullscreen mode Exit fullscreen mode

Alert Setup for Consistent Monitoring 

import smtplib
from email.mime.text import MIMEText

def send_trend_alert(trending_products: list, email: str):
    """Send daily trend report via email"""
    if not trending_products:
        return

    body = "TikTok Shop Trending Products — Daily Report\n\n"
    for p in trending_products[:10]:
        body += f"Product: {p.get('keyword', p.get('name', ''))}\n"
        body += f"  Velocity: {p.get('velocity_ratio', p.get('velocity_score', ''))}x\n"
        body += f"  Videos today: {p.get('videos_24h', '')}\n\n"

    msg = MIMEText(body)
    msg['Subject'] = f"⚡ {len(trending_products)} TikTok products trending today"
    msg['From'] = "monitor@yourdomain.com"
    msg['To'] = email

    with smtplib.SMTP_SSL('smtp.gmail.com', 465) as server:
        server.login("your@gmail.com", "app_password")
        server.send_message(msg)
Enter fullscreen mode Exit fullscreen mode

Related Articles

Get the Complete Apify Scrapers Bundle

Save 10+ hours of setup time. The Apify Scrapers Bundle ($29) includes ready-to-use scrapers for Amazon, Google Maps, LinkedIn, TikTok, Instagram and 30+ more platforms — with documentation, example inputs, and output schemas.

Get the Bundle →

Related Tools

Top comments (0)