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