2026 技术观察：云成本治理进入 FinOps 阶段，资源画像、预算预警和自动优化成为新核心

import json
from datetime import datetime
from collections import defaultdict


CLOUD_RESOURCES = [
    {
        "resource_id": "ecs_001",
        "project": "商城系统",
        "resource_type": "ecs",
        "spec": "4C8G",
        "cpu_usage": 12,
        "memory_usage": 28,
        "daily_cost": 38.5,
        "env": "prod"
    },
    {
        "resource_id": "ecs_002",
        "project": "测试平台",
        "resource_type": "ecs",
        "spec": "8C16G",
        "cpu_usage": 3,
        "memory_usage": 12,
        "daily_cost": 72.0,
        "env": "test"
    },
    {
        "resource_id": "db_001",
        "project": "订单中心",
        "resource_type": "database",
        "spec": "16C32G",
        "cpu_usage": 55,
        "memory_usage": 68,
        "daily_cost": 220.0,
        "env": "prod"
    },
    {
        "resource_id": "gpu_001",
        "project": "AI训练",
        "resource_type": "gpu",
        "spec": "A10",
        "cpu_usage": 6,
        "memory_usage": 20,
        "daily_cost": 480.0,
        "env": "dev"
    }
]


PROJECT_BUDGET = {
    "商城系统": 2000,
    "测试平台": 800,
    "订单中心": 6000,
    "AI训练": 5000
}

def analyze_resource_usage(resource):
    cpu = resource["cpu_usage"]
    memory = resource["memory_usage"]

    if cpu < 10 and memory < 20:
        level = "very_low"
        message = "资源利用率很低，建议重点检查是否可释放或降配。"
    elif cpu < 25 and memory < 35:
        level = "low"
        message = "资源利用率偏低，建议评估是否可以优化规格。"
    else:
        level = "normal"
        message = "资源利用率处于合理范围。"

    return {
        "resource_id": resource["resource_id"],
        "resource_type": resource["resource_type"],
        "project": resource["project"],
        "usage_level": level,
        "cpu_usage": 1867.t.kuaisou.com
        "memory_usage": memory,
        "message": message
    }

def summarize_project_cost(resources):
    project_cost = defaultdict(float)

    for resource in resources:
        project_cost[resource["project"]] += resource["daily_cost"]

    results = []

    for project, daily_cost in project_cost.items():
        monthly_cost = daily_cost * 30
        budget = PROJECT_BUDGET.get(project, 0)

        if budget and monthly_cost > budget:
            status = "over_budget"
            message = "预计月费用超过预算。"
        elif budget and monthly_cost > budget * 0.85:
            status = "near_budget"
            message = "预计月费用接近预算上限。"
        else:
            status = "normal"
            message = "费用处于预算范围内。"

        results.append({
            "project": project,
            "daily_cost": round(daily_cost, 2),
            "estimated_monthly_cost": round(monthly_cost, 2),
            "budget": budget,
            "budget_status": status,
            "message": message
        })

    return results

def generate_resource_optimization(resource, usage_result):
    suggestions = []

    if usage_result["usage_level"] == "very_low":
        if resource["env"] in ["test", "dev"]:
            suggestions.append("该资源属于非生产环境，建议设置自动关停策略。")
        else:
            suggestions.append("该资源利用率极低，建议评估是否可降配。")

    if usage_result["usage_level"] == "low":
        suggestions.append("建议观察连续 7 天利用率，再决定是否降配。")

    if resource["resource_type"] == "gpu" and resource["cpu_usage"] < 10:
        suggestions.append("GPU 资源成本较高，建议确认训练任务是否仍在运行。")

    if not suggestions:
        suggestions.append("当前资源无需立即优化，建议持续监控。")

    return {
        "resource_id": resource["resource_id"],
        "project": resource["project"],
        "suggestions": suggestions
    }

def generate_budget_alerts(project_costs):
    alerts = []

    for item in project_costs:
        if item["budget_status"] == "over_budget":
            alerts.append({
                "project": item["project"],
                "level": "high",
                "message": "项目预计费用已超过预算，建议立即分析成本构成。"
            })

        elif item["budget_status"] == "near_budget":
            alerts.append({
                "project": item["project"],
                "level": "medium",
                "message": "项目费用接近预算上限，建议提前控制新增资源。"
            })

    if not alerts:
        alerts.append({
            "project": "all",
            "level": "normal",
            "message": "当前项目预算状态整体正常。"
        })

    return alerts

def run_cloud_cost_governance():
    usage_results = []
    optimization_results = []

    for resource in CLOUD_RESOURCES:
        usage = analyze_resource_usage(resource)
        optimization = generate_resource_optimization(
            resource,
            usage
        )

        usage_results.append(usage)
        optimization_results.append(optimization)

    project_costs = summarize_project_cost(
        CLOUD_RESOURCES
    )

    budget_alerts = generate_budget_alerts(
        project_costs
    )

    total_daily_cost = sum(
        item["daily_cost"]
        for item in CLOUD_RESOURCES
    )

    report = {
        "report_name": "云成本治理 FinOps 分析报告",
        "total_daily_cost": round(total_daily_cost, 2),
        "estimated_monthly_cost": round(total_daily_cost * 30, 2),
        "usage_results": usage_results,
        "project_costs": project_costs,
        "budget_alerts": 30523.t.kuaisou.com 
        "optimization_results": optimization_results,
        "generate_time": datetime.now().isoformat()
    }

    return report


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

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