ccproxy.pricing

ccproxy.pricing

Dynamic pricing system for Claude models.

This module provides dynamic pricing capabilities by downloading and caching pricing information from external sources like LiteLLM.

PricingCache

PricingCache(settings)

Manages caching of model pricing data from external sources.

Parameters:

Name	Type	Description	Default
settings	PricingSettings	Pricing configuration settings	required

Source code in ccproxy/pricing/cache.py

def __init__(self, settings: PricingSettings) -> None:
    """Initialize pricing cache.

    Args:
        settings: Pricing configuration settings
    """
    self.settings = settings
    self.cache_dir = settings.cache_dir
    self.cache_file = self.cache_dir / "model_pricing.json"

    # Ensure cache directory exists
    self.cache_dir.mkdir(parents=True, exist_ok=True)

is_cache_valid

is_cache_valid()

Check if cached pricing data is still valid.

Returns:

Type	Description
bool	True if cache exists and is not expired

Source code in ccproxy/pricing/cache.py

def is_cache_valid(self) -> bool:
    """Check if cached pricing data is still valid.

    Returns:
        True if cache exists and is not expired
    """
    if not self.cache_file.exists():
        return False

    try:
        stat = self.cache_file.stat()
        age_seconds = time.time() - stat.st_mtime
        age_hours = age_seconds / 3600

        is_valid = age_hours < self.settings.cache_ttl_hours
        return is_valid

    except OSError as e:
        logger.warning("cache_stats_check_failed", error=str(e))
        return False

load_cached_data

load_cached_data()

Load pricing data from cache.

Returns:

Type	Description
dict[str, Any] | None	Cached pricing data or None if cache is invalid/corrupted

Source code in ccproxy/pricing/cache.py

def load_cached_data(self) -> dict[str, Any] | None:
    """Load pricing data from cache.

    Returns:
        Cached pricing data or None if cache is invalid/corrupted
    """
    if not self.is_cache_valid():
        return None

    try:
        with self.cache_file.open(encoding="utf-8") as f:
            data = json.load(f)

        return data  # type: ignore[no-any-return]

    except (OSError, json.JSONDecodeError) as e:
        logger.warning("cache_load_failed", error=str(e))
        return None

download_pricing_data async

download_pricing_data(timeout=None)

Download fresh pricing data from source URL.

Parameters:

Name	Type	Description	Default
timeout	int | None	Request timeout in seconds (uses settings default if None)	None

Returns:

Type	Description
dict[str, Any] | None	Downloaded pricing data or None if download failed

Source code in ccproxy/pricing/cache.py

async def download_pricing_data(
    self, timeout: int | None = None
) -> dict[str, Any] | None:
    """Download fresh pricing data from source URL.

    Args:
        timeout: Request timeout in seconds (uses settings default if None)

    Returns:
        Downloaded pricing data or None if download failed
    """
    if timeout is None:
        timeout = self.settings.download_timeout

    try:
        logger.info("pricing_download_start", url=self.settings.source_url)

        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.get(self.settings.source_url)
            response.raise_for_status()

            data = response.json()
            logger.info("pricing_download_completed", model_count=len(data))
            return data  # type: ignore[no-any-return]

    except (httpx.HTTPError, json.JSONDecodeError) as e:
        logger.error("pricing_download_failed", error=str(e))
        return None

save_to_cache

save_to_cache(data)

Save pricing data to cache.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Pricing data to cache	required

Returns:

Type	Description
bool	True if successfully saved, False otherwise

Source code in ccproxy/pricing/cache.py

def save_to_cache(self, data: dict[str, Any]) -> bool:
    """Save pricing data to cache.

    Args:
        data: Pricing data to cache

    Returns:
        True if successfully saved, False otherwise
    """
    try:
        # Write to temporary file first, then atomic rename
        temp_file = self.cache_file.with_suffix(".tmp")

        with temp_file.open("w", encoding="utf-8") as f:
            json.dump(data, f, indent=2)

        # Atomic rename
        temp_file.replace(self.cache_file)

        return True

    except OSError as e:
        logger.error("cache_save_failed", error=str(e))
        return False

