Building a Competitor Monitoring SaaS with DuckDB: Complete Guide from Data Collection to Automated Reports

import duckdb
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
import random

# Create project directories
Path("competitor_monitor/data").mkdir(parents=True, exist_ok=True)

def generate_competitor_data(days: int = 90):
    """Generate simulated competitor monitoring data"""
    
    # Competitor A: Premium brand, small price fluctuations
    comp_a_prices = []
    base_price_a = 1299
    for day in range(days):
        date = (datetime.now() - timedelta(days=day)).strftime("%Y-%m-%d")
        price = base_price_a + random.randint(-50, 50)
        if day % 14 == 0:  # Bi-weekly promotion
            price = base_price_a - 200
        comp_a_prices.append({
            "competitor": "CompetitorA",
            "date": date,
            "product": "Pro",
            "price": price,
            "url": "https://competitor-a.com/pricing",
            "stock_status": "In Stock" if random.random() > 0.1 else "Out of Stock"
        })
    
    # Competitor B: Mid-range brand, frequent price changes
    comp_b_prices = []
    base_price_b = 799
    for day in range(days):
        date = (datetime.now() - timedelta(days=day)).strftime("%Y-%m-%d")
        price = base_price_b + random.randint(-100, 100)
        if day % 7 == 0:
            price -= 50  # Weekly discount
        comp_b_prices.append({
            "competitor": "CompetitorB",
            "date": date,
            "product": "Standard",
            "price": price,
            "url": "https://competitor-b.com/pricing",
            "stock_status": "In Stock"
        })
    
    # Competitor C: New entrant, aggressive pricing
    comp_c_prices = []
    base_price_c = 499
    for day in range(days):
        date = (datetime.now() - timedelta(days=day)).strftime("%Y-%m-%d")
        price = max(base_price_c - (days - day) // 10, 299)
        comp_c_prices.append({
            "competitor": "CompetitorC",
            "date": date,
            "product": "Basic",
            "price": price,
            "url": "https://competitor-c.com/pricing",
            "stock_status": "In Stock"
        })
    
    # Save to CSV
    all_prices = comp_a_prices + comp_b_prices + comp_c_prices
    pd.DataFrame(all_prices).sort_values("date").to_csv(
        "competitor_monitor/data/competitor_prices.csv", 
        index=False, 
        encoding="utf-8-sig"
    )
    
    print(f"✅ Price data generated: {len(all_prices)} records")
    return "competitor_monitor/data/competitor_prices.csv"

generate_competitor_data()

import duckdb
from pathlib import Path

class CompetitorMonitor:
    """Competitor monitoring and analysis engine"""
    
    def __init__(self, db_path: str = "competitor_monitor/monitor.db"):
        self.con = duckdb.connect(db_path)
        self._setup_schema()
    
    def _setup_schema(self):
        """Create database schema"""
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS competitor_prices (
                competitor VARCHAR,
                date DATE,
                product VARCHAR,
                price DECIMAL(10,2),
                url VARCHAR,
                stock_status VARCHAR
            )
        """)
        
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS feature_updates (
                competitor VARCHAR,
                update_date DATE,
                update_type VARCHAR,
                description VARCHAR,
                source VARCHAR
            )
        """)
        
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS reviews (
                competitor VARCHAR,
                date DATE,
                platform VARCHAR,
                rating INTEGER,
                sentiment VARCHAR,
                summary VARCHAR
            )
        """)

def ingest_data(self, csv_path: str):
    """Import CSV data into DuckDB"""
    self.con.execute(f"""
        COPY competitor_prices 
        FROM '{csv_path}' 
        (FORMAT CSV, HEADER, DELIMITER ',')
    """)
    count = self.con.execute('SELECT COUNT(*) FROM competitor_prices').fetchone()[0]
    print(f"✅ Imported {count} price records")

def _setup_views(self):
    """Create analysis views"""
    # Price aggregation view
    self.con.execute("""
        CREATE OR REPLACE VIEW v_price_analysis AS
        SELECT 
            competitor,
            product,
            COUNT(*) AS data_points,
            ROUND(AVG(price), 2) AS avg_price,
            ROUND(MIN(price), 2) AS min_price,
            ROUND(MAX(price), 2) AS max_price,
            ROUND(STDDEV(price), 2) AS price_volatility,
            ROUND(
                100.0 * (MAX(price) - MIN(price)) / NULLIF(AVG(price), 0), 
                2
            ) AS price_variance_pct,
            MAX(date) AS latest_date
        FROM competitor_prices
        GROUP BY competitor, product
    """)
    
    # Daily change view with window functions
    self.con.execute("""
        CREATE OR REPLACE VIEW v_price_daily AS
        SELECT 
            competitor,
            date,
            product,
            price,
            LAG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date
            ) AS prev_price,
            price - LAG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date
            ) AS price_change,
            ROUND(
                100.0 * (price - LAG(price) OVER (
                    PARTITION BY competitor, product 
                    ORDER BY date
                )) / NULLIF(LAG(price) OVER (
                    PARTITION BY competitor, product 
                    ORDER BY date
                ), 0),
                2
            ) AS price_change_pct
        FROM competitor_prices
    """)
    
    # Moving average view
    self.con.execute("""
        CREATE OR REPLACE VIEW v_price_trend AS
        SELECT 
            competitor,
            date,
            product,
            price,
            ROUND(AVG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date 
                ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
            ), 2) AS ma_7day,
            ROUND(AVG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date 
                ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
            ), 2) AS ma_30day
        FROM competitor_prices
    """)

# View competitor price statistics
print("📊 Competitor Price Overview:")
print(self.con.execute("SELECT * FROM v_price_analysis ORDER BY avg_price").fetchdf())

# View today's price changes
print("\n📈 Today's Price Changes:")
today = datetime.now().strftime("%Y-%m-%d")
print(self.con.execute(f"""
    SELECT competitor, product, price, price_change_pct
    FROM v_price_daily 
    WHERE date = '{today}' 
    AND price_change != 0
    ORDER BY ABS(price_change_pct) DESC
