4. 数据库设计与 SQLite：定价数据表结构设计

定价稳定性监控的核心是数据。没有结构化的存储，价格采集、异常检测与回溯分析都无从谈起。本章从数据库选型出发，设计一张专用于定价监控的 price_records 表，并使用 Python 内置的 sqlite3 模块完成完整的 CRUD（创建、读取、更新、删除）操作。

4.1 为什么选择 SQLite

在课程初期，我们选择 SQLite 而非 MySQL/PostgreSQL，原因如下：

4.2 定价数据表结构设计

一张设计良好的表，应当满足：可追溯（谁、什么时间、什么价格）、可过滤（按产品、来源、时间范围）、可扩展（后续加入币种、折扣字段时不破坏已有数据）。

-- price_records 表 DDL
CREATE TABLE IF NOT EXISTS price_records (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    product_id  TEXT    NOT NULL,
    price       REAL    NOT NULL,
    source      TEXT    NOT NULL DEFAULT 'manual',
    recorded_at TEXT    NOT NULL DEFAULT (datetime('now')),
    created_at  TEXT    NOT NULL DEFAULT (datetime('now'))
);

-- 常用查询索引
CREATE INDEX IF NOT EXISTS idx_product_time
    ON price_records(product_id, recorded_at);

CREATE INDEX IF NOT EXISTS idx_source
    ON price_records(source);

4.3 数据库与表结构关系图

下图展示 SQLite 数据库文件、表、索引与 Python 应用层之间的逻辑关系：

4.4 使用 sqlite3 进行 CRUD 操作

Python 标准库 sqlite3 提供了完整的数据库操作接口。以下代码演示如何连接数据库、创建表、插入、查询、更新和删除记录。

4.4.1 连接与建表

import sqlite3
from datetime import datetime

DB_PATH = "pricing.db"

def get_connection():
    """获取数据库连接，自动提交模式"""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # 支持按列名访问
    conn.execute("PRAGMA journal_mode=WAL;")  # 提升并发读性能
    return conn

def init_database():
    """初始化表与索引"""
    ddl = """
    CREATE TABLE IF NOT EXISTS price_records (
        id          INTEGER PRIMARY KEY AUTOINCREMENT,
        product_id  TEXT    NOT NULL,
        price       REAL    NOT NULL,
        source      TEXT    NOT NULL DEFAULT 'manual',
        recorded_at TEXT    NOT NULL DEFAULT (datetime('now')),
        created_at  TEXT    NOT NULL DEFAULT (datetime('now'))
    );
    CREATE INDEX IF NOT EXISTS idx_product_time
        ON price_records(product_id, recorded_at);
    CREATE INDEX IF NOT EXISTS idx_source
        ON price_records(source);
    """
    conn = get_connection()
    conn.executescript(ddl)
    conn.commit()
    conn.close()
    print("[OK] 数据库初始化完成")

4.4.2 插入记录（Create）

def insert_price(product_id, price, source="manual", recorded_at=None):
    """插入一条价格记录"""
    if recorded_at is None:
        recorded_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    sql = """
        INSERT INTO price_records (product_id, price, source, recorded_at)
        VALUES (?, ?, ?, ?)
    """
    conn = get_connection()
    conn.execute(sql, (product_id, price, source, recorded_at))
    conn.commit()
    conn.close()
    print(f"[INSERT] {product_id} → {price} @ {recorded_at}")

# 使用示例
insert_price("SKU-001", 29.99, source="api")
insert_price("SKU-002", 15.50, source="web", recorded_at="2025-03-01 10:30:00")

4.4.3 查询记录（Read）

def query_prices(product_id, start_date=None, end_date=None, limit=10):
    """按产品ID和时间范围查询价格记录"""
    sql = """
        SELECT id, product_id, price, source, recorded_at, created_at
        FROM price_records
        WHERE product_id = ?
    """
    params = [product_id]
    if start_date:
        sql += " AND recorded_at >= ?"
        params.append(start_date)
    if end_date:
        sql += " AND recorded_at <= ?"
        params.append(end_date)
    sql += " ORDER BY recorded_at DESC LIMIT ?"
    params.append(limit)

    conn = get_connection()
    cursor = conn.execute(sql, params)
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]

# 使用示例
results = query_prices("SKU-001", start_date="2025-03-01", limit=5)
for r in results:
    print(f"{r['recorded_at']} | {r['price']} | {r['source']}")

4.4.4 更新与删除（Update / Delete）

def update_price(record_id, new_price):
    """更新指定记录的价格"""
    sql = "UPDATE price_records SET price = ? WHERE id = ?"
    conn = get_connection()
    conn.execute(sql, (new_price, record_id))
    conn.commit()
    affected = conn.total_changes
    conn.close()
    print(f"[UPDATE] 记录 {record_id} 价格更新为 {new_price}，影响 {affected} 行")

def delete_price(record_id):
    """删除指定记录"""
    sql = "DELETE FROM price_records WHERE id = ?"
    conn = get_connection()
    conn.execute(sql, (record_id,))
    conn.commit()
    affected = conn.total_changes
    conn.close()
    print(f"[DELETE] 删除记录 {record_id}，影响 {affected} 行")

# 使用示例
update_price(1, 31.99)
delete_price(5)

4.5 CRUD 操作流程总览

下图将四个基本操作串联为完整的数据生命周期：

4.6 批量插入与性能提示

当需要一次性写入大量价格记录（如历史数据回填）时，逐条 INSERT 效率极低。应使用 executemany() 配合事务批量提交：

def batch_insert(records):
    """
    批量插入价格记录
    records: list of tuple (product_id, price, source, recorded_at)
    """
    sql = """
        INSERT INTO price_records (product_id, price, source, recorded_at)
        VALUES (?, ?, ?, ?)
    """
    conn = get_connection()
    conn.execute("BEGIN TRANSACTION;")
    conn.executemany(sql, records)
    conn.commit()
    conn.close()
    print(f"[BATCH] 成功插入 {len(records)} 条记录")

# 使用示例
batch_data = [
    ("SKU-001", 30.00, "api", "2025-03-01 08:00:00"),
    ("SKU-001", 30.50, "api", "2025-03-01 09:00:00"),
    ("SKU-002", 16.00, "web", "2025-03-01 08:30:00"),
]
batch_insert(batch_data)

4.7 本章小结

• 表结构：price_records 包含 id、product_id、price、source、recorded_at、created_at 六个字段，覆盖定价监控核心需求。
• 索引策略：idx_product_time 加速按产品+时间范围查询；idx_source 加速按来源过滤。
• CRUD 实现：使用 sqlite3 标准库完成增删改查，所有 SQL 均采用参数化查询防注入。
• 批量写入：executemany() + 显式事务是高性能写入的关键。