4. 数据库设计与ORM：风控规则表、交易流水表、账户表设计、SQLAlchemy异步ORM配置、Alembic迁移管理

实时交易风控系统的核心是数据。所有风控规则的判定、交易流水的记录、账户状态的维护，最终都依赖数据库的稳定与高效。本章将围绕三张核心表（风控规则表、交易流水表、账户表）展开，讲解如何基于 PostgreSQL 进行表结构设计，并配置 SQLAlchemy 异步 ORM 与 Alembic 迁移管理，为后续的 API 开发打下坚实的数据基础。

设计原则： 实时风控场景下，数据库设计需兼顾写入吞吐（交易流水）与查询性能（规则匹配）。优先使用索引覆盖、分区表、避免长事务。

4.1 核心表结构设计

我们采用 三表分离 的设计模式：risk_rules 存储风控规则定义，transactions 存储每笔交易流水，accounts 存储账户摘要与风控状态。三表通过 account_id 关联，形成完整的数据闭环。

表名 核心字段 主键/索引 用途
risk_rules rule_id, rule_name, rule_type, config_json, is_active, priority, created_at PRIMARY KEY (rule_id), INDEX (rule_type, is_active) 定义风控规则（如单笔限额、频率限制、黑名单）
transactions txn_id, account_id, amount, currency, txn_type, status, risk_level, created_at PRIMARY KEY (txn_id), INDEX (account_id, created_at), INDEX (status) 记录每笔交易流水，用于事后审计与实时风控判定
accounts account_id, balance, frozen_amount, risk_score, is_blocked, last_txn_at, version PRIMARY KEY (account_id), UNIQUE (account_id) 维护账户余额、冻结金额、风控评分与乐观锁版本号

4.1.1 风控规则表（risk_rules）

规则表采用 配置化 设计，将规则参数以 JSON 格式存储在 config_json 字段中，避免频繁修改表结构。例如，单笔限额规则可配置为：{"max_amount": 100000, "currency": "CNY"}。

