How Solo Developers Can Build a Personal Data Analytics SaaS with DuckDB

import duckdb

class EcomAnalyzer:
 def __init__(self, db_path="ecom.db"):
 self.con = duckdb.connect(db_path)
 self._init_schema()

 def _init_schema(self):
 """Initialize table structure"""
 self.con.execute("""
 CREATE TABLE IF NOT EXISTS orders (
 order_id VARCHAR,
 order_date DATE,
 product_id VARCHAR,
 category VARCHAR,
 quantity BIGINT,
 unit_price DOUBLE,
 customer_id VARCHAR,
 region VARCHAR,
 status VARCHAR
 )
 """)

 def load_csv(self, file_path: str):
 """Auto-detect schema and load CSV"""
 self.con.execute(f"""
 INSERT INTO orders
 SELECT * FROM read_csv_auto('{file_path}', 
 auto_detect=true,
 header=true,
 sep=','
 )
 """)

 def sales_trend_30d(self) -> str:
 """Sales trend for the last 30 days"""
 result = self.con.execute("""
 SELECT
 order_date,
 COUNT(*) AS order_count,
 SUM(quantity * unit_price) AS daily_revenue,
 AVG(quantity * unit_price) AS avg_order_value
 FROM orders
 WHERE order_date >= CURRENT_DATE - INTERVAL 30 DAY
 AND status != 'refunded'
 GROUP BY order_date
 ORDER BY order_date
 """).fetchdf()
 return result.to_json(orient='records', date_format='iso')

 def profit_ranking(self, top_n: int = 20) -> str:
 """Product profit ranking: find the most and least profitable products"""
 result = self.con.execute("""
 SELECT
 product_id,
 category,
 SUM(quantity) AS total_sold,
 SUM(quantity * unit_price) AS total_revenue,
 COUNT(DISTINCT customer_id) AS unique_customers,
 COUNT(DISTINCT customer_id) * 1.5 AS estimated_cogs,
 SUM(quantity * unit_price) 
 - COUNT(DISTINCT customer_id) * 1.5 AS estimated_profit
 FROM orders
 WHERE status != 'refunded'
 GROUP BY product_id, category
 ORDER BY estimated_profit DESC
 LIMIT ?
 """, [top_n])
 return result.fetchdf().to_json(orient='records')

 def weekly_pattern(self) -> str:
 """Weekly sales pattern by day of week"""
 result = self.con.execute("""
 SELECT
 CASE EXTRACT(DOW FROM order_date)
 WHEN 0 THEN 'Sunday'
 WHEN 1 THEN 'Monday'
 WHEN 2 THEN 'Tuesday'
 WHEN 3 THEN 'Wednesday'
 WHEN 4 THEN 'Thursday'
 WHEN 5 THEN 'Friday'
 WHEN 6 THEN 'Saturday'
 END AS day_name,
 SUM(quantity) AS total_quantity,
 SUM(quantity * unit_price) AS total_revenue,
 COUNT(*) AS order_count
 FROM orders
 GROUP BY EXTRACT(DOW FROM order_date)
 ORDER BY EXTRACT(DOW FROM order_date)
 """).fetchdf()
 return result.to_json(orient='records')

 def repurchase_rate(self) -> str:
 """Customer repurchase rate analysis"""
 result = self.con.execute("""
 WITH customer_orders AS (
 SELECT
 customer_id,
 COUNT(DISTINCT DATE_TRUNC('month', order_date)) AS active_months,
 COUNT(*) AS total_orders,
 SUM(quantity * unit_price) AS lifetime_value
 FROM orders
 WHERE status != 'refunded'
 GROUP BY customer_id
 )
 SELECT
 AVG(CASE WHEN total_orders > 1 THEN 1.0 ELSE 0.0 END) AS repurchase_rate,
 AVG(lifetime_value) AS avg_customer_value,
 PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY lifetime_value) AS top_5p_value
 FROM customer_orders
 """).fetchdf()
 return result.to_json(orient='records')

 def generate_report(self) -> dict:
 """Generate a complete analysis report"""
 return {
 "summary": self.sales_trend_30d(),
 "profit_ranking": self.profit_ranking(),
 "weekly_pattern": self.weekly_pattern(),
 "repurchase_rate": self.repurchase_rate(),
 "generated_at": str(duckdb.query("SELECT NOW()").fetchone()[0])
 }

# Load Excel
self.con.execute("SELECT * FROM read_xlsx('orders.xlsx')")

# Load Parquet (better performance)
self.con.execute("SELECT * FROM read_parquet('orders_*.parquet')")

# Load JSON
self.con.execute("SELECT * FROM read_json_auto('orders.json')")

 def data_quality_check(self) -> dict:
 """Data quality checks"""
 checks = {
 "missing_order_ids": self.con.execute("""
 SELECT COUNT(*) FROM orders WHERE order_id IS NULL
 """).fetchone()[0],
 "invalid_dates": self.con.execute("""
 SELECT COUNT(*) FROM orders 
 WHERE order_date IS NULL OR order_date > CURRENT_DATE
 """).fetchone()[0],
 "negative_prices": self.con.execute("""
 SELECT COUNT(*) FROM orders WHERE unit_price < 0
 """).fetchone()[0],
 }
 return checks

 def cache_monthly_summary(self, year: int, month: int):
 """Cache monthly summary"""
 self.con.execute("""
 CREATE OR REPLACE VIEW monthly_summary_vw AS
 SELECT
 DATE_TRUNC('month', order_date) AS month,
 COUNT(*) AS total_orders,
 SUM(quantity * unit_price) AS total_revenue,
 COUNT(DISTINCT customer_id) AS unique_customers
 FROM orders
 WHERE EXTRACT(YEAR FROM order_date) = ?
 AND EXTRACT(MONTH FROM order_date) = ?
 GROUP BY 1
 """, [year, month])