pg_trickle — Real-time E-commerce Analytics

""" pg_trickle demo — e-commerce analytics scenario: HTML, DAG, and data queries. Exposes: HTML — complete page HTML with {{ dag }} placeholder DAG_DIAGRAM — ASCII art DAG to substitute into {{ dag }} get_data(conn, safe_query, serialize) → dict for /api/data """ # ── Full page HTML ───────────────────────────────────────────────────────────── HTML = r""" pg_trickle — Real-time E-commerce Analytics

pg_trickle Real-time E-commerce Analytics

connecting…

—

Total Orders

—

Total Revenue

—

Avg Order Value

—

Active Customers

📦 Revenue by Category

Category	Orders	Units	Revenue	Avg Price

🏆 Top Products by Revenue

Product	Category	Orders	Units	Revenue

👤 Top 10 Customers DIFFERENTIAL showcase — change ratio ~0.1–0.2

Rank	Customer	🌍	Orders	Total Spent	Avg Order

Fixed cardinality (10 rows): only rank shifts trigger output changes. Mode Advisor recommends KEEP DIFFERENTIAL here.

🌍 Revenue by Country

Country	Customers	Orders	Revenue

🏷️ Catalog Price Impact DIFFERENTIAL showcase — change ratio ~0.07

Product	Category	Base Price	Current Price	Delta $	Delta %	Last Updated

Depends only on product_catalog (1 of 15 products repriced per ~30 cycles). Change ratio ≈ 0.07. Mode Advisor recommends KEEP DIFFERENTIAL here.

⚡ Stream Table Status

Name	Mode	Schedule	Status

📊 DAG Topology — click to expand

{{ dag }}

🏥 Stream Table Health — pgtrickle.pgt_status()

Name	Mode	Schedule	Populated	Last Refresh	Duration	Rows Δ	Status

🌳 Dependency Tree — pgtrickle.dependency_tree()

loading…

📜 Refresh History — pgtrickle.pgt_refresh_history

Table	Action	Started	ms	Rows Δ	Status

⚡ Refresh Efficiency — pgtrickle.refresh_efficiency()

Table	Total	DIFF	FULL	Avg DIFF ms	Avg FULL ms	Speedup	Avg Change Ratio

🤖 Refresh Mode Advisor — pgtrickle.recommend_refresh_mode()

Table	Current	Effective	Recommendation	Confidence	Reason

""" # ── DAG diagram ──────────────────────────────────────────────────────────────── DAG_DIAGRAM = r""" Base tables Layer 1 — Silver Layer 2 — Gold Layer 3 — Platinum ──────────── ────────────────────── ───────────────────── ────────────────────── ┌────────────┐ ┌──────────────────┐ │ customers │──────►│ customer_stats │──────────────────────────────►┌──────────────────┐ └────────────┘ │ (DIFFERENTIAL) │ │ country_revenue │ └──────────────────┘ │ (DIFFERENTIAL) │ │ └──────────────────┘ ┌────────────┐ │ │ orders │────────────────┘ │ (stream) │ └──────┬─────┘ │ ┌────────────────┐ └───────►│ product_sales │ ┌────────────┐ │ (DIFFERENTIAL) │ │ products │─►│ │ │ categories │ └────────────────┘ └──────┬─────┘ │ ┌─────────────────┐ └───────►│ category_revenue│ │ (DIFFERENTIAL) │ └─────────────────┘ ┌────────────────────┐ ┌──────────────────────┐ │ product_catalog │──►│ catalog_price_impact │ ← DIFFERENTIAL SHOWCASE #1 │ (slowly-changing) │ │ (DIFFERENTIAL 5s) │ change ratio ~0.07 └────────────────────┘ └──────────────────────┘ ┌──────────────────┐ ┌──────────────────┐ │ customer_stats │──►│ top_10_customers│ ← DIFFERENTIAL SHOWCASE #2 │ (DIFFERENTIAL) │ │ (DIFFERENTIAL) │ change ratio ~0.1–0.2 └──────────────────┘ │ LIMIT 10 │ (fixed-cardinality output) └──────────────────┘ """ # ── Data queries ─────────────────────────────────────────────────────────────── def get_data(conn, safe_query, serialize) -> dict: category_revenue = safe_query(conn, """ SELECT category_name AS category, order_count, units_sold, revenue, avg_price, unique_customers FROM category_revenue ORDER BY revenue DESC NULLS LAST """) product_sales = safe_query(conn, """ SELECT product_name, category_name AS category, order_count, units_sold, revenue, avg_selling_price FROM product_sales ORDER BY revenue DESC NULLS LAST """) customer_stats = safe_query(conn, """ SELECT customer_name, country, order_count, total_spent, avg_order_value FROM customer_stats WHERE order_count > 0 """) top_10_customers = safe_query(conn, """ SELECT rank, customer_name, country, order_count, total_spent, avg_order_value FROM top_10_customers ORDER BY rank """) country_revenue = safe_query(conn, """ SELECT country, customer_count, total_orders, total_revenue, avg_order_value FROM country_revenue ORDER BY total_revenue DESC NULLS LAST """) catalog_price_impact = safe_query(conn, """ SELECT product_name, category_name AS category, base_price, current_price, price_delta, pct_change, price_last_updated FROM catalog_price_impact ORDER BY product_name """) st_status = safe_query(conn, """ SELECT name, status, refresh_mode, is_populated, schedule FROM pgtrickle.pgt_status() ORDER BY name """) return { "category_revenue": serialize(category_revenue), "product_sales": serialize(product_sales), "customer_stats": serialize(customer_stats), "top_10_customers": serialize(top_10_customers), "country_revenue": serialize(country_revenue), "catalog_price_impact": serialize(catalog_price_impact), "st_status": serialize(st_status), }