Shopify Performance Monitoring: Automated Alerts for Store Speed

A Shopify store that loads in 2 seconds today might load in 6 seconds next week. A new app gets installed. A theme update ships a larger JavaScript bundle. Someone adds an unoptimised hero image. Nobody notices until conversion rates drop and by then you have lost a week of revenue.

Automated performance monitoring catches these regressions the day they happen. Run synthetic speed checks on a schedule, compare against baselines, and get alerts before customers start bouncing.

This guide builds the complete monitoring pipeline: measurement, storage, trend analysis, and alerting.

Who This Is For

• Store owners who have optimised their Shopify speed once and want to keep it that way
• Ecommerce agencies managing multiple client stores who need automated oversight
• Vibe coders building performance monitoring tools for ecommerce clients
• Developers who want to catch performance regressions before they ship to production

Basic Python knowledge and a Shopify store URL are all you need to start.

The Monitoring Architecture

What You Will Need

pip install requests pandas sqlalchemy jinja2

• Google PageSpeed Insights API key (free, 25,000 queries/day)
• Your Shopify store URLs (homepage, collection page, product page)

Step 1: Build the Speed Measurement Client

Use Google’s PageSpeed Insights API to measure Core Web Vitals and Lighthouse scores.

import requests
import pandas as pd
from datetime import datetime
from sqlalchemy import create_engine

class SpeedMonitor:
    """Measure and track Shopify store page speed."""

    PSI_URL = "https://www.googleapis.com/pagespeedonline/v5/runPagespeed"

    def __init__(self, api_key: str, db_url: str = "sqlite:///speed_monitor.db"):
        """Initialise with PSI API key and database."""
        self.api_key = api_key
        self.engine = create_engine(db_url)
        print("Speed monitor initialised")

    def measure_page(self, url: str, strategy: str = "mobile") -> dict:
        """Run a PageSpeed Insights measurement on a single URL."""
        params = {
            "url": url,
            "key": self.api_key,
            "strategy": strategy,
            "category": "performance",
        }

        resp = requests.get(self.PSI_URL, params=params)
        resp.raise_for_status()
        data = resp.json()

        # Extract Lighthouse scores
        lighthouse = data.get("lighthouseResult", {})
        audits = lighthouse.get("audits", {})
        categories = lighthouse.get("categories", {})

        # Core Web Vitals
        metrics = {
            "url": url,
            "strategy": strategy,
            "measured_at": datetime.now().isoformat(),
            "performance_score": int(categories.get("performance", {}).get("score", 0) * 100),
            "fcp_ms": audits.get("first-contentful-paint", {}).get("numericValue", 0),
            "lcp_ms": audits.get("largest-contentful-paint", {}).get("numericValue", 0),
            "cls": audits.get("cumulative-layout-shift", {}).get("numericValue", 0),
            "tbt_ms": audits.get("total-blocking-time", {}).get("numericValue", 0),
            "speed_index_ms": audits.get("speed-index", {}).get("numericValue", 0),
            "tti_ms": audits.get("interactive", {}).get("numericValue", 0),
            "total_byte_weight": audits.get("total-byte-weight", {}).get("numericValue", 0),
            "dom_size": audits.get("dom-size", {}).get("numericValue", 0),
            "third_party_count": len(audits.get("third-party-summary", {}).get("details", {}).get("items", [])),
        }

        print(f"  {url}: score={metrics['performance_score']}, LCP={metrics['lcp_ms']:.0f}ms")
        return metrics

Step 2: Define Pages to Monitor

Monitor the pages that matter most: homepage, top collection, top product, cart.

class StoreMonitor:
    """Monitor multiple pages across a Shopify store."""

    def __init__(self, store_url: str, api_key: str, db_url: str = "sqlite:///speed_monitor.db"):
        """Initialise with store URL and monitoring config."""
        self.store_url = store_url.rstrip("/")
        self.speed = SpeedMonitor(api_key, db_url)
        self.engine = self.speed.engine

        # Default pages to monitor
        self.pages = [
            {"path": "/", "label": "Homepage"},
            {"path": "/collections/all", "label": "Collection"},
            {"path": "/collections", "label": "Collections Index"},
        ]

    def add_page(self, path: str, label: str):
        """Add a page to the monitoring list."""
        self.pages.append({"path": path, "label": label})

    def run_full_check(self) -> pd.DataFrame:
        """Measure all monitored pages."""
        results = []
        print(f"Running speed check on {len(self.pages)} pages...")

        for page in self.pages:
            url = f"{self.store_url}{page['path']}"
            try:
                # Mobile measurement (primary)
                mobile = self.speed.measure_page(url, strategy="mobile")
                mobile["page_label"] = page["label"]
                results.append(mobile)

                # Desktop measurement
                desktop = self.speed.measure_page(url, strategy="desktop")
                desktop["page_label"] = page["label"]
                results.append(desktop)

            except Exception as e:
                print(f"  Error measuring {url}: {e}")
                results.append({
                    "url": url,
                    "page_label": page["label"],
                    "strategy": "error",
                    "measured_at": datetime.now().isoformat(),
                    "performance_score": 0,
                    "error": str(e),
                })

        df = pd.DataFrame(results)
        df.to_sql("speed_measurements", self.engine, if_exists="append", index=False)
        print(f"Stored {len(df)} measurements")
        return df

