SQL 数据库

表、查询以及为什么需要数据库

成交、报价、持仓和参考数据都是表：具有相同列的行。关系型数据库把它们存在磁盘上，允许许多程序同时读写，在操作中途失败时保持一致，并能回答用 SQL 写的问题，而无需你把所有数据载入内存。sqlite3 随 Python 附带，把整个数据库放在一个文件里，足够做研究；生产环境中 PostgreSQL 等服务器做同样的事。基本语句有 CREATE TABLE、INSERT、SELECT ... WHERE ... ORDER BY ... LIMIT、UPDATE 和 DELETE。

# A database in memory with sqlite3: create, insert, query
import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE trades (symbol TEXT, price REAL, qty INTEGER)")
db.executemany("INSERT INTO trades (symbol, price, qty) VALUES (?, ?, ?)",
               [("AAPL", 184.25, 100), ("MSFT", 410.5, 50), ("AAPL", 184.5, 25)])

rows = db.execute("SELECT symbol, price FROM trades WHERE price > ? ORDER BY price DESC LIMIT 2", (184.3,)).fetchall()
print(rows)                                          # [('MSFT', 410.5), ('AAPL', 184.5)]

用参数，而不是拼接字符串

绝不要把值拼接进文本来构造查询：f"... WHERE symbol = '{name}'" 会让像 x' OR '1'='1 这样的值改写查询（SQL 注入），而合法值里的引号也会把它弄坏。在值的位置放一个 ? 占位符，把值另行传入；驱动会把它们当作永远不会被解释为 SQL 的数据发送，数据库还可以复用查询计划。占位符只能代表值。表名和列名不能作为参数，所以如果名称来自外部，在放进文本之前要用严格的模式（字母、数字和下划线）或允许名单检查。

# A pasted value rewrites the query; a placeholder keeps it data
import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE users (name TEXT, secret TEXT)")
db.executemany("INSERT INTO users VALUES (?, ?)", [("alice", "a1"), ("bob", "b2")])

evil = "x' OR '1'='1"
pasted = db.execute(f"SELECT secret FROM users WHERE name = '{evil}'").fetchall()
safe = db.execute("SELECT secret FROM users WHERE name = ?", (evil,)).fetchall()
print(pasted)                                        # [('a1',), ('b2',)]  the injected condition matched every row
print(safe)                                          # []                  the same text is just an unknown name

聚合、连接与索引

GROUP BY 会把共享同一个键的行合并，并为每组计算 SUM、AVG、COUNT、MIN 或 MAX：比如每个代码的总成交量。JOIN 按共同的列把两张表合并；INNER JOIN 只保留两边都有匹配的行，LEFT JOIN 保留左表的全部行。在数据库里做这些事远比在 Python 中循环快。对大表，要在用于过滤或连接的列上加索引（CREATE INDEX idx_trades_symbol ON trades(symbol)）：数据库会直接跳到匹配的行，而不是扫描整张表，EXPLAIN QUERY PLAN 会显示它的选择。把相关的写操作包进一个事务，让它们要么全部发生，要么都不发生。

# GROUP BY and JOIN done by the database
import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE trades (symbol TEXT, qty INTEGER)")
db.execute("CREATE TABLE sectors (symbol TEXT, sector TEXT)")
db.executemany("INSERT INTO trades VALUES (?, ?)", [("AAPL", 100), ("MSFT", 50), ("AAPL", 25), ("XOM", 10)])
db.executemany("INSERT INTO sectors VALUES (?, ?)", [("AAPL", "tech"), ("MSFT", "tech")])

print(db.execute("SELECT symbol, SUM(qty) FROM trades GROUP BY symbol ORDER BY symbol").fetchall())
# [('AAPL', 125), ('MSFT', 50), ('XOM', 10)]
print(db.execute("SELECT s.sector, SUM(t.qty) FROM trades t JOIN sectors s ON t.symbol = s.symbol GROUP BY s.sector").fetchall())
# [('tech', 175)]   XOM has no sector, so the inner join drops it

# An index turns a full scan into a lookup; the query plan shows which one the database uses
import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE ticks (symbol TEXT, ts INTEGER, price REAL)")
db.executemany("INSERT INTO ticks VALUES (?, ?, ?)", [("S%d" % (i % 50), i, 100.0 + i % 7) for i in range(5000)])

query = "EXPLAIN QUERY PLAN SELECT price FROM ticks WHERE symbol = 'S7'"
before = " ".join(row[-1] for row in db.execute(query))
db.execute("CREATE INDEX idx_ticks_symbol ON ticks(symbol)")
after = " ".join(row[-1] for row in db.execute(query))
print("SCAN" in before, "INDEX" in after)            # True True   a full scan before the index, an index search after

• 关系型数据库把表存在磁盘上并回答 SQL 查询；sqlite3 把整个数据库放在一个文件里。
• 把值作为 ? 参数传入，绝不拼进文本（SQL 注入）；用严格的模式检查表名和列名。
• 用 GROUP BY、JOIN 和索引让数据库干活；加上约束，保持 UTC 时间戳和精确金额，并做备份。

下一节：Parquet 与 Arrow，面向大型分析数据的列式格式。

实验：评分内容

• 绝不拼接数值：insert_statement25 分
• 用 build_select 组合查询25 分
• 用 group_sum 按键聚合25 分
• 用 inner_join 连接两张表25 分