DuckDB Personal Finance Automation: Build an Intelligent Financial Assistant with SQL

import duckdb
from pathlib import Path

# Connect to in-memory database (no installation needed)
con = duckdb.connect(':memory:')

# Read China Merchants Bank CSV export
# Assuming file path: ~/Downloads/CMB_202609.csv
cmb_csv = Path.home() / 'Downloads' / 'CMB_202609.csv'

con.execute(f"""
    CREATE TABLE cmb_transactions AS
    SELECT * FROM read_csv_auto('{cmb_csv}', 
        header=true, 
        columns={{
            '交易日期': 'VARCHAR',
            '交易时间': 'VARCHAR', 
            '交易金额': 'DECIMAL(12,2)',
            '余额': 'DECIMAL(12,2)',
            '对方账户': 'VARCHAR',
            '交易摘要': 'VARCHAR'
        }}
    )
""")

print(f"CMB transaction records: {con.execute('SELECT COUNT(*) FROM cmb_transactions').fetchone()[0]}")

-- Read ICBC transactions
CREATE TABLE icbc_transactions AS
SELECT * FROM read_csv_auto('~/Downloads/ICBC_202609.csv', header=true);

-- Read Alipay statements
CREATE TABLE alipay_transactions AS
SELECT * FROM read_csv_auto('~/Downloads/ALIPAY_202609.csv', header=true);

-- Merge all into one unified table
CREATE TABLE all_transactions AS
SELECT 'CMB' AS bank, * FROM cmb_transactions
UNION ALL
SELECT 'ICBC' AS bank, * FROM icbc_transactions
UNION ALL
SELECT 'Alipay' AS bank, * FROM alipay_transactions;

-- Create category rules table
CREATE TABLE category_rules AS
SELECT * FROM (VALUES
    ('Food', '%takeout%'),
    ('Food', '%Meituan%'),
    ('Food', '%Ele.me%'),
    ('Food', '%McDonald%'),
    ('Food', '%KFC%'),
    ('Food', '%Starbucks%'),
    ('Transport', '%DiDi%'),
    ('Transport', '%subway%'),
    ('Transport', '%bus%'),
    ('Shopping', '%Taobao%'),
    ('Shopping', '%JD%'),
    ('Shopping', '%Pinduoduo%'),
    ('Entertainment', '%iQiyi%'),
    ('Entertainment', '%Netflix%'),
    ('Entertainment', '%movie%'),
    ('Housing', '%rent%'),
    ('Housing', '%property%'),
    ('Housing', '%utilities%'),
    ('Salary', '%salary%'),
    ('Salary', '%bonus%'),
    ('Investment', '%fund%'),
    ('Investment', '%stock%'),
    ('Transfer', '%transfer%'),
    ('Refund', '%refund%'),
    ('Refund', '%return%')
) AS t(category_keyword, pattern);

-- Classify each transaction
CREATE TABLE classified_transactions AS
SELECT 
    t.*,
    COALESCE(
        (SELECT r.category_keyword 
         FROM category_rules r 
         WHERE t.交易摘要 ILIKE r.pattern 
         LIMIT 1),
        'Other'
    ) AS category
FROM all_transactions t;

-- Add new category rules
INSERT INTO category_rules (category_keyword, pattern) VALUES
    ('Food', '%Haidilao%'),
    ('Food', '%Xibei%'),
    ('Fitness', '%Keep%'),
    ('Fitness', '%gym%');

-- Monthly spending by category
SELECT 
    category,
    SUM(ABS(交易金额)) AS total_spent,
    COUNT(*) AS transaction_count,
    AVG(ABS(交易金额)) AS avg_amount,
    ROUND(100.0 * SUM(ABS(交易金额)) / SUM(SUM(ABS(交易金额))) OVER(), 1) AS percentage
FROM classified_transactions
WHERE 交易金额 < 0
GROUP BY category
ORDER BY total_spent DESC;

-- Compare spending between consecutive months
WITH monthly_spending AS (
    SELECT 
        strftime(交易日期, '%Y-%m') AS month,
        category,
        SUM(ABS(交易金额)) AS total_spent
    FROM classified_transactions
    WHERE 交易金额 < 0
    GROUP BY strftime(交易日期, '%Y-%m'), category
),
with_change AS (
    SELECT *,
        LAG(total_spent) OVER (PARTITION BY category ORDER BY month) AS prev_month_spent,
        ROUND(total_spent - LAG(total_spent) OVER (PARTITION BY category ORDER BY month), 2) AS change_amount,
        ROUND(100.0 * (total_spent - LAG(total_spent) OVER (PARTITION BY category ORDER BY month)) 
              / NULLIF(LAG(total_spent) OVER (PARTITION BY category ORDER BY month), 0), 1) AS change_pct
    FROM monthly_spending
)
SELECT month, category, total_spent, change_amount, change_pct
FROM with_change
WHERE month = (SELECT MAX(month) FROM monthly_spending)
ORDER BY total_spent DESC;

