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