毕业设计项目：基于容器化的金融交易流水跨区域协同异常检测系统设计与实现-开发日记-01-环境搭建

fastapi
uvicorn[standard]
httpx

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

# Gateway 服务入口：负责接收前端请求，并转发给 Data Service。

import os

import httpx
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse


# 从环境变量读取 Data Service 地址。
# Docker Compose 网络中通常通过服务名 data-service 访问。
DATA_SERVICE_URL = os.getenv(
    "DATA_SERVICE_URL",
    "http://data-service:8000",
)


# 创建 FastAPI 应用实例。
app = FastAPI(title="fincross-risk Gateway")


# 转发 Data Service 的响应，并保留下游返回的状态码。
def forward_response(response: httpx.Response):
    try:
        # Data Service 应返回 JSON 格式的数据。
        content = response.json()
    except ValueError:
        # 下游返回非 JSON 内容时，说明响应格式异常。
        raise HTTPException(
            status_code=502,
            detail="invalid JSON from data service",
        )

    # 保留 404、409、422、503 等下游状态码，
    # 避免 Gateway 把失败请求误返回为成功。
    return JSONResponse(
        status_code=response.status_code,
        content=content,
    )


# Gateway 健康检查接口。
# 用于确认 Gateway 服务是否能够正常响应。
@app.get("/health")
def health():
    return {
        "service": "gateway",
        "status": "ok",
    }


# 提交交易。
# Gateway 接收前端数据，再转发给 Data Service 处理。
@app.post("/api/transactions")
async def create_transaction(payload: dict):
    try:
        # 使用异步 HTTP 客户端调用 Data Service。
        async with httpx.AsyncClient(timeout=15) as client:
            response = await client.post(
                f"{DATA_SERVICE_URL}/transactions",
                json=payload,
            )
    except httpx.HTTPError as exc:
        # 无法连接 Data Service 时，返回 503 错误。
        raise HTTPException(
            status_code=503,
            detail=f"data service unavailable: {exc}",
        )

    return forward_response(response)


# 查询交易。
# 前端通过 external_id 查询交易状态和风险分析结果。
@app.get("/api/transactions/{external_id}")
async def get_transaction(external_id: str):
    try:
        # 根据交易编号请求 Data Service。
        async with httpx.AsyncClient(timeout=15) as client:
            response = await client.get(
                f"{DATA_SERVICE_URL}/transactions/{external_id}",
            )
    except httpx.HTTPError as exc:
        # Data Service 不可用时，向前端返回 503 错误。
        raise HTTPException(
            status_code=503,
            detail=f"data service unavailable: {exc}",
        )

    return forward_response(response)

docker compose build gateway

docker image ls