-- 风控规则表 DDL
CREATE TABLE risk_rules (
    rule_id        BIGSERIAL PRIMARY KEY,
    rule_name      VARCHAR(128) NOT NULL,
    rule_type      VARCHAR(32)  NOT NULL,  -- 'amount_limit', 'frequency', 'blacklist', 'velocity'
    config_json    JSONB        NOT NULL DEFAULT '{}',
    is_active      BOOLEAN      NOT NULL DEFAULT TRUE,
    priority       INT          NOT NULL DEFAULT 0,
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    updated_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_rules_type_active ON risk_rules(rule_type, is_active);

4.1.2 交易流水表（transactions）

交易流水表是系统中最繁忙的表，每秒可能写入数千条记录。我们使用 复合索引 加速按账户和时间范围的查询，并考虑按时间分区以提升维护效率。

-- 交易流水表 DDL（含分区示例）
CREATE TABLE transactions (
    txn_id         BIGSERIAL,
    account_id     BIGINT       NOT NULL,
    amount         NUMERIC(20,4) NOT NULL,
    currency       CHAR(3)      NOT NULL DEFAULT 'CNY',
    txn_type       VARCHAR(16)  NOT NULL,  -- 'deposit', 'withdraw', 'transfer'
    status         VARCHAR(16)  NOT NULL DEFAULT 'pending', -- 'pending','approved','rejected'
    risk_level     VARCHAR(8)   DEFAULT 'low',  -- 'low','medium','high'
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (created_at);

-- 创建分区（示例：按月分区）
CREATE TABLE transactions_202501 PARTITION OF transactions
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

CREATE INDEX idx_tx_account_time ON transactions(account_id, created_at DESC);
CREATE INDEX idx_tx_status ON transactions(status);

4.1.3 账户表（accounts）

账户表使用 乐观锁（version 字段）防止并发更新导致的数据不一致。每次更新余额时，需校验 version 是否与读取时一致。

-- 账户表 DDL
CREATE TABLE accounts (
    account_id     BIGSERIAL PRIMARY KEY,
    balance        NUMERIC(20,4) NOT NULL DEFAULT 0.0000,
    frozen_amount  NUMERIC(20,4) NOT NULL DEFAULT 0.0000,
    risk_score     INT           NOT NULL DEFAULT 0,   -- 0-100
    is_blocked     BOOLEAN       NOT NULL DEFAULT FALSE,
    last_txn_at    TIMESTAMPTZ,
    version        INT           NOT NULL DEFAULT 1,   -- 乐观锁版本号
    created_at     TIMESTAMPTZ   NOT NULL DEFAULT NOW(),
    updated_at     TIMESTAMPTZ   NOT NULL DEFAULT NOW()
);
设计要点： 账户表不存储交易明细，只维护汇总状态。交易流水由 transactions 表独立存储，避免单表过大影响写入性能。

4.2 三表关系与数据流

下图展示了三张核心表之间的关联关系以及实时风控判定的数据流向：

风控系统核心表关系与数据流 风控系统核心表关系与数据流 accounts 账户余额 / 风控状态 transactions 交易流水 / 审计日志 risk_rules 规则配置 / 优先级 account_id 规则匹配 实时风控引擎 读取规则 → 判定交易 → 更新账户状态 触发风控 更新余额/状态 写入判定结果 数据流说明： 1. 交易到达 → 2. 读取 risk_rules 匹配规则 → 3. 写入 transactions 流水 → 4. 更新 accounts 余额/状态

4.3 SQLAlchemy 异步 ORM 配置

在实时交易系统中，同步数据库操作会阻塞事件循环，导致吞吐量下降。我们使用 SQLAlchemy 的 异步扩展（asyncpg 驱动 + asyncio 模式）来避免阻塞。

4.3.1 安装依赖

pip install sqlalchemy[asyncio] asyncpg alembic

4.3.2 创建异步引擎与会话工厂

# config/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/risk_control"

engine = create_async_engine(DATABASE_URL, echo=True, pool_size=10, max_overflow=20)
async_session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

4.3.3 定义 ORM 模型

以 Account 模型为例，展示异步 ORM 的字段定义与乐观锁实现：

# models/account.py
from sqlalchemy import Column, BigInteger, Numeric, Integer, Boolean, DateTime, func
from config.database import Base

class Account(Base):
    __tablename__ = "accounts"

    account_id    = Column(BigInteger, primary_key=True, autoincrement=True)
    balance       = Column(Numeric(20, 4), nullable=False, default=0)
    frozen_amount = Column(Numeric(20, 4), nullable=False, default=0)
    risk_score    = Column(Integer, nullable=False, default=0)
    is_blocked    = Column(Boolean, nullable=False, default=False)
    last_txn_at   = Column(DateTime(timezone=True), nullable=True)
    version       = Column(Integer, nullable=False, default=1)
    created_at    = Column(DateTime(timezone=True), server_default=func.now())
    updated_at    = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

4.3.4 异步 CRUD 示例

# repositories/account_repo.py
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from models.account import Account

async def get_account_by_id(db: AsyncSession, account_id: int) -> Account | None:
    result = await db.execute(select(Account).where(Account.account_id == account_id))
    return result.scalar_one_or_none()

async def update_balance_with_optimistic_lock(
    db: AsyncSession, account_id: int, new_balance: float, current_version: int
) -> bool:
    stmt = (
        update(Account)
        .where(Account.account_id == account_id, Account.version == current_version)
        .values(balance=new_balance, version=current_version + 1)
    )
    result = await db.execute(stmt)
    await db.commit()
    return result.rowcount > 0  # 返回 True 表示更新成功，False 表示版本冲突
注意： 异步 ORM 中，所有数据库操作必须使用 await 调用。不要在异步上下文中混用同步 Session，否则会导致事件循环阻塞。

4.4 Alembic 迁移管理

Alembic 是 SQLAlchemy 的官方迁移工具，支持自动生成迁移脚本、版本回滚、多环境同步。在异步项目中，需配置 run_async=True 以支持异步引擎。

4.4.1 初始化 Alembic

alembic init alembic

4.4.2 配置 alembic.ini 与 env.py

修改 alembic.ini 中的数据库连接字符串：

sqlalchemy.url = postgresql+asyncpg://user:password@localhost:5432/risk_control

修改 alembic/env.py 以支持异步：

# alembic/env.py
from config.database import Base, engine
from models import Account, Transaction, RiskRule  # 导入所有模型

target_metadata = Base.metadata

def run_migrations_online():
    # 使用异步引擎运行迁移
    with engine.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

4.4.3 生成并执行迁移

# 生成迁移脚本（自动检测模型变更）
alembic revision --autogenerate -m "add accounts table"

# 查看生成的迁移脚本
alembic history

# 执行迁移（升级到最新版本）
alembic upgrade head

# 回滚一个版本
alembic downgrade -1
Alembic 命令 说明 常用场景
alembic init 初始化迁移目录 项目首次搭建时
alembic revision --autogenerate 自动生成迁移脚本 模型字段变更后
alembic upgrade head 升级到最新版本 部署/同步数据库
alembic downgrade -1 回滚一个版本 迁移出错时快速回退
alembic history 查看迁移历史 确认当前版本位置

4.5 本章小结

本章完成了实时交易风控系统的数据库层设计：

  • 三表设计：risk_rules（规则配置）、transactions（交易流水）、accounts（账户状态），满足读写分离与高并发需求。
  • 异步 ORM：基于 SQLAlchemy + asyncpg 实现非阻塞数据库操作，避免事件循环阻塞。
  • 迁移管理：使用 Alembic 管理数据库版本，支持自动生成迁移脚本与回滚。
← 上一章 📖 返回目录 下一章 →