How to Competitor Monitoring Api — Step-by-Step

Building a competitor monitoring API gives you real-time visibility into what your competitors are doing — pricing changes, product launches, content updates, and SERP movements. This tutorial shows you how to build one from scratch using Python and SearchHive.

By the end, you will have a working system that monitors competitor websites and search rankings, stores historical data, and sends alerts when significant changes occur.

Key Takeaways

• A competitor monitoring API combines search tracking, page scraping, and change detection
• SearchHive provides all three capabilities (SwiftSearch, ScrapeForge, DeepDive) in one API
• This tutorial builds a complete system: data collection, storage, diff detection, and alerting
• Python + SQLite is enough for most use cases; scale to PostgreSQL when needed
• The system handles anti-bot protection automatically through SearchHive's managed scraping

Prerequisites

Before starting, you need:

• Python 3.8+ with pip
• A SearchHive API key (free tier — 500 credits, no CC)
• SQLite (included with Python)
• Basic familiarity with Python and REST APIs

Install dependencies:

pip install searchhive requests fastapi uvicorn

Step 1: Define Your Monitoring Targets

Start by defining what you want to monitor. Create a configuration file:

# monitor_config.py

COMPETITORS = [
    {
        "name": "Competitor A",
        "domain": "competitor-a.com",
        "pricing_url": "https://competitor-a.com/pricing",
        "blog_url": "https://competitor-a.com/blog",
        "tracking_keywords": ["project management tool", "team collaboration software"]
    },
    {
        "name": "Competitor B",
        "domain": "competitor-b.com",
        "pricing_url": "https://competitor-b.com/pricing",
        "blog_url": "https://competitor-b.com/blog",
        "tracking_keywords": ["task management app", "remote work tools"]
    }
]

ALERT_CONFIG = {
    "slack_webhook": "https://hooks.slack.com/services/...",
    "email_recipient": "team@example.com",
    "check_interval_minutes": 60
}

Step 2: Set Up the Database

Store monitoring data so you can track changes over time:

# database.py
import sqlite3
from datetime import datetime


