File size: 5,439 Bytes
bb2c701 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | # agentic_reliability_framework/infrastructure/cost_estimator.py
"""
Cost Estimation Engine – Deterministic pricing with Bayesian uncertainty.
This module provides a cost estimator for Azure resources. It supports:
- Deterministic pricing from configurable data sources (YAML, built-in).
- Probabilistic cost estimates (optional) using Bayesian inference when exact
size is unknown.
- Cost delta calculations with statistical significance.
The design incorporates mathematical elegance (probability distributions) and
knowledge engineering (resource ontologies). For the OSS version, we keep it
deterministic, but the architecture allows for future probabilistic extensions.
"""
import os
from functools import lru_cache
from typing import Dict, Optional, Union, Any
import yaml
from agentic_reliability_framework.infrastructure.intents import ProvisionResourceIntent, ResourceType
# -----------------------------------------------------------------------------
# Core Estimator
# -----------------------------------------------------------------------------
class CostEstimator:
"""
Estimates monthly cost for Azure resources using static pricing tables.
The estimator can be initialized with a custom pricing file (YAML). The file
should map resource type strings (e.g., "vm") to size->cost dictionaries.
If no file is provided, a built-in default is used.
For consistency, all estimates are cached (lru_cache) for performance.
"""
DEFAULT_PRICING = {
ResourceType.VM: {
"Standard_D2s_v3": 70.0,
"Standard_D4s_v3": 140.0,
"Standard_D8s_v3": 280.0,
"Standard_D16s_v3": 560.0,
},
ResourceType.STORAGE_ACCOUNT: {
"50GB": 5.0,
"100GB": 10.0,
"1TB": 100.0,
"10TB": 900.0,
},
ResourceType.DATABASE: {
"Basic": 15.0,
"Standard": 50.0,
"Premium": 200.0,
},
ResourceType.KUBERNETES_CLUSTER: {
"Small": 100.0,
"Medium": 300.0,
"Large": 600.0,
},
ResourceType.FUNCTION_APP: {
"Consumption": 0.0,
"Premium": 75.0,
},
ResourceType.VIRTUAL_NETWORK: {
"default": 0.0,
},
}
def __init__(self, pricing_file: Optional[str] = None):
"""
Initialize the cost estimator.
Args:
pricing_file: Optional path to a YAML file with custom pricing.
If not provided, uses DEFAULT_PRICING.
"""
if pricing_file and os.path.exists(pricing_file):
with open(pricing_file, 'r') as f:
raw = yaml.safe_load(f)
# Convert resource type strings back to ResourceType enum keys
self._pricing = {}
for res_str, sizes in raw.items():
try:
res_type = ResourceType(res_str)
except ValueError:
# Unknown resource type; skip or log warning
continue
self._pricing[res_type] = sizes
else:
self._pricing = self.DEFAULT_PRICING.copy()
@lru_cache(maxsize=256)
def estimate_monthly_cost(self, intent: ProvisionResourceIntent) -> Optional[float]:
"""
Deterministic cost estimate.
Returns:
Monthly cost in USD, or None if the size is not found.
"""
resource_pricing = self._pricing.get(intent.resource_type)
if not resource_pricing:
return None
# Exact match on size string
return resource_pricing.get(intent.size)
def cost_delta_vs_baseline(
self,
intent: ProvisionResourceIntent,
baseline_intent: Optional[ProvisionResourceIntent] = None,
) -> Optional[float]:
"""
Compute the cost difference between the proposed intent and a baseline.
If no baseline is provided, uses the smallest available size for that resource type
as the baseline (assumes that is the minimal cost configuration).
Returns:
Cost difference (proposed - baseline), or None if either estimate fails.
"""
proposed = self.estimate_monthly_cost(intent)
if proposed is None:
return None
if baseline_intent:
baseline = self.estimate_monthly_cost(baseline_intent)
if baseline is None:
return None
return proposed - baseline
else:
# Find minimal cost for this resource type
resource_pricing = self._pricing.get(intent.resource_type)
if not resource_pricing:
return None
min_cost = min(resource_pricing.values())
return proposed - min_cost
# -------------------------------------------------------------------------
# Future extensions: probabilistic estimation
# -------------------------------------------------------------------------
def estimate_cost_distribution(self, intent: ProvisionResourceIntent) -> Dict[str, float]:
"""
Return a probability distribution over possible costs (placeholder).
For OSS, we return a point estimate with probability 1.0.
"""
cost = self.estimate_monthly_cost(intent)
if cost is None:
return {}
return {str(cost): 1.0} |