4. 数据库模型设计与ORM：预警规则表、客户信息表、预警事件表设计、SQLAlchemy模型定义与迁移

贷后预警系统的核心是围绕“规则-客户-事件”三个实体构建的数据模型。本章将基于关系型数据库（以PostgreSQL为例）设计三张核心表，并使用SQLAlchemy ORM进行模型定义与数据库迁移。良好的模型设计是API性能与数据一致性的基石。

4.1 核心表结构设计

三张表通过外键关联，形成星型模型：预警规则表定义触发条件，客户信息表存储被监控对象，预警事件表记录每次触发的具体实例。

表名	业务含义	主键策略	核心字段数
alert_rule	预警规则定义（如：逾期超过30天触发）	自增整数ID	8
customer_info	客户基本信息与风险画像	业务主键（客户编号）	12
alert_event	每次预警触发的具体事件记录	UUID	10

4.1.1 预警规则表（alert_rule）

该表存储预警规则的静态定义，包括规则名称、触发条件（JSON格式）、阈值、生效状态等。设计时需考虑规则的版本管理与启停控制。

• rule_id: 主键，自增整数。
• rule_name: 规则名称，唯一索引。
• rule_type: 规则类型枚举（逾期预警、额度预警、行为异常等）。
• trigger_condition: JSONB字段，存储触发条件表达式（如 {"days_overdue": ">30", "amount": ">100000"}）。
• severity: 严重级别（高/中/低）。
• is_active: 布尔值，是否启用。
• created_at: 创建时间。
• updated_at: 更新时间。

4.1.2 客户信息表（customer_info）

该表是客户维度的核心表，包含身份信息、联系方式、风险评级等。由于贷后场景需要频繁查询客户状态，建议使用客户编号作为业务主键，并建立复合索引。

• customer_id: 业务主键，字符串类型（如：CUST20240001）。
• name: 客户姓名。
• id_number: 身份证号（加密存储）。
• phone: 手机号。
• risk_level: 风险等级（A/B/C/D）。
• total_loan_amount: 总贷款金额（decimal）。
• overdue_days: 当前逾期天数（整型，默认0）。
• credit_score: 信用评分（整型）。
• status: 客户状态（正常/关注/黑名单）。
• created_at: 创建时间。
• updated_at: 更新时间。
• version: 乐观锁版本号（用于并发控制）。

4.1.3 预警事件表（alert_event）

该表记录每次规则触发的具体事件，是高频写入表。设计时需考虑分区策略（按时间分区）和索引优化，避免写入瓶颈。

• event_id: 主键，UUID类型（避免自增ID在分布式场景下的冲突）。
• rule_id: 外键，关联 alert_rule 表。
• customer_id: 外键，关联 customer_info 表。
• event_time: 事件发生时间（精确到毫秒）。
• event_detail: JSONB字段，存储触发时的上下文数据（如当前逾期天数、贷款余额等）。
• alert_status: 处理状态（待处理/已处理/已忽略）。
• handler: 处理人（可为空）。
• handle_time: 处理时间（可为空）。
• created_at: 记录创建时间。
• updated_at: 更新时间。

4.2 实体关系图（ER图）

下图展示了三张表之间的外键关联关系，以及核心字段的映射。该结构支持从规则到客户再到事件的多维查询。

4.3 SQLAlchemy模型定义

使用SQLAlchemy的声明式基类（declarative_base）定义ORM模型。以下代码展示了三张表的模型定义，包括字段类型、约束、关系映射以及索引声明。

from sqlalchemy import (
    Column, Integer, String, DateTime, Boolean, Enum, 
    Numeric, JSON, ForeignKey, Index, UniqueConstraint
)
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship, declarative_base
from datetime import datetime
import uuid

Base = declarative_base()

