Module: LlmCostTracker::Reconciliation::Sources::AnthropicUsage

Defined in:
lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb

Constant Summary collapse

FINGERPRINT_KEYS = 
%i[
  starting_at ending_at model workspace_id
  service_tier context_window cost_type token_type description
  inference_geo
].freeze
ROW_TYPE_COST = 
"cost"
AUTHORITY_COST_API = 
"cost_api"
DEFAULT_METER = 
"tokens"

Class Method Summary collapse

Class Method Details

.dollars_from_cents(amount) ⇒ Object 



68
69
70
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 68

def dollars_from_cents(amount)
  (BigDecimal(amount.to_s) / 100).to_s("F")
end

.end_inclusive_date(value) ⇒ Object 



137
138
139
140
141
142
143
144
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 137

def end_inclusive_date(value)
  time = case value
         when Numeric then Time.at(value).utc
         when Date then value.to_time.utc
         else Time.parse(value.to_s).utc
         end
  (time - 1).utc.to_date
end

.fingerprint_for(result, starting_at:, ending_at:) ⇒ Object 



124
125
126
127
128
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 124

def fingerprint_for(result, starting_at:, ending_at:)
  attributes = result.merge(starting_at: Coercion.normalized_epoch(starting_at),
                            ending_at: Coercion.normalized_epoch(ending_at))
  Fingerprint.compute(FINGERPRINT_KEYS, attributes)
end

.match_basis_for(result) ⇒ Object 



117
118
119
120
121
122
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 117

def match_basis_for(result)
  return "workspace" if result[:workspace_id]
  return "model" if result[:model]

  "period_only"
end

.metadata_for(result, authority:, row_type:) ⇒ Object 



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 72

def metadata_for(result, authority:, row_type:)
  {
    "row_type" => row_type,
    "meter" => meter_for(result),
    "authority" => authority,
    "match_basis" => match_basis_for(result),
    "model" => result[:model],
    "pricing_mode" => pricing_mode_for(result),
    "context_window" => result[:context_window],
    "cost_type" => result[:cost_type],
    "description" => result[:description],
    "token_type" => result[:token_type],
    "inference_geo" => result[:inference_geo],
    "provider_workspace_id" => result[:workspace_id]
  }.compact
end

.meter_for(result) ⇒ Object 



89
90
91
92
93
94
95
96
97
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 89

def meter_for(result)
  case result[:cost_type].to_s
  when "web_search" then "web_search"
  when "code_execution" then "code_execution_hour"
  when "session_usage" then "session_usage"
  when "tokens" then token_meter(result[:token_type].to_s)
  else DEFAULT_METER
  end
end

.parse(response, authority: AUTHORITY_COST_API, row_type: ROW_TYPE_COST) ⇒ Object 



25
26
27
28
29
30
31
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 25

def parse(response, authority: AUTHORITY_COST_API, row_type: ROW_TYPE_COST)
  payload = Coercion.coerce_hash(response, label: "Anthropic Usage")
  buckets = Array(payload[:data])
  buckets.flat_map do |bucket|
    rows_for_bucket(bucket, authority: authority, row_type: row_type)
  end.compact
end

.parse_date(value) ⇒ Object 



130
131
132
133
134
135
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 130

def parse_date(value)
  return value if value.is_a?(Date)
  return Time.at(value).utc.to_date if value.is_a?(Numeric)

  Time.parse(value.to_s).utc.to_date
end

.pricing_mode_for(result) ⇒ Object 



108
109
110
111
112
113
114
115
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 108

def pricing_mode_for(result)
  modes = []
  modes << "batch" if result[:service_tier].to_s.downcase == "batch"
  if LlmCostTracker::Providers::Anthropic::TierClassification.data_residency_geo?(result[:inference_geo])
    modes << "data_residency"
  end
  modes.empty? ? nil : modes.uniq.join("_")
end

.row_for_result(raw, period_start:, period_end:, starting_at:, ending_at:, authority:, row_type:) ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 52

def row_for_result(raw, period_start:, period_end:, starting_at:, ending_at:, authority:, row_type:)
  result = Coercion.symbolize(raw)
  raw_amount = result[:amount]
  return nil if raw_amount.nil?

  fingerprint = fingerprint_for(result, starting_at: starting_at, ending_at: ending_at)
  {
    external_id: "cost-#{fingerprint}",
    period_start: period_start,
    period_end: period_end,
    billed_amount: dollars_from_cents(raw_amount),
    currency: (result[:currency] || "USD").to_s.upcase,
    metadata: metadata_for(result, authority: authority, row_type: row_type)
  }
end

.rows_for_bucket(bucket, authority:, row_type:) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 33

def rows_for_bucket(bucket, authority:, row_type:)
  bucket = Coercion.symbolize(bucket)
  starting_at = bucket[:starting_at]
  ending_at = bucket[:ending_at]
  return [] unless starting_at && ending_at

  period_start = parse_date(starting_at)
  period_end = end_inclusive_date(ending_at)

  Array(bucket[:results]).filter_map do |raw|
    row_for_result(raw,
                   period_start: period_start, period_end: period_end,
                   starting_at: starting_at, ending_at: ending_at,
                   authority: authority, row_type: row_type)
  end
rescue ArgumentError
  []
end

.token_meter(token_type) ⇒ Object 



99
100
101
102
103
104
105
106
 
# File 'lib/llm_cost_tracker/reconciliation/sources/anthropic_usage.rb', line 99

def token_meter(token_type)
  return "cache_read_input_tokens" if token_type.include?("cache_read")
  return "cache_creation_input_tokens" if token_type.include?("cache_creation")
  return "input_tokens" if token_type.include?("input")
  return "output_tokens" if token_type.include?("output")

  DEFAULT_METER
end