Build a Self-Updating Shopify Analytics Dashboard with Python

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

class ShopifyMetrics:
    """Collect and aggregate Shopify store metrics."""

    def __init__(self, shop_url: str, api_token: str, db_url: str = "sqlite:///shopify_dashboard.db"):
        """Initialise API and database connections."""
        self.base_url = f"https://{shop_url}/admin/api/2026-01"
        self.session = requests.Session()
        self.session.headers["X-Shopify-Access-Token"] = api_token
        self.engine = create_engine(db_url)
        print(f"Connected to {shop_url}")

    def _fetch_all(self, endpoint: str, params: dict) -> list:
        """Fetch all pages from a Shopify endpoint."""
        url = f"{self.base_url}/{endpoint}"
        results = []
        while url:
            resp = self.session.get(url, params=params)
            resp.raise_for_status()
            data = resp.json()
            key = list(data.keys())[0]
            results.extend(data[key])
            url = None
            params = {}
            link = resp.headers.get("Link", "")
            if 'rel="next"' in link:
                for part in link.split(","):
                    if 'rel="next"' in part:
                        url = part.split("<")[1].split(">")[0]
        return results

    def collect_daily_snapshot(self) -> dict:
        """Collect today's metrics snapshot."""
        yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
        today = datetime.now().strftime("%Y-%m-%d")

        # Fetch yesterday's orders
        orders = self._fetch_all("orders.json", {
            "created_at_min": f"{yesterday}T00:00:00Z",
            "created_at_max": f"{today}T00:00:00Z",
            "status": "any",
            "limit": 250,
        })

        # Calculate metrics
        revenue = sum(float(o["total_price"]) for o in orders)
        order_count = len(orders)
        aov = revenue / order_count if order_count > 0 else 0
        items = sum(len(o.get("line_items", [])) for o in orders)
        customers = len(set(o.get("email", "") for o in orders if o.get("email")))

        snapshot = {
            "date": yesterday,
            "revenue": round(revenue, 2),
            "orders": order_count,
            "avg_order_value": round(aov, 2),
            "items_sold": items,
            "unique_customers": customers,
            "collected_at": datetime.now().isoformat(),
        }

        # Store in database
        df = pd.DataFrame([snapshot])
        df.to_sql("daily_metrics", self.engine, if_exists="append", index=False)
        print(f"Snapshot stored for {yesterday}: {order_count} orders, £{revenue:.2f}")

        return snapshot

class InventoryTracker:
    """Track inventory levels and detect low-stock conditions."""

    def __init__(self, metrics: ShopifyMetrics, low_stock_threshold: int = 10):
        """Initialise with a metrics collector."""
        self.metrics = metrics
        self.threshold = low_stock_threshold

    def snapshot_inventory(self) -> pd.DataFrame:
        """Capture current inventory levels for all products."""
        products = self.metrics._fetch_all("products.json", {"limit": 250})

        inventory = []
        for product in products:
            for variant in product.get("variants", []):
                inventory.append({
                    "date": datetime.now().strftime("%Y-%m-%d"),
                    "product_id": product["id"],
                    "product_title": product["title"],
                    "variant_id": variant["id"],
                    "variant_title": variant.get("title", "Default"),
                    "sku": variant.get("sku", ""),
                    "inventory_quantity": variant.get("inventory_quantity", 0),
                    "price": float(variant.get("price", 0)),
                })

        df = pd.DataFrame(inventory)
        df.to_sql("inventory_snapshots", self.metrics.engine, if_exists="append", index=False)
        print(f"Tracked inventory for {len(inventory)} variants")
        return df

    def check_low_stock(self, df: pd.DataFrame) -> pd.DataFrame:
        """Identify variants below the stock threshold."""
        low = df[df["inventory_quantity"] <= self.threshold].copy()
        low["severity"] = low["inventory_quantity"].apply(
            lambda q: "critical" if q == 0 else "warning"
        )
        if len(low) > 0:
            print(f"⚠ {len(low)} variants at low stock:")
            for _, row in low.iterrows():
                print(f"  {row['product_title']} ({row['sku']}): {row['inventory_quantity']} units")
        return low

    def velocity_forecast(self, days: int = 30) -> pd.DataFrame:
        """Estimate days until stockout based on recent sales velocity."""
        query = f"""
            SELECT variant_id, product_title, sku,
                   MIN(inventory_quantity) as current_stock,
                   MAX(inventory_quantity) - MIN(inventory_quantity) as units_sold
            FROM inventory_snapshots
            WHERE date >= date('now', '-{days} days')
            GROUP BY variant_id
            HAVING units_sold > 0
        """
        df = pd.read_sql(query, self.metrics.engine)
        df["daily_velocity"] = (df["units_sold"] / days).round(1)
        df["days_until_stockout"] = (df["current_stock"] / df["daily_velocity"]).round(0)
        return df.sort_values("days_until_stockout")