get_pricing_data async

get_pricing_data(force_refresh=False)

Get pricing data, from cache if valid or by downloading fresh data.

Parameters:

Name	Type	Description	Default
force_refresh	bool	Force download even if cache is valid	False

Returns:

Type	Description
dict[str, Any] | None	Pricing data or None if both cache and download fail

Source code in ccproxy/pricing/cache.py

async def get_pricing_data(
    self, force_refresh: bool = False
) -> dict[str, Any] | None:
    """Get pricing data, from cache if valid or by downloading fresh data.

    Args:
        force_refresh: Force download even if cache is valid

    Returns:
        Pricing data or None if both cache and download fail
    """
    # Try cache first unless forced refresh
    if not force_refresh:
        cached_data = self.load_cached_data()
        if cached_data is not None:
            return cached_data

    # Download fresh data
    fresh_data = await self.download_pricing_data()
    if fresh_data is not None:
        # Save to cache for next time
        self.save_to_cache(fresh_data)
        return fresh_data

    # If download failed, try to use stale cache as fallback
    if not force_refresh:
        logger.warning("pricing_download_failed_using_stale_cache")
        try:
            with self.cache_file.open(encoding="utf-8") as f:
                stale_data = json.load(f)
            logger.warning("stale_cache_used")
            return stale_data  # type: ignore[no-any-return]
        except (OSError, json.JSONDecodeError):
            pass

    logger.error("pricing_data_unavailable")
    return None

clear_cache

clear_cache()

Clear cached pricing data.

Returns:

Type	Description
bool	True if cache was cleared successfully

Source code in ccproxy/pricing/cache.py

def clear_cache(self) -> bool:
    """Clear cached pricing data.

    Returns:
        True if cache was cleared successfully
    """
    try:
        if self.cache_file.exists():
            self.cache_file.unlink()
        return True
    except OSError as e:
        logger.error("cache_clear_failed", error=str(e))
        return False

get_cache_info

get_cache_info()

Get information about cache status.

Returns:

Type	Description
dict[str, Any]	Dictionary with cache information

Source code in ccproxy/pricing/cache.py

def get_cache_info(self) -> dict[str, Any]:
    """Get information about cache status.

    Returns:
        Dictionary with cache information
    """
    info = {
        "cache_file": str(self.cache_file),
        "cache_dir": str(self.cache_dir),
        "source_url": self.settings.source_url,
        "ttl_hours": self.settings.cache_ttl_hours,
        "exists": self.cache_file.exists(),
        "valid": False,
        "age_hours": None,
        "size_bytes": None,
    }

    if self.cache_file.exists():
        try:
            stat = self.cache_file.stat()
            age_seconds = time.time() - stat.st_mtime
            age_hours = age_seconds / 3600

            info.update(
                {
                    "valid": age_hours < self.settings.cache_ttl_hours,
                    "age_hours": age_hours,
                    "size_bytes": stat.st_size,
                }
            )
        except OSError:
            pass

    return info

PricingLoader

Loads and converts pricing data from LiteLLM format to internal format.

extract_claude_models staticmethod

extract_claude_models(litellm_data, verbose=True)

Extract Claude model entries from LiteLLM data.

Parameters:

Name	Type	Description	Default
litellm_data	dict[str, Any]	Raw LiteLLM pricing data	required
verbose	bool	Whether to log individual model discoveries	True

Returns:

Type	Description
dict[str, Any]	Dictionary with only Claude models

Source code in ccproxy/pricing/loader.py

@staticmethod
def extract_claude_models(
    litellm_data: dict[str, Any], verbose: bool = True
) -> dict[str, Any]:
    """Extract Claude model entries from LiteLLM data.

    Args:
        litellm_data: Raw LiteLLM pricing data
        verbose: Whether to log individual model discoveries

    Returns:
        Dictionary with only Claude models
    """
    claude_models = {}

    for model_name, model_data in litellm_data.items():
        # Check if this is a Claude model
        if (
            isinstance(model_data, dict)
            and model_data.get("litellm_provider") == "anthropic"
            and "claude" in model_name.lower()
        ):
            claude_models[model_name] = model_data
            if verbose:
                logger.debug("claude_model_found", model_name=model_name)

    if verbose:
        logger.info(
            "claude_models_extracted",
            model_count=len(claude_models),
            source="LiteLLM",
        )
    return claude_models

