XT NOT NULL, currency TEXT DEFAULT 'USD', fetched_at TEXT NOT NULL ) """) conn.execute(""" CREATE INDEX IF NOT EXISTS idx_target_time ON price_snapshots(target_id, fetched_at DESC) """)

def upsert_snapshot(self, record: PriceRecord) -> None:
    with self._connect() as conn:
        conn.execute(
            "INSERT INTO price_snapshots (target_id, raw_value, normalized_value, source, currency, fetched_at) VALUES (?, ?, ?, ?, ?, ?)",
            (record.target_id, record.raw_value, record.normalized_value, record.source, record.currency, record.fetched_at.isoformat())
        )

def get_latest(self, target_id: str) -> Optional[float]:
    with self._connect() as conn:
        row = conn.execute(
            "SELECT normalized_value FROM price_snapshots WHERE target_id = ? ORDER BY fetched_at DESC LIMIT 1",
            (target_id,)
        ).fetchone()
        return row["normalized_value"] if row else None

**Rationale:** Indexing on `(target_id, fetched_at)` accelerates historical queries. Using `sqlite3.Row` enables dictionary-like access without fragile positional indexing. The `upsert` pattern appends rather than overwrites, preserving audit trails. Change detection happens at the application layer, not the database layer, allowing configurable thresholds.

### 4. Notification Router

Email alerts should trigger only on meaningful shifts. Deduplicate alerts to prevent inbox flooding during rapid competitor adjustments.

```python
import smtplib
import logging
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

logger = logging.getLogger(__name__)

class AlertDispatcher:
    def __init__(self, smtp_host: str, smtp_port: int, sender: str, credentials: tuple):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
        self.sender = sender
        self.credentials = credentials

    def dispatch(self, target_id: str, previous: float, current: float, source: str) -> None:
        delta_pct = ((current - previous) / previous) * 100
        direction = "UP" if delta_pct > 0 else "DOWN"

        body = (
            f"Price Signal Detected\n"
            f"Target: {target_id}\n"
            f"Source: {source}\n"
            f"Previous: ${previous:.2f}\n"
            f"Current: ${current:.2f}\n"
            f"Delta: {delta_pct:+.1f}% ({direction})\n"
            f"Timestamp: {datetime.now(timezone.utc).isoformat()}"
        )

        msg = MIMEMultipart()
        msg["From"] = self.sender
        msg["To"] = self.sender
        msg["Subject"] = f"[PriceIntel] {target_id} {direction} {abs(delta_pct):.1f}%"
        msg.attach(MIMEText(body, "plain"))

        try:
            with smtplib.SMTP_SSL(self.smtp_host, self.smtp_port) as server:
                server.login(*self.credentials)
                server.send_message(msg)
            logger.info("Alert dispatched for %s", target_id)
        except Exception as exc:
            logger.error("Notification failed: %s", exc)

Rationale: SMTP_SSL on port 465 enforces encrypted transport. Structured logging replaces print() statements, enabling integration with log aggregators. The delta calculation uses absolute thresholds in the orchestrator (see below) to filter noise.

5. Orchestrator

Tie components together with configurable thresholds and graceful degradation.

import logging
from typing import List

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")

class PricingEngine:
    def __init__(self, targets: List[PricingTarget], repo: PriceRepository, fetcher: PriceFetcher, notifier: AlertDispatcher, threshold: float = 0.02):
        self.targets = targets
        self.repo = repo
        self.fetcher = fetcher
        self.notifier = notifier
        self.threshold = threshold  # 2% minimum change to trigger alert

    def run_cycle(self) -> None:
        for target in self.targets:
            try:
                raw = self.fetcher.extract(target)
                if raw is None:
                    logger.warning("Parser returned None for %s", target.identifier)
                    continue

                record = PriceRecord(
                    target_id=target.identifier,
                    raw_value=raw,
                    normalized_value=raw,
                    source=target.source_domain,
                    currency=target.currency
                )
                self.repo.upsert_snapshot(record)

                prev = self.repo.get_latest(target.identifier)
                if prev is not None:
                    change_pct = abs((record.normalized_value - prev) / prev)
                    if change_pct >= self.threshold:
                        self.notifier.dispatch(target.identifier, prev, record.normalized_value, target.source_domain)
                        logger.info("Threshold breached: %s (%.2f%%)", target.identifier, change_pct * 100)
                    else:
                        logger.debug("Within threshold: %s", target.identifier)
            except Exception as exc:
                logger.error("Cycle failed for %s: %s", target.identifier, exc)

Rationale: The orchestrator isolates execution per target, ensuring one failure doesn't halt the entire cycle. The threshold parameter prevents alert fatigue from rounding differences or temporary cart adjustments. Structured logging provides observability without console clutter.

Pitfall Guide

Production Bundle

Action Checklist

• Externalize all targets, selectors, and thresholds into a version-controlled configuration file
• Implement UTC normalization at ingestion and validate timezone consistency across all records
• Add a minimum change threshold (2–5%) and alert deduplication window to prevent notification spam
• Replace direct print() statements with structured logging integrated into your observability stack
• Validate parser outputs against expected numeric ranges before storing to catch DOM breakage early
• Schedule the orchestrator via a job runner (cron, systemd timer, or Airflow) with health check monitoring
• Rotate credentials using environment variables or a secrets manager; never commit SMTP or proxy tokens

Decision Matrix

Configuration Template

# pricing_targets.yaml
targets:
  - identifier: "saas_basic_tier"
    url: "https://competitor-a.com/pricing"
    parser_type: "css"
    selector: ".pricing-card.basic .price-amount"
    source_domain: "competitor-a.com"
    currency: "USD"

  - identifier: "ecommerce_pro_bundle"
    url: "https://competitor-b.com/plans"
    parser_type: "regex"
    selector: "Pro Bundle.*?\\$([\\d,]+\\.?\\d*)"
    source_domain: "competitor-b.com"
    currency: "USD"

engine:
  threshold_pct: 0.03
  timeout_seconds: 10
  retries: 3
  alert_window_hours: 24

storage:
  db_path: "./data/price_intel.db"

notifications:
  smtp_host: "smtp.gmail.com"
  smtp_port: 465
  sender: "alerts@yourdomain.com"
  credentials_env: "SMTP_USER,SMTP_PASS"

Quick Start Guide

1. Install dependencies: pip install httpx beautifulsoup4 playwright
2. Initialize browser binaries: playwright install chromium (required only if using dynamic targets)
3. Create configuration: Save the YAML template as pricing_targets.yaml and populate with your targets
4. Run the engine: Execute the orchestrator script. Verify logs show successful fetches, threshold checks, and alert dispatches
5. Schedule execution: Add a cron entry (0 */6 * * * /usr/bin/python3 /path/to/engine.py) or configure a systemd timer for automated cycles

This pipeline transforms pricing from a manual chore into a queryable, alert-driven intelligence layer. By decoupling fetch, parse, store, and notify, you gain observability, resilience, and a foundation for advanced analytics like elasticity modeling or automated rule-based pricing adjustments.