class TrendAnalyser:
    """Compare current metrics to historical periods."""

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

    def weekly_comparison(self) -> dict:
        """Compare this week to last week."""
        query = """
            SELECT
                CASE
                    WHEN date >= date('now', '-7 days') THEN 'this_week'
                    WHEN date >= date('now', '-14 days') THEN 'last_week'
                END as period,
                SUM(revenue) as revenue,
                SUM(orders) as orders,
                AVG(avg_order_value) as aov
            FROM daily_metrics
            WHERE date >= date('now', '-14 days')
            GROUP BY period
        """
        df = pd.read_sql(query, self.engine)

        this_week = df[df["period"] == "this_week"].iloc[0] if len(df[df["period"] == "this_week"]) > 0 else None
        last_week = df[df["period"] == "last_week"].iloc[0] if len(df[df["period"] == "last_week"]) > 0 else None

        if this_week is None or last_week is None:
            return {"status": "insufficient_data"}

        return {
            "revenue": {
                "current": this_week["revenue"],
                "previous": last_week["revenue"],
                "change_pct": round(((this_week["revenue"] - last_week["revenue"]) / last_week["revenue"]) * 100, 1),
            },
            "orders": {
                "current": int(this_week["orders"]),
                "previous": int(last_week["orders"]),
                "change_pct": round(((this_week["orders"] - last_week["orders"]) / last_week["orders"]) * 100, 1),
            },
            "aov": {
                "current": round(this_week["aov"], 2),
                "previous": round(last_week["aov"], 2),
                "change_pct": round(((this_week["aov"] - last_week["aov"]) / last_week["aov"]) * 100, 1),
            },
        }

class AlertEngine:
    """Configurable alert engine for Shopify metrics."""

    DEFAULT_THRESHOLDS = {
        "revenue_drop_pct": -20,         # Alert if revenue drops 20%+ vs last week
        "low_stock_units": 10,           # Alert when stock below 10 units
        "zero_stock_critical": True,     # Always alert on stockouts
        "order_drop_pct": -25,           # Alert if orders drop 25%+ vs last week
        "aov_drop_pct": -15,            # Alert if AOV drops 15%+
    }

    def __init__(self, thresholds: dict = None):
        """Initialise with custom or default thresholds."""
        self.thresholds = {**self.DEFAULT_THRESHOLDS, **(thresholds or {})}

    def evaluate(self, trends: dict, low_stock: pd.DataFrame) -> list[dict]:
        """Evaluate all alert conditions and return triggered alerts."""
        alerts = []

        # Revenue alert
        if trends.get("revenue", {}).get("change_pct", 0) <= self.thresholds["revenue_drop_pct"]:
            alerts.append({
                "type": "revenue_drop",
                "severity": "high",
                "message": f"Revenue dropped {trends['revenue']['change_pct']}% vs last week",
                "current": trends["revenue"]["current"],
                "previous": trends["revenue"]["previous"],
            })

        # Order volume alert
        if trends.get("orders", {}).get("change_pct", 0) <= self.thresholds["order_drop_pct"]:
            alerts.append({
                "type": "order_drop",
                "severity": "high",
                "message": f"Orders dropped {trends['orders']['change_pct']}% vs last week",
            })

        # Stockout alerts
        critical = low_stock[low_stock["severity"] == "critical"]
        if len(critical) > 0 and self.thresholds["zero_stock_critical"]:
            alerts.append({
                "type": "stockout",
                "severity": "critical",
                "message": f"{len(critical)} products out of stock",
                "products": critical["product_title"].tolist(),
            })

        print(f"Evaluated alerts: {len(alerts)} triggered")
        return alerts

import json

def generate_dashboard_data(metrics: ShopifyMetrics, db_url: str) -> dict:
    """Generate a complete dashboard data payload."""
    engine = create_engine(db_url)
    tracker = InventoryTracker(metrics)
    trends = TrendAnalyser(engine)
    alerts = AlertEngine()

    # Collect fresh data
    snapshot = metrics.collect_daily_snapshot()
    inventory = tracker.snapshot_inventory()
    low_stock = tracker.check_low_stock(inventory)
    weekly = trends.weekly_comparison()
    triggered = alerts.evaluate(weekly, low_stock)

    dashboard = {
        "generated_at": datetime.now().isoformat(),
        "today": snapshot,
        "trends": weekly,
        "inventory": {
            "total_variants": len(inventory),
            "low_stock_count": len(low_stock),
            "stockouts": len(low_stock[low_stock["severity"] == "critical"]),
        },
        "alerts": triggered,
    }

    with open("dashboard.json", "w") as f:
        json.dump(dashboard, f, indent=2, default=str)

    print(f"Dashboard data written ({len(triggered)} active alerts)")
    return dashboard