Building an Industry Data Report Product with DuckDB: From Custom Services to Standardized Monetization

import duckdb
import json
from datetime import datetime

class ReportEngine:
    """Industry Data Report Engine"""
    
    def __init__(self, db_path="reports.duckdb"):
        self.con = duckdb.connect(db_path)
        self._setup_schema()
    
    def _setup_schema(self):
        """Initialize data table schema"""
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS company_events (
                event_id VARCHAR PRIMARY KEY,
                company_name VARCHAR,
                industry VARCHAR,
                event_type VARCHAR,
                amount_usd DOUBLE,
                investors VARCHAR[],
                event_date DATE,
                region VARCHAR,
                description VARCHAR
            );
            
            CREATE TABLE IF NOT EXISTS industry_taxonomy (
                code VARCHAR PRIMARY KEY,
                name VARCHAR,
                parent_code VARCHAR
            );
        """)

    def _build_query(self, params):
        """Dynamically build queries—the core!"""
        industry = params.get("industry", "%")
        start_date = params.get("start_date", "1900-01-01")
        end_date = params.get("end_date", "2100-12-31")
        top_n = params.get("top_n", 10)
        
        queries = {}
        
        # 1. Overall overview
        queries["overview"] = f"""
            SELECT 
                COUNT(*) AS total_deals,
                ROUND(SUM(amount_usd), 2) AS total_funding,
                ROUND(AVG(amount_usd), 2) AS avg_ticket,
                ROUND(MEDIAN(amount_usd), 2) AS median_ticket,
                MIN(event_date) AS earliest_date,
                MAX(event_date) AS latest_date
            FROM company_events
            WHERE event_date BETWEEN '{start_date}' AND '{end_date}'
        """
        
        # 2. Monthly trend
        queries["monthly_trend"] = f"""
            SELECT 
                DATE_TRUNC('month', event_date) AS month,
                COUNT(*) AS deal_count,
                ROUND(SUM(amount_usd), 2) AS funding_total,
                ROUND(AVG(amount_usd), 2) AS avg_ticket
            FROM company_events
            WHERE event_date BETWEEN '{start_date}' AND '{end_date}'
            GROUP BY 1
            ORDER BY 1
        """
        
        # 3. Top N investment events
        queries["top_events"] = f"""
            SELECT 
                company_name,
                event_type,
                ROUND(amount_usd, 2) AS amount,
                event_date,
                region
            FROM company_events
            WHERE event_date BETWEEN '{start_date}' AND '{end_date}'
            ORDER BY amount_usd DESC
            LIMIT {top_n}
        """
        
        # 4. Investor activity ranking
        queries["investor_ranking"] = f"""
            SELECT 
                UNNEST(investors) AS investor,
                COUNT(DISTINCT event_id) AS deal_count,
                ROUND(SUM(amount_usd), 2) AS total_invested,
                ROUND(AVG(amount_usd), 2) AS avg_ticket
            FROM company_events
            WHERE event_date BETWEEN '{start_date}' AND '{end_date}'
            GROUP BY 1
            HAVING COUNT(DISTINCT event_id) >= 2
            ORDER BY total_invested DESC
            LIMIT {top_n}
        """
        
        # 5. Sub-sector distribution
        queries["sub_sector_distribution"] = f"""
            SELECT 
                industry,
                COUNT(*) AS deal_count,
                ROUND(SUM(amount_usd), 2) AS total_funding,
                ROUND(SUM(amount_usd) / NULLIF(COUNT(*), 0), 2) AS avg_ticket,
                ROUND(
                    SUM(amount_usd) * 100.0 / 
                    NULLIF(SUM(SUM(amount_usd)) OVER (), 0), 2
                ) AS share_pct
            FROM company_events
            WHERE event_date BETWEEN '{start_date}' AND '{end_date}'
            GROUP BY 1
            ORDER BY total_funding DESC
            LIMIT {top_n}
        """
        
        return queries
    
    def generate_report(self, params):
        """Generate a complete report based on parameters"""
        queries = self._build_query(params)
        results = {}
        
        for metric_name, sql in queries.items():
            df = self.con.execute(sql).fetchdf()
            results[metric_name] = df.to_dict('records')
        
        return {
            "report_date": datetime.now().strftime("%Y-%m-%d"),
            "params": params,
            "data": results
        }

# Initialize the engine
engine = ReportEngine("/tmp/industry_report.duckdb")

# Insert sample data (in production, import from public data sources)
sample_data = [
    ("EVT001", "DeepMindTech", "Artificial Intelligence", "Financing", 50000000, 
     ["Sequoia Capital", "GaoLv Capital", "Google Ventures"], "2026-03-15", "Beijing", "LLM Training Platform"),
    ("EVT002", "CloudScale", "Cloud Computing", "Financing", 30000000,
     ["IDG Capital", "Matrix Partners"], "2026-04-02", "Shanghai", "Edge Computing Infrastructure"),
    ("EVT003", "AutoDriveAI", "Autonomous Driving", "M&A", 120000000,
     ["BYD", "Huawei Investment"], "2026-05-10", "Shenzhen", "L4 Autonomous Driving Solution"),
    ("EVT004", "BioGenLab", "Biotech", "Financing", 80000000,
     ["Qiming Venture", "Eli Lilly Asia Fund"], "2026-02-20", "Suzhou", "mRNA Drug Development"),
    ("EVT005", "DataFlowAI", "Artificial Intelligence", "Financing", 25000000,
     ["Sequoia Capital", "Tencent Investment"], "2026-06-01", "Hangzhou", "AI Data Processing Pipeline"),
    ("EVT006", "GreenEnergyX", "New Energy", "Financing", 45000000,
     ["GaoLv Capital", "CATL"], "2026-01-15", "Hefei", "Solid-State Battery Technology"),
    ("EVT007", "CyberShield", "Cybersecurity", "IPO", 200000000,
     ["SAIF Partners", "SoftBank Vision Fund"], "2026-05-28", "Beijing", "Zero-Trust Security Platform"),
    ("EVT008", "RoboticsPro", "Robotics", "Financing", 35000000,
     ["Matrix Partners", "Xiaomi Strategic"], "2026-04-18", "Shenzhen", "Collaborative Robotic Arm"),
    ("EVT009", "FinTechHub", "FinTech", "M&A", 90000000,
     ["Ant Group", "Ping An Innovation"], "2026-03-25", "Shanghai", "Intelligent Risk Control System"),
    ("EVT010", "SpaceData", "Commercial Aerospace", "Financing", 60000000,
     ["Shenzhen Capital", "China Satellite Network"], "2026-06-10", "Xi'an", "Satellite Remote Sensing Data Processing"),
]

# Bulk insert
for row in sample_data:
    engine.con.execute("""
        INSERT INTO company_events VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, row)