convert_to_internal_format staticmethod

convert_to_internal_format(claude_models, verbose=True)

Convert LiteLLM pricing format to internal format.

LiteLLM format uses cost per token, we use cost per 1M tokens as Decimal.

Parameters:

Name	Type	Description	Default
claude_models	dict[str, Any]	Claude models in LiteLLM format	required
verbose	bool	Whether to log individual model conversions	True

Returns:

Type	Description
dict[str, dict[str, Decimal]]	Dictionary in internal pricing format

Source code in ccproxy/pricing/loader.py

@staticmethod
def convert_to_internal_format(
    claude_models: dict[str, Any], verbose: bool = True
) -> dict[str, dict[str, Decimal]]:
    """Convert LiteLLM pricing format to internal format.

    LiteLLM format uses cost per token, we use cost per 1M tokens as Decimal.

    Args:
        claude_models: Claude models in LiteLLM format
        verbose: Whether to log individual model conversions

    Returns:
        Dictionary in internal pricing format
    """
    internal_format = {}

    for model_name, model_data in claude_models.items():
        try:
            # Extract pricing fields
            input_cost_per_token = model_data.get("input_cost_per_token")
            output_cost_per_token = model_data.get("output_cost_per_token")
            cache_creation_cost = model_data.get("cache_creation_input_token_cost")
            cache_read_cost = model_data.get("cache_read_input_token_cost")

            # Skip models without pricing info
            if input_cost_per_token is None or output_cost_per_token is None:
                if verbose:
                    logger.warning("model_pricing_missing", model_name=model_name)
                continue

            # Convert to per-1M-token pricing (multiply by 1,000,000)
            pricing = {
                "input": Decimal(str(input_cost_per_token * 1_000_000)),
                "output": Decimal(str(output_cost_per_token * 1_000_000)),
            }

            # Add cache pricing if available
            if cache_creation_cost is not None:
                pricing["cache_write"] = Decimal(
                    str(cache_creation_cost * 1_000_000)
                )

            if cache_read_cost is not None:
                pricing["cache_read"] = Decimal(str(cache_read_cost * 1_000_000))

            # Map to canonical model name if needed
            canonical_name = PricingLoader.CLAUDE_MODEL_MAPPINGS.get(
                model_name, model_name
            )
            internal_format[canonical_name] = pricing

            if verbose:
                logger.debug(
                    "model_pricing_converted",
                    original_name=model_name,
                    canonical_name=canonical_name,
                    input_cost=str(pricing["input"]),
                    output_cost=str(pricing["output"]),
                )

        except (ValueError, TypeError) as e:
            if verbose:
                logger.error(
                    "pricing_conversion_failed", model_name=model_name, error=str(e)
                )
            continue

    if verbose:
        logger.info("models_converted", model_count=len(internal_format))
    return internal_format

load_pricing_from_data staticmethod

load_pricing_from_data(litellm_data, verbose=True)

Load and convert pricing data from LiteLLM format.

Parameters:

Name	Type	Description	Default
litellm_data	dict[str, Any]	Raw LiteLLM pricing data	required
verbose	bool	Whether to enable verbose logging	True

Returns:

Type	Description
PricingData | None	Validated pricing data as PricingData model, or None if invalid

Source code in ccproxy/pricing/loader.py

