重要前提
安装AI Skills的关键前提是:必须科学上网,且开启TUN模式,这一点至关重要,直接决定安装能否顺利完成,在此郑重提醒三遍:科学上网,科学上网,科学上网。查看完整安装教程 →
finops-expert by personamanagmentlayer/pcl
npx skills add https://github.com/personamanagmentlayer/pcl --skill finops-expert为云财务运营、成本优化、资源管理和云经济学提供专家指导。
import boto3
from datetime import datetime, timedelta
from typing import Dict, List
import pandas as pd
class AWSCostAnalyzer:
"""使用 Cost Explorer API 分析 AWS 成本"""
def __init__(self):
self.ce_client = boto3.client('ce')
def get_cost_and_usage(self, start_date: str, end_date: str,
granularity: str = 'DAILY',
metrics: List[str] = None) -> Dict:
"""获取成本和用量数据"""
if metrics is None:
metrics = ['UnblendedCost', 'UsageQuantity']
response = self.ce_client.get_cost_and_usage(
TimePeriod={
'Start': start_date,
'End': end_date
},
Granularity=granularity,
Metrics=metrics,
GroupBy=[
{'Type': 'DIMENSION', 'Key': 'SERVICE'}
]
)
return response['ResultsByTime']
def get_top_services_by_cost(self, days: int = 30, top_n: int = 10) -> pd.DataFrame:
"""按成本获取排名靠前的服务"""
end_date = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
results = self.get_cost_and_usage(start_date, end_date, 'MONTHLY')
service_costs = {}
for result in results:
for group in result['Groups']:
service = group['Keys'][0]
cost = float(group['Metrics']['UnblendedCost']['Amount'])
if service in service_costs:
service_costs[service] += cost
else:
service_costs[service] = cost
df = pd.DataFrame(list(service_costs.items()),
columns=['Service', 'Cost'])
return df.nlargest(top_n, 'Cost')
def get_cost_forecast(self, days_ahead: int = 30) -> Dict:
"""获取成本预测"""
start_date = datetime.now().strftime('%Y-%m-%d')
end_date = (datetime.now() + timedelta(days=days_ahead)).strftime('%Y-%m-%d')
response = self.ce_client.get_cost_forecast(
TimePeriod={
'Start': start_date,
'End': end_date
},
Metric='UNBLENDED_COST',
Granularity='MONTHLY'
)
return {
'forecasted_cost': float(response['Total']['Amount']),
'mean_value': float(response['ForecastResultsByTime'][0]['MeanValue'])
}
def get_rightsizing_recommendations(self) -> List[Dict]:
"""获取 EC2 合理调整建议"""
response = self.ce_client.get_rightsizing_recommendation(
Service='AmazonEC2'
)
recommendations = []
for rec in response['RightsizingRecommendations']:
recommendations.append({
'instance_id': rec['CurrentInstance']['ResourceId'],
'current_type': rec['CurrentInstance']['InstanceType'],
'recommended_type': rec['ModifyRecommendationDetail']['TargetInstances'][0]['InstanceType']
if rec.get('ModifyRecommendationDetail') else None,
'estimated_savings': float(rec['EstimatedMonthlySavings']['Value'])
if rec.get('EstimatedMonthlySavings') else 0
})
return recommendations
class CostOptimizer:
"""优化云成本"""
def __init__(self):
self.ec2_client = boto3.client('ec2')
self.rds_client = boto3.client('rds')
self.s3_client = boto3.client('s3')
def find_idle_resources(self) -> Dict[str, List]:
"""查找闲置/未使用的资源"""
idle_resources = {
'ec2_instances': [],
'ebs_volumes': [],
'elastic_ips': [],
'load_balancers': []
}
# 闲置 EC2 实例(停止超过 7 天)
instances = self.ec2_client.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['stopped']}]
)
for reservation in instances['Reservations']:
for instance in reservation['Instances']:
idle_resources['ec2_instances'].append({
'id': instance['InstanceId'],
'type': instance['InstanceType'],
'state': instance['State']['Name']
})
# 未挂载的 EBS 卷
volumes = self.ec2_client.describe_volumes(
Filters=[{'Name': 'status', 'Values': ['available']}]
)
for volume in volumes['Volumes']:
idle_resources['ebs_volumes'].append({
'id': volume['VolumeId'],
'size': volume['Size'],
'type': volume['VolumeType']
})
# 未关联的弹性 IP
addresses = self.ec2_client.describe_addresses()
for address in addresses['Addresses']:
if 'InstanceId' not in address:
idle_resources['elastic_ips'].append({
'allocation_id': address['AllocationId'],
'public_ip': address['PublicIp']
})
return idle_resources
def calculate_reserved_instance_savings(self,
instance_type: str,
count: int,
term: int = 1) -> Dict:
"""计算预留实例节省金额"""
# 简化计算(实际应使用定价 API)
on_demand_hourly = self._get_on_demand_price(instance_type)
ri_hourly = on_demand_hourly * 0.65 # 约 35% 折扣
hours_per_year = 24 * 365
annual_on_demand = on_demand_hourly * hours_per_year * count
annual_ri = ri_hourly * hours_per_year * count
return {
'instance_type': instance_type,
'count': count,
'annual_on_demand_cost': annual_on_demand,
'annual_ri_cost': annual_ri,
'annual_savings': annual_on_demand - annual_ri,
'savings_percentage': ((annual_on_demand - annual_ri) / annual_on_demand) * 100
}
def _get_on_demand_price(self, instance_type: str) -> float:
"""获取按需小时价格(简化版)"""
# 生产环境中,应使用 AWS 定价 API
prices = {
't3.micro': 0.0104,
't3.small': 0.0208,
't3.medium': 0.0416,
'm5.large': 0.096,
'm5.xlarge': 0.192
}
return prices.get(instance_type, 0.10)
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
class CostAllocation:
"""使用标签管理成本分摊"""
def __init__(self):
self.ec2_client = boto3.client('ec2')
self.ce_client = boto3.client('ce')
def define_tagging_strategy(self) -> Dict[str, List[str]]:
"""定义强制标签"""
return {
'environment': ['prod', 'staging', 'dev'],
'team': ['engineering', 'data', 'product'],
'cost_center': ['CC001', 'CC002', 'CC003'],
'project': ['project-a', 'project-b'],
'owner': ['email addresses']
}
def audit_resource_tags(self, resource_type: str = 'instance') -> List[Dict]:
"""审计资源缺失的标签"""
mandatory_tags = ['environment', 'team', 'cost_center']
untagged_resources = []
if resource_type == 'instance':
instances = self.ec2_client.describe_instances()
for reservation in instances['Reservations']:
for instance in reservation['Instances']:
tags = {tag['Key']: tag['Value']
for tag in instance.get('Tags', [])}
missing_tags = [tag for tag in mandatory_tags
if tag not in tags]
if missing_tags:
untagged_resources.append({
'resource_id': instance['InstanceId'],
'missing_tags': missing_tags
})
return untagged_resources
def get_cost_by_tag(self, tag_key: str, start_date: str,
end_date: str) -> pd.DataFrame:
"""按标签分组获取成本"""
response = self.ce_client.get_cost_and_usage(
TimePeriod={
'Start': start_date,
'End': end_date
},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[
{'Type': 'TAG', 'Key': tag_key}
]
)
costs = []
for result in response['ResultsByTime']:
for group in result['Groups']:
costs.append({
'tag_value': group['Keys'][0].split('$')[1]
if '$' in group['Keys'][0] else 'Untagged',
'cost': float(group['Metrics']['UnblendedCost']['Amount'])
})
return pd.DataFrame(costs)
class BudgetManager:
"""管理 AWS 预算和告警"""
def __init__(self):
self.budgets_client = boto3.client('budgets')
self.account_id = boto3.client('sts').get_caller_identity()['Account']
def create_monthly_budget(self, name: str, amount: float,
email: str) -> Dict:
"""创建月度成本预算并设置告警"""
budget = {
'BudgetName': name,
'BudgetLimit': {
'Amount': str(amount),
'Unit': 'USD'
},
'TimeUnit': 'MONTHLY',
'BudgetType': 'COST'
}
# 在 80% 和 100% 时告警
notifications = [
{
'Notification': {
'NotificationType': 'ACTUAL',
'ComparisonOperator': 'GREATER_THAN',
'Threshold': 80,
'ThresholdType': 'PERCENTAGE'
},
'Subscribers': [{
'SubscriptionType': 'EMAIL',
'Address': email
}]
},
{
'Notification': {
'NotificationType': 'ACTUAL',
'ComparisonOperator': 'GREATER_THAN',
'Threshold': 100,
'ThresholdType': 'PERCENTAGE'
},
'Subscribers': [{
'SubscriptionType': 'EMAIL',
'Address': email
}]
}
]
response = self.budgets_client.create_budget(
AccountId=self.account_id,
Budget=budget,
NotificationsWithSubscribers=notifications
)
return response
❌ 无标签策略 ❌ 忽略合理调整建议 ❌ 不使用预留实例 ❌ 无预算告警 ❌ 保留闲置资源 ❌ 手动成本跟踪 ❌ 成本责任孤岛
每周安装量
53
代码仓库
GitHub 星标数
11
首次出现
2026年1月24日
安全审计
已安装于
codex45
opencode44
gemini-cli42
cursor39
github-copilot38
amp35
Expert guidance for cloud financial operations, cost optimization, resource management, and cloud economics.
import boto3
from datetime import datetime, timedelta
from typing import Dict, List
import pandas as pd
class AWSCostAnalyzer:
"""Analyze AWS costs using Cost Explorer API"""
def __init__(self):
self.ce_client = boto3.client('ce')
def get_cost_and_usage(self, start_date: str, end_date: str,
granularity: str = 'DAILY',
metrics: List[str] = None) -> Dict:
"""Get cost and usage data"""
if metrics is None:
metrics = ['UnblendedCost', 'UsageQuantity']
response = self.ce_client.get_cost_and_usage(
TimePeriod={
'Start': start_date,
'End': end_date
},
Granularity=granularity,
Metrics=metrics,
GroupBy=[
{'Type': 'DIMENSION', 'Key': 'SERVICE'}
]
)
return response['ResultsByTime']
def get_top_services_by_cost(self, days: int = 30, top_n: int = 10) -> pd.DataFrame:
"""Get top services by cost"""
end_date = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
results = self.get_cost_and_usage(start_date, end_date, 'MONTHLY')
service_costs = {}
for result in results:
for group in result['Groups']:
service = group['Keys'][0]
cost = float(group['Metrics']['UnblendedCost']['Amount'])
if service in service_costs:
service_costs[service] += cost
else:
service_costs[service] = cost
df = pd.DataFrame(list(service_costs.items()),
columns=['Service', 'Cost'])
return df.nlargest(top_n, 'Cost')
def get_cost_forecast(self, days_ahead: int = 30) -> Dict:
"""Get cost forecast"""
start_date = datetime.now().strftime('%Y-%m-%d')
end_date = (datetime.now() + timedelta(days=days_ahead)).strftime('%Y-%m-%d')
response = self.ce_client.get_cost_forecast(
TimePeriod={
'Start': start_date,
'End': end_date
},
Metric='UNBLENDED_COST',
Granularity='MONTHLY'
)
return {
'forecasted_cost': float(response['Total']['Amount']),
'mean_value': float(response['ForecastResultsByTime'][0]['MeanValue'])
}
def get_rightsizing_recommendations(self) -> List[Dict]:
"""Get EC2 rightsizing recommendations"""
response = self.ce_client.get_rightsizing_recommendation(
Service='AmazonEC2'
)
recommendations = []
for rec in response['RightsizingRecommendations']:
recommendations.append({
'instance_id': rec['CurrentInstance']['ResourceId'],
'current_type': rec['CurrentInstance']['InstanceType'],
'recommended_type': rec['ModifyRecommendationDetail']['TargetInstances'][0]['InstanceType']
if rec.get('ModifyRecommendationDetail') else None,
'estimated_savings': float(rec['EstimatedMonthlySavings']['Value'])
if rec.get('EstimatedMonthlySavings') else 0
})
return recommendations
class CostOptimizer:
"""Optimize cloud costs"""
def __init__(self):
self.ec2_client = boto3.client('ec2')
self.rds_client = boto3.client('rds')
self.s3_client = boto3.client('s3')
def find_idle_resources(self) -> Dict[str, List]:
"""Find idle/unused resources"""
idle_resources = {
'ec2_instances': [],
'ebs_volumes': [],
'elastic_ips': [],
'load_balancers': []
}
# Idle EC2 instances (stopped for > 7 days)
instances = self.ec2_client.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['stopped']}]
)
for reservation in instances['Reservations']:
for instance in reservation['Instances']:
idle_resources['ec2_instances'].append({
'id': instance['InstanceId'],
'type': instance['InstanceType'],
'state': instance['State']['Name']
})
# Unattached EBS volumes
volumes = self.ec2_client.describe_volumes(
Filters=[{'Name': 'status', 'Values': ['available']}]
)
for volume in volumes['Volumes']:
idle_resources['ebs_volumes'].append({
'id': volume['VolumeId'],
'size': volume['Size'],
'type': volume['VolumeType']
})
# Unattached Elastic IPs
addresses = self.ec2_client.describe_addresses()
for address in addresses['Addresses']:
if 'InstanceId' not in address:
idle_resources['elastic_ips'].append({
'allocation_id': address['AllocationId'],
'public_ip': address['PublicIp']
})
return idle_resources
def calculate_reserved_instance_savings(self,
instance_type: str,
count: int,
term: int = 1) -> Dict:
"""Calculate RI savings"""
# Simplified calculation (would use actual pricing API)
on_demand_hourly = self._get_on_demand_price(instance_type)
ri_hourly = on_demand_hourly * 0.65 # ~35% discount
hours_per_year = 24 * 365
annual_on_demand = on_demand_hourly * hours_per_year * count
annual_ri = ri_hourly * hours_per_year * count
return {
'instance_type': instance_type,
'count': count,
'annual_on_demand_cost': annual_on_demand,
'annual_ri_cost': annual_ri,
'annual_savings': annual_on_demand - annual_ri,
'savings_percentage': ((annual_on_demand - annual_ri) / annual_on_demand) * 100
}
def _get_on_demand_price(self, instance_type: str) -> float:
"""Get on-demand hourly price (simplified)"""
# In production, use AWS Pricing API
prices = {
't3.micro': 0.0104,
't3.small': 0.0208,
't3.medium': 0.0416,
'm5.large': 0.096,
'm5.xlarge': 0.192
}
return prices.get(instance_type, 0.10)
class CostAllocation:
"""Manage cost allocation with tags"""
def __init__(self):
self.ec2_client = boto3.client('ec2')
self.ce_client = boto3.client('ce')
def define_tagging_strategy(self) -> Dict[str, List[str]]:
"""Define mandatory tags"""
return {
'environment': ['prod', 'staging', 'dev'],
'team': ['engineering', 'data', 'product'],
'cost_center': ['CC001', 'CC002', 'CC003'],
'project': ['project-a', 'project-b'],
'owner': ['email addresses']
}
def audit_resource_tags(self, resource_type: str = 'instance') -> List[Dict]:
"""Audit resources for missing tags"""
mandatory_tags = ['environment', 'team', 'cost_center']
untagged_resources = []
if resource_type == 'instance':
instances = self.ec2_client.describe_instances()
for reservation in instances['Reservations']:
for instance in reservation['Instances']:
tags = {tag['Key']: tag['Value']
for tag in instance.get('Tags', [])}
missing_tags = [tag for tag in mandatory_tags
if tag not in tags]
if missing_tags:
untagged_resources.append({
'resource_id': instance['InstanceId'],
'missing_tags': missing_tags
})
return untagged_resources
def get_cost_by_tag(self, tag_key: str, start_date: str,
end_date: str) -> pd.DataFrame:
"""Get costs grouped by tag"""
response = self.ce_client.get_cost_and_usage(
TimePeriod={
'Start': start_date,
'End': end_date
},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[
{'Type': 'TAG', 'Key': tag_key}
]
)
costs = []
for result in response['ResultsByTime']:
for group in result['Groups']:
costs.append({
'tag_value': group['Keys'][0].split('$')[1]
if '$' in group['Keys'][0] else 'Untagged',
'cost': float(group['Metrics']['UnblendedCost']['Amount'])
})
return pd.DataFrame(costs)
class BudgetManager:
"""Manage AWS budgets and alerts"""
def __init__(self):
self.budgets_client = boto3.client('budgets')
self.account_id = boto3.client('sts').get_caller_identity()['Account']
def create_monthly_budget(self, name: str, amount: float,
email: str) -> Dict:
"""Create monthly cost budget with alerts"""
budget = {
'BudgetName': name,
'BudgetLimit': {
'Amount': str(amount),
'Unit': 'USD'
},
'TimeUnit': 'MONTHLY',
'BudgetType': 'COST'
}
# Alert at 80% and 100%
notifications = [
{
'Notification': {
'NotificationType': 'ACTUAL',
'ComparisonOperator': 'GREATER_THAN',
'Threshold': 80,
'ThresholdType': 'PERCENTAGE'
},
'Subscribers': [{
'SubscriptionType': 'EMAIL',
'Address': email
}]
},
{
'Notification': {
'NotificationType': 'ACTUAL',
'ComparisonOperator': 'GREATER_THAN',
'Threshold': 100,
'ThresholdType': 'PERCENTAGE'
},
'Subscribers': [{
'SubscriptionType': 'EMAIL',
'Address': email
}]
}
]
response = self.budgets_client.create_budget(
AccountId=self.account_id,
Budget=budget,
NotificationsWithSubscribers=notifications
)
return response
❌ No tagging strategy ❌ Ignoring rightsizing recommendations ❌ Not using Reserved Instances ❌ No budget alerts ❌ Keeping idle resources ❌ Manual cost tracking ❌ Siloed cost responsibility
Weekly Installs
53
Repository
GitHub Stars
11
First Seen
Jan 24, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
codex45
opencode44
gemini-cli42
cursor39
github-copilot38
amp35
Supabase Postgres 最佳实践指南 - 8大类别性能优化规则与SQL示例
86,500 周安装