Tushare 量化实战 05：数据库存储与SQL查询优化

前一篇解决了批量数据获取的速度问题，这篇解决数据持久化：如何将获取的数据存入数据库，并进行高效的SQL查询。

选择数据库

数据库对比

选择：SQLite（开发）+ MySQL（生产）

表结构设计

股票基础信息表

— stock_basic 表
CREATE TABLE stock_basic (
ts_code VARCHAR(20) PRIMARY KEY, — 股票代码
symbol VARCHAR(10) NOT NULL, — 股票代码（不带后缀）
name VARCHAR(50) NOT NULL, — 股票名称
area VARCHAR(50), — 所属地域
industry VARCHAR(50), — 所属行业
list_status VARCHAR(1) NOT NULL, — 上市状态（L 上市/D 退市/P 暂停）
list_date DATE, — 上市日期
delist_date DATE, — 退市日期
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

— 索引
CREATE INDEX idx_industry ON stock_basic(industry);
CREATE INDEX idx_list_status ON stock_basic(list_status);

日线行情表

— daily_quotes 表
CREATE TABLE daily_quotes (
id BIGINT AUTO_INCREMENT PRIMARY KEY, — 自增ID
ts_code VARCHAR(20) NOT NULL, — 股票代码
trade_date DATE NOT NULL, — 交易日期
open DECIMAL(10,4), — 开盘价
high DECIMAL(10,4), — 最高价
low DECIMAL(10,4), — 最低价
close DECIMAL(10,4), — 收盘价
pre_close DECIMAL(10,4), — 前收盘价
change DECIMAL(10,4), — 涨跌额
pct_chg DECIMAL(10,4), — 涨跌幅
vol DECIMAL(15,2), — 成交量（手）
amount DECIMAL(20,2), — 成交额（万元）
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

— 唯一索引，避免重复数据
UNIQUE KEY uk_code_date (ts_code, trade_date)
);

— 索引
CREATE INDEX idx_code_date ON daily_quotes(ts_code, trade_date);
CREATE INDEX idx_date ON daily_quotes(trade_date);
CREATE INDEX idx_pct_chg ON daily_quotes(pct_chg);

股票列表表（沪深300/中证500等）

— index_weight 表
CREATE TABLE index_weight (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
index_code VARCHAR(20) NOT NULL, — 指数代码
ts_code VARCHAR(20) NOT NULL, — 股票代码
trade_date DATE NOT NULL, — 交易日期
weight DECIMAL(10,6), — 权重
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

UNIQUE KEY uk_index_code_date (index_code, ts_code, trade_date)
);

— 索引
CREATE INDEX idx_index_code ON index_weight(index_code, trade_date);
CREATE INDEX idx_ts_code ON index_weight(ts_code, trade_date);

数据库操作封装

基础数据库类

创建 src/db/database.py：

\”\”\”数据库操作封装\”\”\”
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, Session
from typing import Optional, List, Dict, Any
import pandas as pd

class Database:
\”\”\”数据库操作类\”\”\”

def __init__(self, url: str):
\”\”\”
初始化数据库连接

Args:
url: 数据库连接字符串
SQLite: \’sqlite:///data/db/quant.db\’
MySQL: \’mysql+mysqlconnector://user:pass@localhost/quant\’
\”\”\”
self.engine = create_engine(url, pool_pre_ping=True)
self.Session = sessionmaker(bind=self.engine