04_SQLite 数据持久化与 DAO 模式实践：构建高性能本地数据层

04_SQLite 数据持久化与 DAO 模式实践构建高性能本地数据层SQLite 是桌面应用的首选数据库但直接使用容易陷入性能陷阱和并发问题。本文基于 GPFX 项目系统讲解 SQLite 在 Python 桌面应用中的最佳实践包括连接管理、事务处理、DAO 模式、批量操作优化等核心技术。一、为什么选择 SQLite对于桌面量化分析系统SQLite 具有独特优势特性SQLiteMySQL/PostgreSQL部署零配置单文件需安装服务端性能本地读写极快网络开销并发单写多读高并发事务完整 ACID完整 ACID适用场景桌面应用、嵌入式服务端、分布式关键指标5000 只股票 × 250 交易日/年 × 10 年 1250 万条日线数据SQLite 本地查询毫秒级响应单文件便于备份和迁移二、数据库连接管理2.1 连接池与线程安全SQLite 连接不能跨线程使用需要精心设计importsqlite3importthreadingfromcontextlibimportcontextmanagerfrompathlibimportPathclassDatabase:SQLite 数据库连接管理器def__init__(self,db_path:str|Path):self._db_pathstr(db_path)self._conn:sqlite3.Connection|NoneNoneself._lockthreading.RLock()# 可重入锁支持嵌套调用defconnect(self)-sqlite3.Connection:获取连接懒加载 双重检查ifself._connisNone:withself._lock:ifself._connisNone:# 确保目录存在Path(self._db_path).parent.mkdir(parentsTrue,exist_okTrue)# 创建连接self._connsqlite3.connect(self._db_path,check_same_threadFalse,# 允许跨线程使用需配合锁timeout30# 锁等待超时 30 秒)# 返回字典式结果self._conn.row_factorysqlite3.Row# 性能优化 PRAGMAself._conn.execute(PRAGMA journal_modeWAL)# 写前日志self._conn.execute(PRAGMA foreign_keysON)# 外键约束self._conn.execute(PRAGMA synchronousNORMAL)# 平衡安全与性能returnself._conndefclose(self):关闭连接withself._lock:ifself._conn:self._conn.close()self._connNone关键配置解析journal_modeWAL写前日志模式读操作不阻塞写操作崩溃恢复更安全性能提升 2-3 倍synchronousNORMAL同步级别FULL每次事务都刷盘最安全最慢NORMALWAL 模式下安全且快速OFF不保证持久性最快风险高check_same_threadFalse允许跨线程必须配合锁使用由上层保证线程安全2.2 上下文管理器支持classDatabase:def__enter__(self):支持 with 语句self.connect()returnselfdef__exit__(self,exc_type,exc_val,exc_tb):自动关闭连接self.close()returnFalse# 不吞掉异常# 使用示例withDatabase(data/stocks.db)asdb:db.execute(CREATE TABLE IF NOT EXISTS ...)# 退出时自动关闭2.3 事务管理contextmanagerdeftransaction(self):事务上下文管理器 自动提交/回滚 - 正常结束commit - 发生异常rollback withself._lock:connself.connect()try:yieldconn conn.commit()exceptException:try:conn.rollback()exceptException:pass# 回滚失败也要抛出原异常raise# 使用示例defbatch_insert(self,records:list[dict]):withself.transaction()asconn:conn.executemany(INSERT OR REPLACE INTO daily_none VALUES (?, ?, ...),records)# 所有插入在一个事务中要么全成功要么全失败三、DAO 模式实现3.1 通用 SQL 生成器避免手写重复 SQL通过元编程自动生成classStockDAO:股票数据访问对象# 字段定义与 Tushare API 输出参数一致STOCK_BASIC_FIELDS[ts_code,symbol,name,area,industry,market,exchange,list_status,list_date,]DAILY_DATA_FIELDS[ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount,]staticmethoddef_build_upsert_sql(table:str,fields:list[str],conflict_cols:list[str])-str:生成 UPSERT SQLINSERT OR REPLACE SQLite 语法 INSERT INTO table (cols) VALUES (vals) ON CONFLICT(conflict_cols) DO UPDATE SET colexcluded.col col_str, .join(fields)param_str, .join(f:{f}forfinfields)# 命名参数# 冲突时要更新的字段排除主键update_fields[fforfinfieldsiffnotinconflict_cols]conflict_str, .join(conflict_cols)ifnotupdate_fields:# 没有要更新的字段冲突时什么都不做return(fINSERT INTO{table}({col_str}) VALUES ({param_str}) fON CONFLICT({conflict_str}) DO NOTHING)# 冲突时更新所有非主键字段update_str, .join(f{f}excluded.{f}forfinupdate_fields)return(fINSERT INTO{table}({col_str}) VALUES ({param_str}) fON CONFLICT({conflict_str}) DO UPDATE SET{update_str})为什么用命名参数:field而非?可读性更强参数顺序不易错支持字典传参{ts_code: 000001.SZ, ...}便于日志调试3.2 数据规范化处理 pandas 的 NaN 与 SQLite 的 NULL 转换staticmethoddef_normalize_records(records:list[dict],fields:list[str])-list[dict]:规范化记录 - NaN → NoneSQLite NULL - 确保所有字段存在 importmath result[]forrinrecords:row{}forfieldinfields:vr.get(field)# pandas NaN 是 float 类型需要特殊处理ifisinstance(v,float)andmath.isnan(v):row[field]Noneelse:row[field]v result.append(row)returnresult3.3 表名白名单防注入动态表名是 SQL 注入的高危点必须严格校验classDatabase:# 合法表名白名单_VALID_TABLES{users,stock_basic,daily_none,daily_qfq,daily_hfq,adj_factor,trade_cal,selection_runs,selection_results,backtest_runs,backtest_daily,watchlist,intraday_min,index_daily,}def_validate_table(self,name:str)-str:校验表名合法性ifnamenotinself._VALID_TABLES:raiseValueError(f非法表名:{name})returnnamedefquery_table(self,table:str,where:str,params:tuple()):安全的动态表查询tableself._validate_table(table)# 校验表名sqlfSELECT * FROM{table}ifwhere:sqlf WHERE{where}# where 子句仍需参数化returnself.query(sql,params)安全原则表名、列名白名单校验值参数化查询?或:name绝不拼接用户输入四、批量操作优化4.1 批量插入单条插入与批量插入性能对比defupsert_daily_batch(self,records:list[dict],adj_type:str):批量插入日线数据 性能对比5000 条记录 - 单条插入30 秒 - 批量插入0.5 秒60 倍提升 tableself._daily_table(adj_type)fieldsself.DAILY_DATA_FIELDS[updated_at]# 生成 UPSERT SQLsqlself._build_upsert_sql(table,fields,[ts_code,trade_date])# 规范化数据NaN → Nonenormalizedself._normalize_records(records,fields)# 添加更新时间戳fromdatetimeimportdatetime nowdatetime.now().strftime(%Y-%m-%d %H:%M:%S)forrinnormalized:r[updated_at]now# 批量执行关键使用事务withself._db.transaction()asconn:conn.executemany(sql,normalized)性能优化要点executemany vs execute减少 SQL 解析开销事务包裹5000 次提交合并为 1 次预处理语句避免重复解析 SQL4.2 批量查询避免 N1 查询问题defget_daily_batch(self,ts_codes:list[str],start_date:str,end_date:str)-pd.DataFrame:批量查询多只股票日线数据 反模式N1 查询 for code in ts_codes: df get_daily(code, ...) # 5000 次查询 正确做法 一次查询返回所有数据 # 构建 IN 子句的占位符placeholders,.join(?*len(ts_codes))sqlf SELECT * FROM daily_qfq WHERE ts_code IN ({placeholders}) AND trade_date BETWEEN ? AND ? ORDER BY ts_code, trade_date params(*ts_codes,start_date,end_date)rowsself._db.query(sql,params)returnpd.DataFrame([dict(r)forrinrows])五、索引设计5.1 高频查询索引根据查询模式设计索引-- 日线表最常按股票代码和日期范围查询CREATEINDEXidx_daily_none_codeONdaily_none(ts_code);CREATEINDEXidx_daily_none_dateONdaily_none(trade_date);-- 联合索引同时按代码和日期查询覆盖索引CREATEINDEXidx_daily_none_code_dateONdaily_none(ts_code,trade_date);-- 选股结果按运行ID查询明细CREATEINDEXidx_sel_results_runONselection_results(run_id);-- 回测净值按回测ID和日期查询CREATEINDEXidx_bt_daily_btONbacktest_daily(backtest_id,trade_date);5.2 索引使用验证使用EXPLAIN QUERY PLAN验证索引是否生效defexplain_query(self,sql:str,params:tuple()):分析查询计划rowsself.query(fEXPLAIN QUERY PLAN{sql},params)forrowinrows:print(row[detail])# 测试dao.explain_query(SELECT * FROM daily_none WHERE ts_code ? AND trade_date ?,(000001.SZ,20240101))# 输出SEARCH daily_none USING INDEX idx_daily_none_code_date (ts_code? AND trade_date?)六、数据迁移与版本管理6.1 表结构演进支持平滑的表结构升级def_create_stock_basic_table(self):创建/升级 stock_basic 表# 检查表是否存在colsself.query(PRAGMA table_info(stock_basic))ifnotcols:# 表不存在创建新表col_defs[ts_code TEXT PRIMARY KEY]forname,col_typeinself._STOCK_BASIC_COLUMNS.items():col_defs.append(f{name}{col_type})self.execute(fCREATE TABLE stock_basic ({, .join(col_defs)}))return# 表已存在检查是否需要添加新列existing{c[name]forcincols}forname,col_typeinself._STOCK_BASIC_COLUMNS.items():ifnamenotinexisting:self.execute(fALTER TABLE stock_basic ADD COLUMN{name}{col_type})6.2 数据迁移示例从旧表迁移到新表结构def_migrate_daily_tables(self):从旧的 daily_data 表迁移到分表结构 旧结构单表 daily_data用 adj_type 字段区分 新结构daily_none / daily_qfq / daily_hfq 三张表 # 检查旧表是否存在old_colsself.query(PRAGMA table_info(daily_data))ifnotold_cols:# 旧表不存在直接创建新表forsuffixin(none,qfq,hfq):self._create_one_daily_table(suffix)return# 旧表存在执行数据迁移withself.transaction()asconn:forsuffixin(none,qfq,hfq):# 创建新表self._create_one_daily_table(suffix)# 迁移数据conn.execute(f INSERT OR IGNORE INTO daily_{suffix}SELECT * FROM daily_data WHERE adj_type ? ,(suffix,))# 删除旧表conn.execute(DROP TABLE daily_data)迁移原则使用INSERT OR IGNORE避免重复在事务中执行失败可回滚保留旧表备份可选七、并发控制7.1 读写分离SQLite WAL 模式支持单写多读classDatabase:def__init__(self,db_path:str):self._db_pathdb_path self._write_lockthreading.RLock()# 写锁self._read_lockthreading.Lock()# 读锁可选defexecute(self,sql:str,params:tuple()):写操作独占锁withself._write_lock:connself.connect()try:curconn.execute(sql,params)conn.commit()returncurexceptException:conn.rollback()raisedefquery(self,sql:str,params:tuple())-list:读操作WAL 模式下无需加锁# WAL 模式下读操作不会被写操作阻塞connself.connect()returnconn.execute(sql,params).fetchall()7.2 线程局部连接多线程环境下的连接管理classThreadLocalDatabase:线程局部数据库连接def__init__(self,db_path:str):self._db_pathdb_path self._localthreading.local()defget_connection(self)-sqlite3.Connection:获取当前线程的连接ifnothasattr(self._local,conn):self._local.connsqlite3.connect(self._db_path,check_same_threadFalse)self._local.conn.row_factorysqlite3.Rowreturnself._local.conndefclose_current(self):关闭当前线程的连接ifhasattr(self._local,conn):self._local.conn.close()delself._local.conn八、性能监控与调优8.1 慢查询日志importtimeimportlogging loggerlogging.getLogger(__name__)classMonitoredDatabase(Database):带性能监控的数据库SLOW_QUERY_THRESHOLD1.0# 慢查询阈值秒defquery(self,sql:str,params:tuple()):starttime.time()resultsuper().query(sql,params)elapsedtime.time()-startifelapsedself.SLOW_QUERY_THRESHOLD:logger.warning(f慢查询 ({elapsed:.2f}s):{sql[:100]}... f参数:{params})returnresult8.2 数据库统计defget_db_stats(self)-dict:获取数据库统计信息stats{}# 文件大小importos stats[file_size_mb]os.path.getsize(self._db_path)/1024/1024# 各表行数tables[daily_none,daily_qfq,daily_hfq,stock_basic]fortableintables:countself.query_one(fSELECT COUNT(*) as c FROM{table})stats[f{table}_rows]count[c]# 页面统计page_countself.query_one(PRAGMA page_count)[page_count]page_sizeself.query_one(PRAGMA page_size)[page_size]stats[page_count]page_count stats[page_size]page_size stats[total_size_mb]page_count*page_size/1024/1024returnstats九、总结GPFX 的 SQLite 数据层核心实践技术点方案效果连接管理懒加载 双重检查 可重入锁线程安全性能最优事务处理上下文管理器自动提交/回滚数据一致性保证批量操作executemany 事务包裹60 倍性能提升SQL 生成元编程自动生成 UPSERT减少重复代码安全防护表名白名单 参数化查询杜绝 SQL 注入索引优化高频字段 联合索引查询毫秒级响应数据迁移事务包裹 OR IGNORE平滑升级无中断并发控制WAL 模式 写锁单写多读不阻塞性能监控慢查询日志 统计分析问题可追踪这套数据层支撑了 5000 只股票、千万级日线数据的高效存储和查询为上层量化分析提供了坚实的数据基础。核心技术SQLite WAL / Python sqlite3 / 事务管理 / DAO 模式 / 索引优化性能指标批量插入5000 条/0.5s单股查询 10ms全市场扫描 100ms