量化数据库：Python连接与写入

量化数据库：Python连接与写入

小象量化极客小册 · 华尔街量化分析先锋系列

在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

小象量化极客小册 · 华尔街量化分析先锋系列 在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

💡 核心导读

本节将介绍数据库的基础操作技能，包括数据库连接、CRUD操作（增删改查）以及数据批量处理。通过Python的SQLite3和SQLAlchemy实战演示，读者将掌握：1) 使用标准库操作轻量级数据库；2) 通过ORM框架进行高效数据管理；3) 实现结构化数据的持久化存储。所有示例均采用真实业务场景的数据模型，包含完整的异常处理机制。

数据库连接与初始化

本地数据库创建

import sqlite3
from pathlib import Path

db_path = Path('sales_data.db')

if db_path.exists():
    db_path.unlink()

conn = sqlite3.connect(db_path)
cursor = conn.cursor()

cursor.execute('''
CREATE TABLE products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    category TEXT CHECK(category IN ('电子', '量化持仓', '服饰')),
    price REAL DEFAULT 0.0,
    stock INTEGER DEFAULT 0
)''')

conn.commit()
print("数据库初始化完成")

连接状态验证

cursor.execute("PRAGMA table_info(products)")
columns = cursor.fetchall()

print("┌─────┬───────────┬─────────┬─────────┬───────┐")
print("│ CID │   Name    │  Type   │ Not Null│  PK   │")
print("├─────┼───────────┼─────────┼─────────┼───────┤")
for col in columns:
    print(f"│ {col[0]:<3} │ {col[1]:<9} │ {col[2]:<7} │ {col[3]:^7} │ {col[5]:^5} │")
print("└─────┴───────────┴─────────┴─────────┴───────┘")

💡 概念小测：数据库连接与提交

使用 Python sqlite3 将当日行情数据执行 INSERT 写入数据库后，必须执行什么操作才能确保数据真正被持久化写入磁盘，而不是只留在内存缓存中？

• 必须执行 conn.close() 关闭连接。
• 必须在连接对象上调用 conn.commit() 提交事务。
• SQLite 会在每次 execute 后自动保存，无需任何操作。
• 必须运行 conn.rollback() 清空缓存。

📝 选择题

💡 概念小测：数据库连接与提交 使用 Python sqlite3 将当日行情数据执行 INSERT 写入数据库后，必须执行什么操作才能确保数据真正被持久化写入磁盘，而不是只留在内存缓存中？

• A. 必须执行 conn.close() 关闭连接。
• B. 必须在连接对象上调用 conn.commit() 提交事务。
• C. SQLite 会在每次 execute 后自动保存，无需任何操作。
• D. 必须运行 conn.rollback() 清空缓存。

答案：B（必须在连接对象上调用 conn.commit() 提交事务。）

核心CRUD操作

批量数据插入

import pandas as pd

products = [
    ('智能手机', '电子', 5999.0, 100),
    ('蓝牙耳机', '电子', 399.0, 200),
    ('牛肉干', '量化持仓', 89.9, 50),
    ('T恤衫', '服饰', 129.0, 80)
]

cursor.executemany(
    "INSERT INTO products (name, category, price, stock) VALUES (?, ?, ?, ?)",
    products
)
conn.commit()

df = pd.read_sql("SELECT * FROM products", conn)
print("\n插入后的产品表：")
print(df.to_markdown(index=False, tablefmt="grid"))

数据更新与事务

try:
    # 开启事务
    conn.execute("BEGIN TRANSACTION")
    
    # 执行价格调整
    cursor.execute("UPDATE products SET price = price * 0.9 WHERE category = '电子'")
    
    # 模拟业务异常
    if len(products) > 0:
        pass  # 实际业务中可能有条件判断
        
    # 提交事务
    conn.commit()
    print("电子类产品价格已调整9折")
    
except Exception as e:
    conn.rollback()
    print(f"事务回滚: {str(e)}")

ORM框架实战

SQLAlchemy配置

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine(f'sqlite:///{db_path}')
Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    category = Column(String(10))
    price = Column(Float)
    stock = Column(Integer)

Session = sessionmaker(bind=engine)
session = Session()

高级查询示例

results = session.query(Product).filter(
    Product.category == '电子',
    Product.stock < 150
).order_by(Product.price.desc()).all()

print("\n库存紧张的电子产品：")
for p in results:
    print(f"{p.id:02d} | {p.name:<8} | 库存: {p.stock:<3} | 价格: {p.price:.1f}")

💡 代码填空：创建SQLite连接

补全代码，连接到本地名为 quant_data.db 的 SQLite 数据库：

import sqlite3

conn = sqlite3.{blank}('quant_data.db')
cursor = conn.cursor()
print("数据库连接成功！")

• open
• connect
• Cursor

✏️ 填空题

💡 代码填空：创建SQLite连接 补全代码，连接到本地名为 quant_data.db 的 SQLite 数据库：

import sqlite3
# 创建或连接到本地数据库文件
conn = sqlite3.___('quant_data.db')
cursor = conn.cursor()
print("数据库连接成功！")

答案：connect

本章小结

行动清单

• 在SQLite中新建一个客户表(customers)，包含id、name、vip_level字段，并通过Python插入5条测试数据
• 使用SQLAlchemy实现产品库存预警功能，当库存低于阈值时自动发送预警信息（打印到控制台）
• 编写一个事务处理示例，模拟订单创建时同时扣减库存和生成物流记录的操作

「小象量化教研组」