用 DuckDB 搭建个人股票筛选器：三层交叉筛选法，零外部依赖一键出结果

import duckdb

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

# 直接从 CSV 建表，不加载到 pandas
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  -- 成交额 = 成交量 * 收盘价
    FROM read_csv_auto('daily_data.csv')
""")

# 基本面数据表（只做一次）
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
""")

-- 更丰富的基本面筛选
SELECT symbol, pe_ratio, pb_ratio, roe, market_cap
FROM fundamentals
WHERE pe_ratio > 0 
  AND pe_ratio < 20 
  AND roe > 15
  AND market_cap > 5e8  -- 市值大于 5 亿
  AND pb_ratio < 3;     -- 市净率低于 3

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);

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);

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):
        '''增量更新：只追加新数据'''
        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):
        '''执行三层筛选'''
        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 (...) cf ON f.symbol = cf.symbol
            INNER JOIN (...) t ON f.symbol = t.symbol
            ORDER BY f.roe DESC LIMIT 20
        ''').fetchdf()
        return result
    
    def export(self, result, format='csv'):
        '''导出结果'''
        if format == 'csv':
            result.to_csv(
                f'signal_{datetime.now().strftime("%Y%m%d")}.csv',
                index=False
            )
        return result

# 使用示例
scraper = StockScraper()
scraper.update_data('new_data.csv')  # 每天收盘后追加新数据
results = scraper.scan()
results.export(results)
print(f"🎯 今日筛选出 {len(results)} 只标的")

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);

conn = duckdb.connect('stock_scraper.duckdb')
conn.execute("SET threads TO 4")
conn.execute("SET memory_limit='4GB'")

-- 将 CSV 转为 Parquet，压缩比更高、读取更快
COPY daily_kline TO 'kline_data.parquet' (FORMAT PARQUET);