2026 技术观察：API 网关进入智能治理阶段，限流、熔断和异常流量识别成为企业系统新底座

import json
import random
from datetime import datetime
from collections import defaultdict


API_LOGS = [
    {
        "api": "/api/order/create",
        "client_id": "client_A",
        "status_code": 200,
        "latency_ms": 120
    },
    {
        "api": "/api/order/create",
        "client_id": "client_A",
        "status_code": 200,
        "latency_ms": 180
    },
    {
        "api": "/api/payment/pay",
        "client_id": "client_B",
        "status_code": 500,
        "latency_ms": 1200
    },
    {
        "api": "/api/payment/pay",
        "client_id": "client_B",
        "status_code": 502,
        "latency_ms": 1500
    },
    {
        "api": "/api/user/profile",
        "client_id": "client_C",
        "status_code": 200,
        "latency_ms": 90
    }
]

def aggregate_api_metrics(logs):
    metrics = defaultdict(
        lambda: {
            "request_count": 0,
            "error_count": 0,
            "latency_total": 0,
            "clients": set()
        }
    )

    for log in logs:
        api = log["api"]
        metrics[api]["request_count"] += 1
        metrics[api]["latency_total"] += log["latency_ms"]
        metrics[api]["clients"].add(log["client_id"])

        if log["status_code"] >= 500:
            metrics[api]["error_count"] += 1

    results = []

    for api, item in metrics.items():
        request_count = item["request_count"]
        error_rate = item["error_count"] / request_count if request_count else 0
        avg_latency = item["latency_total"] / request_count if request_count else 0

        results.append({
            "api": api,
            "request_count": request_count,
            "error_count": item["error_count"],
            "error_rate": round(error_rate * 100, 2),
            "avg_latency_ms": round(avg_latency, 2),
            "client_count": len(item["clients"])
        })

    return results

def detect_api_risk(metric):
    issues = []
    risk_score = 0

    if metric["error_rate"] >= 30:
        issues.append("接口错误率较高。")
        risk_score += 4

    if metric["avg_latency_ms"] > 1000:
        issues.append("接口平均响应时间过长。")
        risk_score += 3

    if metric["request_count"] > 1000:
        issues.append("接口请求量较高，需要关注容量。")
        risk_score += 2

    if risk_score >= 6:
        level = "high"
    elif risk_score >= 3:
        level = "medium"
    elif risk_score > 0:
        level = "low"
    else:
        level = "normal"

    return {
        "api": metric["api"],
        "risk_score": risk_score,
        "risk_level": level,
        "issues": issues
    }

def analyze_client_behavior(logs):
    client_stats = defaultdict(
        lambda: {
            "request_count": 0,
            "error_count": 0,
            "apis": set()
        }
    )

    for log in logs:
        client_id = log["client_id"]
        client_stats[client_id]["request_count"] += 1
        client_stats[client_id]["apis"].add(log["api"])

        if log["status_code"] >= 500:
            client_stats[client_id]["error_count"] += 1

    results = []

    for client_id, stat in client_stats.items():
        error_rate = stat["error_count"] / stat["request_count"]

        if error_rate >= 0.5:
            level = "high"
            message = "客户端请求错误率较高，建议检查调用逻辑。"
        elif stat["request_count"] > 1000:
            level = "medium"
            message = "客户端请求量较高，建议评估限流策略。"
        else:
            level = "normal"
            message = "客户端调用状态正常。"

        results.append({
            "client_id": client_id,
            "request_count": stat["request_count"],
            "error_count": stat["error_count"],
            "error_rate": round(error_rate * 100, 2),
            "api_count": len(stat["apis"]),
            "risk_level": 31226.t.kuaisou.com 
            "message": message
        })

    return results

def generate_gateway_policy(api_risk):
    level = api_risk["risk_level"]

    if level == "high":
        return {
            "api": api_risk["api"],
            "action": "circuit_breaker",
            "message": "接口风险较高，建议触发熔断并保护后端服务。"
        }

    if level == "medium":
        return {
            "api": api_risk["api"],
            "action": "rate_limit",
            "message": "接口存在中等风险，建议设置动态限流。"
        }

    if level == "low":
        return {
            "api": api_risk["api"],
            "action": "increase_monitoring",
            "message": "接口存在轻微异常，建议提高监控频率。"
        }

    return {
        "api": api_risk["api"],
        "action": "keep",
        "message": "接口状态正常，保持当前策略。"
    }

def run_api_gateway_governance():
    metrics = aggregate_api_metrics(API_LOGS)

    api_risks = [
        detect_api_risk(metric)
        for metric in metrics
    ]

    client_results = analyze_client_behavior(
        API_LOGS
    )

    gateway_policies = [
        generate_gateway_policy(risk)
        for risk in api_risks
    ]

    risk_count = defaultdict(int)

    for risk in api_risks:
        risk_count[risk["risk_level"]] += 1

    report = {
        "report_name": "API 网关智能治理报告",
        "api_metrics": 31227.t.kuaisou.com 
        "api_risks": api_risks,
        "client_results": client_results,
        "gateway_policies": gateway_policies,
        "risk_count": dict(risk_count),
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_api_gateway_governance()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))