-- Set monthly budgets and check for overspending
CREATE TABLE budgets AS
SELECT * FROM (VALUES
    ('Food', 3000.00),
    ('Transport', 800.00),
    ('Shopping', 1500.00),
    ('Entertainment', 500.00),
    ('Housing', 5000.00),
    ('Other', 500.00)
) AS t(category, budget_amount);

-- Check which categories exceeded budget
SELECT 
    b.category,
    b.budget_amount,
    COALESCE(s.total_spent, 0) AS actual_spent,
    b.budget_amount - COALESCE(s.total_spent, 0) AS remaining,
    CASE 
        WHEN COALESCE(s.total_spent, 0) > b.budget_amount THEN 'OVER BUDGET'
        WHEN COALESCE(s.total_spent, 0) > b.budget_amount * 0.8 THEN 'WARNING'
        ELSE 'OK'
    END AS status
FROM budgets b
LEFT JOIN (
    SELECT category, SUM(ABS(交易金额)) AS total_spent
    FROM classified_transactions
    WHERE 交易金额 < 0
    GROUP BY category
) s ON b.category = s.category
ORDER BY remaining ASC;

import duckdb
from pathlib import Path

def generate_monthly_report(year_month: str):
    """Generate monthly financial report"""
    con = duckdb.connect(':memory:')
    
    # 1. Import data
    con.execute(f"""
        CREATE TABLE all_transactions AS
        SELECT 'CMB' AS bank, * FROM read_csv_auto(
            '{Path.home()}/Downloads/CMB_{year_month}.csv', header=true)
        UNION ALL
        SELECT 'Alipay' AS bank, * FROM read_csv_auto(
            '{Path.home()}/Downloads/ALIPAY_{year_month}.csv', header=true);
    """)
    
    # 2. Classify transactions
    con.execute("""
        CREATE TABLE category_rules AS
        SELECT * FROM (VALUES
            ('Food', '%takeout%'), ('Food', '%Meituan%'),
            ('Transport', '%DiDi%'), ('Transport', '%subway%'),
            ('Shopping', '%Taobao%'), ('Shopping', '%JD%'),
            ('Entertainment', '%iQiyi%'), ('Entertainment', '%movie%'),
            ('Housing', '%rent%'), ('Housing', '%property%'),
            ('Salary', '%salary%'), ('Salary', '%bonus%'),
            ('Investment', '%fund%'), ('Investment', '%stock%'),
            ('Refund', '%refund%')
        ) AS t(category_keyword, pattern);
    """)
    
    con.execute("""
        CREATE TABLE classified AS
        SELECT t.*, COALESCE(
            (SELECT r.category_keyword FROM category_rules r 
             WHERE t.交易摘要 ILIKE r.pattern LIMIT 1), 'Other'
        ) AS category
        FROM all_transactions t;
    """)
    
    # 3. Generate report
    print(f"{'='*50}")
    print(f"📊 {year_month} Monthly Financial Report")
    print(f"{'='*50}")
    
    overview = con.execute("""
        SELECT 
            SUM(CASE WHEN 交易金额 >= 0 THEN 交易金额 ELSE 0 END) AS income,
            SUM(CASE WHEN 交易金额 < 0 THEN ABS(交易金额) ELSE 0 END) AS expense,
            SUM(交易金额) AS net_savings
        FROM classified
    ").fetchone()
    
    print(f"\n💰 Income: ${overview[0]:,.2f}")
    print(f"💸 Expenses: ${overview[1]:,.2f}")
    print(f"📈 Net Savings: ${overview[2]:,.2f} ({overview[2]/overview[0]*100:.1f}%)")
    
    print(f"\n📋 Top 5 Spending Categories:")
    top_spending = con.execute("""
        SELECT category, SUM(ABS(交易金额)) AS total
        FROM classified WHERE 交易金额 < 0
        GROUP BY category ORDER BY total DESC LIMIT 5
    ").fetchall()
    for i, (cat, amt) in enumerate(top_spending, 1):
        print(f"  {i}. {cat}: ${amt:,.2f}")
    
    print(f"\n🏪 Top 5 Frequent Merchants:")
    top_merchants = con.execute("""
        SELECT 交易摘要, COUNT(*) AS cnt, SUM(ABS(交易金额)) AS total
        FROM classified WHERE 交易金额 < 0
        GROUP BY 交易摘要 ORDER BY cnt DESC LIMIT 5
    ").fetchall()
    for i, (merch, cnt, total) in enumerate(top_merchants, 1):
        print(f"  {i}. {merch}: {cnt} times, ${total:,.2f}")
    
    print(f"\n{'='*50}")
    con.close()

# Usage
generate_monthly_report('202609')