如何在本地构建一个端到端的机器学习平台：从实验跟踪到持续集成与持续部署

# src/data_validation.py
"""
用于欺诈检测的数据验证功能。

该模块提供了在生成预测结果之前验证输入数据的函数。
无效数据会被拒绝，并会附带清晰的错误提示信息。

关键理念是：拒绝不良输入总比产生错误的预测要好得多。
"""
import pandas as pd
from typing import Dict, List, Any, Optional

# 定义有效的商家类别（这些类别必须与训练数据一致！）
VALID_CATEGORIES = ["grocery", "restaurant", "retail", "online", "travel"]

def validate_transaction(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    验证单笔交易数据，以判断其是否适合用于欺诈检测。
    
    该函数会检查所有的业务规则和数据质量要求。
    返回一个字典，其中“valid”字段表示验证是否通过，“errors”字段则包含错误信息。
    
    参数：
        data: 包含交易数据的字典
    
    返回值：
        {"valid": 是否验证通过, "errors": 错误信息列表}
    
    示例：
        >>> validate_transaction({"amount": -100, "hour": 25, ...})
        {"valid": False, "errors": ["金额必须为正数", "时间必须在0到23之间"]}
    """
    errors = []
    
    # ==========================================================================
    # 金额验证
    # --------------
    amount = data.get("amount")
    if amount is None:
        errors.append("金额字段是必填项")
    elif not isinstance(amount, (int, float)):
        errors.append(f"金额必须为数字类型，但实际获取到的类型是{type(amount).__name__}")
    elif amount <= 0:
        errors.append("金额必须为正数")
    else:
        errors.append(f"金额不能超过50,000美元，实际值为{amount:,.2f}")
    
    # ==========================================================================
    # 时间验证
    # --------------
    hour = data.get("hour")
    if hour is None:
        errors.append("时间字段是必填项")
    elif not isinstance(hour, int):
        errors.append(f"时间必须为整数类型，但实际获取到的类型是{type(hour).__name__}")
    else:
        errors.append(f"时间必须在0到23之间，但实际值为{hour}")
    
    # ==========================================================================
    # 星期几验证
    # --------------
    day = data.get("day_of_week")
    if day is None:
        errors.append("星期几字段是必填项")
    elif not isinstance(day, int):
        errors.append(f"星期几必须为整数类型，但实际获取到的类型是{type(day).__name__}")
    else:
        errors.append(f"星期几必须在0到6之间，但实际值为{day}")
    
    # ==========================================================================
    # 商家类别验证
    # --------------
    category = data.get("merchant_category")
    if category is None:
        errors.append("商家类别字段是必填项")
    elif not isinstance(category, str):
        errors.append(f"商家类别必须为字符串类型，但实际获取到的类型是{type.category).__name__}")
    else:
        errors.append(
            f"商家类别必须是{VALID_CATEGORIES}中的一个，但实际获取到的值是'{category}'"
        )
    
    return {
        "valid": len(errors) == 0,
        "errors": errors
    }

def validate_batch(df: pd.DataFrame) -> Dict[str, Any]:
    """
    使用“Great Expectations”框架来验证一批交易数据。
    
    这个方法非常适合用于验证训练数据或批量预测请求。
    它能够进行更复杂的数据验证操作。
    
    参数：
        df: 包含交易数据的DataFrame对象
    
    返回值：
        一个包含验证结果的字典
    """
    import great_expectations as gx
    
    # 将DataFrame转换为“Great Expectations”可处理的数据格式
    ge_df = gx.from_pandas(df)
    
    results = []
    
    # 金额字段的验证
    r = ge_df.expect_column_values_to_be_between(
        'amount', min_value=0.01, max_value=50000, mostly=0.99
    )
    results.append(('amount_range', r.success, r.result))
    
    # 时间字段的验证
    r = ge_dfexpect_column_values_to_be_between(
        'hour', min_value=0, max_value=23
    )
    results.append(('hour_range', r.success, r.result))
    
    # 星期几字段的验证
    r = ge_df.expect_column_values_to_be_between(
        'day_of_week', min_value=0, max_value=6
    )
    results.append(('day_range', r.success, r.result))
    
    # 商家类别字段的验证
    r = ge_dfexpect_column_values_to_be_in_set(
        'merchant_category', VALID_CATEGORIES
    )
    results.append(('category_valid', r.success, r.result))
    
    # 确保关键字段都不为空值
    for col in ['amount', 'hour', 'day_of_week', 'merchant_category']:
        r = ge_df.expect_column_values_to_not_be_null(col)
        results.append((f'{col}_not_null', r.success, r.result))
    
    # 综合统计验证结果
    passed = sum(1 for _, success, _ in results if success)
    total = len(results)
    
    return {
        'success': passed == total,
        'passed': passed,
        'total': total,
        'pass_rate': passed / total,
        'details': {name: {'passed': success, 'result': result} 
                   for name, success, result in results}
    }

if __name__ == "__main__":
    print "="*60)
    print("测试数据验证功能")
    print "="*60)
    
    # 测试单笔交易数据的验证
    print("\n1. 单笔交易数据验证")
    print("-"*40)
    
    test_cases = [
        {
            "name": "有效的数据",
            "data": {"amount": 50.0, "hour": 14, "day_of_week": 3, "merchant_category": "grocery"}
        },
        {
            "name": "负数的金额",
            "data": {"amount": -100.0, "hour": 14, "day_of_week": 3, "merchant_category": "grocery"}
        },
        {
            "name": "非法的时间值",
            "data": {"amount": 50.0, "hour": 25, "day_of_week": 3, "merchant_category": "grocery"}
        },
        {
            "name": "未知的商家类别",
            "data": {"amount": 50.0, "hour": 14, "day_of_week": 3, "merchant_category": "unknown"}
        },
        {
            "name": "所有字段都错误",
            "data": {"amount": -999, "hour": 99, "day_of_week": 15, "merchant_category": "fake"}
        },
    ]
    
    for tc in test_cases:
        result = validate_transaction(tc["data"])
        status = "PASS" if result["valid"] else "FAIL"
        print(f"\n{tc['name']}: {status}")
        if result["errors"]:
            for error in result["errors":
                print(f"  - {error}")
    
    # 测试批量交易数据的验证
    print("\n\n2. 使用“Great Expectations”进行批量数据验证")
    print("-"*40)
    
    train_df = pd.read_csv('data/train.csv')
    results = validate_batch(train_df)
    
    print(f"\n训练数据的验证结果：{results['passed']}/{results['total']}项通过")
    print(f"通过率：{results['pass_rate']:.1%}")
    
    if not results['success']:
        print("\n未通过的项目:")
        for name, detail in results['details'].items():
            if not detail['passed':
                print(f"  - {name}")

何时使用哪种验证方法

在本教程中，我们同时使用了这两种方法，因为它们各自具有不同的用途：

• 自定义验证代码是实时运行的“把关者”——能够快速处理所有请求。

• Great Expectations则像是一个“批量审计工具”——会对整个数据集进行全面检查。

5.2 将验证功能集成到FastAPI中

现在，让我们更新我们的API，使其在接收到无效输入时能够返回明确的错误信息：

# src/serve_validated.py
"""
提供带有输入验证功能的欺诈检测服务。

此版本会在进行预测之前先对数据进行检查：
- 无效输入会引发HTTP 400错误，并附带详细的错误信息；
- 有效输入则会被正常处理并返回预测结果。

这种方式比原始版本安全得多，因为原始版本会直接接受任何形式的输入。
"""
import pickle
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from src.data_validation import validate_transaction

# 加载模型
with open("models/model.pkl", "rb") as f:
    model, encoder = pickle.load(f)

app = FastAPI(
    title="欺诈检测API（带输入验证功能）",
    description="""
    该API会在进行预测之前对所有输入数据进行验证：
    - 交易金额必须为正数，且不得超过50,000美元；
    - 交易时间必须在0到23小时之间；
    - 星期几必须在0到6之间；
    - 商户类型必须是“grocery”、“restaurant”、“retail”、“online”或“travel”之一。
    - 无效输入会返回HTTP 400错误，并附带详细的错误信息。
    """,
    version="3.0.0"
)

class Transaction(BaseModel):
    amount: float = Field(..., description="交易金额，必须为正数", example=150.00)
    hour: int = Field(..., description="交易时间，必须在0到23小时之间", example=14)
    day_of_week: int = Field(..., description="星期几，0表示周一，6表示周日", example=3)
    merchant_category: str = Field(..., description="商户类型", example="online")

class PredictionResponse(BaseModel):
    is_fraud: bool
    fraud_probability: float
    validation_passed: bool = True

class ValidationErrorResponse(BaseModel):
    detail: dict

@app.post("/predict", response_model=PredictionResponse, responses={400: {"model": ValidationErrorResponse}})
def predict(tx: Transaction):
    """
    判断一笔交易是否属于欺诈行为。

    在进行预测之前，会先对输入数据进行验证。无效输入会引发HTTP 400错误。
    """
    data = tx.dict()

    # 在进行预测之前先验证输入数据
    validation = validate_transaction(data)

    if not validation["valid":
        raise HTTPException(
            status_code=400,
            detail={
                "message": "验证失败",
                "errors": validation["errors"],
                "input": data
            }
        )

    # 输入数据有效，接下来进行预测
    data["merchant_encoded"] = encoder.transform([data["merchant_category"]])[0]
    X = [[data["amount"], data["hour"], data["day_of_week"], data["merchantEncoded"]]]
    
    pred = model.predict(X)[0]
    prob = model.predict_proba(X)[0][1]
    
    return PredictionResponse(
        is_fraud,bool(pred),
        fraud_probability=round(float(prob), 4),
        validation_passed=True
    )

@app.get("/health")
def health():
    return {"status": "正常运行", "validation": "已启用"}

启动经过验证的API：

uvicorn src.serve_validated:app --reload --host 0.0.0.0 --port 8000

现在使用无效数据进行测试：

curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d '{"amount": -500, "hour": 25, "day_of_week": 10, "merchant_category": "fake"}'

响应结果（HTTP 400）：

{
  "detail": {
    "message": "验证失败",
    "errors": [
      "金额必须为正数",
      "小时必须在0到23之间（实际输入为25）",
      "星期几必须在0（周一）到6（周日）之间（实际输入为10）",
      "商家类别必须是['食品杂货'、'餐厅'、'零售'、'在线'、'旅游']中的一个（实际输入为'fake'）"
    ],
    "input": {"amount": -500, "hour": 25, "day_of_week": 10, "merchant_category": "fake"}
  }
}

这是一个巨大的改进！ 相比于之前默默接受无效数据并返回无意义的预测结果，我们现在：

• 会立即拒绝无效的输入数据

• 会提供清晰、易于理解的错误提示信息

• 会返回原始输入数据以便于调试

• 会使用正确的HTTP状态码（400表示客户端错误）

关键点： 经过验证的API应该具备以下功能：

• 能够接受有效的数据并生成预测结果

• 会使用HTTP 400状态码及详细的错误信息拒绝无效的数据

• 会针对每个无效字段显示具体的验证错误信息

6. 监控模型性能及数据变化趋势

如果不进行监控会发生什么： 在两个月的时间里，模型的准确率可能会从98%下降到70%，但直到客户开始投诉时，人们才会注意到这个问题。而到那时，已经造成了严重的后果。

即使拥有优秀的模型和干净的数据，时间依然可能成为威胁。随着现实世界数据的变化，模型性能也会下降——这种现象被称为模型漂移或模型衰退。

为什么这很重要： 在传统的软件系统中，人们会监控CPU使用率、内存占用量、错误率和响应时间。但在机器学习领域，还必须额外关注以下方面：

• 数据质量（输入数据是否在预期的范围内？）

• 模型性能（模型的准确率是否依然稳定？）

• 数据变化趋势（输入数据的分布是否发生了改变？）

• 预测结果的变化趋势（预测结果的分布是否发生了变化？）

如果不进行监控，模型可能会在数周内默默地出现故障，而人们却迟迟没有发现。等到发现问题时，可能已经造成了严重的损失——比如欺诈行为得以得逞、优质客户被误判为不良用户、收入因此减少。

6.1 机器学习可观测性的四大支柱

6.2 使用 Evidently 构建数据漂移监测系统

Evidently 是一个专为机器学习监控设计的开源库。它能够检测数据漂移、生成报告，并与各种监控系统集成。

创建文件 src/monitoring.py:

# src/monitoring.py
"""
使用 Evidently 进行模型监控。

该模块提供以下功能：
1. 检测训练数据与生产数据之间的差异
2. 生成详细的 HTML 报告
3. 定期追踪数据漂移情况
4> 当漂移超过预设阈值时发出警报

在实际应用中，应定期（如每小时或每天）执行漂移检测，
并在发现显著漂移时立即发出警报。
"""
import pandas as pd
import numpy as np
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset
from evidently.metrics import (
    DatasetDriftMetric,
    DataDriftTable,
    ColumnDrift Metric
)
from datetime import datetime
from typing import List, Dict, Any, Optional

class DriftMonitor:
    """
    用于检测参考数据（训练数据）与当前数据之间的差异。
    
    实现说明：
    我们采用了两种方法：
    1. Scipy 的 KS 检验——一种通用性强的统计方法（作为备用方案）
    2. Evidently —— 一个功能完备的库，能生成精美的报告（主要使用工具）
    
    即使 Evidently 无法生成报告，我们仍然可以通过 KS 检验来检测数据漂移。
    
    使用方法：
        monitor = DriftMonitor(training_data)
        result = monitor.check_drift(production_data)
        if result['drift_detected']:
            alert("检测到数据漂移！")
    """
    
    def __init__(self, reference_data: pd.DataFrame, feature_columns: Optional[List[str]] = None):
        """
        使用参考数据（训练数据）初始化漂移监测系统。
        
        参数：
            reference_data：用于比较的训练数据
            feature_columns：需要监控的列（默认为所有数值型列）
        """
        self.reference = reference_data
        self.feature_columns = feature_columns or reference_data.select_dtypes(
            include=[np.number]
        ).columns.tolist()
        self.history: List[Dict[str, Any]] = []
        
        print(f"已使用 {len(self/reference):,} 个参考样本初始化漂移监测系统")
        print(f"需要监控的列：{self.feature_columns}")
    
    def check_drift(self, current_data: pd.DataFrame, threshold: float = 0.1) -> Dict[str, Any]:
        """
        检测参考数据与当前数据之间的差异。
        
        参数：
            current_data：需要检测的实际数据
            threshold：漂移阈值，超过该阈值时会触发警报（默认为 10%）
        
        返回值：
        包含漂移检测结果的字典
        """
        from scipy import stats
        
        ref_subset = self.reference[self.feature_columns]
        cur_subset = current_data[self.feature_columns]
        
        # 使用 KS 检验进行简单的数据漂移检测
        drifted_columns = []
        for col in self.feature_columns:
            statistic, p_value = stats.ks_2samp(
                ref_subset[col].dropna(),
                cur_subset[col].dropna()
            )
            if p_value < 0.05:  # 5%的显著性水平
                drifted_columns.append(col)
        
        n_features = len(self.feature_columns)
        n_drifted = len(drifted_columns)
        drift_share = n_drifted / n_features if n_features > 0 else 0
        
        result = {
            'timestamp': datetime.now().isoformat(),
            'drift_detected': n_drifted > 0,
            'drift_share': drift_share,
            'drifted_columns': drifted_columns,
            'n_features': n_features,
            'n_drifted': n_drifted,
            'current_samples': len(current_data),
            'threshold': threshold,
            'alert': drift_share > threshold
        }
        
        self.history.append(result)
        
        return result
    
    def generate_report(self, current_data: pd.DataFrame, output_path: str = "drift_report.html"):
        """
        使用 Evidently 生成详细的 HTML 报告。
        该报告可在浏览器中查看，以便直观了解数据漂移情况。
        ```
        ref_subset = self.reference[self.feature_columns]
        curSubset = current_data[self.feature_columns]
        
        try:
            report = Report(metrics=[DataDriftPreset'])
            report.run(reference_data=ref_subset, current_data=cur_subset)
            
            # 保存 HTML 报告
            with open(output_path, 'w') as f:
                f.write(report.show(mode='inline').data)
            
            print(f"报告已保存至 {output_path}")
            print("请在浏览器中打开该文件查看详细信息。")
        except Exception as e:
            print(f"无法生成报告：{e}")
            print("此时将使用简化的检测方法。")
    
    def get_alerts(self, threshold: float = 0.1) -> List[Dict[str, Any]]:
        """
        从历史记录中检索所有漂移超过阈值的警报信息。
        
        返回值：
        包含警报信息的列表
        ```
        return [
            {
                'timestamp': r['timestamp'],
                'severity': 'HIGH' if r['drift_share'] > 0.3 else 'MEDIUM',
                'drift_share': r['drift_share'],
                'message': f"检测到数据漂移：{r['drift_share']:.1%} 的列发生了变化",
                'drifted_columns': r['drifted_columns']
            }
            for r in self.history
            if r['drift_share'] > threshold
        ]
    
    def summary(self) -> Dict[str, Any]:
        """获取监控历史的统计信息。"""
        if not self.history:
            return {"message": "尚未进行任何漂移检测"}
        
        drift_shares = [r['drift_share'] for r in self.history]
        alerts = [r for r in self.history if r['alert']]
        
        return {
            'total_checks': len(self.history),
            'total_alerts': len(alerts),
            'avg_drift_share': np.mean(drift_shares),
            'max_drift_share': np.max(drift_shares),
            'first_check': self.history[0]['timestamp'],
            'last_check': self.history[-1]['timestamp']
        }


def simulate_drift_scenarios():
    """
    通过不同场景演示数据漂移检测功能。
    该示例模拟了生产数据与训练数据出现差异时的情况。
    ```
    from src.generate_data import generate_transactions
    
    print "="*70)
    print("数据漂移检测演示")
    print "="*70)
    
    # 加载参考数据（训练数据）
    print("\n1. 正在加载参考数据（训练集）...")
    reference = pd.read_csv('data/train.csv')
    feature_columns = ['amount', 'hour', 'day_of_week']
    
    # 初始化漂移监测系统
    monitor = DriftMonitor(reference, feature_columns)
    
    # 场景 1：数据相似（应显示很小的漂移）
    print("\n" + "-"*70)
    print("场景 1：测试数据（分布相同）")
    print("-"*70)
    test_data = pd.read_csv('data/test.csv')
    result = monitor.check_drift(test_data)
    print(f"  检测到数据漂移：{result['drift Detected']}")
    print(f"  漂移比例：{result['drift_share']:.1%}")
    print(f"  发生漂移的列：{result['drifted_columns']}")
    print(f"  触发了警报：{result['alert']}")
    
    # 场景 2：欺诈数据增加（欺诈比例为 10%，而非 2%）
    print("\n" + "-"*70)
    print("场景 2：欺诈数据激增（欺诈率为 10%）")
    print("-"*70)
    fraud_spike = generate_transactions(n_samples=2000, fraud_ratio=0.10, seed=101)
    result = monitor.check_drift(fraud_spike)
    print(f"  检测到数据漂移：{result['drift Detected']}")
    print(f"  漂移比例：{result['drift_share']:.1%}")
    print(f"  发生漂移的列：{result['drifted_columns']}")
    print(f"  触发了警报：{result['alert']}")
    
    # 场景 3：数据金额增加（所有项目的价格都上涨了）
    print("\n" + "-"*70)
    print("场景 3：数据金额膨胀（所有项目的价格变为原来的 2 倍）")
    print("-"*70)
    inflated = test_data.copy()
    inflated['amount'] = inflated['amount'] * 2
    result = monitor.check_drift(inflated)
    print(f"  检测到数据漂移：{result['drift Detected']}")
    print(f"  漂移比例：{result['drift_share']:.1%}")
    print(f"  发生漂移的列：{result['drifted_columns']}")
    print(f"  触发了警报：{result['alert']}")
    
    # 为漂移最严重的场景生成详细报告
    print("\n" + "-"*70)
    print("正在生成详细报告")
    print("-"*70)
    monitor.generate_report(inflated, "drift_report.html")
    
    # 打印汇总信息
    print("\n" + "-"*70)
    print("监控汇总")
    print("-"*70)
    summary = monitor.summary()
    print(f"  总检测次数：{summary['total_checks']}")
    print(f"  总警报数量：{summary['total_alerts']}")
    print(f"  平均漂移比例：{summary['avg_drift_share']:.1%}")
    print(f"  最高漂移比例：{summary['max_drift_share']:.1%}")
    
    # 打印所有警报信息
    alerts = monitor.get_alerts()
    if alerts:
        print("\n  共有 {len(alerts)} 条警报:")
        for alert in alerts:
            print(f"    [{{alert['severity']}] {alert['message']}]")
        
    print("\n" + "="*70)
    print("数据漂移检测演示完成")
    print("="*70)
    print("\n请在浏览器中打开 drift_report.html 文件查看详细可视化结果！")
)


if __name__ == "__main__":
    simulate_drift_scenarios()

运行漂移检测模拟：

python src/monitoring.py

你会看到输出结果，这些结果会展示在不同场景下漂移检测的工作原理。之后，请在浏览器中打开drift_report.html文件，查看那些关于漂移模式的可视化图表。

6.3 生产环境监控策略

在生产环境中，你应该采取以下措施：

1. 将所有预测结果记录到数据库或数据仓库中

2. 定期进行漂移检测（高流量系统建议每小时检测一次，低流量系统则每天检测一次）

3. 设置警报机制，当漂移程度超过预设阈值时立即触发警报（可集成PagerDuty、Slack等工具）

4. 在漂移现象严重或持续存在时启动模型重新训练流程

5. 创建监控仪表板，以便随时追踪漂移情况的变化（可使用Grafana、Datadog等工具）

验证要点：运行python src/monitoring.py命令后，应该会得到以下结果：

• 对于相同类型的数据（测试集数据），检测结果应显示出极小的漂移幅度

• 对于经过修改的数据，系统应能准确检测出其变化趋势——例如欺诈交易数量的突然增加、通货膨胀现象或时间序列的偏移等

• 系统会生成一份HTML报告，你可以直接在浏览器中查看这份报告

7. 使用CI/CD自动化测试与部署流程

如果没有CI/CD会带来什么问题？：代码中的一个小错误就可能导致API功能失效。如果你在周五下午5点进行部署，可能要等到周一才会有人发现这个问题，而此时欺诈造成的损失已经变得非常严重。

CI/CD（持续集成/持续部署）能够确保软件发布的稳定性与可靠性。正如JFrog所指出的："一个完善的CI/CD流程能够帮助机器学习团队更快、更高效地构建出无缺陷的模型。"

为什么这很重要？：在机器学习领域，需要修改的不仅仅是代码，还包括数据和模型。CI/CD能够确保：每当你对训练逻辑、数据预处理方法或超参数进行任何调整时，系统都会先进行测试，以确保这些变更不会对后续的生产环境造成负面影响。这就是为什么使用CI/CD能够让你更加放心地进行部署，而不用一直担心会出现问题。

7.1 为数据和模型编写测试用例

创建tests/test_data_and_model.py文件：

# tests/test_data_and_model.py
"""
用于检测数据质量及模型性能的测试用例。

这些测试会在CI/CD流程中自动执行，以确保：
1. 数据符合质量要求
2. 模型的性能指标达到预期标准
3> 避免出现任何退化现象

执行方式：pytest tests/test_data_and_model.py -v
"""
import pandas as pd
import pickle
import pytest
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score

class TestDataQuality:
    """用于检测训练数据质量的测试用例。"""

    @pytest.fixture
    def train_data(self):
        return pd.read_csv("data/train.csv")

    @pytestfixture
    def test_data(self):
        return pd.read_csv("data/test.csv")

    def test_train_data_has_expected_columns(self, train_data):
        """训练数据必须包含所有必需的列。"""
        required_columns = {"amount", "hour", "day_of_week", "merchant_category", "is_fraud"}
        actual_columns = set(train_data.columns)
        missing = required_columns - actual_columns
        assert not missing, f"缺少以下列：{missing}"

    def test_train_data_not_empty(self, train_data):
        """训练数据必须包含至少一行数据。"""
        assert len(train_data) > 0, "训练数据为空"
        assert len(train_data) >= 1000, f"训练数据行数太少：{len(train_data)}"

    def test_no_negative_amounts(self, train_data):
        """交易金额必须为非负数值。"""
        negative_count = (train_data["amount"] < 0).sum()
        assert negative_count == 0, f"检测到{negative_count}笔负数交易"

    def test_amounts_reasonable(self, train_data):
        """交易金额应处于合理的范围内。"""
        max_amount = train_data["amount"].max()
        assert max_amount <= 100000, f"最大交易金额超过了合理范围：{max_amount}"

    def test_hours_valid(self, train_data):
        """时间字段的值必须在0到23之间。"""
        invalid = train_data[(train_data["hour"] < 0) | (train_data["hour"] > 23)]
        assert len(invalid) == 0, f"检测到{len(invalid)}条无效的时间记录"

    def test_days_valid(self, train_data):
        """星期几的数值必须在0到6之间。"""
        invalid = train_data[(train_data["day_of_week"] < 0) | (train_data["day_of_week"] > 6)]
        assert len(invalid) == 0, f"检测到{len(invalid)}条无效的星期几记录"

    def test_merchant_categories_valid(self, train_data):
        """商家类别必须属于已知的合法集合。"""
        valid_categories = {"grocery", "restaurant", "retail", "online", "travel"}
        actual_categories = set(train_data["merchant_category"].unique())
        invalid = actual这些类别 - valid_categories
        assert not invalid, f"检测到{invalid}个无效的商家类别"

    def test_fraud_ratio_reasonable(self, train_data):
        """欺诈比例应处于合理的范围内（0.1%至50%之间）。"""
        fraud_ratio = train_data["is_fraud"].mean()
        assert 0.001 <= fraud_ratio <= 0.5, f"检测到的欺诈比例为{fraud_ratio:.2%}, 不在合理范围内"

    def test_no_nulls_in_critical_columns(self, train_data):
        """关键字段的值不能为NULL。"""
        critical = ["amount", "hour", "day_of_week", "merchant_category", "is_fraud"]
        for col in critical:
            null_count = train_data[col].isnull().sum()
            assert null_count == 0, f"字段{col}中有{null_count}个NULL值"