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

小象实战讲义 · 量化交易Python入门：从咖啡馆到华尔街

在量化交易中，历史回测数据往往以多种异构本地文件（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()
conn.close()
print("数据库初始化完成")

数据库初始化完成

连接状态验证

import sqlite3
from pathlib import Path

db_path = Path('sales_data.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()

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("└─────┴───────────┴─────────┴─────────┴───────┘")
conn.close()

┌─────┬───────────┬─────────┬─────────┬───────┐
│ CID │   Name    │  Type   │ Not Null│  PK   │
├─────┼───────────┼─────────┼─────────┼───────┤
│ 0   │ id        │ INTEGER │    0    │   1   │
│ 1   │ name      │ TEXT    │    1    │   0   │
│ 2   │ category  │ TEXT    │    0    │   0   │
│ 3   │ price     │ REAL    │    0    │   0   │
│ 4   │ stock     │ INTEGER │    0    │   0   │
└─────┴───────────┴─────────┴─────────┴───────┘

核心CRUD操作

批量数据插入

import sqlite3
import pandas as pd
from pathlib import Path

db_path = Path('sales_data.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()

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_string(index=False))
conn.close()

插入后的产品表：
 id name category  price  stock
  1 智能手机       电子 5999.0    100
  2 蓝牙耳机       电子  399.0    200
  3  牛肉干     量化持仓   89.9     50
  4  T恤衫       服饰  129.0     80

数据更新与事务

import sqlite3
from pathlib import Path

db_path = Path('sales_data.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()

try:
    conn.execute("BEGIN TRANSACTION")
    cursor.execute("UPDATE products SET price = price * 0.9 WHERE category = '电子'")
    conn.commit()
    print("电子类产品价格已调整9折")
    
except Exception as e:
    conn.rollback()
    print(f"事务回滚: {str(e)}")
finally:
    conn.close()

电子类产品价格已调整9折

ORM框架实战

SQLAlchemy配置

import sqlite3
from pathlib import Path

db_path = Path('sales_data.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()

print("使用原生SQL查询库存紧张的电子产品：")
cursor.execute('''
    SELECT id, name, stock, price
    FROM products
    WHERE category = '电子' AND stock < 150
    ORDER BY price DESC
''')
results = cursor.fetchall()

for row in results:
    print(f"{row[0]:02d} | {row[1]:<8} | 库存: {row[2]:<3} | 价格: {row[3]:.1f}")
conn.close()

使用原生SQL查询库存紧张的电子产品：
01 | 智能手机     | 库存: 100 | 价格: 5399.1

高级查询示例

import sqlite3
import pandas as pd
from pathlib import Path

db_path = Path('sales_data.db')
conn = sqlite3.connect(db_path)

df = pd.read_sql("SELECT * FROM products", conn)
print("\n最终产品表数据：")
print(df.to_string(index=False))
conn.close()

最终产品表数据：
 id name category  price  stock
  1 智能手机       电子 5399.1    100
  2 蓝牙耳机       电子  359.1    200
  3  牛肉干     量化持仓   89.9     50
  4  T恤衫       服饰  129.0     80

📝 动手练一练

第 1 题 · 概念应用

你正在编写一个量化回测程序，需要将每日计算出的策略持仓数据（包含股票代码、持仓数量、持仓成本）批量写入 SQLite 数据库。为了提高写入性能，你决定使用事务机制。以下哪种做法是正确且高效的？

A. 在循环中，对每一条持仓记录都执行 cursor.execute(insert_sql, data)，并在每次执行后立即调用 conn.commit()。 B. 先执行 conn.execute('BEGIN TRANSACTION')，然后使用 cursor.executemany(insert_sql, all_data) 一次性插入所有数据，最后再调用 conn.commit()。 C. 使用 cursor.executemany(insert_sql, all_data) 插入数据，然后立即调用 conn.rollback() 以确保数据安全。 D. 在循环中，对每一条持仓记录都执行 cursor.execute(insert_sql, data)，循环结束后再调用 conn.close()，系统会自动提交。

第 2 题 · 动手题（代码填空）

你正在构建一个简易的股票行情数据库。请补全以下代码，使用 sqlite3 模块创建一个名为 stock_prices 的表，并一次性插入三条模拟的日行情数据。

表结构要求：

• symbol (TEXT): 股票代码，非空。
• trade_date (TEXT): 交易日期，格式 ‘YYYY-MM-DD’，非空。
• close_price (REAL): 收盘价。
• volume (INTEGER): 成交量。

import sqlite3
import tempfile
import os

# 1. 创建一个临时数据库文件（唯一名，避免与历史练习文件冲突；实盘可换成固定业务路径）
fd, db_file = tempfile.mkstemp(suffix='.db', prefix='market_data_')
os.close(fd)

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

# 2. 创建表
create_table_sql = '''
CREATE TABLE stock_prices (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    symbol TEXT NOT NULL,
    trade_date TEXT NOT NULL,
    close_price REAL,
    volume INTEGER
)
'''
cursor.execute(create_table_sql)

# 3. 准备要插入的数据（三条记录）
daily_records = [
    ('000001.SZ', '2023-10-25', 14.32, 12543210),
    ('600519.SH', '2023-10-25', 1680.50, 3452100),
    ('000858.SZ', '2023-10-25', 35.88, 9876543)
]

# 4. 补全此处代码，实现批量插入
# 提示：使用 cursor 对象的一个方法，该方法专为执行带参数的多条相同SQL语句设计。
cursor.______(
    "INSERT INTO stock_prices (symbol, trade_date, close_price, volume) VALUES (?, ?, ?, ?)",
    daily_records
)

# 5. 提交事务并验证
conn.commit()

# 查询验证
cursor.execute("SELECT COUNT(*) FROM stock_prices")
count = cursor.fetchone()[0]
print(f"成功插入 {count} 条行情记录。")

cursor.close()
conn.close()

import sqlite3
import tempfile
import os

# 1. 创建一个临时数据库文件（唯一名，避免与历史练习文件冲突；实盘可换成固定业务路径）
fd, db_file = tempfile.mkstemp(suffix='.db', prefix='market_data_')
os.close(fd)

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

# 2. 创建表
create_table_sql = '''
CREATE TABLE stock_prices (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    symbol TEXT NOT NULL,
    trade_date TEXT NOT NULL,
    close_price REAL,
    volume INTEGER
)
'''
cursor.execute(create_table_sql)

# 3. 准备要插入的数据（三条记录）
daily_records = [
    ('000001.SZ', '2023-10-25', 14.32, 12543210),
    ('600519.SH', '2023-10-25', 1680.50, 3452100),
    ('000858.SZ', '2023-10-25', 35.88, 9876543)
]

# 4. 批量插入：executemany 用同一 SQL 模板一次性绑定多行参数
cursor.executemany(
    "INSERT INTO stock_prices (symbol, trade_date, close_price, volume) VALUES (?, ?, ?, ?)",
    daily_records
)

# 5. 提交事务并验证
conn.commit()

# 查询验证
cursor.execute("SELECT COUNT(*) FROM stock_prices")
count = cursor.fetchone()[0]
print(f"成功插入 {count} 条行情记录。")

cursor.close()
conn.close()

成功插入 3 条行情记录。

本章小结

📋 行动清单

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

—— 小象教研组