# Generate a complete report
report = engine.generate_report({
    "industry": "%",
    "start_date": "2026-01-01",
    "end_date": "2026-06-25",
    "top_n": 5,
})

# View overview data
print(json.dumps(report["data"]["overview"], indent=2, ensure_ascii=False))

{
  "total_deals": 10,
  "total_funding": 835000000.0,
  "avg_ticket": 83500000.0,
  "median_ticket": 47500000.0,
  "earliest_date": "2026-01-15",
  "latest_date": "2026-06-10"
}

# Pseudocode: automated weekly execution
def weekly_report(industry):
    engine = ReportEngine(f"/data/reports/{industry}.duckdb")
    
    # 1. Update data from public sources
    engine.con.execute("CALL http_get('https://api.example.com/events')")
    
    # 2. Generate report
    report = engine.generate_report({
        "industry": industry,
        "start_date": "last_week_monday",
        "end_date": "last_week_sunday",
        "top_n": 10,
    })
    
    # 3. Export as PDF or HTML
    export_to_html(report, f"/output/{industry}_weekly.html")
    
    # 4. Send email/Telegram notification
    send_notification(industry, report)

# Telegram Bot integration example
@bot.message_handler(commands=['report'])
def handle_report(message):
    parts = message.text.split()
    if len(parts) < 2:
        bot.reply(message, "Usage: /report industry_name [start_date] [end_date]")
        return
    
    industry = parts[1]
    engine = ReportEngine("/tmp/report_cache.duckdb")
    
    report = engine.generate_report({
        "industry": industry,
        "start_date": parts[2] if len(parts) > 2 else "2026-01-01",
        "end_date": parts[3] if len(parts) > 3 else "2026-06-25",
        "top_n": 10,
    })
    
    # Send structured report
    bot.reply(message, format_telegram_report(report))

from fastapi import FastAPI, Query
app = FastAPI()

@app.get("/api/report")
def get_report(
    industry: str = Query(default="%"),
    start_date: str = Query(default="2026-01-01"),
    end_date: str = Query(default="2026-12-31"),
    top_n: int = Query(default=10)
):
    engine = ReportEngine("/tmp/api_reports.duckdb")
    return engine.generate_report({
        "industry": industry,
        "start_date": start_date,
        "end_date": end_date,
        "top_n": top_n,
    })

import duckdb

# Batch ingest from scraped JSON data
import json
with open("scraped_events.json", "r") as f:
    events = json.load(f)

con = duckdb.connect("events.db")
con.execute("CREATE TABLE IF NOT EXISTS scraped_events AS SELECT * FROM read_json_auto('scraped_events.json')")
con.execute("""
    INSERT INTO company_events
    SELECT * FROM scraped_events
    ON CONFLICT (event_id) DO NOTHING
""")

# Query remote JSON API directly, no local storage needed
con = duckdb.connect()
result = con.execute("""
    SELECT * FROM read_json_auto(
        'https://api.crunchbase.com/v4/fundings?size=100'
    )
""").fetchdf()

# Mount multiple data sources for joint querying
con = duckdb.connect("master_report.duckdb")
con.execute("ATTACH 'events.db' AS events")
con.execute("ATTACH 'financials.parquet' AS fin")
con.execute("ATTACH 'https://data.open-source.com/company.csv' AS src")

# Cross-source JOIN query
con.execute("""
    SELECT c.company_name, c.industry, f.revenue
    FROM events.company_events c
    LEFT JOIN fin.financials f ON c.company_name = f.name
    WHERE c.event_date > '2026-01-01'
""")