def init_db():
    conn = sqlite3.connect("monitor.db")
    c = conn.cursor()
    
    c.execute("""
        CREATE TABLE IF NOT EXISTS page_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            competitor TEXT NOT NULL,
            url TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            title TEXT,
            content_length INTEGER,
            captured_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    c.execute("""
        CREATE TABLE IF NOT EXISTS serp_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            keyword TEXT NOT NULL,
            competitor_domain TEXT NOT NULL,
            position INTEGER,
            url TEXT,
            title TEXT,
            captured_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    c.execute("""
        CREATE TABLE IF NOT EXISTS alerts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            competitor TEXT NOT NULL,
            alert_type TEXT NOT NULL,
            message TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    conn.commit()
    conn.close()

init_db()
print("Database initialized")

Step 3: Build the Page Monitor

This module scrapes competitor pages and detects changes:

# page_monitor.py
import hashlib
import sqlite3
from searchhive import ScrapeForge

client = ScrapeForge(api_key="your_api_key")


def get_content_hash(content):
    return hashlib.sha256(content.encode()).hexdigest()[:32]


def get_last_snapshot(conn, competitor, url):
    c = conn.cursor()
    c.execute(
        "SELECT content_hash, title FROM page_snapshots "
        "WHERE competitor=? AND url=? ORDER BY captured_at DESC LIMIT 1",
        (competitor, url)
    )
    return c.fetchone()


def save_snapshot(conn, competitor, url, content_hash, title, content_length):
    conn.execute(
        "INSERT INTO page_snapshots (competitor, url, content_hash, title, content_length) "
        "VALUES (?, ?, ?, ?, ?)",
        (competitor, url, content_hash, title, content_length)
    )
    conn.commit()


def monitor_page(competitor_name, url):
    """Scrape a page and detect changes. Returns alert if changed."""
    conn = sqlite3.connect("monitor.db")
    
    try:
        page = client.scrape(url, render_js=True)
        current_hash = get_content_hash(page.content)
        last = get_last_snapshot(conn, competitor_name, url)
        
        if last and last[0] != current_hash:
            # Page changed!
            alert = {
                "competitor": competitor_name,
                "url": url,
                "type": "page_change",
                "old_title": last[1],
                "new_title": page.title,
                "hash": current_hash
            }
            print(f"[CHANGE] {competitor_name}: {url} - Title: {last[1]} -> {page.title}")
        else:
            alert = None
            print(f"[OK] {competitor_name}: {url} - No changes")
        
        save_snapshot(conn, competitor_name, url, current_hash, page.title, len(page.content))
        return alert
    
    except Exception as e:
        print(f"[ERROR] {competitor_name}: {url} - {e}")
        return {"competitor": competitor_name, "type": "error", "message": str(e)}
    finally:
        conn.close()

Step 4: Build the SERP Monitor

Track where competitors rank for your target keywords:

# serp_monitor.py
import sqlite3
from searchhive import SwiftSearch

client = SwiftSearch(api_key="your_api_key")


def track_keyword(keyword, competitor_domain):
    """Search for a keyword and find the competitor's position."""
    conn = sqlite3.connect("monitor.db")
    
    try:
        results = client.search(query=keyword, num=20)
        
        position = None
        for r in results.organic:
            if competitor_domain in r.url:
                position = r.position
                break
        
        # Save snapshot
        conn.execute(
            "INSERT INTO serp_snapshots (keyword, competitor_domain, position, url, title) "
            "VALUES (?, ?, ?, ?, ?)",
            (keyword, competitor_domain, position, r.url if position else None, r.title if position else None)
        )
        conn.commit()
        
        # Check for significant rank changes
        c = conn.cursor()
        c.execute(
            "SELECT position FROM serp_snapshots "
            "WHERE keyword=? AND competitor_domain=? "
            "ORDER BY captured_at DESC LIMIT 2",
            (keyword, competitor_domain)
        )
        rows = c.fetchall()
        
        alert = None
        if len(rows) == 2 and rows[0][0] is not None and rows[1][0] is not None:
            old_pos, new_pos = rows[1][0], rows[0][0]
            if abs(new_pos - old_pos) >= 3:
                direction = "up" if new_pos < old_pos else "down"
                alert = {
                    "competitor": competitor_domain,
                    "keyword": keyword,
                    "type": "rank_change",
                    "old_position": old_pos,
                    "new_position": new_pos,
                    "direction": direction
                }
                print(f"[RANK] {competitor_domain}: '{keyword}' moved {direction} from {old_pos} to {new_pos}")
        
        status = f"Position {position}" if position else "Not in top 20"
        print(f"[OK] {competitor_domain}: '{keyword}' - {status}")
        return alert
    
    finally:
        conn.close()

Step 5: Build the Alerting System

# alerting.py
import requests
import sqlite3


def send_slack_alert(alert, webhook_url):
    if alert["type"] == "page_change":
        message = (
            f"*Competitor Alert: {alert['competitor']}*\n"
            f"Page changed: {alert['url']}\n"
            f"Title: {alert.get('old_title', 'N/A')} -> {alert['new_title']}"
        )
    elif alert["type"] == "rank_change":
        emoji = "\U0001F7E2" if alert["direction"] == "up" else "\U0001F534"
        message = (
            f"*Rank Change: {alert['competitor']}*\n"
            f"Keyword: '{alert['keyword']}'\n"
            f"Position: {alert['old_position']} -> {alert['new_position']} {emoji}"
        )
    else:
        message = f"Alert: {alert}"
    
    requests.post(webhook_url, json={"text": message})


def log_alert(alert):
    conn = sqlite3.connect("monitor.db")
    conn.execute(
        "INSERT INTO alerts (competitor, alert_type, message) VALUES (?, ?, ?)",
        (alert.get("competitor", "unknown"), alert["type"], str(alert))
    )
    conn.commit()
    conn.close()

Step 6: Build the REST API

Expose monitoring data through a REST API with FastAPI:

# api.py
from fastapi import FastAPI
import sqlite3

app = FastAPI(title="Competitor Monitor API")


@app.get("/snapshots/{competitor}")
def get_snapshots(competitor: str, limit: int = 50):
    conn = sqlite3.connect("monitor.db")
    conn.row_factory = sqlite3.Row
    c = conn.cursor()
    c.execute(
        "SELECT * FROM page_snapshots WHERE competitor=? ORDER BY captured_at DESC LIMIT ?",
        (competitor, limit)
    )
    rows = [dict(r) for r in c.fetchall()]
    conn.close()
    return {"competitor": competitor, "snapshots": rows}


@app.get("/rankings/{competitor}")
def get_rankings(competitor: str, limit: int = 100):
    conn = sqlite3.connect("monitor.db")
    conn.row_factory = sqlite3.Row
    c = conn.cursor()
    c.execute(
        "SELECT * FROM serp_snapshots WHERE competitor_domain=? ORDER BY captured_at DESC LIMIT ?",
        (competitor, limit)
    )
    rows = [dict(r) for r in c.fetchall()]
    conn.close()
    return {"competitor": competitor, "rankings": rows}


@app.get("/alerts")
def get_alerts(limit: int = 50):
    conn = sqlite3.connect("monitor.db")
    conn.row_factory = sqlite3.Row
    c = conn.cursor()
    c.execute("SELECT * FROM alerts ORDER BY created_at DESC LIMIT ?", (limit,))
    rows = [dict(r) for r in c.fetchall()]
    conn.close()
    return {"alerts": rows}


@app.post("/check")
def run_check():
    from monitor_config import COMPETITORS, ALERT_CONFIG
    from page_monitor import monitor_page
    from serp_monitor import track_keyword
    from alerting import send_slack_alert, log_alert
    
    all_alerts = []
    
    for comp in COMPETITORS:
        # Check pricing page
        alert = monitor_page(comp["name"], comp["pricing_url"])
        if alert:
            all_alerts.append(alert)
            log_alert(alert)
            send_slack_alert(alert, ALERT_CONFIG["slack_webhook"])
        
        # Check keywords
        for keyword in comp["tracking_keywords"]:
            alert = track_keyword(keyword, comp["domain"])
            if alert:
                all_alerts.append(alert)
                log_alert(alert)
                send_slack_alert(alert, ALERT_CONFIG["slack_webhook"])
    
    return {"checked": len(COMPETITORS), "alerts": len(all_alerts), "details": all_alerts}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Start the API:

python api.py
# API running at http://localhost:8000
# Docs at http://localhost:8000/docs

Step 7: Schedule Automated Checks

Set up a cron expression generator to run checks automatically:

# Run competitor monitoring every hour
0 * * * * cd /path/to/monitor && python -c "
import requests
response = requests.post('http://localhost:8000/check')
print(response.json())
"

Or use Python's schedule library for in-process scheduling:

import schedule
import time
import requests

def run_monitor():
    resp = requests.post("http://localhost:8000/check")
    data = resp.json()
    print(f"Checked {data['checked']} competitors, {data['alerts']} alerts")


schedule.every().hour.do(run_monitor)

while True:
    schedule.run_pending()
    time.sleep(60)

Common Issues

Competitor Sites Block Your Scrapes

SearchHive handles anti-bot bypass automatically with a 99.3% success rate. If you still get occasional failures, the built-in retry logic handles transient issues. For persistently difficult sites, try adding specific wait selectors:

page = client.scrape(url, render_js=True, wait_for=".pricing-table")

Too Many False Alerts

Some pages change frequently (timestamps, analytics IDs, etc.) without meaningful content changes. Use content fingerprinting on specific sections instead of full-page hashes:

# Hash only the pricing section, not the whole page
pricing_section = page.css(".pricing-table").text_content()
content_hash = get_content_hash(pricing_section)

Database Growing Too Fast

For high-frequency monitoring, add a cleanup job that keeps only the last 30 days of snapshots:

import sqlite3

conn = sqlite3.connect("monitor.db")
conn.execute("DELETE FROM page_snapshots WHERE captured_at < datetime('now', '-30 days')")
conn.execute("DELETE FROM serp_snapshots WHERE captured_at < datetime('now', '-30 days')")
conn.commit()
conn.close()

Next Steps

Your competitor monitoring API is running. Here are ways to extend it:

• Add SPA scraping for JavaScript-heavy competitor sites
• Connect your AI agent via MCP integration for intelligent analysis
• Learn more automation for monitoring patterns and best practices

Start building today — get your free SearchHive API key with 500 credits. No credit card required. Full API docs at searchhive.dev/docs.