量化数据库：SQL常用查询语法

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

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

💡 核心导读

本节将介绍数据库基础概念和SQL核心语法，帮助读者掌握数据采集场景下的数据库操作技能。重点包括：数据库连接方法、CRUD操作、聚合查询以及Python与数据库的交互实践。通过本节学习，读者将能够使用SQL语句高效处理本地数据库中的多源数据，并实现与Python工作流的无缝衔接。

数据库基础操作

连接数据库

import sqlite3
import pandas as pd

conn = sqlite3.connect(':memory:')

data = {'id': [1, 2, 3], 'name': ['小象咖啡', '腾讯控股', '贵州茅台']}
df = pd.DataFrame(data)
df.to_sql('users', conn, index=False, if_exists='replace')

result = pd.read_sql('SELECT * FROM users', conn)
print(result)

执行结果：

id  name
0   1  小象咖啡
1   2  腾讯控股
2   3  贵州茅台

表结构操作

-- 创建新表
CREATE TABLE products (
    product_id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    price REAL CHECK(price > 0)
);

-- 查看表结构（注意：.schema 是 sqlite3 命令行工具的点命令，并非标准 SQL，仅能在 sqlite3 交互式命令行中执行）
.schema products

SQL核心语法实战

数据查询基础

import sqlite3
import pandas as pd

conn = sqlite3.connect(':memory:')

users_data = {'id': [1, 2, 3], 'name': ['小象咖啡', '腾讯控股', '贵州茅台']}
pd.DataFrame(users_data).to_sql('users', conn, index=False, if_exists='replace')

products_data = {'product_id': [1, 2, 3], 'name': ['咖啡', '股票', '白酒'], 'price': [30.0, 450.0, 1800.0]}
pd.DataFrame(products_data).to_sql('products', conn, index=False, if_exists='replace')

orders_data = pd.DataFrame([
    (1, 1, 3, 2),
    (2, 3, 1, 1),
    (3, 2, 2, 3)
], columns=['order_id', 'user_id', 'product_id', 'quantity'])
orders_data.to_sql('orders', conn, index=False, if_exists='replace')

query = '''
SELECT 
    u.name as user_name,
    p.name as product_name,
    o.quantity
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN products p ON o.product_id = p.product_id
WHERE o.quantity > 1
ORDER BY o.quantity DESC
'''

result = pd.read_sql(query, conn)
print(result)

user_name product_name  quantity
0      腾讯控股           股票         3
1      小象咖啡           白酒         2

数据修改操作

import sqlite3
import pandas as pd

conn = sqlite3.connect(':memory:')

products_data = {'product_id': [1, 2, 3], 'name': ['咖啡', '股票', '白酒'], 'price': [30.0, 450.0, 1800.0]}
pd.DataFrame(products_data).to_sql('products', conn, index=False, if_exists='replace')

orders_data = pd.DataFrame([
    (1, 1, 3, 2),
    (2, 3, 1, 1),
    (3, 2, 2, 3)
], columns=['order_id', 'user_id', 'product_id', 'quantity'])
orders_data.to_sql('orders', conn, index=False, if_exists='replace')

try:
    cursor = conn.cursor()
    cursor.execute("BEGIN TRANSACTION")
    cursor.execute("UPDATE products SET price = price * 0.9 WHERE product_id = 1")
    cursor.execute("INSERT INTO orders VALUES (4, 1, 2, 1)")
    conn.commit()
    print("事务执行成功")
except Exception as e:
    conn.rollback()
    print(f"事务失败: {e}")

result = pd.read_sql("SELECT * FROM products WHERE product_id = 1", conn)
print("更新后的产品价格:")
print(result)

事务执行成功
更新后的产品价格:
   product_id name  price
0           1   咖啡   27.0

Python与数据库交互

使用Pandas批量操作

import sqlite3
import pandas as pd

conn = sqlite3.connect(':memory:')

orders_data = pd.DataFrame([
    (1, 1, 3, 2),
    (2, 3, 1, 1),
    (3, 2, 2, 3)
], columns=['order_id', 'user_id', 'product_id', 'quantity'])
orders_data.to_sql('orders', conn, index=False, if_exists='replace')

new_products = pd.DataFrame({
    'product_id': [4, 5],
    'name': ['键盘', '鼠标'],
    'price': [299.0, 159.0]
})
new_products.to_sql('products', conn, if_exists='append', index=False)

user_id = 2
orders = pd.read_sql('SELECT * FROM orders WHERE user_id = ?', conn, params=(user_id,))
print(f"用户 {user_id} 的订单:")
print(orders)

all_products = pd.read_sql('SELECT * FROM products', conn)
print("\n所有产品:")
print(all_products)

用户 2 的订单:
   order_id  user_id  product_id  quantity
0         3        2           2         3

所有产品:
   product_id name  price
0           4   键盘  299.0
1           5   鼠标  159.0

