DuckDB Zero-Deploy Report System: Build Automated Financial Analysis Tools with read_csv_auto + CTE

import duckdb
import os
import yaml
from datetime import datetime
from pathlib import Path


class FinancialReportGenerator:
 def __init__(self, config_path="config/settings.yaml"):
 with open(config_path, 'r', encoding='utf-8') as f:
 self.config = yaml.safe_load(f)
 self.conn = duckdb.connect(":memory:")
 
 def load_data(self, data_dir: str):
 """Auto-load all CSV files in directory"""
 data_path = Path(data_dir)
 for csv_file in data_path.glob("*.csv"):
 table_name = csv_file.stem
 # read_csv_auto infers column types automatically
 self.conn.execute(f"CREATE TABLE {table_name} AS SELECT * FROM read_csv_auto('{csv_file}')")
 print(f"✅ Loaded: {csv_file.name} ({self.conn.execute(f'SELECT COUNT(*) FROM {table_name}').fetchone()[0]} rows)")

 def generate_financial_analysis(self) -> dict:
 """Generate core financial metrics analysis"""
 
 # Monthly profit analysis: CTE + JOIN in one step
 profit_query = """
 WITH monthly_sales AS (
 SELECT 
 strftime(date, '%Y-%m') as month, 
 SUM(amount) as revenue
 FROM sales 
 GROUP BY month
 ),
 monthly_expenses AS (
 SELECT 
 strftime(date, '%Y-%m') as month, 
 SUM(amount) as cost
 FROM expenses 
 GROUP BY month
 )
 SELECT 
 s.month,
 ROUND(s.revenue, 2) as revenue,
 ROUND(e.cost, 2) as expenses,
 ROUND(s.revenue - e.cost, 2) as profit,
 ROUND((s.revenue - e.cost) / s.revenue * 100, 2) as margin_pct,
 CASE 
 WHEN (s.revenue - e.cost) / s.revenue > 0.8 THEN '🟢 Excellent'
 WHEN (s.revenue - e.cost) / s.revenue > 0.6 THEN '🟡 Good'
 ELSE '🔴 Needs Attention'
 END as health_status
 FROM monthly_sales s
 LEFT JOIN monthly_expenses e ON s.month = e.month
 ORDER BY s.month
 """
 
 profit_data = self.conn.execute(profit_query).fetchall()
 return {
 "monthly_profit": profit_data,
 "total_revenue": sum(row[1] for row in profit_data),
 "total_expenses": sum(row[2] for row in profit_data),
 "total_profit": sum(row[3] for row in profit_data),
 "avg_margin": sum(row[4] for row in profit_data) / len(profit_data) if profit_data else 0
 }

 def generate_region_analysis(self) -> list:
 """Regional sales analysis"""
 query = """
 SELECT 
 region,
 COUNT(*) as orders,
 ROUND(SUM(amount), 2) as revenue,
 ROUND(AVG(amount), 2) as avg_order,
 ROUND(SUM(amount) / (SELECT SUM(amount) FROM sales) * 100, 2) as pct
 FROM sales
 GROUP BY region
 ORDER BY revenue DESC
 """
 return self.conn.execute(query).fetchall()
 
 def generate_category_analysis(self) -> list:
 """Business category performance analysis"""
 query = """
 SELECT 
 category,
 COUNT(*) as orders,
 ROUND(SUM(amount), 2) as revenue,
 ROUND(AVG(amount), 2) as avg_order,
 ROUND(SUM(amount) / (SELECT SUM(amount) FROM sales) * 100, 2) as pct
 FROM sales
 GROUP BY category
 ORDER BY revenue DESC
 """
 return self.conn.execute(query).fetchall()
 
 def generate_insights(self) -> list:
 """Automatically generate data insights"""
 insights = []
 
 # Find anomalous months
 anomaly_query = """
 WITH monthly_stats AS (
 SELECT 
 strftime(date, '%Y-%m') as month,
 SUM(amount) as amount
 FROM sales GROUP BY month
 )
 SELECT 
 month,
 amount,
 AVG(amount) OVER () as avg_amount,
 STDDEV(amount) OVER () as stddev_amount
 FROM monthly_stats
 """
 stats = self.conn.execute(anomaly_query).fetchall()
 
 if stats:
 avg = stats[0][2]
 std = stats[0][3] if stats[0][3] else 0
 for row in stats:
 if std > 0 and abs(row[1] - avg) > 2 * std:
 direction = "below" if row[1] < avg else "above"
 insights.append(f"⚠️ {row[0]} sales are {direction} average by {abs(row[1]-avg)/avg*100:.1f}%, investigate cause")
 
 # Find top performer
 top_query = """
 SELECT region, SUM(amount) as total
 FROM sales GROUP BY region ORDER BY total DESC LIMIT 1
 """
 top_region = self.conn.execute(top_query).fetchone()
 if top_region:
 insights.append(f"🏆 {top_region[0]} is the largest revenue source, consider increasing investment")
 
 return insights

 def generate_markdown_report(self, output_path: str):
 """Generate professional Markdown report"""
 
 # Core financial metrics
 financial = self.generate_financial_analysis()
 
 report = f"""# 📊 Financial Report - {datetime.now().strftime('%B %Y')}

## 1. Core Financial Metrics

| Metric | Value |
|--------|-------|
| Total Revenue | ${financial['total_revenue']:,.2f} |
| Total Expenses | ${financial['total_expenses']:,.2f} |
| Net Profit | ${financial['total_profit']:,.2f} |
| Average Margin | {financial['avg_margin']:.2f}% |

## 2. Monthly Profit Trend

| Month | Revenue | Expenses | Profit | Margin | Status |
|-------|---------|----------|--------|--------|--------|
"""
 
 for row in financial['monthly_profit']:
 report += f"| {row[0]} | ${row[1]:,.2f} | ${row[2]:,.2f} | ${row[3]:,.2f} | {row[4]}% | {row[5]} |\n"
 
 # Regional analysis
 report += "\n## 3. Regional Sales Distribution\n\n"
 for row in self.generate_region_analysis():
 report += f"- **{row[0]}**: {row[1]} orders, ${row[2]:,.2f} revenue ({row[4]}%)\n"
 
 # Category analysis
 report += "\n## 4. Business Category Performance\n\n"
 for row in self.generate_category_analysis():
 report += f"- **{row[0]}**: {row[1]} orders, ${row[2]:,.2f} revenue ({row[3]}%)\n"
 
 # Insights
 report += "\n## 5. Data Insights\n\n"
 for insight in self.generate_insights():
 report += f"{insight}\n"
 
 # Action recommendations
 report += "\n## 6. Action Recommendations\n\n"
 if financial['avg_margin'] > 70:
 report += "1. ✅ Margin is healthy, consider scaling the business\n"
 if financial['avg_margin'] < 50:
 report += "1. ⚠️ Margin is low, optimize cost structure\n"
 report += "2. 📈 Monitor anomalous months regularly, establish alert mechanisms\n"
 report += "3. 🎯 Focus on high-margin regions and business categories\n"
 
 with open(output_path, 'w', encoding='utf-8') as f:
 f.write(report)
 
 print(f"✅ Report generated: {output_path}")
 return output_path

 def generate_json_output(self) -> dict:
 """Generate JSON format for API integration"""
 return {
 "generated_at": datetime.now().isoformat(),
 "summary": self.generate_financial_analysis(),
 "regions": [
 {"region": row[0], "orders": row[1], "revenue": row[2], "avg_order": row[3], "pct": row[4]}
 for row in self.generate_region_analysis()
 ],
 "categories": [
 {"category": row[0], "orders": row[1], "revenue": row[2], "avg_order": row[3], "pct": row[4]}
 for row in self.generate_category_analysis()
 ],
 "insights": self.generate_insights()
 }

# Use parallel reading
conn.execute("PRAGMA threads=8")

# Use persistent storage (avoid re-parsing)
conn = duckdb.connect("cache.duckdb")
conn.execute("CREATE TABLE IF NOT EXISTS sales AS SELECT * FROM read_csv_auto('large_file.csv')")

# Use cron for scheduled report generation
import schedule
import time

def daily_report():
 generator = FinancialReportGenerator()
 generator.load_data("./data")
 generator.generate_markdown_report(f"./reports/report_{datetime.now().strftime('%Y%m%d')}.md")

schedule.every().day.at("09:00").do(daily_report)

while True:
 schedule.run_pending()
 time.sleep(60)

# Convert to Pandas
df = conn.execute("SELECT * FROM sales").fetchdf()

# Work with other SQL engines
conn.execute("ATTACH 'existing.db' AS other_db")
result = conn.execute("SELECT * FROM sales JOIN other_db.customers ON ...").fetchall()