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 TABLE 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'")