4、数据存储与管理：使用CSV存储数据，使用SQLite存储数据，数据读取与更新策略

数据存储，说白了就是给交易系统找个靠谱的“仓库”。

我刚开始做量化那会儿，觉得数据存哪儿都一样。后来吃过亏才明白——选错存储方式，后面全是坑。今天咱们就聊聊两种最常用的方案：CSV 和 SQLite。

4.1 为什么需要数据存储？

你想想看，策略跑起来，每天产生成千上万条行情记录、交易记录、资金曲线。如果每次重启程序都重新拉数据，不仅慢，还容易触发交易所的限流。

我个人习惯是：原始数据落地，计算数据缓存。这样即使网络断了，本地数据也能支撑回测和分析。

核心原则： 数据存储要解决三个问题——存什么、怎么存、怎么取。

4.2 使用 CSV 存储数据

CSV 是最简单的存储格式。说白了就是个表格文件，用逗号分隔。我早期做回测时，所有历史K线都存成CSV，方便用Excel直接查看。

4.2.1 CSV 的优缺点

优点 缺点
人类可读，Excel直接打开 没有索引，查询慢
跨平台，任何语言都能解析 不支持并发写入
文件小，传输方便 数据量大时加载慢
适合做数据交换格式 没有数据类型约束

4.2.2 实战：用 Python 读写 CSV

嗯，这里要注意：别自己手写解析逻辑，用标准库 csv 模块就行。

import csv
from datetime import datetime

