Coverage for src / local_deep_research / metrics / pricing / pricing_fetcher.py: 91%
107 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +0000
1"""
2LLM Pricing Data Fetcher
4Fetches real-time pricing data from various LLM providers.
5Supports multiple providers and fallback to static pricing.
6"""
8from typing import Any, Dict, Optional
10import aiohttp
11from loguru import logger
13from ...llm.providers.base import normalize_provider
16class PricingFetcher:
17 """Fetches LLM pricing data from various sources."""
19 def __init__(self):
20 self.session = None
21 self.static_pricing = self._load_static_pricing()
23 async def __aenter__(self):
24 self.session = aiohttp.ClientSession()
25 return self
27 async def __aexit__(self, exc_type, exc_val, exc_tb):
28 if self.session:
29 await self.session.close()
31 def _load_static_pricing(self) -> Dict[str, Dict[str, float]]:
32 """Load static pricing as fallback (per 1K tokens in USD)."""
33 return {
34 # OpenAI Models
35 "gpt-4": {"prompt": 0.03, "completion": 0.06},
36 "gpt-4-turbo": {"prompt": 0.01, "completion": 0.03},
37 "gpt-4o": {"prompt": 0.005, "completion": 0.015},
38 "gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
39 "gpt-3.5-turbo": {"prompt": 0.001, "completion": 0.002},
40 # Anthropic Models
41 "claude-3-opus": {"prompt": 0.015, "completion": 0.075},
42 "claude-3-sonnet": {"prompt": 0.003, "completion": 0.015},
43 "claude-3-haiku": {"prompt": 0.00025, "completion": 0.00125},
44 "claude-3-5-sonnet": {"prompt": 0.003, "completion": 0.015},
45 # Google Models
46 "gemini-pro": {"prompt": 0.0005, "completion": 0.0015},
47 "gemini-pro-vision": {"prompt": 0.0005, "completion": 0.0015},
48 "gemini-1.5-pro": {"prompt": 0.0035, "completion": 0.0105},
49 "gemini-1.5-flash": {"prompt": 0.00035, "completion": 0.00105},
50 # Local/Open Source (free)
51 "ollama": {"prompt": 0.0, "completion": 0.0},
52 "llama": {"prompt": 0.0, "completion": 0.0},
53 "mistral": {"prompt": 0.0, "completion": 0.0},
54 "gemma": {"prompt": 0.0, "completion": 0.0},
55 "qwen": {"prompt": 0.0, "completion": 0.0},
56 "codellama": {"prompt": 0.0, "completion": 0.0},
57 "vicuna": {"prompt": 0.0, "completion": 0.0},
58 "alpaca": {"prompt": 0.0, "completion": 0.0},
59 "lmstudio": {"prompt": 0.0, "completion": 0.0},
60 "llamacpp": {"prompt": 0.0, "completion": 0.0},
61 }
63 async def fetch_openai_pricing(self) -> Optional[Dict[str, Any]]:
64 """Fetch OpenAI pricing from their API (if available)."""
65 try:
66 # Note: OpenAI doesn't have a public pricing API
67 # This would need to be web scraping or manual updates
68 logger.info("Using static OpenAI pricing (no public API available)")
69 return None
70 except Exception:
71 logger.warning("Failed to fetch OpenAI pricing")
72 return None
74 async def fetch_anthropic_pricing(self) -> Optional[Dict[str, Any]]:
75 """Fetch Anthropic pricing."""
76 try:
77 # Note: Anthropic doesn't have a public pricing API
78 # This would need to be web scraping or manual updates
79 logger.info(
80 "Using static Anthropic pricing (no public API available)"
81 )
82 return None
83 except Exception:
84 logger.warning("Failed to fetch Anthropic pricing")
85 return None
87 async def fetch_google_pricing(self) -> Optional[Dict[str, Any]]:
88 """Fetch Google/Gemini pricing."""
89 try:
90 # Note: Google doesn't have a dedicated pricing API for individual models
91 # This would need to be web scraping or manual updates
92 logger.info("Using static Google pricing (no public API available)")
93 return None
94 except Exception:
95 logger.warning("Failed to fetch Google pricing")
96 return None
98 async def fetch_huggingface_pricing(self) -> Optional[Dict[str, Any]]:
99 """Fetch HuggingFace Inference API pricing."""
100 try:
101 if not self.session:
102 return None
104 # HuggingFace has some pricing info but not a structured API
105 # This is more for hosted inference endpoints
106 url = "https://huggingface.co/pricing"
107 async with self.session.get(url) as response:
108 if response.status == 200:
109 # Would need to parse HTML for pricing info
110 logger.info(
111 "HuggingFace pricing would require HTML parsing"
112 )
113 return None
114 except Exception:
115 logger.warning("Failed to fetch HuggingFace pricing")
116 return None
118 async def get_model_pricing(
119 self, model_name: str, provider: str = None
120 ) -> Optional[Dict[str, float]]:
121 """Get pricing for a specific model and provider."""
122 # Normalize inputs
123 model_name = model_name.lower() if model_name else ""
124 provider = normalize_provider(provider) or ""
126 # Provider-first approach: Check if provider indicates local/free models
127 local_providers = ["ollama", "lmstudio", "llamacpp"]
128 if provider in local_providers:
129 logger.debug(
130 f"Local provider '{provider}' detected - returning zero cost"
131 )
132 return {"prompt": 0.0, "completion": 0.0}
134 # Try to fetch live pricing first (most providers don't have APIs)
135 if (
136 provider == "openai"
137 or "gpt" in model_name
138 or "openai" in model_name
139 ):
140 await self.fetch_openai_pricing()
141 elif (
142 provider == "anthropic"
143 or "claude" in model_name
144 or "anthropic" in model_name
145 ):
146 await self.fetch_anthropic_pricing()
147 elif (
148 provider == "google"
149 or "gemini" in model_name
150 or "google" in model_name
151 ):
152 await self.fetch_google_pricing()
154 # Fallback to static pricing with provider priority
155 if provider:
156 # First try provider-specific lookup with exact matching
157 provider_models = self._get_models_by_provider(provider)
158 # Try exact match
159 if model_name in provider_models:
160 return provider_models[model_name]
161 # Try exact match without provider prefix
162 if "/" in model_name: 162 ↛ 169line 162 didn't jump to line 169 because the condition on line 162 was always true
163 model_only = model_name.split("/")[-1]
164 if model_only in provider_models: 164 ↛ 169line 164 didn't jump to line 169 because the condition on line 164 was always true
165 return provider_models[model_only]
167 # Exact model name matching only
168 # First try exact match
169 if model_name in self.static_pricing:
170 return self.static_pricing[model_name]
172 # Try exact match without provider prefix (e.g., "openai/gpt-4o-mini" -> "gpt-4o-mini")
173 if "/" in model_name:
174 model_only = model_name.split("/")[-1]
175 if model_only in self.static_pricing: 175 ↛ 179line 175 didn't jump to line 179 because the condition on line 175 was always true
176 return self.static_pricing[model_only]
178 # No pricing found - return None instead of default pricing
179 logger.warning(
180 f"No pricing found for model: {model_name}, provider: {provider}"
181 )
182 return None
184 def _get_models_by_provider(
185 self, provider: str
186 ) -> Dict[str, Dict[str, float]]:
187 """Get models for a specific provider."""
188 provider = normalize_provider(provider) or ""
189 provider_models = {}
191 if provider == "openai":
192 provider_models = {
193 k: v
194 for k, v in self.static_pricing.items()
195 if k.startswith("gpt")
196 }
197 elif provider == "anthropic":
198 provider_models = {
199 k: v
200 for k, v in self.static_pricing.items()
201 if k.startswith("claude")
202 }
203 elif provider == "google":
204 provider_models = {
205 k: v
206 for k, v in self.static_pricing.items()
207 if k.startswith("gemini")
208 }
209 elif provider in ["ollama", "lmstudio", "llamacpp"]:
210 # All local models are free
211 provider_models = {
212 k: v
213 for k, v in self.static_pricing.items()
214 if v["prompt"] == 0.0 and v["completion"] == 0.0
215 }
217 return provider_models
219 async def get_all_pricing(self) -> Dict[str, Dict[str, float]]:
220 """Get pricing for all known models."""
221 # In the future, this could aggregate from multiple live sources
222 return self.static_pricing.copy()
224 def get_provider_from_model(self, model_name: str) -> str:
225 """Determine the provider from model name."""
226 model_name = model_name.lower()
228 if "gpt" in model_name or "openai" in model_name:
229 return "openai"
230 if "claude" in model_name or "anthropic" in model_name:
231 return "anthropic"
232 if "gemini" in model_name or "google" in model_name:
233 return "google"
234 if "llama" in model_name or "meta" in model_name:
235 return "meta"
236 if "mistral" in model_name:
237 return "mistral"
238 if "ollama" in model_name: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 return "ollama"
240 return "unknown"