Commerce Trend Scout

# /commerce-trend-scout/main.py
# Runs on MuleRun Computer as a scheduled task (cron: 0 6 * * *)

import json, datetime, time
from pathlib import Path

DATE = datetime.date.today().isoformat()
DRIVE_BASE = f"/commerce-trend-scout/{DATE}"
MAX_RETRIES = 3
RETRY_DELAY = 30  # seconds between retries

def run_with_retry(func, name, retries=MAX_RETRIES):
    """Execute step with exponential backoff for rate limits."""
    for attempt in range(retries):
        try:
            result = func()
            log_step(name, "success", attempt + 1)
            return result
        except RateLimitError:
            wait = RETRY_DELAY * (2 ** attempt)
            log_step(name, f"rate-limited, retry in {wait}s", attempt + 1)
            time.sleep(wait)
        except Exception as e:
            log_step(name, f"error: {e}", attempt + 1)
            if attempt == retries - 1:
                raise
    raise RuntimeError(f"{name} failed after {retries} attempts")

# === PIPELINE ===
amazon_data  = run_with_retry(scan_amazon, "amazon_scan")    # Step 2
tiktok_data  = run_with_retry(scan_tiktok, "tiktok_scan")    # Step 3
reviews      = run_with_retry(extract_reviews, "reviews")    # Step 4
competitors  = run_with_retry(compare_prices, "competitors") # Step 5
scores       = score_opportunities(amazon_data, tiktok_data, reviews, competitors)

# Archive to MuleRun Drive
for name, data in [
    ("amazon-products.json", amazon_data),
    ("tiktok-products.json", tiktok_data),
    ("competitors.json", competitors),
    ("reviews.json", reviews),
    ("opportunity-score.json", scores),
]:
    drive_upload(f"{DRIVE_BASE}/{name}", json.dumps(data))

# Amazon scan using MuleRun amazon-ecommerce skill
# Capabilities: product discovery, BSR rankings, review mining, competitor tracking

def scan_amazon():
    """Scan Amazon for trending products across target categories."""
    categories = ["Health & Wellness", "Home Decor", "Kitchen",
                  "Beauty", "Fitness", "Mobile Accessories"]
    products = []

    for cat in categories:
        # Product discovery with BSR ranking data  [EXAMPLE]
        results = mulerun.skill("amazon-ecommerce", action="product_discovery",
            category=cat, sort_by="sales_rank", min_reviews=50, max_price=150)

        for p in results["products"]:
            # Review mining for sentiment & pain points  [EXAMPLE]
            review_data = mulerun.skill("amazon-ecommerce", action="review_analysis",
                asin=p["asin"], period_days=30)

            products.append({
                "asin": p["asin"], "title": p["title"],
                "price": p["price"], "bsr": p["sales_rank"],
                "review_velocity": review_data["weekly_new_reviews"],
                "sentiment": review_data["avg_sentiment"],
                "category": cat,
                "margin_proxy": estimate_margin(p["price"], cat),
            })
    return {"products": products, "scanned_at": utc_now()}

# TikTok Shop scan using MuleRun tiktok-ecommerce skill
# Capabilities: trending products, creator velocity, content trend scanning

def scan_tiktok():
    """Scan TikTok Shop for viral products and creator momentum."""
    # Product ranking by sales velocity  [EXAMPLE]
    trending = mulerun.skill("tiktok-ecommerce", action="trending_products",
        region="US", period="7d", limit=100)

    products = []
    for item in trending["products"]:
        # Content trend scanning — creator velocity  [EXAMPLE]
        creators = mulerun.skill("tiktok-ecommerce", action="product_creators",
            product_id=item["id"], period="7d")

        products.append({
            "id": item["id"], "title": item["title"],
            "price": item["price"],
            "creator_velocity": creators["new_creators_this_week"],
            "total_videos": creators["total_videos"],
            "category": item["category"],
            "margin_proxy": estimate_margin(item["price"], item["category"]),
        })
    return {"products": products, "scanned_at": utc_now()}

# Opportunity scoring model — combines signals from both platforms

WEIGHTS = {"demand": 0.30, "margin": 0.25, "competition": 0.20, "trend": 0.25}

def score_opportunities(amazon, tiktok, reviews, competitors):
    """Score each product 0-100 using weighted composite model."""
    scored = []
    all_products = amazon["products"] + tiktok["products"]

    for p in all_products:
        # Market research: competitor density check  [EXAMPLE]
        comp = mulerun.skill("amazon-ecommerce", action="competitor_analysis",
            keyword=p["title"], max_results=20)

        demand_score   = normalize(p.get("review_velocity", 0) + p.get("creator_velocity", 0))
        margin_score   = normalize(p["margin_proxy"])
        comp_score     = 100 - normalize(comp["competitor_count"])
        trend_score    = compute_trend(p)
        risk           = assess_risk(p, comp, reviews)

        total = (WEIGHTS["demand"] * demand_score +
                 WEIGHTS["margin"] * margin_score +
                 WEIGHTS["competition"] * comp_score +
                 WEIGHTS["trend"] * trend_score)

        scored.append({**p, "opp_score": round(total), "risk": risk})
    return sorted(scored, key=lambda x: x["opp_score"], reverse=True)

#!/bin/bash
# Set up the Commerce Trend Scout scheduled task on MuleRun Computer

# Create scheduled task via MuleRun Computer  [EXAMPLE]
mulerun computer schedule create \
  --name "commerce-trend-scout" \
  --cron "0 6 * * *" \
  --command "cd /workspace/commerce-trend-scout && python main.py" \
  --timeout 3600 \
  --retry-on-failure 2

# Publish dashboard to MuleRun Pages  [EXAMPLE]
mulerun page deploy \
  --source "/workspace/commerce-trend-scout/output/" \
  --name "commerce-trend-scout" \
  --title "Commerce Trend Scout Dashboard"

# Verify Drive archive  [EXAMPLE]
mulerun drive ls /commerce-trend-scout/$(date +%Y-%m-%d)/

# Data freshness and integrity checks

def validate_data_freshness(drive_path, max_age_hours=26):
    """Reject stale data — ensure snapshots are from current run."""
    meta = drive_stat(drive_path)
    age_hours = (utc_now() - meta["modified_at"]).total_seconds() / 3600
    if age_hours > max_age_hours:
        raise StaleDataError(
            f"Data at {drive_path} is {age_hours:.1f}h old (max: {max_age_hours}h). "
            f"Pipeline may have failed — check run-log.json.")
    return True

def validate_row_counts(data, min_expected=10):
    """Ensure scan returned minimum viable data."""
    count = len(data.get("products", []))
    if count < min_expected:
        log_warning(f"Only {count} products returned (expected >= {min_expected})")
        return False
    return True