# 写入数据
def save_to_csv(data, filename):
    with open(filename, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        for row in data:
            writer.writerow(row)

# 读取数据
def load_from_csv(filename):
    data = []
    with open(filename, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            row['timestamp'] = datetime.fromisoformat(row['timestamp'])
            data.append(row)
    return data

# 示例：保存5条K线数据
sample_data = [
    ['2024-01-01 09:30:00', 100.0, 101.5, 99.8, 101.2, 15000],
    ['2024-01-01 09:31:00', 101.2, 102.0, 100.5, 101.8, 12000],
]
save_to_csv(sample_data, 'btc_1min.csv')
print(load_from_csv('btc_1min.csv'))
我的经验： CSV 适合存“一次性写入、多次读取”的数据，比如历史K线。但别用它存交易订单——并发写入会出问题。

4.3 使用 SQLite 存储数据

当数据量上来后，CSV 就不够用了。你想想看，几百万条数据用 CSV 加载，内存直接爆炸。这时候就该 SQLite 上场了。

SQLite 是个轻量级数据库，不需要安装服务端，一个文件就是一个数据库。我在项目中遇到过，用 SQLite 存了3年的1分钟K线，查询速度比 CSV 快了几十倍。

4.3.1 SQLite 的核心优势

  • 支持 SQL 查询：可以按时间范围、价格条件筛选数据
  • 事务支持：写入失败可以回滚，数据不会损坏
  • 索引机制：给时间字段加索引，查询秒级响应
  • 并发读：多个进程可以同时读取

4.3.2 实战：创建表并写入数据

import sqlite3
from datetime import datetime

# 创建数据库连接
conn = sqlite3.connect('trading.db')
cursor = conn.cursor()

# 创建K线表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS kline_1min (
        timestamp TEXT PRIMARY KEY,
        open REAL NOT NULL,
        high REAL NOT NULL,
        low REAL NOT NULL,
        close REAL NOT NULL,
        volume INTEGER NOT NULL
    )
''')

# 插入数据（使用参数化查询防SQL注入）
def insert_kline(data):
    cursor.execute('''
        INSERT OR REPLACE INTO kline_1min 
        VALUES (?, ?, ?, ?, ?, ?)
    ''', data)
    conn.commit()

# 批量插入（性能更好）
def batch_insert(data_list):
    cursor.executemany('''
        INSERT OR REPLACE INTO kline_1min 
        VALUES (?, ?, ?, ?, ?, ?)
    ''', data_list)
    conn.commit()

# 示例
insert_kline(['2024-01-01 09:30:00', 100.0, 101.5, 99.8, 101.2, 15000])
batch_insert([
    ['2024-01-01 09:31:00', 101.2, 102.0, 100.5, 101.8, 12000],
    ['2024-01-01 09:32:00', 101.8, 102.3, 101.0, 102.1, 18000],
])

4.3.3 查询与更新策略

数据存进去，关键是怎么取。我总结了一套“懒加载+增量更新”的策略。

# 按时间范围查询（配合索引）
def query_by_time(start_time, end_time):
    cursor.execute('''
        SELECT * FROM kline_1min 
        WHERE timestamp BETWEEN ? AND ?
        ORDER BY timestamp ASC
    ''', (start_time, end_time))
    return cursor.fetchall()

# 增量更新：只更新最近N条数据
def incremental_update(new_data):
    # 先删除重复的时间戳
    timestamps = [row[0] for row in new_data]
    placeholders = ','.join(['?'] * len(timestamps))
    cursor.execute(f'''
        DELETE FROM kline_1min 
        WHERE timestamp IN ({placeholders})
    ''', timestamps)
    
    # 再批量插入新数据
    batch_insert(new_data)

# 查询最新一条数据的时间戳
def get_latest_timestamp():
    cursor.execute('SELECT MAX(timestamp) FROM kline_1min')
    result = cursor.fetchone()[0]
    return result if result else '2000-01-01 00:00:00'

# 使用示例
latest = get_latest_timestamp()
print(f'数据库中最新数据时间：{latest}')

# 只拉取最新数据
new_kline_data = fetch_from_exchange(start_time=latest)
incremental_update(new_kline_data)
我曾经踩过的坑： 别在循环里一条一条 insert，速度慢到怀疑人生。用 executemany 批量插入，性能提升100倍以上。

4.4 数据读取与更新策略

策略跑起来，数据是动态增长的。你不能每次重启都全量加载，那样太慢了。我常用的策略就三个字：增量、缓存、校验。

4.4.1 增量读取

只读取上次处理之后的新数据。比如你上次处理到 10:00:00，这次就只读 10:00:00 之后的数据。

def incremental_read(last_processed_time):
    """只读取新数据"""
    cursor.execute('''
        SELECT * FROM kline_1min 
        WHERE timestamp > ?
        ORDER BY timestamp ASC
    ''', (last_processed_time,))
    return cursor.fetchall()

4.4.2 内存缓存

频繁查询数据库？没必要。把最近的数据缓存到内存里，查询速度从毫秒级降到微秒级。

class DataCache:
    def __init__(self):
        self.cache = {}  # {timestamp: kline_data}
        self.max_size = 10000  # 最多缓存1万条
    
    def update(self, new_data):
        for row in new_data:
            self.cache[row[0]] = row
        # 如果缓存太大，删除最旧的数据
        if len(self.cache) > self.max_size:
            sorted_keys = sorted(self.cache.keys())
            for key in sorted_keys[:len(sorted_keys) - self.max_size]:
                del self.cache[key]
    
    def get(self, timestamp):
        return self.cache.get(timestamp)

# 使用
cache = DataCache()
cache.update(query_by_time('2024-01-01', '2024-01-02'))

4.4.3 数据校验

数据源偶尔会出问题，比如某根K线价格异常。我习惯在写入前做一次校验。

def validate_kline(row):
    """校验K线数据合理性"""
    timestamp, open_, high, low, close, volume = row
    
    # 价格不能为负
    if any(p <= 0 for p in [open_, high, low, close]):
        return False
    
    # 最高价必须大于等于最低价
    if high < low:
        return False
    
    # 开盘价和收盘价必须在最高最低之间
    if not (low <= open_ <= high and low <= close <= high):
        return False
    
    # 成交量不能为负
    if volume < 0:
        return False
    
    return True

# 写入前校验
valid_data = [row for row in new_data if validate_kline(row)]
if len(valid_data) != len(new_data):
    print(f'警告：过滤了 {len(new_data) - len(valid_data)} 条异常数据')
batch_insert(valid_data)
小技巧： 每天收盘后做一次全量校验，对比交易所的官方数据。我吃过亏，某次数据源漏了10分钟的数据，回测结果完全失真。

4.5 本章知识体系

下面这张图，帮你理清数据存储与管理的整体脉络。

数据存储与管理知识体系 数据源（交易所API） CSV存储 适合历史数据、数据交换 SQLite存储 适合高频查询、增量更新 读取策略：增量读取 + 内存缓存 更新策略：增量更新 + 数据校验 策略引擎：稳定、高效的数据供给

说白了，数据存储就三步：收进来、存下来、拿出去。CSV 和 SQLite 各有各的用武之地。我个人习惯是：原始数据用 SQLite 存，方便查询；导出报告时用 CSV，方便分享。

嗯，这一章的内容就到这儿。记住一点：数据是量化交易的地基。地基没打好，上面盖多高的楼都白搭。

专注资料整理

← 上一章 📖 返回目录 下一章 →