""").fetchdf())

# View 7-day moving average trend
print("\n📉 Price Trend (7-day MA):")
print(self.con.execute(f"""
    SELECT competitor, date, product, price, ma_7day
    FROM v_price_trend 
    WHERE date >= '{(datetime.now()-timedelta(days=7)).strftime("%Y-%m-%d")}'
    ORDER BY competitor, date
""").fetchdf())

def detect_price_anomalies(self, threshold_pct: float = 5.0):
    """Detect unusual price fluctuations"""
    result = self.con.execute(f"""
        SELECT 
            competitor,
            date,
            product,
            price,
            price_change_pct,
            'Price Anomaly' AS alert_type
        FROM v_price_daily
        WHERE ABS(price_change_pct) >= {threshold_pct}
        ORDER BY ABS(price_change_pct) DESC
    """).fetchdf()
    return result

# Usage example
anomalies = monitor.detect_price_anomalies(5.0)
if len(anomalies) > 0:
    print(f"⚠️ Found {len(anomalies)} price anomalies:")
    print(anomalies.head(10))

def check_alerts(self) -> list:
    """Check and generate alerts"""
    alerts = []
    
    # Price drop alert
    price_drop = self.con.execute("""
        SELECT competitor, date, product, price, price_change_pct
        FROM v_price_daily
        WHERE date = (SELECT MAX(date) FROM v_price_daily)
        AND price_change_pct <= -5
    """).fetchall()
    
    for row in price_drop:
        alerts.append({
            "type": "PRICE_DROP",
            "competitor": row[0],
            "date": row[1],
            "product": row[2],
            "price": row[3],
            "change_pct": row[4],
            "message": f"⚠️ {row[0]}'s {row[2]} dropped {abs(row[4]):.1f}%, now at ${row[3]}"
        })
    
    # Out of stock alert
    out_of_stock = self.con.execute("""
        SELECT competitor, date, product
        FROM competitor_prices
        WHERE stock_status = 'Out of Stock'
        AND date = (SELECT MAX(date) FROM competitor_prices)
    """).fetchall()
    
    for row in out_of_stock:
        alerts.append({
            "type": "OUT_OF_STOCK",
            "competitor": row[0],
            "date": row[1],
            "product": row[2],
            "message": f"📦 {row[0]}'s {row[2]} is out of stock"
        })
    
    return alerts

def generate_report(self, output_path: str = "report.html"):
    """Generate HTML analysis report"""
    from jinja2 import Template
    
    template_str = """
    <!DOCTYPE html>
    <html>
    <head>
        <meta charset="utf-8">
        <title>Competitor Monitor Daily Report - {{ date }}</title>
        <style>
            body { font-family: -apple-system, sans-serif; max-width: 900px; margin: 0 auto; padding: 20px; }
            h1 { color: #1a1a2e; border-bottom: 3px solid #4facfe; padding-bottom: 10px; }
            h2 { color: #16213e; margin-top: 30px; }
            table { width: 100%; border-collapse: collapse; margin: 15px 0; }
            th { background: #1a1a2e; color: white; padding: 10px; text-align: left; }
            td { padding: 8px 10px; border-bottom: 1px solid #ddd; }
            tr:hover { background: #f5f5f5; }
            .alert { background: #fff3cd; border-left: 4px solid #ffc107; padding: 10px; margin: 10px 0; }
            .good { color: #28a745; }
            .bad { color: #dc3545; }
            .neutral { color: #6c757d; }
        </style>
    </head>
    <body>
        <h1>📊 Competitor Monitor Daily Report - {{ date }}</h1>
        
        <h2>🔔 Alerts</h2>
        {% for alert in alerts %}
        <div class="alert">{{ alert.message }}</div>
        {% endfor %}
        {% if not alerts %}
        <p class="good">✅ No alerts today</p>
        {% endif %}
        
        <h2>📈 Price Overview</h2>
        <table>
            <tr><th>Competitor</th><th>Product</th><th>Avg Price</th><th>Min</th><th>Max</th><th>Variance</th></tr>
            {% for row in price_analysis %}
            <tr>
                <td>{{ row[0] }}</td>
                <td>{{ row[1] }}</td>
                <td>${{ row[2] }}</td>
                <td>${{ row[3] }}</td>
                <td>${{ row[4] }}</td>
                <td>{{ row[7] }}%</td>
            </tr>
            {% endfor %}
        </table>
        
        <h2>📉 Latest Changes</h2>
        <table>
            <tr><th>Competitor</th><th>Date</th><th>Product</th><th>Current</th><th>Change</th></tr>
            {% for row in daily_changes %}
            <tr>
                <td>{{ row[0] }}</td>
                <td>{{ row[1] }}</td>
                <td>{{ row[2] }}</td>
                <td>${{ row[3] }}</td>
                <td class="{% if row[5] < 0 %}bad{% elif row[5] > 0 %}good{% else %}neutral{% endif %}">
                    {{ '%+.2f%%'|format(row[5]) }}
                </td>
            </tr>
            {% endfor %}
        </table>
        
        <p style="color: #999; font-size: 12px; margin-top: 30px;">
            Generated by DuckDB Competitor Monitor | Data from public sources
        </p>
    </body>
    </html>
    """
    
    template = Template(template_str)
    today = datetime.now().strftime("%Y-%m-%d")
    
    alerts = self.check_alerts()
    price_analysis = self.con.execute("SELECT * FROM v_price_analysis ORDER BY avg_price").fetchall()
    daily_changes = self.con.execute(f"""
        SELECT competitor, date, product, price, prev_price, price_change_pct
        FROM v_price_daily 
        WHERE date = '{today}'
        ORDER BY ABS(price_change_pct) DESC
    """).fetchall()
    
    html = template.render(
        date=today,
        alerts=alerts,
        price_analysis=price_analysis,
        daily_changes=daily_changes
    )
    
    Path(output_path).write_text(html, encoding="utf-8")
    print(f"✅ Report generated: {output_path}")
    return output_path

# Main entry point
if __name__ == "__main__":
    # 1. Initialize monitoring engine
    monitor = CompetitorMonitor()
    
    # 2. Generate/import data
    csv_path = generate_competitor_data()
    monitor.ingest_data(csv_path)
    
    # 3. Create analysis views
    monitor._setup_views()
    
    # 4. Run analysis
    print("\n" + "="*50)
    print("Competitor Price Analysis Report")
    print("="*50)
    
    alerts = monitor.check_alerts()
    if alerts:
        print(f"\n⚠️ Found {len(alerts)} alerts:")
        for alert in alerts:
            print(f"  - {alert['message']}")
    else:
        print("\n✅ No alerts")
    
    # 5. Generate report
    monitor.generate_report("competitor_monitor/report.html")
    
    # 6. Export CSV for further analysis
    monitor.con.execute("""
        COPY (SELECT * FROM v_price_analysis) 
        TO 'competitor_monitor/price_summary.csv' 
        (HEADER, DELIMITER ',')
    """)
    print("\n✅ Data exported: competitor_monitor/price_summary.csv")

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Competitor Monitor API")

class PriceAlert(BaseModel):
    competitor: str
    min_price: float
    max_price: float
    alert_email: str

@app.post("/alerts")
def set_price_alert(alert: PriceAlert):
    """Set price alert"""
    current = monitor.con.execute(f"""
        SELECT price FROM competitor_prices 
        WHERE competitor = '{alert.competitor}'
        AND date = (SELECT MAX(date) FROM competitor_prices)
    """).fetchone()
    
    if current and (current[0] <= alert.min_price or current[0] >= alert.max_price):
        send_alert_email(alert.alert_email, alert.competitor, current[0])
        return {"status": "alert_sent", "price": current[0]}
    
    return {"status": "no_alert", "current_price": current[0] if current else None}

# Use cron for daily execution
0 9 * * * cd /path/to/competitor_monitor && python3 monitor.py >> logs/cron.log 2>&1

# Or use Python schedule library
import schedule
import time

schedule.every().day.at("09:00").do(run_monitor)
schedule.every().hour.do(check_alerts)

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