📚 课程目录第4章数据采集与API设计

4. 数据采集与API设计：RESTful API基础、使用Flask构建模拟数据接口、实时数据流的概念(Kafka/WebSocket简介)

在保险赔付率实时预测系统中，数据采集是连接业务系统与预测模型的桥梁。本章将系统讲解如何通过RESTful API设计标准化的数据接口，使用Flask快速构建模拟数据源，并理解实时数据流的核心概念。掌握这些技能后，你将能够为后续的模型训练与实时预测提供稳定、高效的数据管道。

4.1 RESTful API 基础

REST（Representational State Transfer）是一种基于HTTP协议的软件架构风格，广泛应用于微服务与数据接口设计。在保险场景中，RESTful API用于暴露赔付记录、保单信息、理赔状态等资源。

4.1.1 核心原则

原则	说明	保险示例
资源导向	每个URL代表一个资源（名词）	`/api/claims` 表示理赔资源集合
统一接口	使用标准HTTP方法操作资源	GET查询、POST创建、PUT更新、DELETE删除
无状态	每次请求包含所有必要信息，服务端不保存客户端状态	请求头携带认证Token，不依赖Session
表现层	资源可以有多种表示形式（JSON/XML）	默认返回JSON格式的赔付数据

4.1.2 保险赔付率API设计规范

以下是一个典型的赔付率预测系统API端点设计：

GET /api/v1/claims — 获取理赔记录列表（支持分页、日期过滤）
POST /api/v1/claims — 提交新的理赔案件
GET /api/v1/claims/{id} — 获取单笔理赔详情
GET /api/v1/policies/{policy_id}/loss-ratio — 查询指定保单的累计赔付率
POST /api/v1/predict — 提交特征数据，返回实时赔付率预测结果

  设计要点： 使用版本前缀（如 /api/v1/）保证向后兼容；响应体统一包含 status、data、message 字段；错误码使用标准HTTP状态码（200成功、400参数错误、500服务端异常）。

4.2 使用Flask构建模拟数据接口

Flask是一个轻量级Python Web框架，非常适合快速搭建数据模拟服务。我们将构建一个返回模拟赔付数据的RESTful接口，用于后续模型开发与测试。

4.2.1 环境准备与基础结构

# 安装依赖
pip install flask flask-cors

# 项目结构
insurance_api/
├── app.py              # 主应用入口
├── mock_data.py        # 模拟数据生成器
├── requirements.txt    # 依赖清单
└── config.py           # 配置文件

4.2.2 实现模拟赔付数据接口

# app.py
from flask import Flask, jsonify, request
from flask_cors import CORS
from mock_data import generate_claims, generate_single_claim

app = Flask(__name__)
CORS(app)  # 允许跨域访问

# 模拟数据存储（生产环境应使用数据库）
claims_db = generate_claims(100)  # 预生成100条模拟理赔记录

@app.route('/api/v1/claims', methods=['GET'])
def get_claims():
    """获取理赔列表，支持分页与日期过滤"""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    start_date = request.args.get('start_date')
    end_date = request.args.get('end_date')

    filtered = claims_db
    if start_date:
        filtered = [c for c in filtered if c['claim_date'] >= start_date]
    if end_date:
        filtered = [c for c in filtered if c['claim_date'] <= end_date]

    total = len(filtered)
    start = (page - 1) * per_page
    end = start + per_page
    page_data = filtered[start:end]

    return jsonify({
        'status': 'success',
        'data': page_data,
        'total': total,
        'page': page,
        'per_page': per_page
    })

@app.route('/api/v1/claims', methods=['POST'])
def create_claim():
    """提交新理赔案件"""
    new_claim = request.get_json()
    if not new_claim:
        return jsonify({'status': 'error', 'message': '请求体不能为空'}), 400

    # 模拟生成ID与时间戳
    new_claim['claim_id'] = len(claims_db) + 1
    new_claim['created_at'] = '2025-01-01T00:00:00Z'
    claims_db.append(new_claim)

    return jsonify({'status': 'success', 'data': new_claim}), 201

