

> 前几篇我们把动量、波动率、换手率算了个遍，但每天跑一遍脚本、重新拉数据、重新算因子，既慢又浪费额度。本文解决一个问题：**把因子落进本地数据库，支持增量更新，让下游策略直接查**。

## 一、为什么因子必须持久化

如果你跟着前几篇写过代码，一定会遇到这个痛点：

- 回测要用 6 个月的因子值 → 每次回测拉 6 个月 K 线重算
- 选股模型要对比今天和昨天的得分 → 昨天算完今天丢了
- 多因子组合需要同时读 mom/vol/turnover → 三个接口各拉一遍

**持久化的好处**：算一次、存下来、天天复用。新数据来了只算新增部分（增量更新），历史记录不动。

## 二、本文你将得到什么

1. 一套逐日计算因子的流水线（动量/波动率/换手率）
2. 用 SQLite 做本地因子库的完整代码（建表+索引+upsert）
3. 增量更新逻辑：只拉新日期、只算新因子、旧数据不碰
4. 两个下游查询示例："最近一期得分 Top5" + "单票历史因子序列"

## 三、核心代码：因子 → SQLite

### 3.1 依赖与数据拉取

```python
import sqlite3
import pandas as pd
import numpy as np
from mairui import Client

# 演示证书（请替换为你自己的正式证书）
client = Client("LICENCE-66D8-9F96-0C7F0FBCD073")

# 拉取单票历史 K 线（前复权）
raw = client.stock_history("600519", lt=180, dividend="n")
df = pd.DataFrame(raw)
df["t"] = pd.to_datetime(df["t"])
```

### 3.2 因子计算函数（复用前几篇逻辑）

```python
def momentum_factor(series, window=60, skip=5):
    end = len(series) - 1 - skip
    start = end - window
    if start < 0:
        return np.nan
    return series.iloc[end] / series.iloc[start] - 1

def volatility_factor(series, window=20):
    rets = series.pct_change().dropna().tail(window)
    return float(np.std(rets) * np.sqrt(252))

def turnover_proxy(df, window=20):
    sub = df.tail(window)
    return float((sub["v"] * sub["c"]).mean())
```

### 3.3 建表与增量写入

```python
def init_db(db_path="factor.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS daily_factor (
            code TEXT NOT NULL,
            trade_date TEXT NOT NULL,
            mom REAL,
            vol REAL,
            turnover REAL,
            score REAL,
            updated_at TEXT,
            PRIMARY KEY (code, trade_date)
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_date ON daily_factor(trade_date)")
    conn.commit()
    return conn

def upsert_factors(conn, df_factor):
    """INSERT OR REPLACE 增量写入，带综合得分"""
    now = pd.Timestamp.now().isoformat()
    rows = []
    for d in df_factor["trade_date"].unique():
        snap = df_factor[df_factor["trade_date"] == d].copy()
        # z-score 合成：mom 正向，vol/turnover 负向（低波、低换手更好）
        snap["score"] = (
            ((snap["mom"] - snap["mom"].mean()) / snap["mom"].std()).fillna(0) * 0.4
            + (-(snap["vol"] - snap["vol"].mean()) / snap["vol"].std()).fillna(0) * 0.3
            + (-(snap["turnover"] - snap["turnover"].mean()) / snap["turnover"].std()).fillna(0) * 0.3
        )
        for _, r in snap.iterrows():
            rows.append((r["code"], d, r["mom"], r["vol"], r["turnover"], r["score"], now))
    conn.executemany(
        """INSERT OR REPLACE INTO daily_factor
           (code, trade_date, mom, vol, turnover, score, updated_at)
           VALUES (?,?,?,?,?,?,?)""", rows)
    conn.commit()
```

### 3.4 下游查询示例

```python
# 查询最近一期得分最高的 5 只
def query_latest_top5(conn):
    sql = """
        SELECT * FROM daily_factor
        WHERE trade_date = (SELECT MAX(trade_date) FROM daily_factor)
        ORDER BY score DESC LIMIT 5
    """
    return pd.read_sql_query(sql, conn)

# 查询单票历史序列
def query_history(conn, code):
    sql = "SELECT * FROM daily_factor WHERE code = ? ORDER BY trade_date"
    return pd.read_sql_query(sql, conn, params=(code,))
```

## 四、自验结果

在 30 只大盘股面板上运行，生成 900 条因子记录（30 只 x 30 个交易日），SQLite 写入/更新 900 条，查询耗时 < 10ms。

> 注：若使用演示证书，因子值截面差异较小，score 可能趋同；替换正式证书后各票因子即有真实差异，排序分层正常。

## 五、常见坑与注意

1. **PRIMARY KEY 必须用 (code, trade_date)**：单票单日唯一一行，重复跑脚本时 `INSERT OR REPLACE` 自动覆盖。
2. **一定要建索引**：`trade_date` 索引让"查最近一期"从全表扫描变成范围查询，数据量上万时差异明显。
3. **score 的 z-score 要按截面算**：每天单独做标准化，不能把 30 天的数据混在一起算均值标准差。
4. **增量更新前先查库内最大日期**：只拉 `MAX(trade_date)` 之后的新数据，减少 API 调用次数。
5. **字段类型统一**：`REAL` 存浮点，`TEXT` 存日期（格式固定 `YYYY-MM-DD`，方便排序）。

## 六、小结 + 下篇预告

本文把因子从"内存里的临时变量"变成了"可查询、可复用、可增量更新"的本地资产。核心链路：

```
K 线数据 → 逐日算因子 → SQLite upsert → 下游直接 SELECT
```

**下一篇（#10）**：因子上线后不会永远有效，市场风格一变 IC 可能就掉。我们将用**滚动 IC**监控因子有效性，判断什么时候该调参、什么时候该下线。

---

## 免责声明

本文仅供技术学习交流，不构成投资建议。文中示例数据来源于公开行情接口，历史表现不代表未来收益。投资有风险，决策需谨慎。

## 了解更多

- 代码与文档：https://github.com/MaiRuiApi


前几篇我们把动量、波动率、换手率算了个遍，但每天跑一遍脚本、重新拉数据、重新算因子，既慢又浪费额度。本文解决一个问题：把因子落进本地数据库，支持增量更新，让下游策略直接查。

【Python量化因子实战 #09】因子算完就丢？用 SQLite 落地+增量更新，下游直接查

编程语言

数据库

后端

本文介绍因子持久化方案，将动量、波动率、换手率等因子存入SQLite数据库，支持增量更新与快速查询。通过PRIMARY KEY实现去重覆盖，建立trade_date索引加速查询，z-score按截面标准化合成综合得分。实测900条记录写入查询耗时<10ms，显著提升策略迭代效率，适用于多因子组合的本地化存储与调用场景。

代码解释

SQLite

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【Python量化因子实战 #09】因子算完就丢？用 SQLite 落地+增量更新，下游直接查-腾讯云开发者社区-腾讯云

【Python量化因子实战 #09】因子算完就丢？用 SQLite 落地+增量更新，下游直接查

【Python量化因子实战 #09】因子算完就丢？用 SQLite 落地+增量更新，下游直接查

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