@staticmethod
def load_pricing_from_data(
    litellm_data: dict[str, Any], verbose: bool = True
) -> PricingData | None:
    """Load and convert pricing data from LiteLLM format.

    Args:
        litellm_data: Raw LiteLLM pricing data
        verbose: Whether to enable verbose logging

    Returns:
        Validated pricing data as PricingData model, or None if invalid
    """
    try:
        # Extract Claude models
        claude_models = PricingLoader.extract_claude_models(
            litellm_data, verbose=verbose
        )

        if not claude_models:
            if verbose:
                logger.warning("claude_models_not_found", source="LiteLLM")
            return None

        # Convert to internal format
        internal_pricing = PricingLoader.convert_to_internal_format(
            claude_models, verbose=verbose
        )

        if not internal_pricing:
            if verbose:
                logger.warning("pricing_data_invalid")
            return None

        # Validate and create PricingData model
        pricing_data = PricingData.from_dict(internal_pricing)

        if verbose:
            logger.info("pricing_data_loaded", model_count=len(pricing_data))

        return pricing_data

    except ValidationError as e:
        if verbose:
            logger.error("pricing_validation_failed", error=str(e))
        return None
    except Exception as e:
        if verbose:
            logger.error("pricing_load_failed", source="LiteLLM", error=str(e))
        return None

validate_pricing_data staticmethod

validate_pricing_data(pricing_data, verbose=True)

Validate pricing data using Pydantic models.

Parameters:

Name	Type	Description	Default
pricing_data	Any	Pricing data to validate (dict or PricingData)	required
verbose	bool	Whether to enable verbose logging	True

Returns:

Type	Description
PricingData | None	Valid PricingData model or None if validation fails

Source code in ccproxy/pricing/loader.py

@staticmethod
def validate_pricing_data(
    pricing_data: Any, verbose: bool = True
) -> PricingData | None:
    """Validate pricing data using Pydantic models.

    Args:
        pricing_data: Pricing data to validate (dict or PricingData)
        verbose: Whether to enable verbose logging

    Returns:
        Valid PricingData model or None if validation fails
    """
    try:
        # If already a PricingData instance, return it
        if isinstance(pricing_data, PricingData):
            if verbose:
                logger.debug(
                    "pricing_already_validated", model_count=len(pricing_data)
                )
            return pricing_data

        # If it's a dict, try to create PricingData from it
        if isinstance(pricing_data, dict):
            if not pricing_data:
                if verbose:
                    logger.warning("pricing_data_empty")
                return None

            # Try to create PricingData model
            validated_data = PricingData.from_dict(pricing_data)

            if verbose:
                logger.debug(
                    "pricing_data_validated", model_count=len(validated_data)
                )

            return validated_data

        # Invalid type
        if verbose:
            logger.error(
                "pricing_data_invalid_type",
                actual_type=type(pricing_data).__name__,
                expected_types=["dict", "PricingData"],
            )
        return None

    except ValidationError as e:
        if verbose:
            logger.error("pricing_validation_failed", error=str(e))
        return None
    except Exception as e:
        if verbose:
            logger.error("pricing_validation_unexpected_error", error=str(e))
        return None

get_model_aliases staticmethod

get_model_aliases()

Get mapping of model aliases to canonical names.

Returns:

Type	Description
dict[str, str]	Dictionary mapping aliases to canonical model names

Source code in ccproxy/pricing/loader.py

@staticmethod
def get_model_aliases() -> dict[str, str]:
    """Get mapping of model aliases to canonical names.

    Returns:
        Dictionary mapping aliases to canonical model names
    """
    return PricingLoader.CLAUDE_MODEL_MAPPINGS.copy()

get_canonical_model_name staticmethod

get_canonical_model_name(model_name)

Get canonical model name for a given model name.

Parameters:

Name	Type	Description	Default
model_name	str	Model name (possibly an alias)	required

Returns:

Type	Description
str	Canonical model name

Source code in ccproxy/pricing/loader.py

@staticmethod
def get_canonical_model_name(model_name: str) -> str:
    """Get canonical model name for a given model name.

    Args:
        model_name: Model name (possibly an alias)

    Returns:
        Canonical model name
    """
    return PricingLoader.CLAUDE_MODEL_MAPPINGS.get(model_name, model_name)

ModelPricing

Bases: BaseModel

Pricing information for a single Claude model.

All costs are in USD per 1 million tokens.

convert_to_decimal classmethod

