pg_trickle — Real-time Fraud Detection

""" pg_trickle demo — fraud detection 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 Fraud Detection

pg_trickle Real-time Fraud Detection Pipeline

connecting…

—

Total Transactions

—

LOW Risk

—

MEDIUM Risk

—

HIGH Risk

LOW MEDIUM HIGH

⚠️ Recent HIGH / MEDIUM Risk Alerts

#	User	Merchant	Category	Amount	Risk

🏪 Merchant Risk Leaderboard

Merchant	Cat.	High	Med	Risk%

👤 User Velocity (top 10)

User	🌍	Txns	Avg $

🌍 Country Overview

Country	Users	Txns	Volume $

📦 Category Volume

Category	Txns	Avg $	Users

🔬 Merchant Tier Stats DIFFERENTIAL showcase — change ratio ~0.07

Merchant	Cat.	Tier	Risk	Last Changed

Only tier rotations trigger output changes (~every 30 cycles). No fast-growing sources — only merchant_risk_tier (15 rows). Mode Advisor recommends KEEP DIFFERENTIAL here.

🏷️ Live Merchant Risk Tiers rotates ~every 30 cycles

Merchant	Category	Tier

🏆 Top 10 Risky Merchants Leaderboard DIFFERENTIAL showcase — change ratio ~0.2–0.3

Rank	Merchant	Category	Total Txns	High	Medium	Risk%

Fixed cardinality (10 rows) means only rank shifts change, not 100% of output rows. Change ratio ≈ 0.2–0.3 (typically 2–3 merchants move in/out per cycle). Mode Advisor recommends KEEP DIFFERENTIAL here.

🎯 Alert Summary

Risk Level	Transactions	Avg Amount	Total Volume

🔍 Risk Distribution by Category

Category	Low	Medium	High	High%

⚡ 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 ──────────── ────────────────────── ───────────────────── ────────────────────── ┌────────────┐ ┌──────────────────┐ │ users │────────►│ user_velocity │─────────────────────────►┌──────────────────┐ └────────────┘ │ (DIFFERENTIAL) │ │ country_risk │ └──────┬───────────┘ │ (DIFFERENTIAL) │ │ └──────────────────┘ ┌────────────┐ │ ┌──────────────────┐ │transactions│────────────────┼─►│ merchant_stats │ │ (stream) │ │ │ (DIFFERENTIAL) │ └────────────┘ │ └──────┬───────────┘ │ │ │ │ ┌─────────────┼─────────┘ ← DIAMOND DEPENDENCY │ │ │ │ ▼ ▼ ┌───────────────────────┐ │ ┌────────────────────────┐ │ alert_summary │ │ │ risk_scores │────────────────────────────────►│ (DIFFERENTIAL) │ │ │ (FULL, calculated) │ └───────────────────────┘ │ └────────────────────────┘ │ ┌───────────────────────┐ │ │ top_risky_merchants │ └──────────────────────────────────────────────────────────────►│ (DIFFERENTIAL) │ └───────────────────────┘ ┌────────────┐ ┌──────────────────┐ │ merchants │────────►│ category_volume │ └────────────┘ │ (DIFFERENTIAL) │ └──────────────────┘ ┌───────────────────┐ ┌─────────────────────┐ │ merchant_risk_tier│──►│ merchant_tier_stats │ ← DIFFERENTIAL SHOWCASE #1 │ (slowly-changing) │ │ (DIFFERENTIAL 5s) │ change ratio ~0.07 └───────────────────┘ └─────────────────────┘ ┌──────────────────────┐ ┌─────────────────────┐ │ top_risky_merchants │──►│ top_10_risky_ │ ← DIFFERENTIAL SHOWCASE #2 │ (DIFFERENTIAL) │ │ merchants │ change ratio ~0.2–0.3 └──────────────────────┘ │ (DIFFERENTIAL 5s) │ └─────────────────────┘ """ # ── Data queries ─────────────────────────────────────────────────────────────── def get_data(conn, safe_query, serialize) -> dict: recent_alerts = safe_query(conn, """ SELECT txn_id, user_name, merchant_name, merchant_category, amount, risk_level FROM risk_scores WHERE risk_level IN ('HIGH', 'MEDIUM') ORDER BY txn_id DESC LIMIT 15 """) alert_summary = safe_query(conn, """ SELECT risk_level, txn_count, total_amount, avg_amount FROM alert_summary ORDER BY CASE risk_level WHEN 'HIGH' THEN 0 WHEN 'MEDIUM' THEN 1 ELSE 2 END """) top_risky_merchants = safe_query(conn, """ SELECT merchant_name, merchant_category, total_txns, high_risk_count, medium_risk_count, risk_rate_pct FROM top_risky_merchants """) user_velocity = safe_query(conn, """ SELECT user_name, country, txn_count, total_spent, avg_txn_amount FROM user_velocity WHERE txn_count > 0 """) country_risk = safe_query(conn, """ SELECT country, user_count, total_txns, total_volume FROM country_risk """) category_volume = safe_query(conn, """ SELECT category, txn_count, total_volume, avg_txn_amount, unique_users FROM category_volume WHERE txn_count > 0 """) st_status = safe_query(conn, """ SELECT name, status, refresh_mode, is_populated, schedule FROM pgtrickle.pgt_status() ORDER BY name """) risk_by_category = safe_query(conn, """ SELECT merchant_category, COUNT(*) FILTER (WHERE risk_level = 'LOW') AS low_count, COUNT(*) FILTER (WHERE risk_level = 'MEDIUM') AS medium_count, COUNT(*) FILTER (WHERE risk_level = 'HIGH') AS high_count, ROUND(100.0 * COUNT(*) FILTER (WHERE risk_level = 'HIGH') / NULLIF(COUNT(*), 0), 1) AS high_pct FROM risk_scores GROUP BY merchant_category ORDER BY high_pct DESC NULLS LAST """) merchant_tier_stats = safe_query(conn, """ SELECT merchant_id, merchant_name, category, merchant_tier, risk_score, tier_last_changed FROM merchant_tier_stats ORDER BY merchant_id """) merchant_tiers = safe_query(conn, """ SELECT mrt.merchant_id, m.name AS merchant_name, mrt.tier, mrt.updated_at FROM merchant_risk_tier mrt JOIN merchants m ON m.id = mrt.merchant_id ORDER BY mrt.merchant_id """) top_10_risky = safe_query(conn, """ SELECT rank, merchant_name, merchant_category, total_txns, high_risk_count, medium_risk_count, risk_rate_pct FROM top_10_risky_merchants ORDER BY rank """) return { "recent_alerts": serialize(recent_alerts), "alert_summary": serialize(alert_summary), "top_risky_merchants": serialize(top_risky_merchants), "user_velocity": serialize(user_velocity), "country_risk": serialize(country_risk), "category_volume": serialize(category_volume), "st_status": serialize(st_status), "risk_by_category": serialize(risk_by_category), "merchant_tier_stats": serialize(merchant_tier_stats), "merchant_tiers": serialize(merchant_tiers), "top_10_risky": serialize(top_10_risky), }