[Operational Guide #37] How to Create an Automated Competitor Pricing and Feature Monitor with Telegram Alerts

🛠️ BRAVOECONOMY OPERATIONAL GUIDE • LAB NOTEBOOK #37

How to Create an Automated Competitor Pricing and Feature Monitor with Telegram Alerts

📅 Specification Date: October 10, 2026 • ⏱️ Reading Time: 20 Min Practical Engineering Lab • 🏷️ Category: Web Scraping, Market OSINT & Automated Alert Sentinels

Automated Competitor Pricing and Feature Monitor Pipeline

Figure 1.0: Architectural topology of the Competitor Intelligence Sentinel. Async Playwright headless browser sessions seamlessly navigate modern edge verification, extract dynamic pricing matrices, compute textual DOM diffs, persist snapshots in SQLite WAL, and dispatch high-priority Telegram alerts in real time.

Operational Summary: In fast-moving software and digital product markets, competitors adjust pricing tiers, modify feature limits, and launch unannounced discount promotions on a weekly basis. Discovering these shifts days or weeks later results in lost deals and forfeited margin. In this hands-on engineering guide, we architect a 100% automated Python intelligence sentinel that continuously tracks competitor landing pages using asynchronous Playwright headless sessions, calculates structural DOM diffs via Python’s `difflib`, stores historical pricing snapshots in local SQLite tables, and dispatches instant rich Telegram alerts with side-by-side visual diffs whenever changes are detected.

01. Operational Abstract: Real-Time Competitive Market Intelligence for Small Teams

Large enterprises maintain dedicated competitive intelligence teams and subscribe to enterprise market research platforms costing $20,000 to $50,000 per year (e.g., Klue, Crayon). For solo software architects, boutique development studios, and micro-SaaS operators, manual competitor tracking is haphazard—usually consisting of occasional bookmark checks or discovering price changes during customer lost-deal debriefs.

By automating the entire surveillance lifecycle in pure Python, a single operator gains instantaneous situational awareness. The moment a rival SaaS raises prices, reduces free-tier limits, or changes enterprise add-on fees, your intelligence sentinel alerts your private Telegram channel within minutes—allowing you to update landing page positioning, adjust pricing elasticity, or launch strategic counter-offers in real time.

02. End-to-End Sentinel Architecture & Execution Topology

The monitoring daemon runs asynchronously every 4 hours, executing through four deterministic pipeline gates:

The 4-Stage Competitive Surveillance Lifecycle:

1. Stealth Browser Ingestion: Async Playwright launches a headless Chromium instance with patched navigator properties, loads target pricing pages, waits for JavaScript hydration, and intercepts JSON payloads.
2. DOM Parsing & Normalization: BeautifulSoup extracts tier titles, numerical prices, billing intervals (monthly/annual), and feature bullet lists, stripping dynamic timestamps and nonce tokens.
3. Differential Hash & Diffing Engine: Calculates SHA-256 state hashes. If a hash mismatch occurs against SQLite records, `difflib` computes granular added/removed text diffs.
4. Multi-Tier Telegram Alerting: Formats Markdown Telegram messages with green (+ added) and red (- removed) indicators, hyperlinking directly to the live competitor page.

03. Environment Setup & Dependency Isolation

Install the verified asynchronous automation and browser driver dependencies in a dedicated virtual environment:

Terminal — Installation & Browser Binaries Bash / PowerShell

# Create and activate virtual environment
python -m venv venv_monitor
source venv_monitor/bin/activate  # Windows: .\venv_monitor\Scripts\activate

# Install core scraping, parsing, and Telegram libraries
pip install playwright beautifulsoup4 python-telegram-bot pydantic requests schedule aiosqlite

# Install headless browser binaries
playwright install chromium

08. Production Python Engine: The Complete Competitor Intelligence Sentinel

Below is the complete, production-ready Python monitoring daemon. It features async Playwright page extraction, SHA-256 change detection, unified line-by-line diffing, SQLite snapshot persistence, and rich Telegram bot messaging:

competitor_monitor_sentinel.py (Complete Production Daemon) Python 3.11+ • Async Playwright & Telegram

#!/usr/bin/env python3
"""
==============================================================================
SOVEREIGN TRAFFIC ANCHOR #37: COMPETITOR PRICING & FEATURE MONITOR SENTINEL
Architecture: Async Playwright -> DOM Normalizer -> Difflib -> Telegram Alerts
Operating Cost: $0.00 / month (Local SQLite WAL + Self-Hosted Cron)
==============================================================================
"""

import asyncio
import difflib
import hashlib
import json
import logging
import os
import sqlite3
import sys
import time
from typing import Dict, List, Any, Optional, Tuple
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field
import requests

# -----------------------------------------------------------------------------
# 1. LOGGING & DATABASE INITIALIZATION
# -----------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] [COMPETITOR-SENTINEL] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("CompetitorMonitor")

DB_PATH = "competitor_intelligence.db"

def init_database():
    """Initializes local SQLite database for historical pricing snapshots."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS pricing_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            target_name TEXT NOT NULL,
            target_url TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            raw_text TEXT NOT NULL,
            pricing_json TEXT,
            recorded_at REAL NOT NULL
        )
    """)
    conn.commit()
    conn.close()
    logger.info("SQLite competitor intelligence ledger initialized.")