(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk> docker image ls
                                                                                                                                                                                    i Info →   U  In Use
IMAGE                          ID             DISK USAGE   CONTENT SIZE   EXTRA
apache/kafka:4.3.1             77e3df905404        686MB          239MB    U   
fincross-risk-gateway:latest   8c1444c4da57        258MB         62.2MB        
postgres:18.6                  4ef4dbc939d6        650MB          168MB    U   
redis:8.10.1                   298e5b3bc566        212MB         57.5MB    U   
(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk>

fastapi
uvicorn[standard]
psycopg[binary]
kafka-python
pydantic

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

import json
import math
import os
from datetime import datetime

import psycopg
from fastapi import FastAPI, HTTPException
from kafka import KafkaProducer
from pydantic import BaseModel

# 配置 PostgreSQL 和 Kafka。
DATABASE_URL = os.environ["DATABASE_URL"]
KAFKA_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:19092")
KAFKA_TOPIC = os.getenv("KAFKA_TOPIC", "transaction-analysis")

# 支付方式编码，与训练脚本保持一致。
PAYMENT_FORMAT_CODES = {
    "ACH": 1,
    "CREDIT CARD": 2,
    "CHEQUE": 3,
    "CASH": 4,
    "WIRE": 5,
}

# 创建 FastAPI 应用。
app = FastAPI(title="fincross-risk Data Service")

# 创建 Kafka 生产者。
producer = KafkaProducer(
    bootstrap_servers=KAFKA_SERVERS,
    # 将任务序列化为 JSON 字节。
    value_serializer=lambda value: json.dumps(value).encode("utf-8"),
)

# 定义交易请求模型。
class TransactionIn(BaseModel):
    external_id: str
    transaction_timestamp: datetime
    from_bank: str
    from_account: str
    to_bank: str
    to_account: str
    amount_received: float
    receiving_currency: str
    amount_paid: float
    payment_currency: str
    payment_format: str
    # 可选的真实标签。
    is_laundering: bool | None = None

# 创建数据库连接。
def db():
    return psycopg.connect(DATABASE_URL)

# 将支付方式转换为数值编码，未知方式记为 0。
def payment_format_code(value: str) -> float:
    return float(PAYMENT_FORMAT_CODES.get(value.strip().upper(), 0))

# 生成交易特征。
def make_features(item: TransactionIn) -> dict:
    # 提取交易小时和星期。
    hour = item.transaction_timestamp.hour
    weekday = item.transaction_timestamp.weekday()
    return {
        # 保留原始金额，并计算非负金额的对数特征。
        "amount_paid": float(item.amount_paid),
        "amount_received": float(item.amount_received),
        "amount_paid_log": math.log1p(max(item.amount_paid, 0.0)),
        "amount_received_log": math.log1p(max(item.amount_received, 0.0)),
        "hour": float(hour),
        "weekday": float(weekday),
        # 生成跨行和跨币种标记。
        "cross_bank_flag": float(item.from_bank != item.to_bank),
        "cross_currency_flag": float(
            item.payment_currency.strip().upper()
            != item.receiving_currency.strip().upper()
        ),
        "payment_format_code": payment_format_code(item.payment_format),
    }

# Data Service 健康检查。
@app.get("/health")
def health():
    with db() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT 1")
            cur.fetchone()
    return {"service": "data-service", "status": "ok"}

# 提交交易。
@app.post("/transactions")
def create_transaction(item: TransactionIn):
    features = make_features(item)
    # 保存交易，初始状态为 RECEIVED。
    try:
        with db() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """INSERT INTO transactions
                    (external_id, source_type, transaction_timestamp,
                     from_bank, from_account, to_bank, to_account,
                     amount_received, receiving_currency, amount_paid,
                     payment_currency, payment_format, is_laundering, status)
                    VALUES (%s, 'ONLINE', %s, %s, %s, %s, %s, %s, %s,
                            %s, %s, %s, %s, 'RECEIVED')
                    RETURNING id""",
                    (
                        item.external_id,
                        item.transaction_timestamp,
                        item.from_bank,
                        item.from_account,
                        item.to_bank,
                        item.to_account,
                        item.amount_received,
                        item.receiving_currency,
                        item.amount_paid,
                        item.payment_currency,
                        item.payment_format,
                        item.is_laundering,
                    ),
                )
                # 获取内部交易编号并提交事务。
                transaction_id = cur.fetchone()[0]
                conn.commit()
    except psycopg.errors.UniqueViolation:
        # 交易编号重复时返回 409。
        raise HTTPException(status_code=409, detail="external_id already exists")

    # 组装交易分析任务。
    message = {
        "transaction_id": transaction_id,
        "external_id": item.external_id,
        "features": features,
    }
    try:
        # 发布任务并等待 Kafka 确认。
        producer.send(KAFKA_TOPIC, message).get(timeout=10)
    except Exception as exc:
        # Kafka 发布失败时返回 503。
        raise HTTPException(status_code=503, detail=f"kafka publish failed: {exc}")

    # 将 RECEIVED 更新为 QUEUED，避免覆盖已完成状态。
    with db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE transactions SET status='QUEUED' WHERE id=%s AND status='RECEIVED'",
                (transaction_id,),
            )
            conn.commit()
    # 返回任务受理信息。
    return {
        "external_id": item.external_id,
        "transaction_id": transaction_id,
        "status": "QUEUED",
    }