convert_to_decimal(v)

Convert numeric values to Decimal for precision.

Source code in ccproxy/pricing/models.py

@field_validator("*", mode="before")
@classmethod
def convert_to_decimal(cls, v: Any) -> Decimal:
    """Convert numeric values to Decimal for precision."""
    if isinstance(v, int | float | str):
        return Decimal(str(v))
    if isinstance(v, Decimal):
        return v
    raise TypeError(f"Cannot convert {type(v)} to Decimal")

PricingData

Bases: RootModel[dict[str, ModelPricing]]

Complete pricing data for all Claude models.

This is a wrapper around a dictionary of model name to ModelPricing that provides dict-like access while maintaining type safety.

items

items()

Get model name and pricing pairs.

Source code in ccproxy/pricing/models.py

def items(self) -> Iterator[tuple[str, ModelPricing]]:
    """Get model name and pricing pairs."""
    return iter(self.root.items())

keys

keys()

Get model names.

Source code in ccproxy/pricing/models.py

def keys(self) -> Iterator[str]:
    """Get model names."""
    return iter(self.root.keys())

values

values()

Get pricing objects.

Source code in ccproxy/pricing/models.py

def values(self) -> Iterator[ModelPricing]:
    """Get pricing objects."""
    return iter(self.root.values())

get

get(model_name, default=None)

Get pricing for a model with optional default.

Source code in ccproxy/pricing/models.py

def get(
    self, model_name: str, default: ModelPricing | None = None
) -> ModelPricing | None:
    """Get pricing for a model with optional default."""
    return self.root.get(model_name, default)

model_names

model_names()

Get list of all model names.

Source code in ccproxy/pricing/models.py

def model_names(self) -> list[str]:
    """Get list of all model names."""
    return list(self.root.keys())

to_dict

to_dict()

Convert to legacy dict format for backward compatibility.

Source code in ccproxy/pricing/models.py

def to_dict(self) -> dict[str, dict[str, Decimal]]:
    """Convert to legacy dict format for backward compatibility."""
    return {
        model_name: {
            "input": pricing.input,
            "output": pricing.output,
            "cache_read": pricing.cache_read,
            "cache_write": pricing.cache_write,
        }
        for model_name, pricing in self.root.items()
    }

from_dict classmethod

from_dict(data)

Create PricingData from legacy dict format.

Source code in ccproxy/pricing/models.py

@classmethod
def from_dict(cls, data: dict[str, dict[str, Any]]) -> "PricingData":
    """Create PricingData from legacy dict format."""
    models = {
        model_name: ModelPricing(**pricing_dict)
        for model_name, pricing_dict in data.items()
    }
    return cls(root=models)

PricingUpdater

PricingUpdater(cache, settings)

Manages periodic updates of pricing data.

Parameters:

Name	Type	Description	Default
cache	PricingCache	Pricing cache instance	required
settings	PricingSettings	Pricing configuration settings	required

Source code in ccproxy/pricing/updater.py

def __init__(
    self,
    cache: PricingCache,
    settings: PricingSettings,
) -> None:
    """Initialize pricing updater.

    Args:
        cache: Pricing cache instance
        settings: Pricing configuration settings
    """
    self.cache = cache
    self.settings = settings
    self._cached_pricing: PricingData | None = None
    self._last_load_time: float = 0
    self._last_file_check_time: float = 0
    self._cached_file_mtime: float = 0

get_current_pricing async

get_current_pricing(force_refresh=False)

Get current pricing data with automatic updates.

Parameters:

Name	Type	Description	Default
force_refresh	bool	Force refresh even if cache is valid	False

Returns:

Type	Description
PricingData | None	Current pricing data as PricingData model

Source code in ccproxy/pricing/updater.py