数据库性能优化

import sqlite3
import pandas as pd

conn = sqlite3.connect(':memory:')

orders_data = pd.DataFrame([
    (1, 1, 3, 2),
    (2, 3, 1, 1),
    (3, 2, 2, 3),
    (4, 1, 2, 1)
], columns=['order_id', 'user_id', 'product_id', 'quantity'])
orders_data.to_sql('orders', conn, index=False, if_exists='replace')

conn.execute("CREATE INDEX idx_user_id ON orders(user_id)")

plan = conn.execute("EXPLAIN QUERY PLAN SELECT * FROM orders WHERE user_id = 1").fetchall()
print("查询计划:")
for line in plan:
    print(line)

查询计划:
(3, 0, 62, 'SEARCH orders USING INDEX idx_user_id (user_id=?)')

📝 动手练一练

第 1 题 · 概念应用

在构建量化策略时，你正在分析一个包含多只股票日行情的数据表 stock_daily，其字段包括 ticker（股票代码）、date（日期）、close（收盘价）和 volume（成交量）。为了计算策略的每日总持仓市值，你需要查询在 2023-12-01 这一天，收盘价高于 100 元 且 成交量超过 100 万股 的所有股票。

以下哪个 SQL 查询语句能够正确实现这个需求？

• A. SELECT ticker, close FROM stock_daily WHERE date = '2023-12-01' AND close > 100 OR volume > 1000000
• B. SELECT ticker, close FROM stock_daily WHERE date = '2023-12-01' HAVING close > 100 AND volume > 1000000
• C. SELECT ticker, close FROM stock_daily WHERE date = '2023-12-01' AND (close > 100 AND volume > 1000000)
• D. SELECT ticker, close FROM stock_daily WHERE date = '2023-12-01' OR close > 100 AND volume > 1000000

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

你正在编写一个数据预处理脚本，需要从 SQLite 数据库中读取“小象咖啡”某门店的日营收数据表 daily_sales，并计算其过去7天的移动平均营收。已知数据库文件为 cafe_data.db，表结构包含 date（日期，TEXT）和 revenue（营收，REAL）。

请补全以下 Python 代码中的关键步骤，实现连接数据库、查询数据并计算移动平均。

import sqlite3
import pandas as pd

# 1. 连接到指定的 SQLite 数据库文件
conn = sqlite3.______('cafe_data.db')

# 2. 使用 pandas 读取 `daily_sales` 表中的所有数据
query = "SELECT date, revenue FROM daily_sales ORDER BY date"
df = pd.read_sql(query, conn)

# 3. 确保日期列为 datetime 类型并设置为索引
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)

# 4. 计算营收的7天简单移动平均（SMA）
df['revenue_7d_sma'] = df['revenue'].rolling(window=7, min_periods=1).mean()

# 5. 打印最后10天的数据，包括原始营收和移动平均
print(df[['revenue', 'revenue_7d_sma']].tail(10))

# 6. 关闭数据库连接
conn.close()

import sqlite3
import pandas as pd

# 准备示例数据：实际场景中 cafe_data.db 与 daily_sales 表已由业务系统建好
setup = sqlite3.connect('cafe_data.db')
pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=30).strftime('%Y-%m-%d'),
    'revenue': [1000 + i*50 + (i%7)*100 for i in range(30)]  # 构造一些有波动的营收数据
}).to_sql('daily_sales', setup, index=False, if_exists='replace')
setup.close()

# 1. 连接到指定的 SQLite 数据库文件
conn = sqlite3.connect('cafe_data.db')

# 2. 使用 pandas 读取 `daily_sales` 表中的所有数据
query = "SELECT date, revenue FROM daily_sales ORDER BY date"
df = pd.read_sql(query, conn)

# 3. 确保日期列为 datetime 类型并设置为索引
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)

# 4. 计算营收的7天简单移动平均（SMA）
df['revenue_7d_sma'] = df['revenue'].rolling(window=7, min_periods=1).mean()

# 5. 打印最后10天的数据，包括原始营收和移动平均
print(df[['revenue', 'revenue_7d_sma']].tail(10))

# 6. 关闭数据库连接
conn.close()

revenue  revenue_7d_sma
date
2024-01-21     2600          2150.0
2024-01-22     2050          2200.0
2024-01-23     2200          2250.0
2024-01-24     2350          2300.0
2024-01-25     2500          2350.0
2024-01-26     2650          2400.0
2024-01-27     2800          2450.0
2024-01-28     2950          2500.0
2024-01-29     2400          2550.0
2024-01-30     2550          2600.0

本章小结

📋 行动清单

• 使用SQLite创建包含3个关联表的数据库，并实现跨表联合查询
• 编写Python脚本实现数据库事务操作，包含成功提交和异常回滚两种情况
• 对10万条测试数据创建适当索引，比较索引前后查询性能差异

—— 小象教研组