-- Create the financial analytics database
ATTACH 'finance.duckdb' AS finance;
USE finance;

-- Organize into schemas
CREATE SCHEMA IF NOT EXISTS raw;
CREATE SCHEMA IF NOT EXISTS analytics;

-- Import raw sales data from client CSV files
-- read_csv_auto auto-detects types and supports glob patterns
CREATE TABLE raw.sales AS
SELECT * FROM read_csv_auto(
 'client_data/sales_*.csv',
 header=true,
 types={
 'order_id': 'VARCHAR',
 'date': 'DATE',
 'amount': 'DOUBLE',
 'cost': 'DOUBLE',
 'category': 'VARCHAR',
 'customer_id': 'VARCHAR',
 'channel': 'VARCHAR'
 },
 dateformat='%Y-%m-%d'
);

-- Google Sheets data is also supported directly:
-- CREATE TABLE raw.sales AS
-- SELECT * FROM st_read(
-- 'https://docs.google.com/spreadsheets/d/SPREADSHEET_ID/export?format=csv'
-- );

-- Safe loading: treat everything as text first
CREATE TABLE raw.sales_safe AS
SELECT * FROM read_csv_auto(
 'client_data/sales_*.csv',
 header=true,
 all_varchar=true,
 filename=true -- track which file each row came from
);

-- Clean and convert
CREATE TABLE raw.sales_clean AS
SELECT
 order_id,
 TRY_CAST(date AS DATE) AS date,
 TRY_CAST(amount AS DOUBLE) AS amount,
 TRY_CAST(cost AS DOUBLE) AS cost,
 category,
 customer_id,
 channel
FROM raw.sales_safe
WHERE TRY_CAST(amount AS DOUBLE) IS NOT NULL;

-- Daily aggregation layer
CREATE TABLE analytics.agg_daily AS
SELECT
 date_trunc('day', date)::DATE AS day,
 category,
 channel,
 count(DISTINCT order_id) AS order_count,
 sum(amount) AS revenue,
 sum(cost) AS total_cost,
 sum(amount) - sum(cost) AS gross_profit,
 round(
 (sum(amount) - sum(cost)) / nullif(sum(amount), 0) * 100,
 2
 ) AS profit_margin
FROM raw.sales_clean
GROUP BY day, category, channel;

-- Rolling metrics layer (window functions in action)
CREATE TABLE analytics.metrics_rolling AS
SELECT
 day,
 category,
 revenue,
 profit_margin,
 -- 30-day moving average profit margin
 avg(profit_margin) OVER (
 PARTITION BY category
 ORDER BY day
 ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
 ) AS ma30_profit_margin,
 -- Week-over-week revenue change
 revenue - lag(revenue, 7) OVER (
 PARTITION BY category ORDER BY day
 ) AS wow_revenue_change,
 -- Month-to-date cumulative revenue
 sum(revenue) OVER (
 PARTITION BY category, date_trunc('month', day)
 ORDER BY day
 ) AS mtd_revenue
FROM analytics.agg_daily
ORDER BY category, day;

# update_pipeline.py — deployable on GitHub Actions or any Linux server
"""
How to deploy:
1. Run on the client's server at 2:00 AM daily
2. Or run on your own server, pulling client data via API
"""
import duckdb
from datetime import datetime

DB_PATH = 'finance.duckdb'
FEED_DIR = '/data/daily_feed/' # client drops daily data here

conn = duckdb.connect(DB_PATH)

# 1. Read today's incremental data
today = datetime.now().strftime('%Y-%m-%d')
conn.execute(f"""
 CREATE OR REPLACE TEMP TABLE delta AS
 SELECT * FROM read_csv_auto(
 '{FEED_DIR}/*.csv',
 header=true,
 all_varchar=true
 );
""")

# 2. Merge into main table (deduplicate by order_id)
conn.execute("""
 INSERT INTO raw.sales_clean BY NAME
 SELECT
 order_id,
 TRY_CAST(date AS DATE) AS date,
 TRY_CAST(amount AS DOUBLE) AS amount,
 TRY_CAST(cost AS DOUBLE) AS cost,
 category,
 customer_id,
 channel
 FROM delta d
 WHERE NOT EXISTS (
 SELECT 1 FROM raw.sales_clean r
 WHERE r.order_id = d.order_id
 );
""")

# 3. Incrementally update aggregation layer (only affected days)
affected_days = conn.execute("""
 SELECT DISTINCT TRY_CAST(date AS DATE) AS day
 FROM delta
 WHERE TRY_CAST(date AS DATE) IS NOT NULL
""").fetchall()