class AlertRule(Base):
    __tablename__ = 'alert_rule'
    
    rule_id = Column(Integer, primary_key=True, autoincrement=True)
    rule_name = Column(String(128), unique=True, nullable=False, index=True)
    rule_type = Column(Enum('overdue', 'quota', 'behavior', name='rule_type_enum'), nullable=False)
    trigger_condition = Column(JSON, nullable=False)
    severity = Column(Enum('high', 'medium', 'low', name='severity_enum'), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # 关系映射：一个规则对应多个事件
    events = relationship('AlertEvent', back_populates='rule', lazy='dynamic')

class CustomerInfo(Base):
    __tablename__ = 'customer_info'
    
    customer_id = Column(String(32), primary_key=True)
    name = Column(String(64), nullable=False, index=True)
    id_number = Column(String(18), nullable=False)  # 生产环境需加密
    phone = Column(String(20), nullable=False)
    risk_level = Column(Enum('A', 'B', 'C', 'D', name='risk_level_enum'), default='C')
    total_loan_amount = Column(Numeric(15, 2), default=0.00)
    overdue_days = Column(Integer, default=0)
    credit_score = Column(Integer, default=600)
    status = Column(Enum('normal', 'watch', 'blacklist', name='cust_status_enum'), default='normal')
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    version = Column(Integer, default=1)
    
    # 关系映射：一个客户对应多个事件
    events = relationship('AlertEvent', back_populates='customer', lazy='dynamic')
    
    # 复合索引：风险等级+逾期天数，用于快速筛选高风险客户
    __table_args__ = (
        Index('idx_risk_overdue', 'risk_level', 'overdue_days'),
    )

class AlertEvent(Base):
    __tablename__ = 'alert_event'
    
    event_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    rule_id = Column(Integer, ForeignKey('alert_rule.rule_id'), nullable=False)
    customer_id = Column(String(32), ForeignKey('customer_info.customer_id'), nullable=False)
    event_time = Column(DateTime, nullable=False, index=True)
    event_detail = Column(JSON, nullable=True)
    alert_status = Column(Enum('pending', 'handled', 'ignored', name='alert_status_enum'), default='pending')
    handler = Column(String(64), nullable=True)
    handle_time = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # 关系映射
    rule = relationship('AlertRule', back_populates='events')
    customer = relationship('CustomerInfo', back_populates='events')
    
    # 复合索引：按规则+状态查询
    __table_args__ = (
        Index('idx_rule_status', 'rule_id', 'alert_status'),
    )

4.4 数据库迁移（Alembic）

使用Alembic进行数据库迁移管理，确保模型变更可追溯、可回滚。以下是初始化迁移环境的步骤和示例迁移脚本。

4.4.1 初始化Alembic

# 安装依赖
pip install alembic

# 初始化迁移目录
alembic init alembic

# 配置 alembic.ini 中的数据库连接
# sqlalchemy.url = postgresql://user:password@localhost:5432/alert_db

# 在 alembic/env.py 中导入 Base 元数据
from app.models import Base
target_metadata = Base.metadata

4.4.2 生成并执行迁移

# 自动生成迁移脚本（对比模型与数据库）
alembic revision --autogenerate -m "init_alert_tables"

# 查看生成的迁移文件（位于 alembic/versions/ 目录）
# 示例迁移脚本内容如下：

"""init_alert_tables

Revision ID: 1234abcd
Revises: 
Create Date: 2024-01-15 10:00:00
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID, JSON

revision = '1234abcd'
down_revision = None

def upgrade():
    op.create_table('alert_rule',
        sa.Column('rule_id', sa.Integer(), autoincrement=True, nullable=False),
        sa.Column('rule_name', sa.String(length=128), nullable=False),
        sa.Column('rule_type', sa.Enum('overdue', 'quota', 'behavior', name='rule_type_enum'), nullable=False),
        sa.Column('trigger_condition', JSON(), nullable=False),
        sa.Column('severity', sa.Enum('high', 'medium', 'low', name='severity_enum'), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.Column('updated_at', sa.DateTime(), nullable=False),
        sa.PrimaryKeyConstraint('rule_id'),
        sa.UniqueConstraint('rule_name')
    )
    op.create_index('idx_rule_name', 'alert_rule', ['rule_name'])
    
    op.create_table('customer_info',
        sa.Column('customer_id', sa.String(length=32), nullable=False),
        sa.Column('name', sa.String(length=64), nullable=False),
        sa.Column('id_number', sa.String(length=18), nullable=False),
        sa.Column('phone', sa.String(length=20), nullable=False),
        sa.Column('risk_level', sa.Enum('A', 'B', 'C', 'D', name='risk_level_enum'), nullable=False),
        sa.Column('total_loan_amount', sa.Numeric(15, 2), nullable=False),
        sa.Column('overdue_days', sa.Integer(), nullable=False),
        sa.Column('credit_score', sa.Integer(), nullable=False),
        sa.Column('status', sa.Enum('normal', 'watch', 'blacklist', name='cust_status_enum'), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.Column('updated_at', sa.DateTime(), nullable=False),
        sa.Column('version', sa.Integer(), nullable=False),
        sa.PrimaryKeyConstraint('customer_id')
    )
    op.create_index('idx_risk_overdue', 'customer_info', ['risk_level', 'overdue_days'])
    
    op.create_table('alert_event',
        sa.Column('event_id', UUID(), nullable=False),
        sa.Column('rule_id', sa.Integer(), nullable=False),
        sa.Column('customer_id', sa.String(length=32), nullable=False),
        sa.Column('event_time', sa.DateTime(), nullable=False),
        sa.Column('event_detail', JSON(), nullable=True),
        sa.Column('alert_status', sa.Enum('pending', 'handled', 'ignored', name='alert_status_enum'), nullable=False),
        sa.Column('handler', sa.String(length=64), nullable=True),
        sa.Column('handle_time', sa.DateTime(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.Column('updated_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(['rule_id'], ['alert_rule.rule_id'], ),
        sa.ForeignKeyConstraint(['customer_id'], ['customer_info.customer_id'], ),
        sa.PrimaryKeyConstraint('event_id')
    )
    op.create_index('idx_event_time', 'alert_event', ['event_time'])
    op.create_index('idx_rule_status', 'alert_event', ['rule_id', 'alert_status'])

def downgrade():
    op.drop_table('alert_event')
    op.drop_table('customer_info')
    op.drop_table('alert_rule')
    # 删除自定义枚举类型
    op.execute('DROP TYPE IF EXISTS rule_type_enum')
    op.execute('DROP TYPE IF EXISTS severity_enum')
    op.execute('DROP TYPE IF EXISTS risk_level_enum')
    op.execute('DROP TYPE IF EXISTS cust_status_enum')
    op.execute('DROP TYPE IF EXISTS alert_status_enum')

4.4.3 执行迁移

# 升级到最新版本
alembic upgrade head

# 回滚一个版本
alembic downgrade -1

# 查看当前版本
alembic current

4.5 模型与API的映射关系

下图展示了数据库模型如何映射到RESTful API的请求/响应结构，以及ORM在其中的桥梁作用。

4.6 性能优化建议

基于上述模型设计，在API开发阶段需关注以下性能要点：

• 批量写入：预警事件表的高频写入场景，使用 bulk_insert_mappings 替代逐条 add，减少数据库连接往返。
• 懒加载与预加载：在查询事件列表时，使用 joinedload 或 subqueryload 预加载关联的规则和客户信息，避免N+1查询问题。
• 分页查询：所有列表接口必须支持游标分页或偏移分页，防止一次加载大量数据导致内存溢出。
• 连接池配置：在SQLAlchemy引擎中配置合理的连接池大小（如 pool_size=20, max_overflow=10），避免高并发下的连接耗尽。

# 批量写入示例
from sqlalchemy.dialects.postgresql import insert

def batch_insert_events(session, events_data: list):
    """批量插入预警事件，使用ON CONFLICT处理重复"""
    stmt = insert(AlertEvent).values(events_data)
    stmt = stmt.on_conflict_do_nothing(index_elements=['event_id'])
    session.execute(stmt)
    session.commit()

# 预加载关联对象示例
from sqlalchemy.orm import joinedload

def get_events_with_relations(session, rule_id: int, page: int, size: int):
    query = session.query(AlertEvent).options(
        joinedload(AlertEvent.rule),
        joinedload(AlertEvent.customer)
    ).filter(AlertEvent.rule_id == rule_id)
    return query.offset((page-1)*size).limit(size).all()