@app.route('/api/v1/predict', methods=['POST'])
def predict_loss_ratio():
    """模拟赔付率预测接口"""
    features = request.get_json()
    if not features:
        return jsonify({'status': 'error', 'message': '缺少特征数据'}), 400

    # 模拟预测逻辑（后续章节将替换为真实模型）
    import random
    predicted_ratio = round(random.uniform(0.3, 0.8), 4)

    return jsonify({
        'status': 'success',
        'data': {
            'predicted_loss_ratio': predicted_ratio,
            'confidence': 0.85,
            'model_version': 'v1.0.0'
        }
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

4.2.3 模拟数据生成器

# mock_data.py
import random
from datetime import datetime, timedelta

def generate_claims(count=100):
    """生成模拟理赔记录"""
    claim_types = ['车险', '健康险', '寿险', '财产险', '意外险']
    statuses = ['已结案', '处理中', '待审核', '拒赔']
    claims = []
    for i in range(1, count + 1):
        claim_date = datetime.now() - timedelta(days=random.randint(0, 365))
        claims.append({
            'claim_id': i,
            'policy_id': f'POL-{random.randint(1000, 9999)}',
            'claim_type': random.choice(claim_types),
            'claim_amount': round(random.uniform(1000, 100000), 2),
            'settlement_amount': round(random.uniform(500, 80000), 2),
            'claim_date': claim_date.strftime('%Y-%m-%d'),
            'status': random.choice(statuses),
            'customer_age': random.randint(20, 70),
            'is_expedited': random.choice([True, False])
        })
    return claims

def generate_single_claim():
    """生成单条模拟理赔"""
    return generate_claims(1)[0]

启动服务： 运行 python app.py 后，访问 http://localhost:5000/api/v1/claims 即可看到JSON格式的模拟赔付数据。使用 curl 或Postman测试POST接口。

4.3 实时数据流的概念

传统API基于请求-响应模式，客户端需要主动轮询才能获取最新数据。在赔付率实时预测场景中，理赔数据持续产生，需要更高效的推送机制。实时数据流技术应运而生。

4.3.1 实时数据流 vs 传统API

特性	传统REST API	实时数据流
数据获取方式	客户端主动请求（Pull）	服务端主动推送（Push）
延迟	取决于轮询间隔（秒~分钟级）	毫秒~秒级
资源消耗	频繁轮询浪费带宽与计算	连接复用，按需推送
适用场景	查询历史数据、批量操作	实时监控、告警、动态预测

4.3.2 Apache Kafka 简介

Kafka是一个分布式消息队列系统，专为高吞吐量实时数据流设计。在保险赔付率系统中，Kafka作为数据管道核心组件，负责收集各业务系统的理赔事件。

Producer（生产者） — 理赔系统、核保系统等发送赔付事件
Topic（主题） — 按业务分类的数据通道，如 claim-events、policy-changes
Consumer（消费者） — 预测服务、数据仓库等订阅并处理数据
Partition（分区） — 实现数据并行处理与水平扩展

注意： Kafka适合跨服务、跨团队的大规模数据集成，对于单机原型或小规模应用，直接使用WebSocket或Redis Pub/Sub更为轻量。

4.3.3 WebSocket 简介

WebSocket在客户端与服务器之间建立持久连接，实现全双工通信。在保险赔付率预测中，WebSocket常用于前端实时展示预测结果。

# 使用Flask-SocketIO实现WebSocket推送
from flask_socketio import SocketIO, emit
import random
import time

socketio = SocketIO(app)

@socketio.on('connect')
def handle_connect():
    print('客户端已连接')
    emit('message', {'data': '连接成功，开始推送赔付率数据'})

def background_push():
    """后台线程持续推送模拟赔付率"""
    while True:
        ratio = round(random.uniform(0.3, 0.8), 4)
        socketio.emit('loss_ratio_update', {
            'timestamp': time.time(),
            'predicted_ratio': ratio,
            'policy_count': random.randint(100, 500)
        })
        time.sleep(2)  # 每2秒推送一次

# 启动后台推送线程
import threading
thread = threading.Thread(target=background_push)
thread.daemon = True
thread.start()

4.4 数据采集架构总览

下图展示了保险赔付率实时预测系统中数据采集与API设计的整体架构，涵盖RESTful接口、模拟数据服务与实时数据流通道。

4.5 本章小结

RESTful API 是保险数据接口的标准范式，遵循资源导向、统一接口、无状态原则。
Flask 可快速构建模拟数据接口，支持分页、过滤、POST提交等完整CRUD操作，适合原型开发与测试。
Kafka 适用于高吞吐量分布式数据流场景，Producer/Topic/Consumer模型清晰；WebSocket 适合低延迟双向通信，常用于前端实时展示。
实际项目中，通常组合使用RESTful API（查询历史数据）与实时数据流（推送新事件），形成完整的数据采集管道。