async def get_current_pricing(
    self, force_refresh: bool = False
) -> PricingData | None:
    """Get current pricing data with automatic updates.

    Args:
        force_refresh: Force refresh even if cache is valid

    Returns:
        Current pricing data as PricingData model
    """
    import time

    current_time = time.time()

    # Return cached pricing if recent and not forced
    if (
        not force_refresh
        and self._cached_pricing is not None
        and (current_time - self._last_load_time) < self.settings.memory_cache_ttl
    ):
        # Only check file changes every 30 seconds to reduce I/O
        if (current_time - self._last_file_check_time) > 30:
            if self._has_cache_file_changed():
                logger.info("cache_file_changed")
                # File changed, need to reload
                pricing_data = await self._load_pricing_data()
                self._cached_pricing = pricing_data
                self._last_load_time = current_time
                return pricing_data
            self._last_file_check_time = current_time

        return self._cached_pricing

    # Check if we need to refresh
    should_refresh = force_refresh or (
        self.settings.auto_update and not self.cache.is_cache_valid()
    )

    if should_refresh:
        logger.info("pricing_refresh_start")
        await self._refresh_pricing()

    # Load pricing data
    pricing_data = await self._load_pricing_data()

    # Cache the result
    self._cached_pricing = pricing_data
    self._last_load_time = current_time
    self._last_file_check_time = current_time

    return pricing_data

force_refresh async

force_refresh()

Force a refresh of pricing data.

Returns:

Type	Description
bool	True if refresh was successful

Source code in ccproxy/pricing/updater.py

async def force_refresh(self) -> bool:
    """Force a refresh of pricing data.

    Returns:
        True if refresh was successful
    """
    logger.info("pricing_force_refresh_start")

    # Clear cached pricing
    self._cached_pricing = None
    self._last_load_time = 0

    # Refresh from external source
    success = await self._refresh_pricing()

    if success:
        # Reload pricing data
        await self.get_current_pricing(force_refresh=True)

    return success

clear_cache

clear_cache()

Clear all cached pricing data.

Returns:

Type	Description
bool	True if cache was cleared successfully

Source code in ccproxy/pricing/updater.py

def clear_cache(self) -> bool:
    """Clear all cached pricing data.

    Returns:
        True if cache was cleared successfully
    """
    logger.info("pricing_cache_clear_start")

    # Clear in-memory cache
    self._cached_pricing = None
    self._last_load_time = 0

    # Clear file cache
    return self.cache.clear_cache()

get_pricing_info async

get_pricing_info()

Get information about current pricing state.

Returns:

Type	Description
dict[str, Any]	Dictionary with pricing information

Source code in ccproxy/pricing/updater.py

async def get_pricing_info(self) -> dict[str, Any]:
    """Get information about current pricing state.

    Returns:
        Dictionary with pricing information
    """
    cache_info = self.cache.get_cache_info()

    pricing_data = await self.get_current_pricing()

    return {
        "models_loaded": len(pricing_data) if pricing_data else 0,
        "model_names": pricing_data.model_names() if pricing_data else [],
        "auto_update": self.settings.auto_update,
        "fallback_to_embedded": self.settings.fallback_to_embedded,
        "has_cached_pricing": self._cached_pricing is not None,
    }

validate_external_source async

validate_external_source()

Validate that external pricing source is accessible.

Returns:

Type	Description
bool	True if external source is accessible and has valid data

Source code in ccproxy/pricing/updater.py

async def validate_external_source(self) -> bool:
    """Validate that external pricing source is accessible.

    Returns:
        True if external source is accessible and has valid data
    """
    try:
        logger.debug("external_pricing_validation_start")

        # Try to download data
        raw_data = await self.cache.download_pricing_data(timeout=10)
        if raw_data is None:
            return False

        # Try to parse Claude models
        claude_models = PricingLoader.extract_claude_models(raw_data)
        if not claude_models:
            logger.warning("claude_models_not_found_in_external")
            return False

        # Try to load and validate using Pydantic
        pricing_data = PricingLoader.load_pricing_from_data(raw_data, verbose=False)
        if not pricing_data:
            logger.warning("external_pricing_load_failed")
            return False

        logger.info(
            "external_pricing_validation_completed", model_count=len(pricing_data)
        )
        return True

    except Exception as e:
        logger.error("external_pricing_validation_failed", error=str(e))
        return False