Skip to content

ccproxy.plugins.pricing.utils

ccproxy.plugins.pricing.utils

Cost calculation utilities for token-based pricing (plugin-owned).

These helpers live inside the pricing plugin to avoid coupling core to pricing logic. They accept an optional PricingService instance for callers that already have one; otherwise they create a default service on demand.

calculate_token_cost async 

calculate_token_cost(
    tokens_input,
    tokens_output,
    model,
    cache_read_tokens=None,
    cache_write_tokens=None,
    pricing_service=None,
)

Calculate total cost in USD for the given token usage.

If no pricing_service is provided, a default PricingService is created using PricingConfig(). Returns None if model or tokens are missing or if pricing information is unavailable.

Source code in ccproxy/plugins/pricing/utils.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
async def calculate_token_cost(
    tokens_input: int | None,
    tokens_output: int | None,
    model: str | None,
    cache_read_tokens: int | None = None,
    cache_write_tokens: int | None = None,
    pricing_service: PricingService | None = None,
) -> float | None:
    """Calculate total cost in USD for the given token usage.

    If no pricing_service is provided, a default PricingService is created
    using PricingConfig(). Returns None if model or tokens are missing or if
    pricing information is unavailable.
    """
    if not model or (
        not tokens_input
        and not tokens_output
        and not cache_read_tokens
        and not cache_write_tokens
    ):
        return None

    service = pricing_service or PricingService(PricingConfig())

    try:
        cost_decimal = await service.calculate_cost(
            model_name=model,
            input_tokens=tokens_input or 0,
            output_tokens=tokens_output or 0,
            cache_read_tokens=cache_read_tokens or 0,
            cache_write_tokens=cache_write_tokens or 0,
        )
        return float(cost_decimal) if cost_decimal is not None else None
    except Exception:
        return None

calculate_cost_breakdown async 

calculate_cost_breakdown(
    tokens_input,
    tokens_output,
    model,
    cache_read_tokens=None,
    cache_write_tokens=None,
    pricing_service=None,
)

Return a detailed cost breakdown using current pricing data.

If no pricing_service is provided, a default PricingService is created. Returns None if inputs are insufficient or model pricing is unavailable.

Source code in ccproxy/plugins/pricing/utils.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
async def calculate_cost_breakdown(
    tokens_input: int | None,
    tokens_output: int | None,
    model: str | None,
    cache_read_tokens: int | None = None,
    cache_write_tokens: int | None = None,
    pricing_service: PricingService | None = None,
) -> dict[str, float | str] | None:
    """Return a detailed cost breakdown using current pricing data.

    If no pricing_service is provided, a default PricingService is created.
    Returns None if inputs are insufficient or model pricing is unavailable.
    """
    if not model or (
        not tokens_input
        and not tokens_output
        and not cache_read_tokens
        and not cache_write_tokens
    ):
        return None

    service = pricing_service or PricingService(PricingConfig())

    try:
        model_pricing = await service.get_model_pricing(model)
        if not model_pricing:
            return None

        input_cost = ((tokens_input or 0) / 1_000_000) * float(model_pricing.input)
        output_cost = ((tokens_output or 0) / 1_000_000) * float(model_pricing.output)
        cache_read_cost = ((cache_read_tokens or 0) / 1_000_000) * float(
            model_pricing.cache_read
        )
        cache_write_cost = ((cache_write_tokens or 0) / 1_000_000) * float(
            model_pricing.cache_write
        )

        total_cost = input_cost + output_cost + cache_read_cost + cache_write_cost

        return {
            "input_cost": input_cost,
            "output_cost": output_cost,
            "cache_read_cost": cache_read_cost,
            "cache_write_cost": cache_write_cost,
            "total_cost": total_cost,
            "model": model,
        }
    except Exception:
        return None