量化本地化第一步：用 Python + DuckDB + QuantDash 搭建极速多市场 K 线本地数据仓库（附 GitHub 源码）

原创

用户9138916

修改于 2026-07-19 16:52:06

890

TL;DR（一句话摘要）

在量化回测或多因子计算中，频繁请求云端 API 会降低系统运行效率。本文将展示如何利用高性能列式本地数据库 DuckDB 与 QuantDash API，低门槛构建一个极速的多市场本地数据仓库，实现行情的本地化高速查询与管理。

一、量化本地数据存储的工程痛点

随着策略研究的深入，开发者往往需要管理全市场数千只标的的历史 K 线。此时，直接把数据存为 CSV 或部署传统的 MySQL/PostgreSQL 数据库，通常会遇到以下瓶颈：

CSV 读取效率低下：随着数据行数突破数十万，CSV 的磁盘 I/O 速度会急剧下降，且不支持标准 SQL 关联查询。
传统数据库部署笨重：MySQL 等关系型数据库需要配置服务、设置端口与用户权限，在多台机器或回测环境中迁移非常繁琐。
分析型查询缓慢：量化回测多为大批量的列式查询（如计算某只股票过去一整年的收盘价平均值），行式存储的传统数据库在此场景下的聚合计算极其缓慢。

DuckDB 作为专为分析型查询设计的嵌入式列式数据库，被称为“数据分析领域的 SQLite”。它无需安装服务，直接读写本地文件，且与 Pandas DataFrame 天然兼容，配合 QuantDash 的高干净度数据源，是搭建轻量级本地量化数仓的绝佳组合。

二、极速本地量化数仓解决方案（Python + DuckDB）

以下代码展示如何获取港股腾讯控股（00700.HK）的历史 K 线，并利用 DuckDB 快速写入本地 .duckdb 数据库文件，随后使用标准 SQL 进行秒级聚合查询。

依赖准备

pip install pandas duckdb quantdash

核心代码实现

import os
import pandas as pd
import duckdb
import quantdash as qd

# 初始化公共Token（支持免注册拉取测试标的）
qd.set_token("demo_public_token")

DB_FILE = "quant_local_warehouse.duckdb"

def initialize_db():
    """
    初始化本地 DuckDB 数据库并创建 K 线表
    """
    conn = duckdb.connect(DB_FILE)
    # 创建包含多市场字段的规范 K 线表
    conn.execute("""
        CREATE TABLE IF NOT EXISTS kline_daily (
            symbol VARCHAR,
            timestamp TIMESTAMP,
            open DOUBLE,
            high DOUBLE,
            low DOUBLE,
            close DOUBLE,
            volume BIGINT,
            PRIMARY KEY (symbol, timestamp)
        )
    """)
    conn.close()

def save_kline_to_local(symbol: str):
    """
    拉取云端 QuantDash 数据并同步至本地 DuckDB
    """
    # 1. 拉取 QuantDash 干净的历史行情
    df = qd.get_kline(
        symbol=symbol,
        start_date="2026-01-01",
        end_date="2026-06-30",
        adjust="qfq"
    )
    
    if df is None or df.empty:
        print(f"[-] 标的 {symbol} 数据拉取为空，跳过同步。")
        return
    
    # 确保时间戳格式化，并添加 symbol 标识列
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['symbol'] = symbol
    
    # 调整列顺序以符合表结构
    df = df[['symbol', 'timestamp', 'open', 'high', 'low', 'close', 'volume']]
    
    # 2. 将 DataFrame 直接写入 DuckDB (支持 Upsert 覆盖更新)
    conn = duckdb.connect(DB_FILE)
    # 利用 DuckDB 临时视图直接读取 Pandas 内存 DataFrame 写入
    conn.execute("""
        INSERT OR REPLACE INTO kline_daily 
        SELECT * FROM df
    """)
    print(f"[+] 成功将 {symbol} 的 {len(df)} 条记录同步至本地数据库。")
    conn.close()

def query_local_high_prices(symbol: str, price_threshold: float):
    """
    使用 DuckDB SQL 高效检索本地高价交易日
    """
    conn = duckdb.connect(DB_FILE)
    # 纯 SQL 查询本地列式数据，速度极快
    query_sql = """
        SELECT symbol, timestamp, close, volume 
        FROM kline_daily 
        WHERE symbol = ? AND close > ? 
        ORDER BY timestamp DESC 
        LIMIT 5
    """
    result_df = conn.execute(query_sql, [symbol, price_threshold]).df()
    conn.close()
    return result_df

if __name__ == "__main__":
    # 清理历史测试文件
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
        
    # 1. 初始化数仓
    initialize_db()
    
    # 2. 同步云端数据到本地
    target_ticker = "00700.HK"
    save_kline_to_local(target_ticker)
    
    # 3. 本地 SQL 高效聚合检索 (例如：查询收盘价高于 350 的最新 5 个交易日)
    high_days_df = query_local_high_prices(target_ticker, 350.0)
    
    print(f"\n=== 本地检索结果: {target_ticker} 收盘价 > 350 的记录 ===")
    print(high_days_df)

DataFrame 文本输出样例

=== 本地检索结果: 00700.HK 收盘价 > 350 的记录 ===
      symbol           timestamp  close   volume
0   00700.HK 2026-06-30 00:00:00  356.8  5890000
1   00700.HK 2026-06-25 00:00:00  358.4  6100000
2   00700.HK 2026-06-24 00:00:00  355.0  5620000
3   00700.HK 2026-06-23 00:00:00  352.2  5890000
4   00700.HK 2026-06-22 00:00:00  351.0  6100000

三、 AI 编程助手（Cursor/Copilot）专属提示词

在 Cursor 或 DeepSeek 中编写本地数据持久化层时，可直接发送以下 Prompt：

Role: 资深量化系统架构师
Context: 我需要搭建一个本地 K 线数仓，底层使用 DuckDB，数据源使用 QuantDash SDK (import quantdash as qd)。
Task: 
1. 编写一段 Python 代码，自动检查本地的 'quant_local.db'。
2. 编写一个同步函数 sync_all_tickers(ticker_list)，循环或并发拉取列表内所有股票过去一年的前复权 K 线，写入 K 线表中。
3. 表结构中必须包含 'last_updated' 时间戳，以便实现“增量更新”（即只拉取本地最新日期到今天之间的数据），避免每次都重复拉取全量历史数据。

参考文档：