用 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
""")

-- fundamentals 表已经建好，条件是 PE < 20 且 ROE > 15
-- 这步在数据导入时完成，高效且只需执行一次

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)

-- 三层取交集，得到今日推荐股票
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 (
                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'):
        """导出结果"""
        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)} 只标的")
print(results[['symbol', 'pe_ratio', 'roe', 'close', 'break_pct']])

-- 在 daily_kline 表上创建索引，加速增量更新时的 MAX(date) 查询
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
-- ... 完整的三层 JOIN 逻辑 ...
WHERE t.date = (SELECT MAX(date) FROM daily_kline);

-- 下次直接查询即可
SELECT * FROM today_signals ORDER BY roe DESC LIMIT 20;

conn = duckdb.connect('stock_scraper.duckdb')
# 启用并行：根据 CPU 核心数自动设置
conn.execute("SET threads TO 4")
conn.execute("SET memory_limit='4GB'")