Build a Production-Ready Analytics Dashboard with DuckDB + Streamlit

# data_layer.py
import duckdb
import pandas as pd
import numpy as np
from pathlib import Path

class SaaSDashboard:
 """Core data layer for a SaaS analytics dashboard"""
 
 def __init__(self, data_dir: str = "/tmp/saas_data"):
 self.data_dir = Path(data_dir)
 self.data_dir.mkdir(exist_ok=True)
 self.con = duckdb.connect(":memory:")
 self._load_data()
 
 def _load_data(self):
 """Load sample data (replace with your real data source)"""
 
 # Customer table
 np.random.seed(42)
 n_customers = 500
 plans = np.random.choice(
 ['free', 'starter', 'pro', 'enterprise'],
 n_customers,
 p=[0.3, 0.35, 0.25, 0.1]
 )
 
 customers = pd.DataFrame({
 'customer_id': range(1, n_customers + 1),
 'name': [f'Customer{i}' for i in range(1, n_customers + 1)],
 'plan': plans,
 'signup_date': pd.date_range('2025-01-01', periods=n_customers, freq='d'),
 'country': np.random.choice(
 ['US', 'CN', 'JP', 'DE', 'UK', 'KR', 'BR', 'IN'],
 n_customers
 )
 })
 
 # Usage data
 usage_records = []
 base_queries = {'free': 10, 'starter': 50, 'pro': 200, 'enterprise': 1000}
 
 for _ in range(10000):
 cid = np.random.randint(1, n_customers + 1)
 plan = plans[cid - 1]
 usage_records.append({
 'customer_id': cid,
 'usage_date': (pd.Timestamp.now() - pd.Timedelta(days=np.random.randint(0, 90))).strftime('%Y-%m-%d'),
 'query_count': max(0, int(np.random.normal(base_queries[plan], base_queries[plan] * 0.3))),
 'storage_gb': round(np.random.uniform(0.1, 50.0), 2),
 'api_calls': max(0, int(np.random.normal(base_queries[plan] * 2, base_queries[plan] * 0.5))),
 'error_rate': round(max(0, np.random.normal(0.02, 0.01)), 4)
 })
 
 usage = pd.DataFrame(usage_records)
 
 # Save to Parquet (faster queries in production)
 customers.to_parquet(self.data_dir / 'customers.parquet')
 usage.to_parquet(self.data_dir / 'usage.parquet')
 
 # Register in DuckDB
 self.con.execute("CREATE TABLE customers AS SELECT * FROM read_parquet(?)", 
 [str(self.data_dir / 'customers.parquet')])
 self.con.execute("CREATE TABLE usage AS SELECT * FROM read_parquet(?)",
 [str(self.data_dir / 'usage.parquet')])
 
 print(f"✅ Data loaded: {len(customers)} customers, {len(usage)} usage records")
 
 def get_mrr_dashboard(self) -> pd.DataFrame:
 """MRR core metrics"""
 return self.con.execute("""
 SELECT 
 plan,
 COUNT(DISTINCT customer_id) as customer_count,
 SUM(CASE 
 WHEN plan = 'free' THEN 0
 WHEN plan = 'starter' THEN 29
 WHEN plan = 'pro' THEN 99
 WHEN plan = 'enterprise' THEN 499
 END) as total_mrr,
 ROUND(
 SUM(CASE 
 WHEN plan = 'free' THEN 0
 WHEN plan = 'starter' THEN 29
 WHEN plan = 'pro' THEN 99
 WHEN plan = 'enterprise' THEN 499
 END) * 12, 0
 ) as arr_projection
 FROM customers
 GROUP BY plan
 ORDER BY total_mrr DESC
 """).fetchdf()
 
 def get_churn_risk(self, top_n: int = 20) -> pd.DataFrame:
 """Churn risk customer ranking"""
 return self.con.execute("""
 SELECT 
 u.customer_id,
 c.name,
 c.plan,
 c.country,
 COUNT(DISTINCT u.usage_date) as active_days,
 ROUND(AVG(u.query_count), 1) as avg_daily_queries,
 ROUND(AVG(u.error_rate) * 100, 2) as avg_error_rate_pct,
 CASE 
 WHEN AVG(u.query_count) < 5 THEN '🔴 High Risk'
 WHEN AVG(u.query_count) < 20 THEN '🟡 Medium Risk'
 ELSE '🟢 Normal'
 END as risk_level
 FROM usage u
 JOIN customers c ON u.customer_id = c.customer_id
 GROUP BY u.customer_id, c.name, c.plan, c.country
 ORDER BY avg_daily_queries ASC
 LIMIT ?
 """, [top_n]).fetchdf()
 
 def get_usage_trend(self, days: int = 30) -> pd.DataFrame:
 """Usage trend (daily aggregation)"""
 return self.con.execute("""
 SELECT 
 usage_date,
 COUNT(DISTINCT customer_id) as active_customers,
 SUM(query_count) as total_queries,
 ROUND(AVG(query_count), 1) as avg_queries_per_user,
 SUM(api_calls) as total_api_calls,
 ROUND(AVG(error_rate) * 100, 2) as avg_error_rate_pct
 FROM usage
 WHERE usage_date >= DATE(CURRENT_DATE - INTERVAL ? DAYS)
 GROUP BY usage_date
 ORDER BY usage_date
 """, [days]).fetchdf()
 
 def get_country_breakdown(self) -> pd.DataFrame:
 """Country/region breakdown"""
 return self.con.execute("""
 SELECT 
 c.country,
 COUNT(DISTINCT c.customer_id) as customers,
 SUM(CASE 
 WHEN c.plan = 'starter' THEN 29
 WHEN c.plan = 'pro' THEN 99
 WHEN c.plan = 'enterprise' THEN 499
 ELSE 0
 END) as mrr,
 ROUND(AVG(u.query_count), 1) as avg_usage
 FROM customers c
 LEFT JOIN usage u ON c.customer_id = u.customer_id
 GROUP BY c.country
 ORDER BY mrr DESC
 """).fetchdf()
 
 def get_customer_lifecycle(self) -> pd.DataFrame:
 """Customer lifecycle analysis"""
 return self.con.execute("""
 SELECT 
 DATE_TRUNC('month', signup_date) as signup_month,
 COUNT(*) as new_customers,
 SUM(CASE WHEN plan = 'free' THEN 0
 WHEN plan = 'starter' THEN 29
 WHEN plan = 'pro' THEN 99
 WHEN plan = 'enterprise' THEN 499
 ELSE 0 END) as month_mrr,
 ROUND(
 SUM(CASE WHEN plan IN ('starter','pro','enterprise') THEN 1 ELSE 0 END) * 100.0 / COUNT(*),
 1
 ) as paid_rate_pct
 FROM customers
 GROUP BY DATE_TRUNC('month', signup_date)
 ORDER BY signup_month
 """).fetchdf()
 
 def get_revenue_forecast(self) -> pd.DataFrame:
 """Revenue forecast (simple linear)"""
 current_mrr = self.get_mrr_dashboard()['total_mrr'].sum()
 
 forecast = pd.DataFrame({
 'month': ['This Month', 'Next Month', '3 Months', '6 Months'],
 'mrr': [
 current_mrr,
 round(current_mrr * 1.05, 0),
 round(current_mrr * 1.15, 0),
 round(current_mrr * 1.30, 0)
 ]
 })
 return forecast

