Building a Personal Stock Screener with DuckDB: Three-Layer Filtering in Under 100 Lines

import duckdb

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

# Create table directly from CSV, no pandas needed
conn.execute("""
CREATE TABLE IF NOT EXISTS daily_kline AS
SELECT 
    symbol,
    date,
    close,
    volume,
    (close - open) / open * 100 AS daily_return,
    volume * close AS turnover  -- Turnover = Volume × Close Price
FROM read_csv_auto('daily_data.csv')
""")

# Fundamental data table
conn.execute("""
CREATE TABLE IF NOT EXISTS fundamentals AS
SELECT 
    symbol,
    pe_ratio,
    pb_ratio,
    roe,
    market_cap
FROM read_csv_auto('fundamentals.csv')
WHERE pe_ratio > 0 AND pe_ratio < 20
  AND roe > 15
""")

-- fundamentals table already created with PE < 20 AND ROE > 15
-- This step is done at import time, efficient and one-time only

CREATE TABLE IF NOT EXISTS capital_flow AS
WITH daily_turnover AS (
    SELECT 
        symbol,
        date,
        turnover,
        AVG(turnover) OVER (
            PARTITION BY symbol 
            ORDER BY date 
            ROWS BETWEEN 4 PRECEDING AND CURRENT ROW
        ) AS ma5_turnover,
        AVG(turnover) OVER (
            PARTITION BY symbol 
            ORDER BY date 
            ROWS BETWEEN 19 PRECEDING AND CURRENT ROW
        ) AS ma20_turnover
    FROM daily_kline
)
SELECT DISTINCT symbol
FROM daily_turnover
WHERE ma5_turnover / NULLIF(ma20_turnover, 0) > 1.3
  AND date = (SELECT MAX(date) FROM daily_kline)

CREATE TABLE IF NOT EXISTS technical_signal AS
WITH ma_data AS (
    SELECT 
        symbol,
        date,
        close,
        AVG(close) OVER (
            PARTITION BY symbol 
            ORDER BY date 
            ROWS BETWEEN 19 PRECEDING AND CURRENT ROW
        ) AS ma20
    FROM daily_kline
)
SELECT DISTINCT symbol
FROM ma_data
WHERE close > ma20
  AND date = (SELECT MAX(date) FROM daily_kline)

-- Three-layer intersection: get today's recommended stocks
SELECT 
    f.symbol,
    f.pe_ratio,
    f.roe,
    ROUND(f.market_cap / 1e8, 2) AS market_cap_yi,
    t.date AS signal_date,
    t.close,
    ROUND(t.close / t.ma20 - 1, 4) AS break_pct
FROM fundamentals f
INNER JOIN (
    SELECT DISTINCT symbol
    FROM daily_turnover
    WHERE ma5_turnover / NULLIF(ma20_turnover, 0) > 1.3
      AND date = (SELECT MAX(date) FROM daily_kline)
) cf ON f.symbol = cf.symbol
INNER JOIN (
    SELECT symbol, date, close, ma20
    FROM ma_data
    WHERE close > ma20
      AND date = (SELECT MAX(date) FROM daily_kline)
) t ON f.symbol = t.symbol
ORDER BY f.roe DESC
LIMIT 20;

import duckdb
from datetime import datetime

class StockScraper:
    def __init__(self, db_path='stock_scraper.duckdb'):
        self.conn = duckdb.connect(db_path)
    
    def update_data(self, csv_path):
        """Incremental update: only append new data"""
        self.conn.execute(f"""
            INSERT INTO daily_kline
            SELECT symbol, date, close, open, volume,
                   (close - open) / open * 100 AS daily_return,
                   volume * close AS turnover
            FROM read_csv_auto('{csv_path}')
            WHERE date > (SELECT COALESCE(MAX(date), '1900-01-01') FROM daily_kline)
        """)
    
    def scan(self):
        """Execute three-layer filtering"""
        max_date = self.conn.execute(
            "SELECT MAX(date) FROM daily_kline"
        ).fetchone()[0]
        
        result = self.conn.execute(f"""
            SELECT 
                f.symbol, 
                f.pe_ratio, 
                f.roe, 
                ROUND(f.market_cap / 1e8, 2) AS market_cap_yi,
                t.date AS signal_date,
                t.close,
                ROUND(t.close / t.ma20 - 1, 4) AS break_pct
            FROM fundamentals f
            INNER JOIN (
                SELECT DISTINCT symbol
                FROM daily_turnover
                WHERE ma5_turnover / NULLIF(ma20_turnover, 0) > 1.3
                  AND date = '{max_date}'
            ) cf ON f.symbol = cf.symbol
            INNER JOIN (
                SELECT symbol, date, close, ma20
                FROM ma_data
                WHERE close > ma20
                  AND date = '{max_date}'
            ) t ON f.symbol = t.symbol
            ORDER BY f.roe DESC
            LIMIT 20
        """).fetchdf()
        return result
    
    def export(self, result, format='csv'):
        """Export results"""
        if format == 'csv':
            result.to_csv(
                f'signal_{datetime.now().strftime("%Y%m%d")}.csv', 
                index=False
            )
        return result

# Usage
scraper = StockScraper()
scraper.update_data('new_data.csv')  # Append new data after daily close
results = scraper.scan()
results.export(results)
print(f"🎯 Today's screen found {len(results)} candidates")
print(results[['symbol', 'pe_ratio', 'roe', 'close', 'break_pct']])

-- Create indexes on daily_kline to accelerate incremental update's MAX(date) query
CREATE INDEX IF NOT EXISTS idx_kline_date ON daily_kline(date);
CREATE INDEX IF NOT EXISTS idx_kline_symbol_date ON daily_kline(symbol, date);

CREATE MATERIALIZED VIEW IF NOT EXISTS today_signals AS
SELECT f.symbol, f.pe_ratio, f.roe, t.date, t.close, 
       ROUND(t.close / t.ma20 - 1, 4) AS break_pct
FROM fundamentals f
-- ... full three-layer JOIN logic ...
WHERE t.date = (SELECT MAX(date) FROM daily_kline);

-- Next time, query directly
SELECT * FROM today_signals ORDER BY roe DESC LIMIT 20;

conn = duckdb.connect('stock_scraper.duckdb')
# Enable parallelism: auto-set based on CPU cores
conn.execute("SET threads TO 4")
conn.execute("SET memory_limit='4GB'")