Step 3: Establish Performance Baselines

A regression needs a baseline. Calculate rolling averages to compare against.

class BaselineCalculator:
    """Calculate and maintain performance baselines."""

    def __init__(self, engine):
        """Initialise with database engine."""
        self.engine = engine

    def calculate_baselines(self, days: int = 14) -> pd.DataFrame:
        """Calculate baseline metrics from recent measurements."""
        query = f"""
            SELECT
                url,
                page_label,
                strategy,
                AVG(performance_score) as baseline_score,
                AVG(lcp_ms) as baseline_lcp,
                AVG(fcp_ms) as baseline_fcp,
                AVG(cls) as baseline_cls,
                AVG(tbt_ms) as baseline_tbt,
                AVG(speed_index_ms) as baseline_speed_index,
                PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY lcp_ms) as p75_lcp,
                PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY fcp_ms) as p75_fcp,
                COUNT(*) as measurement_count
            FROM speed_measurements
            WHERE measured_at >= datetime('now', '-{days} days')
              AND strategy != 'error'
            GROUP BY url, page_label, strategy
        """
        df = pd.read_sql(query, self.engine)
        df.to_sql("performance_baselines", self.engine, if_exists="replace", index=False)
        print(f"Baselines calculated from {days} days of data")
        return df

    def detect_regressions(self, current: pd.DataFrame) -> list[dict]:
        """Compare current measurements against baselines."""
        baselines = pd.read_sql("SELECT * FROM performance_baselines", self.engine)
        regressions = []

        for _, measurement in current.iterrows():
            baseline = baselines[
                (baselines["url"] == measurement["url"]) &
                (baselines["strategy"] == measurement["strategy"])
            ]

            if len(baseline) == 0:
                continue

            b = baseline.iloc[0]

            # Score regression (drop of 10+ points)
            score_diff = measurement["performance_score"] - b["baseline_score"]
            if score_diff <= -10:
                regressions.append({
                    "url": measurement["url"],
                    "page_label": measurement.get("page_label", ""),
                    "strategy": measurement["strategy"],
                    "metric": "performance_score",
                    "current": measurement["performance_score"],
                    "baseline": round(b["baseline_score"], 1),
                    "change": round(score_diff, 1),
                    "severity": "critical" if score_diff <= -20 else "warning",
                })

            # LCP regression (increase of 500ms+)
            lcp_diff = measurement["lcp_ms"] - b["baseline_lcp"]
            if lcp_diff >= 500:
                regressions.append({
                    "url": measurement["url"],
                    "page_label": measurement.get("page_label", ""),
                    "strategy": measurement["strategy"],
                    "metric": "lcp_ms",
                    "current": round(measurement["lcp_ms"]),
                    "baseline": round(b["baseline_lcp"]),
                    "change": round(lcp_diff),
                    "severity": "critical" if lcp_diff >= 1000 else "warning",
                })

            # CLS regression (increase of 0.05+)
            cls_diff = measurement["cls"] - b["baseline_cls"]
            if cls_diff >= 0.05:
                regressions.append({
                    "url": measurement["url"],
                    "page_label": measurement.get("page_label", ""),
                    "strategy": measurement["strategy"],
                    "metric": "cls",
                    "current": round(measurement["cls"], 3),
                    "baseline": round(b["baseline_cls"], 3),
                    "change": round(cls_diff, 3),
                    "severity": "critical" if cls_diff >= 0.1 else "warning",
                })

        print(f"Detected {len(regressions)} regressions")
        return regressions

Step 4: Build the Alert Engine

Send actionable alerts when regressions are detected.

import json
import os