# app.py
import streamlit as st
import pandas as pd
from data_layer import SaaSDashboard

st.set_page_config(page_title="SaaS Analytics Dashboard", layout="wide")

@st.cache_data(ttl=60)
def get_dashboard():
 return SaaSDashboard()

dash = get_dashboard()

st.title("📊 SaaS Analytics Dashboard")
st.markdown("Built with DuckDB + Streamlit")

# Top KPI cards
col1, col2, col3, col4 = st.columns(4)
mrr_df = dash.get_mrr_dashboard()
total_mrr = mrr_df['total_mrr'].sum()
total_customers = mrr_df['customer_count'].sum()
paid_customers = mrr_df[mrr_df['plan'] != 'free']['customer_count'].sum()
arr = mrr_df['arr_projection'].sum()

col1.metric("Total MRR", f"${total_mrr:,}")
col2.metric("Active Customers", f"{total_customers:,}")
col3.metric("Paying Customers", f"{paid_customers:,}")
col4.metric("ARR Projection", f"${arr:,}")

# Tab navigation
tab1, tab2, tab3, tab4 = st.tabs(["💰 Revenue", "⚠️ Churn Risk", "📈 Trends", "🌍 Regions"])

# Tab 1: Revenue Overview
with tab1:
 st.subheader("MRR by Plan")
 st.dataframe(mrr_df, use_container_width=True)
 
 col_a, col_b = st.columns(2)
 with col_a:
 import plotly.express as px
 st.plotly_chart(
 px.pie(mrr_df, values='total_mrr', labels='plan', title='MRR Distribution'),
 use_container_width=True
 )
 with col_b:
 forecast = dash.get_revenue_forecast()
 st.subheader("Revenue Forecast")
 st.dataframe(forecast, use_container_width=True)

# Tab 2: Churn Risk
with tab2:
 st.subheader("Top 20 Churn Risk Customers")
 churn_df = dash.get_churn_risk(20)
 st.dataframe(churn_df, use_container_width=True)
 
 risk_counts = churn_df['risk_level'].value_counts()
 st.bar_chart(risk_counts)

# Tab 3: Usage Trends
with tab3:
 days = st.slider("Last N days", 7, 90, 30)
 trend_df = dash.get_usage_trend(days)
 
 col_x, col_y = st.columns(2)
 with col_x:
 import plotly.express as px
 st.plotly_chart(
 px.line(trend_df, x='usage_date', y='active_customers', title='Active Customers'),
 use_container_width=True
 )
 with col_y:
 import plotly.express as px
 st.plotly_chart(
 px.line(trend_df, x='usage_date', y='total_queries', title='Total Queries'),
 use_container_width=True
 )

# Tab 4: Regional Breakdown
with tab4:
 st.subheader("Country-wise MRR & Usage")
 country_df = dash.get_country_breakdown()
 st.dataframe(country_df, use_container_width=True)
 
 import plotly.express as px
 st.plotly_chart(
 px.bar(country_df, x='country', y='mrr', title='MRR by Country'),
 use_container_width=True
 )

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8080
CMD ["streamlit", "run", "app.py", "--server.port=8080", "--server.headless=true"]

streamlit run app.py