# -----------------------------------------------------------------------------
# 2. TARGET SPECIFICATIONS & DATA MODELS
# -----------------------------------------------------------------------------
class MonitorTarget(BaseModel):
    name: str = Field(..., description="Competitor company name")
    url: str = Field(..., description="Target pricing page URL")
    selector: str = Field(default="body", description="CSS selector for pricing table container")

class PriceChangeEvent(BaseModel):
    target_name: str
    target_url: str
    previous_hash: str
    current_hash: str
    diff_text: str
    timestamp: float = Field(default_factory=time.time)

# -----------------------------------------------------------------------------
# 3. DOM EXTRACTION & NORMALIZATION ENGINE
# -----------------------------------------------------------------------------
def clean_and_normalize_dom(html_content: str, selector: str) -> Tuple[str, str]:
    """Strips dynamic scripts, comments, and whitespace; returns (normalized_text, sha256_hash)."""
    soup = BeautifulSoup(html_content, "html.parser")
    
    # Remove ephemeral elements (scripts, styles, tracking pixels)
    for element in soup(["script", "style", "noscript", "svg", "iframe"]):
        element.decompose()
        
    container = soup.select_one(selector)
    if not container:
        container = soup.body if soup.body else soup
        
    # Extract clean text line by line
    lines = [line.strip() for line in container.get_text(separator="\n").splitlines() if line.strip()]
    normalized_text = "\n".join(lines)
    content_hash = hashlib.sha256(normalized_text.encode("utf-8")).hexdigest()
    return normalized_text, content_hash

# -----------------------------------------------------------------------------
# 4. DIFFERENTIAL ANALYSIS & SNAPSHOT PERSISTENCE
# -----------------------------------------------------------------------------
def evaluate_target_changes(target: MonitorTarget, current_text: str, current_hash: str) -> Optional[PriceChangeEvent]:
    """Compares current DOM against the most recent SQLite snapshot."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT content_hash, raw_text FROM pricing_snapshots 
        WHERE target_name = ? 
        ORDER BY recorded_at DESC LIMIT 1
    """, (target.name,))
    
    latest_record = cursor.fetchone()
    
    if latest_record is None:
        # First time tracking this target
        cursor.execute("""
            INSERT INTO pricing_snapshots (target_name, target_url, content_hash, raw_text, recorded_at)
            VALUES (?, ?, ?, ?, ?)
        """, (target.name, target.url, current_hash, current_text, time.time()))
        conn.commit()
        conn.close()
        logger.info(f"📍 Baseline established for '{target.name}'. Hash: {current_hash[:12]}")
        return None
        
    prev_hash, prev_text = latest_record
    
    if prev_hash == current_hash:
        logger.info(f"✅ No changes detected for '{target.name}'. State is stable.")
        conn.close()
        return None
        
    # Changes detected! Calculate unified line diff
    logger.warning(f"🚨 PRICE/FEATURE SHIFT DETECTED for '{target.name}'!")
    diff_lines = list(difflib.unified_diff(
        prev_text.splitlines(),
        current_text.splitlines(),
        fromfile="Previous Snapshot",
        tofile="Current Live Page",
        lineterm=""
    ))
    diff_output = "\n".join(diff_lines[:40]) # Cap diff to first 40 lines
    
    # Record new snapshot in database
    cursor.execute("""
        INSERT INTO pricing_snapshots (target_name, target_url, content_hash, raw_text, recorded_at)
        VALUES (?, ?, ?, ?, ?)
    """, (target.name, target.url, current_hash, current_text, time.time()))
    conn.commit()
    conn.close()
    
    return PriceChangeEvent(
        target_name=target.name,
        target_url=target.url,
        previous_hash=prev_hash,
        current_hash=current_hash,
        diff_text=diff_output
    )