# 查询交易。
@app.get("/transactions/{external_id}")
def get_transaction(external_id: str):
    # 查询交易和最新预测，尚未预测的交易也保留。
    with db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT t.external_id, t.transaction_timestamp,
                          t.from_bank, t.from_account, t.to_bank, t.to_account,
                          t.amount_received, t.receiving_currency,
                          t.amount_paid, t.payment_currency, t.payment_format,
                          t.is_laundering, t.status,
                          p.risk_probability, p.risk_level,
                          p.model_version, p.predicted_at
                   FROM transactions t
                   LEFT JOIN LATERAL (
                       SELECT * FROM predictions p0
                       WHERE p0.transaction_id=t.id
                       ORDER BY p0.predicted_at DESC LIMIT 1
                   ) p ON TRUE
                   WHERE t.external_id=%s""",
                (external_id,),
            )
            row = cur.fetchone()
    if row is None:
        # 交易不存在时返回 404。
        raise HTTPException(status_code=404, detail="transaction not found")
    # 按查询字段顺序组装响应字典。
    keys = [
        "external_id", "transaction_timestamp", "from_bank", "from_account",
        "to_bank", "to_account", "amount_received", "receiving_currency",
        "amount_paid", "payment_currency", "payment_format", "is_laundering",
        "status", "risk_probability", "risk_level", "model_version",
        "predicted_at",
    ]
    return dict(zip(keys, row))

docker compose build data-service
docker compose up -d data-service

(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk> docker ps
CONTAINER ID   IMAGE                        COMMAND                   CREATED         STATUS                    PORTS                      NAMES
fdb6ebe2177a   fincross-risk-data-service   "uvicorn app:app --h…"   8 seconds ago   Up 6 seconds (healthy)    127.0.0.1:8001->8000/tcp   fincross-data-service
4be8212e7fc4   postgres:18.6                "docker-entrypoint.s…"   4 days ago      Up 24 minutes (healthy)   127.0.0.1:5432->5432/tcp   fincross-postgres
82d1657de82e   redis:8.10.1                 "docker-entrypoint.s…"   4 days ago      Up 24 minutes (healthy)   6379/tcp                   fincross-redis
ce11e5f44bbf   apache/kafka:4.3.1           "/__cacert_entrypoin…"   4 days ago      Up 24 minutes (healthy)   127.0.0.1:9092->9092/tcp   fincross-kafka
(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk>

{
  "service": "data-service",
  "status": "ok"
}

row = [[float(payload[name]) for name in FEATURE_NAMES]]

# 模型接口接收的是“一批交易”，外层放多笔交易，内层放一笔交易的 9 个特征，这里一批交易中只有一笔交易。
row = [[9000, 9000, 9.105, 9.105, 2, 5, 1, 0, 5]]

probability = float(model.predict_proba(row)[0][1])

# 训练结束后保存
joblib.dump(models[best_name], MODEL_DIR / 'model.pkl')

# 服务启动时加载
model = joblib.load(MODEL_PATH)

model.predict_proba(row)
# [[0.246, 0.754]]
#   正常    风险

result = model.predict_proba(row)  # 得到 [[0.246, 0.754]]
first = result[0]                 # 取出第一笔交易的结果：[0.246, 0.754]
probability = float(first[1])     # 取第二个数 0.754，并转为普通浮点数

probability = float(model.predict_proba(row)[0][1])

{
  "risk_probability": 0.754,
  "risk_level": "HIGH",
  "model_version": "aml-model-v1"
}

fastapi
uvicorn[standard]
scikit-learn
xgboost
joblib
numpy

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

import os
from pathlib import Path

# 加载模型，创建接口并返回 HTTP 错误
import joblib
from fastapi import FastAPI, HTTPException

# 固定九个特征的排列顺序，与训练时一致
FEATURE_NAMES = [
    "amount_paid", "amount_received", "amount_paid_log",
    "amount_received_log", "hour", "weekday", "cross_bank_flag",
    "cross_currency_flag", "payment_format_code",
]

# 读取环境变量；未设置时使用默认值
MODEL_PATH = os.getenv("MODEL_PATH", "/models/model.pkl")
MODEL_VERSION = os.getenv("MODEL_VERSION", "aml-model-v1")

# 创建模型服务的 API 应用
app = FastAPI(title="fincross-risk Model Service")

# 启动时加载一次模型；文件不存在时暂记为 None
model = joblib.load(MODEL_PATH) if Path(MODEL_PATH).exists() else None

# 健康检查接口，供外部判断模型服务是否就绪
@app.get("/health")
def health():
    # 模型未加载时返回 503，表示服务尚不可用
    if model is None:
        raise HTTPException(status_code=503, detail="model is not loaded; train model first")

    # 模型已加载，返回服务状态和版本标识
    return {
        "service": "model-service",
        "status": "ok",
        "loaded": model is not None,
        "model_version": MODEL_VERSION,
    }

# 预测接口，payload 接收请求中的 JSON 特征数据
@app.post("/predict")
def predict(payload: dict):
    # 没有可用模型时拒绝预测
    if model is None:
        raise HTTPException(status_code=503, detail="model is not loaded; train model first")

    # 检查九个特征是否齐全，缺失时返回 400 和字段名
    missing = [name for name in FEATURE_NAMES if name not in payload]
    if missing:
        raise HTTPException(status_code=400, detail={"missing": missing})

    # 按训练时的顺序取值并转为浮点数，外层列表表示一批交易
    row = [[float(payload[name]) for name in FEATURE_NAMES]]

    # 计算概率；[0] 取第一笔交易，[1] 取风险类别的概率
    probability = float(model.predict_proba(row)[0][1])

    # 根据风险概率设定风险等级
    risk_level = "HIGH" if probability >= 0.70 else "MEDIUM" if probability >= 0.30 else "LOW"

    # 返回概率、等级和版本；数据库写回由 Worker 负责
    return {
        "risk_probability": probability,
        "risk_level": risk_level,
        "model_version": MODEL_VERSION,
    }

docker compose build model-service

docker image ls

(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk> docker image ls
                                                                                                                                                                                    i Info →   U  In Use
IMAGE                                ID             DISK USAGE   CONTENT SIZE   EXTRA
apache/kafka:4.3.1                   77e3df905404        686MB          239MB    U   
fincross-risk-data-service:latest    c5381d7a71d9        291MB         68.8MB    U   
fincross-risk-gateway:latest         8c1444c4da57        258MB         62.2MB        
fincross-risk-model-service:latest   24048d1ba4e3       1.31GB          451MB        
postgres:18.6                        4ef4dbc939d6        650MB          168MB    U   
redis:8.10.1                         298e5b3bc566        212MB         57.5MB    U   
(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk>

requests
psycopg[binary]
kafka-python
redis

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY worker.py .
CMD ["python", "worker.py"]

import json
import os
import time

import psycopg
import redis
import requests
from kafka import KafkaConsumer


# 读取数据库、Redis、Kafka 和模型服务的连接配置。
DATABASE_URL = os.environ["DATABASE_URL"]
REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0")
KAFKA_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:19092")
KAFKA_TOPIC = os.getenv("KAFKA_TOPIC", "transaction-analysis")
MODEL_SERVICE_URL = os.getenv("MODEL_SERVICE_URL", "http://model-service:8000")


# 创建 Redis 客户端，将读取的内容解码为字符串。
redis_client = redis.from_url(REDIS_URL, decode_responses=True)


# 创建 PostgreSQL 连接。
def db():
    return psycopg.connect(DATABASE_URL)


# 处理一条交易分析任务。
def process_message(message):
    # 读取业务交易编号和数据库内部编号。
    payload = message.value
    external_id = payload["external_id"]
    transaction_id = int(payload["transaction_id"])

    # 使用业务交易编号生成 Redis 完成标记的键名。
    done_key = f"fincross:transaction:{external_id}:done"

    # 已有完成标记时跳过预测和数据库写入。
    if redis_client.exists(done_key):
        print(f"skip duplicated message external_id={external_id}", flush=True)
        return

    # 将任务中的九个特征发送给模型服务，等待预测结果。
    response = requests.post(
        f"{MODEL_SERVICE_URL}/predict",
        json=payload["features"],
        timeout=15,
    )

    # HTTP 请求失败时抛出异常，成功时读取 JSON 结果。
    response.raise_for_status()
    result = response.json()

    # 在同一个事务中保存预测结果并更新交易状态。
    with db() as conn:
        with conn.cursor() as cur:
            # 将模型版本、风险概率和风险等级写入预测表。
            cur.execute(
                """INSERT INTO predictions
                   (transaction_id, model_version, risk_probability, risk_level)
                   VALUES (%s,%s,%s,%s)""",
                (transaction_id, result["model_version"],
                 result["risk_probability"], result["risk_level"]),
            )

            # 将交易标记为已完成评分。
            cur.execute(
                "UPDATE transactions SET status='SCORED' WHERE id=%s",
                (transaction_id,),
            )

        # 两项写入均成功后提交事务；发生异常则回滚。
        conn.commit()

    # 数据库提交成功后保存完成标记，有效期为一天。
    redis_client.set(done_key, "1", ex=86400)

    # 输出处理结果，flush=True 让容器日志及时显示。
    print(
        f"scored external_id={external_id} "
        f"probability={result['risk_probability']:.6f}",
        flush=True,
    )


# 启动消费者，持续读取 Kafka 任务。
def main():
    consumer = KafkaConsumer(
        # 订阅 Data Service 发布任务的主题。
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_SERVERS,

        # 同一消费组的 Worker 按分区分担任务。
        group_id="fincross-risk-worker",

        # 关闭自动提交，由代码在处理成功后提交进度。
        enable_auto_commit=False,

        # 没有有效消费进度时，从仍保留的最早消息开始读取。
        auto_offset_reset="earliest",

        # 将消息中的 JSON 字节解码为 Python 字典。
        value_deserializer=lambda value: json.loads(value.decode("utf-8")),
    )
    print("worker started", flush=True)

    # 逐条处理收到的消息。
    for message in consumer:
        try:
            process_message(message)

            # 处理完成后提交消费进度，供后续恢复消费使用。
            consumer.commit()
        except Exception as exc:
            # 记录错误并等待三秒；此处不会自动重试当前消息。
            print(f"processing failed: {exc}", flush=True)
            time.sleep(3)


# 直接运行 worker.py 时启动消费循环。
if __name__ == "__main__":
    main()

docker compose build worker

docker image ls

(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk> docker image ls            
                                                                                                                                                                                    i Info →   U  In Use
IMAGE                                ID             DISK USAGE   CONTENT SIZE   EXTRA
apache/kafka:4.3.1                   77e3df905404        686MB          239MB    U   
fincross-risk-data-service:latest    c5381d7a71d9        291MB         68.8MB    U   
fincross-risk-gateway:latest         8c1444c4da57        258MB         62.2MB        
fincross-risk-model-service:latest   24048d1ba4e3       1.31GB          451MB        
fincross-risk-worker:latest          02ab908e4d9e        239MB         56.5MB        
postgres:18.6                        4ef4dbc939d6        650MB          168MB    U   
redis:8.10.1                         298e5b3bc566        212MB         57.5MB    U   
(.venv) PS C:\Users\Ezekielx\PycharmProjects\fincross-risk>