class PerformanceAlertEngine:
    """Send performance regression alerts via Slack or email."""

    def __init__(self, slack_webhook: str = None):
        """Initialise with notification channels."""
        self.slack_webhook = slack_webhook or os.environ.get("SLACK_WEBHOOK_URL")

    def send_regression_alert(self, regressions: list[dict], store_url: str):
        """Send a formatted regression alert."""
        if not regressions:
            print("No regressions — no alert needed")
            return

        critical = [r for r in regressions if r["severity"] == "critical"]
        warnings = [r for r in regressions if r["severity"] == "warning"]

        # Build message
        lines = [f"*Performance Alert: {store_url}*"]
        lines.append(f"Detected {len(regressions)} regressions ({len(critical)} critical)")
        lines.append("")

        if critical:
            lines.append("*Critical:*")
            for r in critical:
                lines.append(f"• {r['page_label']} ({r['strategy']}): "
                             f"{r['metric']} changed from {r['baseline']} → {r['current']} "
                             f"({r['change']:+})")

        if warnings:
            lines.append("*Warnings:*")
            for r in warnings:
                lines.append(f"• {r['page_label']} ({r['strategy']}): "
                             f"{r['metric']} changed from {r['baseline']} → {r['current']} "
                             f"({r['change']:+})")

        message = "\n".join(lines)

        # Send to Slack
        if self.slack_webhook:
            requests.post(self.slack_webhook, json={"text": message})
            print("Slack alert sent")
        else:
            print("No Slack webhook configured")
            print(message)

    def generate_report(self, measurements: pd.DataFrame, regressions: list[dict]) -> str:
        """Generate an HTML performance report."""
        summary = measurements.groupby(["page_label", "strategy"]).agg(
            score=("performance_score", "mean"),
            lcp=("lcp_ms", "mean"),
            cls=("cls", "mean"),
        ).round(1).reset_index()

        report = f"""<h2>Performance Report — {datetime.now().strftime('%Y-%m-%d')}</h2>
<table>
<tr><th>Page</th><th>Device</th><th>Score</th><th>LCP (ms)</th><th>CLS</th></tr>
"""
        for _, row in summary.iterrows():
            score_colour = "#22c55e" if row["score"] >= 90 else "#f59e0b" if row["score"] >= 50 else "#ef4444"
            report += f"""<tr>
<td>{row['page_label']}</td>
<td>{row['strategy']}</td>
<td style="color:{score_colour}">{row['score']}</td>
<td>{row['lcp']:.0f}</td>
<td>{row['cls']:.3f}</td>
</tr>"""
        report += "</table>"

        if regressions:
            report += f"<p><strong>{len(regressions)} regressions detected</strong></p>"

        return report

Step 5: Schedule the Monitor

Run checks twice daily and send alerts when thresholds are breached.

from prefect import flow

@flow(name="shopify-performance-monitor")
def monitor_store_performance(
    store_url: str,
    psi_api_key: str,
    db_url: str = "sqlite:///speed_monitor.db",
):
    """Daily performance monitoring pipeline."""
    # Measure
    monitor = StoreMonitor(store_url, psi_api_key, db_url)
    measurements = monitor.run_full_check()

    # Analyse
    from sqlalchemy import create_engine
    engine = create_engine(db_url)
    baselines = BaselineCalculator(engine)
    baselines.calculate_baselines(days=14)
    regressions = baselines.detect_regressions(measurements)

    # Alert
    alerter = PerformanceAlertEngine()
    alerter.send_regression_alert(regressions, store_url)

    # Report
    report = alerter.generate_report(measurements, regressions)

    print(f"Monitoring complete: {len(measurements)} measurements, {len(regressions)} regressions")
    return {"measurements": len(measurements), "regressions": len(regressions)}


# Schedule twice daily
monitor_store_performance.serve(
    name="speed-monitor-twice-daily",
    schedules=[
        {"cron": "0 6 * * *", "timezone": "Europe/London"},
        {"cron": "0 18 * * *", "timezone": "Europe/London"},
    ],
)

What This Replaces

Manual Approach	Automated Monitoring
Run Lighthouse manually every few weeks	Automated daily measurements
No historical data — each test is isolated	Full history with trend analysis
Notice speed issues after conversion drops	Alerts on the day regression happens
Test homepage only	Multi-page monitoring with baselines
No threshold configuration	Configurable alert rules per metric
No team visibility into store health	Shared reports and Slack alerts

Next Steps

For the complete guide to diagnosing and fixing Shopify performance issues, see Why Is My Shopify Store Slow? Diagnose and Fix Performance. For fixing the most common Shopify speed problems, see How to Fix Slow Shopify Stores.

For building a self-updating dashboard from the monitoring data, see Shopify Automated Reporting. For scheduling the monitoring pipeline with Prefect, see Schedule and Orchestrate Workflows with Prefect.

Ecommerce optimisation services include performance monitoring setup, Core Web Vitals optimisation, and ongoing store speed management.

Get in touch to discuss setting up automated performance monitoring for your store.