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.