Competitive Intelligence API: The Complete Guide to Automating Competitor Research

# Store your API key as an environment variable
export RIVALCHECK_API_KEY="rc_live_abc123def456..."

{
  "competitors": [
    {
      "id": 42,
      "name": "Acme Corp",
      "website_url": "https://acmecorp.com",
      "status": "active",
      "monitored_pages": 6,
      "last_checked_at": "2026-03-23T14:30:00Z",
      "changes_count": 23,
      "has_battle_card": true
    },
    {
      "id": 43,
      "name": "Nexus Software",
      "website_url": "https://nexus.io",
      "status": "active",
      "monitored_pages": 4,
      "last_checked_at": "2026-03-23T14:30:00Z",
      "changes_count": 15,
      "has_battle_card": true
    }
  ],
  "meta": { "total": 2, "page": 1, "per_page": 25 }
}

{
  "id": 891,
  "competitor": { "id": 42, "name": "Acme Corp" },
  "page_url": "https://acmecorp.com/pricing",
  "category": "pricing",
  "severity": "high",
  "title": "Enterprise plan price increased from $299 to $349/mo",
  "summary": "Acme Corp raised their Enterprise tier price by 17%. The Pro tier
    remains unchanged. They also added a new 'Starter' tier at $29/mo,
    suggesting a move to capture smaller accounts.",
  "strategic_analysis": "This price increase signals confidence in their
    enterprise value proposition. The new Starter tier opens a new segment
    they previously ignored. Consider whether your mid-market positioning
    now has more room to compete on value.",
  "detected_at": "2026-03-22T09:15:00Z",
  "screenshot_before_url": "https://cdn.rivalcheck.com/shots/...",
  "screenshot_after_url": "https://cdn.rivalcheck.com/shots/..."
}

# Get only pricing changes, high severity
curl -s -H "Authorization: Bearer $RIVALCHECK_API_KEY" \
  "https://app.rivalcheck.com/api/v1/changes?category=pricing&severity=high"

#!/usr/bin/env python3
"""
Weekly Competitive Intelligence Report Generator
Pulls data from the RivalCheck API and produces a formatted summary.
"""

import os
import json
import requests
from datetime import datetime, timedelta
from collections import defaultdict

API_BASE = "https://app.rivalcheck.com/api/v1"
API_KEY = os.environ["RIVALCHECK_API_KEY"]

headers = {"Authorization": f"Bearer {API_KEY}"}


def get_competitors():
    """Fetch all active competitors."""
    resp = requests.get(f"{API_BASE}/competitors?status=active", headers=headers)
    resp.raise_for_status()
    return resp.json()["competitors"]


def get_changes(since_date):
    """Fetch all changes since a given date."""
    all_changes = []
    page = 1
    while True:
        resp = requests.get(
            f"{API_BASE}/changes",
            headers=headers,
            params={"since": since_date, "page": page, "limit": 100}
        )
        resp.raise_for_status()
        data = resp.json()
        all_changes.extend(data["changes"])
        if len(all_changes) >= data["meta"]["total"]:
            break
        page += 1
    return all_changes


def get_battle_card(competitor_id):
    """Fetch the current battle card for a competitor."""
    resp = requests.get(
        f"{API_BASE}/competitors/{competitor_id}/battle_card",
        headers=headers
    )
    if resp.status_code == 404:
        return None
    resp.raise_for_status()
    return resp.json()


def generate_report():
    """Build the weekly CI report."""
    since = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d")
    competitors = get_competitors()
    changes = get_changes(since)

    # Group changes by competitor
    by_competitor = defaultdict(list)
    for change in changes:
        by_competitor[change["competitor"]["name"]].append(change)

    # Count by category
    by_category = defaultdict(int)
    for change in changes:
        by_category[change["category"]] += 1

    # Build report
    report = []
    report.append(f"# Weekly Competitive Intelligence Report")
    report.append(f"Period: {since} to {datetime.utcnow().strftime('%Y-%m-%d')}")
    report.append(f"Competitors monitored: {len(competitors)}")
    report.append(f"Total changes detected: {len(changes)}")
    report.append("")

    # Summary by category
    report.append("## Changes by Category")
    for category, count in sorted(by_category.items(),
                                   key=lambda x: x[1], reverse=True):
        report.append(f"  - {category.title()}: {count}")
    report.append("")

    # High-severity alerts
    high_severity = [c for c in changes if c["severity"] == "high"]
    if high_severity:
        report.append("## High-Priority Alerts")
        for change in high_severity:
            report.append(f"  [{change['competitor']['name']}] {change['title']}")
            report.append(f"  Analysis: {change['strategic_analysis']}")
            report.append("")

    # Per-competitor breakdown
    report.append("## Competitor Details")
    for comp_name, comp_changes in sorted(by_competitor.items()):
        report.append(f"\n### {comp_name} ({len(comp_changes)} changes)")
        for change in comp_changes:
            severity_marker = "!" if change["severity"] == "high" else "-"
            report.append(
                f"  {severity_marker} [{change['category']}] {change['title']}"
            )

    return "\n".join(report)


if __name__ == "__main__":
    report = generate_report()
    print(report)

    # Optionally save to file
    filename = f"ci_report_{datetime.utcnow().strftime('%Y%m%d')}.md"
    with open(filename, "w") as f:
        f.write(report)
    print(f"\nReport saved to {filename}")

# Create a Linear issue when a competitor ships a new feature
import requests

def create_competitive_insight_issue(change):
    """Create a Linear issue in the 'Competitive Insights' project."""
    linear_api = "https://api.linear.app/graphql"
    headers = {"Authorization": LINEAR_API_KEY}

    mutation = """
    mutation CreateIssue($input: IssueCreateInput!) {
        issueCreate(input: $input) {
            issue { id identifier url }
        }
    }
    """

    variables = {
        "input": {
            "teamId": PRODUCT_TEAM_ID,
            "projectId": CI_PROJECT_ID,
            "title": f"[CI] {change['competitor']['name']}: {change['title']}",
            "description": (
                f"## Competitive Change Detected\n\n"
                f"**Competitor:** {change['competitor']['name']}\n"
                f"**Category:** {change['category']}\n"
                f"**Severity:** {change['severity']}\n\n"
                f"### Summary\n{change['summary']}\n\n"
                f"### Strategic Analysis\n{change['strategic_analysis']}\n\n"
                f"*Auto-generated by RivalCheck API*"
            ),
            "priority": 2 if change["severity"] == "high" else 3,
            "labelIds": [COMPETITIVE_LABEL_ID]
        }
    }

    resp = requests.post(linear_api, json={
        "query": mutation, "variables": variables
    }, headers=headers)
    return resp.json()