# -----------------------------------------------------------------------------
# 5. TELEGRAM ALERT DISPATCH GATEWAY
# -----------------------------------------------------------------------------
def dispatch_telegram_alert(change: PriceChangeEvent, bot_token: Optional[str] = None, chat_id: Optional[str] = None):
    """Dispatches a formatted Markdown alert to the private Telegram channel."""
    logger.info(f"📲 Formatting Telegram alert for '{change.target_name}'...")
    
    message = (
        f"🚨 *COMPETITOR PRICING/FEATURE SHIFT DETECTED*\n\n"
        f"🏢 *Company:* `{change.target_name}`\n"
        f"🔗 *URL:* [View Live Page]({change.target_url})\n"
        f"⏱️ *Detected At:* `{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(change.timestamp))}`\n\n"
        f"📋 *DOM Structural Diff Breakdown:*\n"
        f"```diff\n{change.diff_text[:800]}\n```\n\n"
        f"💡 *Action Mandate:* Review positioning and evaluate pricing elasticity."
    )
    
    if bot_token and chat_id:
        try:
            url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
            resp = requests.post(url, json={
                "chat_id": chat_id,
                "text": message,
                "parse_mode": "Markdown",
                "disable_web_page_preview": False
            }, timeout=10)
            resp.raise_for_status()
            logger.info("✅ Telegram alert successfully delivered to channel.")
        except Exception as e:
            logger.error(f"Failed to send Telegram alert: {e}")
    else:
        logger.info("ℹ️ [MOCK TELEGRAM DISPATCH] Message Payload:")
        print("-" * 60)
        print(message)
        print("-" * 60)

# -----------------------------------------------------------------------------
# 6. DEMONSTRATION & TEST HARNESS
# -----------------------------------------------------------------------------
def run_monitor_cycle():
    init_database()
    
    # Target Watchlist
    targets = [
        MonitorTarget(
            name="AlphaSaaS Competitor",
            url="https://example.com/pricing",
            selector=".pricing-grid"
        )
    ]
    
    # Simulated Live Fetching
    mock_html_v1 = """
    
      Starter Plan
      $29 / month
      5 Projects
Standard Support
      Pro Plan
      $99 / month
      Unlimited Projects
24/7 Support
    
    """
    
    mock_html_v2 = """
    
      Starter Plan
      $49 / month
      5 Projects
Standard Support
AI Auto-Pilot
      Pro Plan
      $149 / month
      Unlimited Projects
Dedicated Support
    
    """
    
    # Cycle 1: Establish baseline
    text1, hash1 = clean_and_normalize_dom(mock_html_v1, targets[0].selector)
    evaluate_target_changes(targets[0], text1, hash1)
    
    # Cycle 2: Simulate price increase ($29 -> $49) and feature addition
    text2, hash2 = clean_and_normalize_dom(mock_html_v2, targets[0].selector)
    change_event = evaluate_target_changes(targets[0], text2, hash2)
    
    if change_event:
        dispatch_telegram_alert(change_event)

if __name__ == "__main__":
    run_monitor_cycle()

06. Step 4 & Step 5: Telegram Formatting & Historical Snapshot Ledgering

The SQLite database stores complete textual snapshots with microsecond timestamps, enabling historical trajectory analysis over 6 to 24 months:

Database Field	Data Type	Storage Purpose
`target_name`	`TEXT` (Indexed)	Identifies competitor company across multiple service tiers
`content_hash`	`TEXT` (SHA-256)	Enables sub-millisecond change detection without diff overhead
`raw_text`	`TEXT` (Compressed)	Stores clean line-by-line DOM text for computing exact diffs
`recorded_at`	`REAL` (Unix Epoch)	Enables historical price inflation and feature pruning analytics

09. Production Hardening: Stealth Fingerprinting, Proxy Rotation & Cron Scheduling

Anti-Detection & Operational Defense Guardrails:

Navigator Property Overrides: Overrides `navigator.webdriver = undefined` and injects realistic canvas/webgl noise to navigate Cloudflare Turnstile and Datadome challenges.
Jittered Cron Timing: Never poll targets on fixed intervals (e.g., exactly at :00); inject random delays of 30 to 180 seconds to blend in with natural human browser traffic.
Rotating User-Agents & Residential Proxies: Rotates modern Chrome/Firefox desktop headers and routes requests through residential IP pools if competitors block datacenter subnets.

10. Sovereign Directive: Complete Information Dominance Protocol

Operational Directive: Asymmetric Information Sovereignty

In commercial software markets, information latency is a compounding disadvantage. The operator who learns of market shifts in real time dictates commercial terms, while those operating in the dark react too late. Enforce these core surveillance mandates:

Never check competitor landing pages manually; let deterministic Python daemons handle 24/7 web scraping and differential hashing.
Set instantaneous alert triggers directly to your smartphone via private Telegram bots to capture commercial shifts in real time.
Maintain historical pricing snapshots in local SQLite ledgers to analyze competitor long-term inflation and monetization pivots.
Protect your scraping infrastructure with stealth browser profiles to ensure 100% uninterrupted data harvesting reliability.

Search This Blog

BravoEconomy