cost_tracker

cost_tracker

Cost tracking for LLM API calls.

Provides accurate cost tracking with thread safety and detailed breakdowns.

CostEntry dataclass

CostEntry(tokens_in: int, tokens_out: int, cost: Decimal, model: str, timestamp: float)

Single cost tracking entry.

CostTracker

CostTracker(input_cost_per_1k: Decimal | None = None, output_cost_per_1k: Decimal | None = None)

Detailed cost accounting with thread-safety and per-stage breakdowns.

Scope: Detailed financial tracking and reporting Pattern: Accumulator with thread-safe operations

Use CostTracker for: - Stage-by-stage cost breakdowns - Detailed entry logging (timestamp, model, tokens) - Thread-safe accumulation in concurrent execution - Cost reporting and analytics - Budget enforcement (via BudgetController)

NOT for: - Simple orchestration state (use ExecutionContext for that)

Why separate from ExecutionContext? - CostTracker = detailed accounting system (entries, breakdowns, thread-safety) - ExecutionContext = orchestration state (progress, session, timing) - Different concerns: accounting vs execution control

Thread Safety: - All methods protected by threading.Lock - Safe for concurrent LLM invocations

Example

tracker = CostTracker( input_cost_per_1k=Decimal("0.00015"), output_cost_per_1k=Decimal("0.0006") ) cost = tracker.add(tokens_in=1000, tokens_out=500, model="gpt-4o-mini") breakdown = tracker.get_stage_costs() # {"llm_invocation": Decimal("0.00045")}

See Also: - ExecutionContext: For orchestration-level state - BudgetController: For cost limit enforcement - docs/TECHNICAL_REFERENCE.md: Cost tracking architecture

Initialize cost tracker.

Parameters:

Name	Type	Description	Default
input_cost_per_1k	Decimal | None	Input token cost per 1K tokens	None
output_cost_per_1k	Decimal | None	Output token cost per 1K tokens	None

Source code in ondine/utils/cost_tracker.py

def __init__(
    self,
    input_cost_per_1k: Decimal | None = None,
    output_cost_per_1k: Decimal | None = None,
):
    """
    Initialize cost tracker.

    Args:
        input_cost_per_1k: Input token cost per 1K tokens
        output_cost_per_1k: Output token cost per 1K tokens
    """
    self.input_cost_per_1k = input_cost_per_1k or Decimal("0.0")
    self.output_cost_per_1k = output_cost_per_1k or Decimal("0.0")

    self._total_input_tokens = 0
    self._total_output_tokens = 0
    self._total_cost = Decimal("0.0")
    self._entries: list[CostEntry] = []
    self._stage_costs: dict[str, Decimal] = {}
    self._lock = threading.Lock()

total_cost property

total_cost: Decimal

Get total accumulated cost.

total_tokens property

total_tokens: int

Get total token count.

input_tokens property

input_tokens: int

Get total input tokens.

output_tokens property

output_tokens: int

Get total output tokens.

add

add(tokens_in: int, tokens_out: int, model: str, timestamp: float, stage: str | None = None) -> Decimal

Add cost entry.

Parameters:

Name	Type	Description	Default
tokens_in	int	Input tokens used	required
tokens_out	int	Output tokens used	required
model	str	Model identifier	required
timestamp	float	Timestamp of request	required
stage	str | None	Optional stage name	None

Returns:

Type	Description
Decimal	Cost for this entry

Source code in ondine/utils/cost_tracker.py

def add(
    self,
    tokens_in: int,
    tokens_out: int,
    model: str,
    timestamp: float,
    stage: str | None = None,
) -> Decimal:
    """
    Add cost entry.

    Args:
        tokens_in: Input tokens used
        tokens_out: Output tokens used
        model: Model identifier
        timestamp: Timestamp of request
        stage: Optional stage name

    Returns:
        Cost for this entry
    """
    cost = self.calculate_cost(tokens_in, tokens_out)

    with self._lock:
        entry = CostEntry(
            tokens_in=tokens_in,
            tokens_out=tokens_out,
            cost=cost,
            model=model,
            timestamp=timestamp,
        )
        self._entries.append(entry)

        self._total_input_tokens += tokens_in
        self._total_output_tokens += tokens_out
        self._total_cost += cost

        if stage:
            self._stage_costs[stage] = (
                self._stage_costs.get(stage, Decimal("0.0")) + cost
            )

    return cost

calculate_cost

calculate_cost(tokens_in: int, tokens_out: int) -> Decimal

Calculate cost for given token counts.

Parameters:

Name	Type	Description	Default
tokens_in	int	Input tokens	required
tokens_out	int	Output tokens	required

Returns:

Type	Description
Decimal	Total cost

Source code in ondine/utils/cost_tracker.py

def calculate_cost(self, tokens_in: int, tokens_out: int) -> Decimal:
    """
    Calculate cost for given token counts.

    Args:
        tokens_in: Input tokens
        tokens_out: Output tokens

    Returns:
        Total cost
    """
    from ondine.utils.cost_calculator import CostCalculator

    return CostCalculator.calculate(
        tokens_in=tokens_in,
        tokens_out=tokens_out,
        input_cost_per_1k=self.input_cost_per_1k,
        output_cost_per_1k=self.output_cost_per_1k,
    )

get_estimate

get_estimate(rows: int = 0) -> CostEstimate

Get cost estimate.

Parameters:

Name	Type	Description	Default
rows	int	Number of rows processed	0

Returns:

Type	Description
CostEstimate	CostEstimate object

Source code in ondine/utils/cost_tracker.py

def get_estimate(self, rows: int = 0) -> CostEstimate:
    """
    Get cost estimate.

    Args:
        rows: Number of rows processed

    Returns:
        CostEstimate object
    """
    with self._lock:
        total_tokens = self._total_input_tokens + self._total_output_tokens
        return CostEstimate(
            total_cost=self._total_cost,
            total_tokens=total_tokens,
            input_tokens=self._total_input_tokens,
            output_tokens=self._total_output_tokens,
            rows=rows,
            breakdown_by_stage=dict(self._stage_costs),
            confidence="actual",
        )

reset

reset() -> None

Reset all tracking.

Source code in ondine/utils/cost_tracker.py

def reset(self) -> None:
    """Reset all tracking."""
    with self._lock:
        self._total_input_tokens = 0
        self._total_output_tokens = 0
        self._total_cost = Decimal("0.0")
        self._entries.clear()
        self._stage_costs.clear()

get_stage_costs

get_stage_costs() -> dict[str, Decimal]

Get costs breakdown by stage.

Source code in ondine/utils/cost_tracker.py

def get_stage_costs(self) -> dict[str, Decimal]:
    """Get costs breakdown by stage."""
    with self._lock:
        return dict(self._stage_costs)