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)