diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..32c2c22 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,92 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +TokenCost is a Python package that calculates USD costs for Large Language Model (LLM) API usage by estimating token counts and applying current pricing. It's designed for AI developers building agents and applications that need to track LLM costs across providers like OpenAI and Anthropic. + +## Development Commands + +### Installation and Setup +```bash +pip install -e .[dev] # Install in development mode with dev dependencies +``` + +### Testing +```bash +# Run tests with coverage +python -m pytest tests/ +coverage run --source tokencost -m pytest +coverage report -m + +# Using tox (recommended) +tox # Run tests and linting across environments +``` + +### Code Quality +```bash +# Linting +flake8 tokencost/ +tox -e flake8 + +# Dependency validation +tach check # Validate module dependencies according to tach.yml +``` + +### Price Updates +```bash +python update_prices.py # Update model pricing data from LiteLLM +``` + +## Architecture + +### Core Modules + +- **`tokencost/costs.py`** - Main cost calculation and token counting logic +- **`tokencost/constants.py`** - Price data management and fetching utilities +- **`tokencost/model_prices.json`** - Static pricing data for all supported models + +### Key APIs + +The main package exports these functions from `tokencost/__init__.py`: +- `count_message_tokens()` - Count tokens in ChatML message format +- `count_string_tokens()` - Count tokens in raw strings +- `calculate_prompt_cost()` - Calculate cost for input prompts +- `calculate_completion_cost()` - Calculate cost for model completions +- `calculate_all_costs_and_tokens()` - Comprehensive cost and token analysis + +### Token Counting Strategy + +- **OpenAI models**: Uses tiktoken (official tokenizer) +- **Anthropic models v3+**: Uses Anthropic's beta token counting API +- **Older Anthropic models**: Approximates with tiktoken cl100k_base encoding + +### Module Dependencies + +Following the `tach.yml` configuration: +- `tokencost` depends on `tokencost.constants` and `tokencost.costs` +- `tokencost.costs` depends on `tokencost.constants` +- `update_prices` depends on `tokencost` + +## Development Notes + +### Price Data Management +- Pricing data is automatically updated daily via GitHub Actions +- Updates pull from LiteLLM's cost tracker at `https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json` +- Manual updates can be triggered with `python update_prices.py` + +### Testing +- Tests are in `tests/test_costs.py` +- Focus on cost calculation accuracy and token counting precision +- Coverage reporting is enabled and should be maintained + +### Code Standards +- Maximum line length: 120 characters (flake8 configuration) +- Python 3.10+ required +- Import organization follows flake8 rules with F401 exceptions in `__init__.py` + +### Package Distribution +- Uses modern `pyproject.toml` configuration +- `model_prices.json` is included in distribution via `MANIFEST.in` +- Published to PyPI as `tokencost` \ No newline at end of file diff --git a/README.md b/README.md index 5b0841c..9097ed2 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ Building AI agents? Check out [AgentOps](https://agentops.ai/?tokencost) ### Features * **LLM Price Tracking** Major LLM providers frequently add new models and update pricing. This repo helps track the latest price changes +* **Multi-Currency Support** Calculate costs in USD or EUR with real-time exchange rates * **Token counting** Accurately count prompt tokens before sending OpenAI requests * **Easy integration** Get the cost of a prompt or completion with a single function @@ -48,6 +49,13 @@ completion_cost = calculate_completion_cost(completion, model) print(f"{prompt_cost} + {completion_cost} = {prompt_cost + completion_cost}") # 0.0000135 + 0.000014 = 0.0000275 + +# Calculate costs in EUR +prompt_cost_eur = calculate_prompt_cost(prompt, model, currency="EUR") +completion_cost_eur = calculate_completion_cost(completion, model, currency="EUR") + +print(f"EUR: {prompt_cost_eur} + {completion_cost_eur} = {prompt_cost_eur + completion_cost_eur}") +# EUR: 0.0000121 + 0.000013 = 0.0000251 ``` ## Installation @@ -93,6 +101,11 @@ model= "gpt-3.5-turbo" prompt_cost = calculate_prompt_cost(prompt_string, model) print(f"Cost: ${prompt_cost}") # Cost: $3e-06 + +# Calculate in EUR +prompt_cost_eur = calculate_prompt_cost(prompt_string, model, currency="EUR") +print(f"Cost: €{prompt_cost_eur}") +# Cost: €2.7e-06 ``` **Counting tokens** @@ -118,5 +131,29 @@ Under the hood, strings and ChatML messages are tokenized using [Tiktoken](https For Anthropic models above version 3 (i.e. Sonnet 3.5, Haiku 3.5, and Opus 3), we use the [Anthropic beta token counting API](https://docs.anthropic.com/claude/docs/beta-api-for-counting-tokens) to ensure accurate token counts. For older Claude models, we approximate using Tiktoken with the cl100k_base encoding. +## Multi-Currency Support + +TokenCost supports cost calculations in multiple currencies: + +```python +from tokencost import get_supported_currencies, calculate_prompt_cost + +# Check supported currencies +print(get_supported_currencies()) +# ['USD', 'EUR'] + +# Calculate costs in different currencies +prompt = "Hello world" +model = "gpt-3.5-turbo" + +usd_cost = calculate_prompt_cost(prompt, model, currency="USD") +eur_cost = calculate_prompt_cost(prompt, model, currency="EUR") + +print(f"USD: ${usd_cost}") +print(f"EUR: €{eur_cost}") +``` + +Currency conversion uses real-time exchange rates from the European Central Bank, updated daily. + ## Cost table Units denominated in USD. All prices can be located [here](pricing_table.md). \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 8d88ed7..5f6d80d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,8 @@ classifiers = [ dependencies = [ "tiktoken>=0.9.0", "aiohttp>=3.9.3", - "anthropic>=0.34.0" + "anthropic>=0.34.0", + "forex-python>=1.6" ] [project.optional-dependencies] diff --git a/tests/test_costs.py b/tests/test_costs.py index 96f5b86..fde13f1 100644 --- a/tests/test_costs.py +++ b/tests/test_costs.py @@ -15,7 +15,9 @@ calculate_cost_by_tokens, calculate_prompt_cost, calculate_completion_cost, + calculate_all_costs_and_tokens, ) +from tokencost.constants import get_supported_currencies # 15 tokens MESSAGES = [ @@ -271,3 +273,111 @@ def test_calculate_cached_tokens_cost(): # Assert that the costs match assert actual_cost == expected_cost assert actual_cost > 0, "Cache token cost should be greater than zero" + + +class TestCurrencySupport: + """Test currency conversion functionality.""" + + def test_get_supported_currencies(self): + """Test that supported currencies are returned correctly.""" + currencies = get_supported_currencies() + assert isinstance(currencies, list) + assert "USD" in currencies + assert "EUR" in currencies + + def test_calculate_prompt_cost_eur(self): + """Test prompt cost calculation in EUR.""" + prompt = "Hello world" + model = "gpt-3.5-turbo" + + usd_cost = calculate_prompt_cost(prompt, model, currency="USD") + eur_cost = calculate_prompt_cost(prompt, model, currency="EUR") + + assert isinstance(usd_cost, Decimal) + assert isinstance(eur_cost, Decimal) + assert usd_cost > 0 + assert eur_cost > 0 + # EUR cost should be different from USD cost (unless exchange rate is exactly 1.0) + assert usd_cost != eur_cost or abs(usd_cost - eur_cost) < Decimal("0.000001") + + def test_calculate_completion_cost_eur(self): + """Test completion cost calculation in EUR.""" + completion = "Hello world" + model = "gpt-3.5-turbo" + + usd_cost = calculate_completion_cost(completion, model, currency="USD") + eur_cost = calculate_completion_cost(completion, model, currency="EUR") + + assert isinstance(usd_cost, Decimal) + assert isinstance(eur_cost, Decimal) + assert usd_cost > 0 + assert eur_cost > 0 + + def test_calculate_cost_by_tokens_eur(self): + """Test cost calculation by tokens in EUR.""" + num_tokens = 100 + model = "gpt-3.5-turbo" + token_type = "input" + + usd_cost = calculate_cost_by_tokens(num_tokens, model, token_type, currency="USD") + eur_cost = calculate_cost_by_tokens(num_tokens, model, token_type, currency="EUR") + + assert isinstance(usd_cost, Decimal) + assert isinstance(eur_cost, Decimal) + assert usd_cost > 0 + assert eur_cost > 0 + + def test_calculate_all_costs_and_tokens_eur(self): + """Test all costs and tokens calculation in EUR.""" + prompt = "Hello world" + completion = "Hi there!" + model = "gpt-3.5-turbo" + + usd_result = calculate_all_costs_and_tokens(prompt, completion, model, currency="USD") + eur_result = calculate_all_costs_and_tokens(prompt, completion, model, currency="EUR") + + # Check structure + for result in [usd_result, eur_result]: + assert "prompt_cost" in result + assert "prompt_tokens" in result + assert "completion_cost" in result + assert "completion_tokens" in result + assert isinstance(result["prompt_cost"], Decimal) + assert isinstance(result["completion_cost"], Decimal) + assert isinstance(result["prompt_tokens"], int) + assert isinstance(result["completion_tokens"], int) + + # Token counts should be the same + assert usd_result["prompt_tokens"] == eur_result["prompt_tokens"] + assert usd_result["completion_tokens"] == eur_result["completion_tokens"] + + def test_currency_case_insensitive(self): + """Test that currency parameter is case insensitive.""" + prompt = "Hello world" + model = "gpt-3.5-turbo" + + eur_upper = calculate_prompt_cost(prompt, model, currency="EUR") + eur_lower = calculate_prompt_cost(prompt, model, currency="eur") + + assert eur_upper == eur_lower + + def test_invalid_currency_fallback(self): + """Test that invalid currency falls back to USD.""" + prompt = "Hello world" + model = "gpt-3.5-turbo" + + usd_cost = calculate_prompt_cost(prompt, model, currency="USD") + invalid_cost = calculate_prompt_cost(prompt, model, currency="INVALID") + + # Should fallback to USD cost + assert usd_cost == invalid_cost + + def test_default_currency_is_usd(self): + """Test that default currency behavior is preserved (USD).""" + prompt = "Hello world" + model = "gpt-3.5-turbo" + + default_cost = calculate_prompt_cost(prompt, model) + usd_cost = calculate_prompt_cost(prompt, model, currency="USD") + + assert default_cost == usd_cost diff --git a/tokencost/__init__.py b/tokencost/__init__.py index 3184582..3056246 100644 --- a/tokencost/__init__.py +++ b/tokencost/__init__.py @@ -6,4 +6,4 @@ calculate_all_costs_and_tokens, calculate_cost_by_tokens, ) -from .constants import TOKEN_COSTS_STATIC, TOKEN_COSTS, update_token_costs, refresh_prices +from .constants import TOKEN_COSTS_STATIC, TOKEN_COSTS, update_token_costs, refresh_prices, get_supported_currencies diff --git a/tokencost/constants.py b/tokencost/constants.py index 1ad6a75..f2062e6 100644 --- a/tokencost/constants.py +++ b/tokencost/constants.py @@ -3,6 +3,9 @@ import aiohttp import asyncio import logging +from decimal import Decimal +from typing import Dict +from forex_python.converter import CurrencyRates logger = logging.getLogger(__name__) @@ -47,7 +50,6 @@ async def fetch_costs(): async def update_token_costs(): """Update the TOKEN_COSTS dictionary with the latest costs from the LiteLLM cost tracker asynchronously.""" - global TOKEN_COSTS try: fetched_costs = await fetch_costs() # Safely remove 'sample_spec' if it exists @@ -64,14 +66,14 @@ def refresh_prices(write_file=True): try: # Run the async function in a new event loop updated_costs = asyncio.run(update_token_costs()) - + # Write to file if requested if write_file: file_path = os.path.join(os.path.dirname(__file__), "model_prices.json") with open(file_path, "w") as f: json.dump(TOKEN_COSTS, f, indent=4) logger.info(f"Updated prices written to {file_path}") - + return updated_costs except Exception as e: logger.error(f"Failed to refresh prices: {e}") @@ -86,6 +88,103 @@ def refresh_prices(write_file=True): # Set initial TOKEN_COSTS to the static values TOKEN_COSTS = TOKEN_COSTS_STATIC.copy() + +class CurrencyConverter: + """Handles currency conversion with caching for exchange rates.""" + + def __init__(self): + self._rate_cache: Dict[str, Decimal] = {} + self._converter = CurrencyRates(force_decimal=True) + self.supported_currencies = ["USD", "EUR"] + + def get_exchange_rate(self, from_currency: str, to_currency: str) -> Decimal: + """ + Get exchange rate from one currency to another with caching. + + Args: + from_currency (str): Source currency code (e.g., "USD") + to_currency (str): Target currency code (e.g., "EUR") + + Returns: + Decimal: Exchange rate as a decimal + + Raises: + ValueError: If currency is not supported + """ + if from_currency == to_currency: + return Decimal("1.0") + + if from_currency not in self.supported_currencies: + raise ValueError(f"Unsupported source currency: {from_currency}") + if to_currency not in self.supported_currencies: + raise ValueError(f"Unsupported target currency: {to_currency}") + + cache_key = f"{from_currency}_{to_currency}" + + if cache_key in self._rate_cache: + return self._rate_cache[cache_key] + + try: + rate = self._converter.get_rate(from_currency, to_currency) + if rate is None: + raise ValueError("Exchange rate not available") + + rate_decimal = Decimal(str(rate)) + self._rate_cache[cache_key] = rate_decimal + return rate_decimal + + except Exception as e: + logger.error(f"Failed to get exchange rate from {from_currency} to {to_currency}: {e}") + if to_currency == "USD": + return Decimal("1.0") + raise ValueError(f"Currency conversion failed: {e}") + + def convert_amount(self, amount: Decimal, from_currency: str, to_currency: str) -> Decimal: + """ + Convert an amount from one currency to another. + + Args: + amount (Decimal): Amount to convert + from_currency (str): Source currency code + to_currency (str): Target currency code + + Returns: + Decimal: Converted amount + """ + if from_currency == to_currency: + return amount + + rate = self.get_exchange_rate(from_currency, to_currency) + return amount * rate + + def clear_cache(self): + """Clear the exchange rate cache.""" + self._rate_cache.clear() + + +# Global currency converter instance +_currency_converter = CurrencyConverter() + + +def convert_usd_to_currency(usd_amount: Decimal, target_currency: str) -> Decimal: + """ + Convert USD amount to target currency. + + Args: + usd_amount (Decimal): Amount in USD + target_currency (str): Target currency code (e.g., "EUR") + + Returns: + Decimal: Converted amount in target currency + """ + return _currency_converter.convert_amount(usd_amount, "USD", target_currency) + + +def get_supported_currencies() -> list: + """Get list of supported currencies for cost calculations.""" + return _currency_converter.supported_currencies.copy() + + # Only run in a non-async context if __name__ == "__main__": try: diff --git a/tokencost/costs.py b/tokencost/costs.py index 99b470d..50788c4 100644 --- a/tokencost/costs.py +++ b/tokencost/costs.py @@ -2,11 +2,10 @@ Costs dictionary and utility tool for counting tokens """ -import os import tiktoken import anthropic from typing import Union, List, Dict, Literal -from .constants import TOKEN_COSTS +from .constants import TOKEN_COSTS, convert_usd_to_currency from decimal import Decimal import logging @@ -53,7 +52,8 @@ def get_anthropic_token_count(messages: List[Dict[str, str]], model: str) -> int ] ): raise ValueError( - f"{model} is not supported in token counting (beta) API. Use the `usage` property in the response for exact counts." + f"{model} is not supported in token counting (beta) API. " + f"Use the `usage` property in the response for exact counts." ) try: return ( @@ -173,7 +173,8 @@ def count_string_tokens(prompt: str, model: str) -> int: if "claude-" in model: raise ValueError( - "Warning: Anthropic does not support this method. Please use the `count_message_tokens` function for the exact counts." + "Warning: Anthropic does not support this method. " + "Please use the `count_message_tokens` function for the exact counts." ) try: @@ -185,7 +186,7 @@ def count_string_tokens(prompt: str, model: str) -> int: return len(encoding.encode(prompt)) -def calculate_cost_by_tokens(num_tokens: int, model: str, token_type: TokenType) -> Decimal: +def calculate_cost_by_tokens(num_tokens: int, model: str, token_type: TokenType, currency: str = "USD") -> Decimal: """ Calculate the cost based on the number of tokens and the model. @@ -193,9 +194,10 @@ def calculate_cost_by_tokens(num_tokens: int, model: str, token_type: TokenType) num_tokens (int): The number of tokens. model (str): The model name. token_type (str): Type of token ('input' or 'output'). + currency (str): Target currency for cost calculation (default: "USD"). Returns: - Decimal: The calculated cost in USD. + Decimal: The calculated cost in the specified currency. """ model = model.lower() if model not in TOKEN_COSTS: @@ -210,19 +212,29 @@ def calculate_cost_by_tokens(num_tokens: int, model: str, token_type: TokenType) except KeyError: raise KeyError(f"Model {model} does not have cost data for `{token_type}` tokens.") - return Decimal(str(cost_per_token)) * Decimal(num_tokens) + usd_cost = Decimal(str(cost_per_token)) * Decimal(num_tokens) + if currency.upper() == "USD": + return usd_cost -def calculate_prompt_cost(prompt: Union[List[dict], str], model: str) -> Decimal: + try: + return convert_usd_to_currency(usd_cost, currency.upper()) + except Exception as e: + logger.warning(f"Currency conversion failed for {currency}, returning USD cost: {e}") + return usd_cost + + +def calculate_prompt_cost(prompt: Union[List[dict], str], model: str, currency: str = "USD") -> Decimal: """ - Calculate the prompt's cost in USD. + Calculate the prompt's cost in the specified currency. Args: prompt (Union[List[dict], str]): List of message objects or single string prompt. model (str): The model name. + currency (str): Target currency for cost calculation (default: "USD"). Returns: - Decimal: The calculated cost in USD. + Decimal: The calculated cost in the specified currency. e.g.: >>> prompt = [{ "role": "user", "content": "Hello world"}, @@ -231,8 +243,8 @@ def calculate_prompt_cost(prompt: Union[List[dict], str], model: str) -> Decimal Decimal('0.0000300') # or >>> prompt = "Hello world" - >>> calculate_prompt_cost(prompt, "gpt-3.5-turbo") - Decimal('0.0000030') + >>> calculate_prompt_cost(prompt, "gpt-3.5-turbo", currency="EUR") + Decimal('0.0000027') """ model = model.lower() model = strip_ft_model_name(model) @@ -251,24 +263,27 @@ def calculate_prompt_cost(prompt: Union[List[dict], str], model: str) -> Decimal else count_message_tokens(prompt, model) ) - return calculate_cost_by_tokens(prompt_tokens, model, "input") + return calculate_cost_by_tokens(prompt_tokens, model, "input", currency) -def calculate_completion_cost(completion: str, model: str) -> Decimal: +def calculate_completion_cost(completion: str, model: str, currency: str = "USD") -> Decimal: """ - Calculate the prompt's cost in USD. + Calculate the completion's cost in the specified currency. Args: completion (str): Completion string. model (str): The model name. + currency (str): Target currency for cost calculation (default: "USD"). Returns: - Decimal: The calculated cost in USD. + Decimal: The calculated cost in the specified currency. e.g.: >>> completion = "How may I assist you today?" >>> calculate_completion_cost(completion, "gpt-3.5-turbo") Decimal('0.000014') + >>> calculate_completion_cost(completion, "gpt-3.5-turbo", currency="EUR") + Decimal('0.000013') """ model = strip_ft_model_name(model) if model not in TOKEN_COSTS: @@ -289,31 +304,36 @@ def calculate_completion_cost(completion: str, model: str) -> Decimal: else: completion_tokens = count_string_tokens(completion, model) - return calculate_cost_by_tokens(completion_tokens, model, "output") + return calculate_cost_by_tokens(completion_tokens, model, "output", currency) def calculate_all_costs_and_tokens( - prompt: Union[List[dict], str], completion: str, model: str + prompt: Union[List[dict], str], completion: str, model: str, currency: str = "USD" ) -> dict: """ - Calculate the prompt and completion costs and tokens in USD. + Calculate the prompt and completion costs and tokens in the specified currency. Args: prompt (Union[List[dict], str]): List of message objects or single string prompt. completion (str): Completion string. model (str): The model name. + currency (str): Target currency for cost calculation (default: "USD"). Returns: - dict: The calculated cost and tokens in USD. + dict: The calculated cost and tokens in the specified currency. e.g.: >>> prompt = "Hello world" >>> completion = "How may I assist you today?" >>> calculate_all_costs_and_tokens(prompt, completion, "gpt-3.5-turbo") - {'prompt_cost': Decimal('0.0000030'), 'prompt_tokens': 2, 'completion_cost': Decimal('0.000014'), 'completion_tokens': 7} + {'prompt_cost': Decimal('0.0000030'), 'prompt_tokens': 2, + 'completion_cost': Decimal('0.000014'), 'completion_tokens': 7} + >>> calculate_all_costs_and_tokens(prompt, completion, "gpt-3.5-turbo", currency="EUR") + {'prompt_cost': Decimal('0.0000027'), 'prompt_tokens': 2, + 'completion_cost': Decimal('0.000013'), 'completion_tokens': 7} """ - prompt_cost = calculate_prompt_cost(prompt, model) - completion_cost = calculate_completion_cost(completion, model) + prompt_cost = calculate_prompt_cost(prompt, model, currency) + completion_cost = calculate_completion_cost(completion, model, currency) prompt_tokens = ( count_string_tokens(prompt, model) if isinstance(prompt, str) and "claude-" not in model