for (day,) in affected_days:
 conn.execute(f"DELETE FROM analytics.agg_daily WHERE day = '{day}'")
 conn.execute(f"""
 INSERT INTO analytics.agg_daily
 SELECT
 date_trunc('day', date)::DATE AS day,
 category,
 channel,
 count(DISTINCT order_id) AS order_count,
 sum(amount) AS revenue,
 sum(cost) AS total_cost,
 sum(amount) - sum(cost) AS gross_profit,
 round((sum(amount) - sum(cost)) / nullif(sum(amount), 0) * 100, 2) AS profit_margin
 FROM raw.sales_clean
 WHERE date_trunc('day', date)::DATE = '{day}'
 GROUP BY day, category, channel;
 """)

# 4. Full rebuild of rolling metrics
# At sub-million-row scale, full rebuild takes <5 seconds — no need for incremental
conn.execute("DELETE FROM analytics.metrics_rolling;")
conn.execute("""
 INSERT INTO analytics.metrics_rolling
 SELECT
 day, category, revenue, profit_margin,
 avg(profit_margin) OVER (
 PARTITION BY category ORDER BY day
 ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
 ) AS ma30_profit_margin,
 revenue - lag(revenue, 7) OVER (
 PARTITION BY category ORDER BY day
 ) AS wow_revenue_change,
 sum(revenue) OVER (
 PARTITION BY category, date_trunc('month', day)
 ORDER BY day
 ) AS mtd_revenue
 FROM analytics.agg_daily
 ORDER BY category, day;
""")

# 5. Export data for the frontend
conn.execute("""
 COPY (
 SELECT * FROM analytics.metrics_rolling
 WHERE day >= current_date - INTERVAL '90 days'
 ORDER BY day DESC
 ) TO '/var/www/dashboard/data/metrics.parquet'
 (FORMAT PARQUET, COMPRESSION ZSTD);
""")

conn.close()
print(f"✅ Update complete: {today}, affected {len(affected_days)} days")

# crontab -e
0 2 * * * cd /opt/client-finance && /usr/bin/python3 update_pipeline.py >> /var/log/duckdb_update.log 2>&1

# .github/workflows/daily_update.yml
name: Daily Finance Data Update
on:
 schedule:
 - cron: '0 2 * * *'
 workflow_dispatch:

jobs:
 update:
 runs-on: ubuntu-latest
 steps:
 - uses: actions/checkout@v4
 - uses: actions/setup-python@v5
 with:
 python-version: '3.11'
 - run: pip install duckdb
 - run: python update_pipeline.py
 - name: Deploy Parquet to CDN
 run: |
 aws s3 cp /var/www/dashboard/data/metrics.parquet \
 s3://client-dashboard-data/metrics.parquet

-- Export last 90 days of core metrics
COPY (
 SELECT * FROM analytics.metrics_rolling
 WHERE day >= current_date - INTERVAL '90 days'
 ORDER BY day DESC
) TO '/var/www/dashboard/data/metrics.parquet'
(FORMAT PARQUET, COMPRESSION ZSTD);

// Browser-side code
const duckdb = await DuckDBClient.create({
 metrics: 'https://your-cdn.com/data/metrics.parquet'
});
const result = await duckdb.query(`
 SELECT day, revenue, profit_margin
 FROM metrics
 WHERE category = 'Retail'
 ORDER BY day DESC
 LIMIT 30
`);

INSTALL httpserver;
LOAD httpserver;

-- Start HTTP service (port 8338)
-- duckdb finance.duckdb -c "LOAD httpserver; SELECT httpserve_start('0.0.0.0', 8338);"

-- Then the frontend queries:
-- GET http://localhost:8338/?sql=SELECT+day,revenue+FROM+metrics_rolling+LIMIT+10
-- Returns JSON results

# dashboard.py
import streamlit as st
import duckdb
import plotly.express as px

st.set_page_config(page_title="Financial Dashboard", layout="wide")

conn = duckdb.connect('finance.duckdb')

df = conn.execute("""
 SELECT day, category, revenue, profit_margin, wow_revenue_change
 FROM analytics.metrics_rolling
 WHERE day >= current_date - INTERVAL '30 days'
 ORDER BY day
""").fetchdf()

fig = px.line(df, x='day', y='revenue', color='category',
 title='Daily Revenue Trend (Last 30 Days)')
st.plotly_chart(fig, use_container_width=True)

latest = df[df['day'] == df['day'].max()]
col1, col2, col3 = st.columns(3)
col1.metric("Today's Margin", f"{latest['profit_margin'].mean():.1f}%")
col2.metric("Yesterday's Margin",
 f"{df[df['day'] == df['day'].max() - pd.Timedelta(days=1)]['profit_margin'].mean():.1f}%")
col3.metric("WoW Revenue Change",
 f"{latest['wow_revenue_change'].sum():+.0f}")