diff --git a/config/config.exs b/config/config.exs
index ac30e78..3b3d55e 100644
--- a/config/config.exs
+++ b/config/config.exs
@@ -54,7 +54,10 @@ config :arcadia_cloud, Oban,
        # ProjectsWorker first so attribution is fresh before resource syncs
        {"*/15 * * * *", ArcadiaCloud.Sync.ProjectsWorker},
        {"*/15 * * * *", ArcadiaCloud.Sync.DropletsWorker},
-       {"*/15 * * * *", ArcadiaCloud.Sync.DomainsWorker}
+       {"*/15 * * * *", ArcadiaCloud.Sync.DomainsWorker},
+       # Billing: hourly balance, daily invoice discovery
+       {"7 * * * *", ArcadiaCloud.Sync.BalanceWorker},
+       {"23 2 * * *", ArcadiaCloud.Sync.BillingHistoryWorker}
      ]}
   ],
   repo: ArcadiaCloud.Repo
diff --git a/lib/arcadia_cloud/billing.ex b/lib/arcadia_cloud/billing.ex
new file mode 100644
index 0000000..46bc8e7
--- /dev/null
+++ b/lib/arcadia_cloud/billing.ex
@@ -0,0 +1,166 @@
+defmodule ArcadiaCloud.Billing do
+  @moduledoc """
+  Context for cloud cost ingestion: balance snapshots, monthly invoices,
+  per-line-item COGS, and resource matching.
+
+  Pipeline:
+    BalanceWorker         → cloud_balance_snapshots          (hourly)
+    BillingHistoryWorker  → cloud_invoices (headers)          (daily)
+    InvoiceIngestWorker   → cloud_cost_lines + resource match (per invoice)
+
+  Phase 1 stops here. Phase 1+ pushes matched cost_lines into skyai-finance
+  as expense lines (skyai-internal) or AR (tenant).
+  """
+
+  import Ecto.Query, warn: false
+
+  alias ArcadiaCloud.Repo
+  alias ArcadiaCloud.Billing.{CloudBalanceSnapshot, CloudInvoice, CloudCostLine}
+  alias ArcadiaCloud.Cloud.CloudResource
+
+  # ---- balance --------------------------------------------------------------
+
+  def record_balance(attrs) do
+    %CloudBalanceSnapshot{}
+    |> CloudBalanceSnapshot.changeset(attrs)
+    |> Repo.insert()
+  end
+
+  def latest_balance(provider \\ "digitalocean") do
+    from(s in CloudBalanceSnapshot,
+      where: s.provider == ^provider,
+      order_by: [desc: s.generated_at],
+      limit: 1
+    )
+    |> Repo.one()
+  end
+
+  # ---- invoices -------------------------------------------------------------
+
+  def upsert_invoice(attrs) do
+    provider = attrs[:provider] || attrs["provider"]
+    invoice_id = attrs[:provider_invoice_id] || attrs["provider_invoice_id"]
+
+    case Repo.get_by(CloudInvoice, provider: provider, provider_invoice_id: invoice_id) do
+      nil ->
+        %CloudInvoice{}
+        |> CloudInvoice.changeset(attrs)
+        |> Repo.insert()
+
+      existing ->
+        existing
+        |> CloudInvoice.changeset(attrs)
+        |> Repo.update()
+    end
+  end
+
+  def list_invoices_needing_ingest(provider \\ "digitalocean") do
+    from(i in CloudInvoice,
+      where: i.provider == ^provider and is_nil(i.lines_ingested_at),
+      order_by: [desc: i.invoice_period]
+    )
+    |> Repo.all()
+  end
+
+  def mark_invoice_ingested(invoice) do
+    now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+    invoice
+    |> CloudInvoice.changeset(%{lines_ingested_at: now, csv_fetched_at: now})
+    |> Repo.update()
+  end
+
+  # ---- cost lines -----------------------------------------------------------
+
+  @doc """
+  Replace all cost lines for an invoice in one transaction. CSV is the
+  authoritative source for an invoice's content; if we re-fetch, we
+  replace not merge.
+  """
+  def replace_cost_lines(%CloudInvoice{} = invoice, line_attrs_list) when is_list(line_attrs_list) do
+    Repo.transaction(fn ->
+      Repo.delete_all(from(l in CloudCostLine, where: l.invoice_id == ^invoice.id))
+
+      Enum.each(line_attrs_list, fn attrs ->
+        %CloudCostLine{}
+        |> CloudCostLine.changeset(Map.put(attrs, :invoice_id, invoice.id))
+        |> Repo.insert!()
+      end)
+    end)
+  end
+
+  @doc """
+  Match unmatched cost lines for an invoice to cloud_resources by
+  (kind, description=name) — case-insensitive. Updates matched_at + resource_id.
+  Returns count of newly-matched lines.
+  """
+  def match_cost_lines_to_resources(%CloudInvoice{id: invoice_id}) do
+    unmatched =
+      from(l in CloudCostLine,
+        where: l.invoice_id == ^invoice_id and is_nil(l.resource_id) and not is_nil(l.kind)
+      )
+      |> Repo.all()
+
+    now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+    Enum.reduce(unmatched, 0, fn line, acc ->
+      case find_resource(line) do
+        %CloudResource{id: rid} ->
+          line
+          |> CloudCostLine.changeset(%{resource_id: rid, matched_at: now})
+          |> Repo.update!()
+
+          acc + 1
+
+        nil ->
+          acc
+      end
+    end)
+  end
+
+  defp find_resource(%CloudCostLine{kind: kind, description: desc}) when is_binary(desc) do
+    name_lower = desc |> extract_name() |> String.downcase()
+
+    from(r in CloudResource,
+      where:
+        r.kind == ^kind and is_nil(r.deleted_at) and
+          fragment("LOWER(?) = ?", r.name, ^name_lower),
+      limit: 1
+    )
+    |> Repo.one()
+  end
+
+  defp find_resource(_), do: nil
+
+  # DO CSV description is often "name (size_slug)" or "name-1234 (region) NGB Snapshot".
+  # Strip everything after the first " (" — best-effort name extraction.
+  defp extract_name(desc) do
+    case String.split(desc, " (", parts: 2) do
+      [name | _] -> String.trim(name)
+      _ -> desc
+    end
+  end
+
+  def list_cost_lines(opts \\ []) do
+    base =
+      from(l in CloudCostLine,
+        order_by: [desc: l.invoice_period, desc: l.amount_cents]
+      )
+
+    base
+    |> maybe_filter(:invoice_period, opts[:period])
+    |> maybe_filter(:kind, opts[:kind])
+    |> maybe_filter(:resource_id, opts[:resource_id])
+    |> maybe_limit(opts[:limit])
+    |> Repo.all()
+  end
+
+  defp maybe_filter(query, _field, nil), do: query
+
+  defp maybe_filter(query, field, value) do
+    from(l in query, where: field(l, ^field) == ^value)
+  end
+
+  defp maybe_limit(query, nil), do: query
+  defp maybe_limit(query, n) when is_integer(n), do: from(q in query, limit: ^n)
+end
diff --git a/lib/arcadia_cloud/billing/cloud_balance_snapshot.ex b/lib/arcadia_cloud/billing/cloud_balance_snapshot.ex
new file mode 100644
index 0000000..833ebfa
--- /dev/null
+++ b/lib/arcadia_cloud/billing/cloud_balance_snapshot.ex
@@ -0,0 +1,27 @@
+defmodule ArcadiaCloud.Billing.CloudBalanceSnapshot do
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  schema "cloud_balance_snapshots" do
+    field :provider, :string
+    field :month_to_date_balance_cents, :integer
+    field :account_balance_cents, :integer
+    field :month_to_date_usage_cents, :integer
+    field :generated_at, :utc_datetime
+    field :raw, :map
+
+    timestamps(type: :utc_datetime, updated_at: false)
+  end
+
+  @required ~w(provider generated_at)a
+  @optional ~w(month_to_date_balance_cents account_balance_cents month_to_date_usage_cents raw)a
+
+  def changeset(snap, attrs) do
+    snap
+    |> cast(attrs, @required ++ @optional)
+    |> validate_required(@required)
+  end
+end
diff --git a/lib/arcadia_cloud/billing/cloud_cost_line.ex b/lib/arcadia_cloud/billing/cloud_cost_line.ex
new file mode 100644
index 0000000..a1496c1
--- /dev/null
+++ b/lib/arcadia_cloud/billing/cloud_cost_line.ex
@@ -0,0 +1,38 @@
+defmodule ArcadiaCloud.Billing.CloudCostLine do
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  schema "cloud_cost_lines" do
+    field :invoice_period, :date
+    field :kind, :string
+    field :description, :string
+    field :qty, :decimal
+    field :unit, :string
+    field :unit_cost_cents, :integer
+    field :amount_cents, :integer
+    field :start_at, :utc_datetime
+    field :end_at, :utc_datetime
+    field :project_name, :string
+    field :category, :string
+    field :matched_at, :utc_datetime
+    field :raw, :map
+
+    belongs_to :invoice, ArcadiaCloud.Billing.CloudInvoice
+    belongs_to :resource, ArcadiaCloud.Cloud.CloudResource
+
+    timestamps(type: :utc_datetime, updated_at: false)
+  end
+
+  @required ~w(invoice_id invoice_period amount_cents)a
+  @optional ~w(resource_id kind description qty unit unit_cost_cents start_at end_at
+               project_name category matched_at raw)a
+
+  def changeset(line, attrs) do
+    line
+    |> cast(attrs, @required ++ @optional)
+    |> validate_required(@required)
+  end
+end
diff --git a/lib/arcadia_cloud/billing/cloud_invoice.ex b/lib/arcadia_cloud/billing/cloud_invoice.ex
new file mode 100644
index 0000000..058c275
--- /dev/null
+++ b/lib/arcadia_cloud/billing/cloud_invoice.ex
@@ -0,0 +1,31 @@
+defmodule ArcadiaCloud.Billing.CloudInvoice do
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  schema "cloud_invoices" do
+    field :provider, :string
+    field :provider_invoice_id, :string
+    field :invoice_period, :date
+    field :amount_cents, :integer
+    field :status, :string, default: "open"
+    field :issued_at, :utc_datetime
+    field :csv_fetched_at, :utc_datetime
+    field :lines_ingested_at, :utc_datetime
+    field :raw, :map
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @required ~w(provider provider_invoice_id invoice_period)a
+  @optional ~w(amount_cents status issued_at csv_fetched_at lines_ingested_at raw)a
+
+  def changeset(invoice, attrs) do
+    invoice
+    |> cast(attrs, @required ++ @optional)
+    |> validate_required(@required)
+    |> unique_constraint([:provider, :provider_invoice_id])
+  end
+end
diff --git a/lib/arcadia_cloud/digital_ocean/client.ex b/lib/arcadia_cloud/digital_ocean/client.ex
index ab3d040..1e3e1a7 100644
--- a/lib/arcadia_cloud/digital_ocean/client.ex
+++ b/lib/arcadia_cloud/digital_ocean/client.ex
@@ -21,6 +21,52 @@ defmodule ArcadiaCloud.DigitalOcean.Client do
   def list_volumes(opts \\ []), do: list_paginated("/volumes", "volumes", opts)
   def list_floating_ips(opts \\ []), do: list_paginated("/floating_ips", "floating_ips", opts)
 
+  # ---- billing --------------------------------------------------------------
+
+  def get_balance(opts \\ []) do
+    request(:get, "/customers/my/balance", purpose: opts[:purpose] || "billing")
+  end
+
+  def list_billing_history(opts \\ []) do
+    list_paginated("/customers/my/billing_history", "billing_history",
+      Keyword.put(opts, :purpose, opts[:purpose] || "billing"))
+  end
+
+  def get_invoice_summary(invoice_uuid, opts \\ []) do
+    request(:get, "/customers/my/invoices/#{invoice_uuid}/summary",
+      purpose: opts[:purpose] || "billing")
+  end
+
+  @doc """
+  Fetch the CSV body for an invoice. Returns {:ok, csv_string} | {:error, _}.
+  """
+  def fetch_invoice_csv(invoice_uuid, opts \\ []) do
+    purpose = opts[:purpose] || "billing"
+
+    with {:ok, token} <- Tokens.fetch(purpose) do
+      case Req.request(
+             method: :get,
+             url: @base <> "/customers/my/invoices/#{invoice_uuid}/csv",
+             headers: [
+               {"authorization", "Bearer " <> token},
+               {"accept", "text/csv"}
+             ],
+             retry: :transient,
+             max_retries: 3,
+             decode_body: false
+           ) do
+        {:ok, %Req.Response{status: 200, body: body}} when is_binary(body) ->
+          {:ok, body}
+
+        {:ok, %Req.Response{status: status, body: body}} ->
+          {:error, {:http, status, body}}
+
+        {:error, e} ->
+          {:error, {:transport, e}}
+      end
+    end
+  end
+
   def create_project(name, purpose, description \\ "", opts \\ []) do
     body = %{
       name: name,
diff --git a/lib/arcadia_cloud/sync/balance_worker.ex b/lib/arcadia_cloud/sync/balance_worker.ex
new file mode 100644
index 0000000..ebffc93
--- /dev/null
+++ b/lib/arcadia_cloud/sync/balance_worker.ex
@@ -0,0 +1,47 @@
+defmodule ArcadiaCloud.Sync.BalanceWorker do
+  @moduledoc """
+  Hourly poll of `/v2/customers/my/balance`. Records a snapshot row so
+  the cost dashboard can show MTD usage in real time.
+  """
+
+  use Oban.Worker, queue: :cloud_billing, max_attempts: 3
+
+  alias ArcadiaCloud.Billing
+  alias ArcadiaCloud.DigitalOcean.Client
+
+  @impl Oban.Worker
+  def perform(_job) do
+    with {:ok, body} <- Client.get_balance() do
+      Billing.record_balance(%{
+        provider: "digitalocean",
+        month_to_date_balance_cents: dollars_to_cents(body["month_to_date_balance"]),
+        account_balance_cents: dollars_to_cents(body["account_balance"]),
+        month_to_date_usage_cents: dollars_to_cents(body["month_to_date_usage"]),
+        generated_at: parse_iso(body["generated_at"]),
+        raw: body
+      })
+
+      :ok
+    end
+  end
+
+  defp dollars_to_cents(nil), do: nil
+
+  defp dollars_to_cents(value) when is_binary(value) do
+    case Float.parse(value) do
+      {float, _} -> round(float * 100)
+      :error -> nil
+    end
+  end
+
+  defp dollars_to_cents(value) when is_number(value), do: round(value * 100)
+
+  defp parse_iso(nil), do: DateTime.utc_now() |> DateTime.truncate(:second)
+
+  defp parse_iso(str) when is_binary(str) do
+    case DateTime.from_iso8601(str) do
+      {:ok, dt, _} -> DateTime.truncate(dt, :second)
+      _ -> DateTime.utc_now() |> DateTime.truncate(:second)
+    end
+  end
+end
diff --git a/lib/arcadia_cloud/sync/billing_history_worker.ex b/lib/arcadia_cloud/sync/billing_history_worker.ex
new file mode 100644
index 0000000..af325f3
--- /dev/null
+++ b/lib/arcadia_cloud/sync/billing_history_worker.ex
@@ -0,0 +1,97 @@
+defmodule ArcadiaCloud.Sync.BillingHistoryWorker do
+  @moduledoc """
+  Daily discovery of new invoices via `/v2/customers/my/billing_history`.
+
+  Upserts an invoice header per row and enqueues an InvoiceIngestWorker
+  job for any invoice not yet ingested (CSV parse + line-item write +
+  resource match). Idempotent — re-running re-enqueues nothing new.
+  """
+
+  use Oban.Worker, queue: :cloud_billing, max_attempts: 3
+
+  alias ArcadiaCloud.Billing
+  alias ArcadiaCloud.DigitalOcean.Client
+  alias ArcadiaCloud.Sync.InvoiceIngestWorker
+
+  @impl Oban.Worker
+  def perform(_job) do
+    with {:ok, history} <- Client.list_billing_history() do
+      Enum.each(history, fn item ->
+        case extract_invoice_attrs(item) do
+          {:ok, attrs} ->
+            {:ok, invoice} = Billing.upsert_invoice(attrs)
+
+            if is_nil(invoice.lines_ingested_at) do
+              %{invoice_id: invoice.id}
+              |> InvoiceIngestWorker.new()
+              |> Oban.insert!()
+            end
+
+          :skip ->
+            :ok
+        end
+      end)
+
+      :ok
+    end
+  end
+
+  # DO billing_history items have shape:
+  #   {"type" => "Invoice", "description" => "Invoice for September 2024",
+  #    "amount" => "-100.00", "date" => "2024-10-01T00:00:00Z",
+  #    "invoice_id" => "uuid-here", "invoice_uuid" => "uuid-here"}
+  # We only care about Invoice rows (skip Payment, Credit, etc).
+  defp extract_invoice_attrs(%{"type" => "Invoice"} = item) do
+    case item["invoice_uuid"] || item["invoice_id"] do
+      nil ->
+        :skip
+
+      uuid ->
+        {:ok,
+         %{
+           provider: "digitalocean",
+           provider_invoice_id: uuid,
+           invoice_period: derive_period(item["date"]),
+           amount_cents: dollars_to_cents_abs(item["amount"]),
+           status: "issued",
+           issued_at: parse_iso(item["date"]),
+           raw: item
+         }}
+    end
+  end
+
+  defp extract_invoice_attrs(_), do: :skip
+
+  # Invoice dated YYYY-MM-01 covers the previous month. Derive period as
+  # first-of-the-previous-month.
+  defp derive_period(nil), do: Date.utc_today() |> Date.add(-30) |> first_of_month()
+
+  defp derive_period(iso) do
+    case DateTime.from_iso8601(iso) do
+      {:ok, dt, _} -> dt |> DateTime.to_date() |> Date.add(-1) |> first_of_month()
+      _ -> Date.utc_today() |> first_of_month()
+    end
+  end
+
+  defp first_of_month(%Date{year: y, month: m}), do: Date.new!(y, m, 1)
+
+  defp dollars_to_cents_abs(nil), do: nil
+
+  defp dollars_to_cents_abs(value) when is_binary(value) do
+    case Float.parse(value) do
+      {float, _} -> abs(round(float * 100))
+      :error -> nil
+    end
+  end
+
+  defp dollars_to_cents_abs(value) when is_number(value), do: abs(round(value * 100))
+
+  defp parse_iso(nil), do: nil
+
+  defp parse_iso(str) when is_binary(str) do
+    case DateTime.from_iso8601(str) do
+      {:ok, dt, _} -> DateTime.truncate(dt, :second)
+      _ -> nil
+    end
+  end
+end
diff --git a/lib/arcadia_cloud/sync/invoice_ingest_worker.ex b/lib/arcadia_cloud/sync/invoice_ingest_worker.ex
new file mode 100644
index 0000000..75cd2f2
--- /dev/null
+++ b/lib/arcadia_cloud/sync/invoice_ingest_worker.ex
@@ -0,0 +1,194 @@
+defmodule ArcadiaCloud.Sync.InvoiceIngestWorker do
+  @moduledoc """
+  Fetch a single invoice's CSV, parse line items, replace cost_lines,
+  then match each line to a cloud_resource by (kind, name).
+
+  Enqueued per invoice by BillingHistoryWorker. Per-invoice idempotency —
+  re-runs replace the line set in one transaction. Marks invoice
+  `lines_ingested_at` on success.
+  """
+
+  use Oban.Worker, queue: :cloud_billing, max_attempts: 3
+
+  alias ArcadiaCloud.Billing
+  alias ArcadiaCloud.Billing.CloudInvoice
+  alias ArcadiaCloud.DigitalOcean.Client
+  alias ArcadiaCloud.Repo
+
+  NimbleCSV.define(InvoiceCsv, separator: ",", escape: "\"")
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"invoice_id" => invoice_id}}) do
+    invoice = Repo.get!(CloudInvoice, invoice_id)
+    do_ingest(invoice)
+  end
+
+  defp do_ingest(%CloudInvoice{provider_invoice_id: uuid} = invoice) do
+    with {:ok, csv} <- Client.fetch_invoice_csv(uuid) do
+      lines = parse_csv(csv, invoice.invoice_period)
+
+      {:ok, _} = Billing.replace_cost_lines(invoice, lines)
+      matched = Billing.match_cost_lines_to_resources(invoice)
+      {:ok, _} = Billing.mark_invoice_ingested(invoice)
+
+      {:ok, %{lines: length(lines), matched: matched}}
+    end
+  end
+
+  # ---- CSV parsing ----------------------------------------------------------
+
+  # DO invoice CSV columns (as of v2 API):
+  #   product, group description, description, hours, start, end, USD,
+  #   project name, category
+  #
+  # Header is on the first line; we use it to find columns rather than
+  # rely on order (DO occasionally adds columns).
+  defp parse_csv(csv, period) do
+    rows =
+      csv
+      |> InvoiceCsv.parse_string(skip_headers: false)
+
+    case rows do
+      [headers | data] ->
+        index = build_index(headers)
+        Enum.map(data, &row_to_line_attrs(&1, index, period))
+
+      _ ->
+        []
+    end
+  end
+
+  defp build_index(headers) do
+    headers
+    |> Enum.with_index()
+    |> Enum.into(%{}, fn {h, i} -> {String.downcase(String.trim(h)), i} end)
+  end
+
+  defp row_to_line_attrs(row, index, period) do
+    product = at(row, index, "product")
+    description = at(row, index, "description") || at(row, index, "group_description")
+    hours = at(row, index, "hours")
+    usd = at(row, index, "usd")
+    start_at = at(row, index, "start")
+    end_at = at(row, index, "end")
+    project_name = at(row, index, "project_name")
+    category = at(row, index, "category")
+
+    %{
+      invoice_period: period,
+      kind: derive_kind(product, category),
+      description: description,
+      qty: parse_decimal(hours),
+      unit: if(hours, do: "hours", else: nil),
+      amount_cents: parse_cents(usd),
+      unit_cost_cents: nil,
+      start_at: parse_datetime(start_at),
+      end_at: parse_datetime(end_at),
+      project_name: project_name,
+      category: category,
+      raw: %{
+        "product" => product,
+        "category" => category,
+        "row" => row
+      }
+    }
+  end
+
+  defp at(row, index, key) do
+    case Map.get(index, key) do
+      nil -> nil
+      i -> Enum.at(row, i) |> blank_to_nil()
+    end
+  end
+
+  defp blank_to_nil(""), do: nil
+  defp blank_to_nil(other), do: other
+
+  # Best-effort mapping from DO product/category strings to our cloud_resources.kind.
+  defp derive_kind(product, _category) when is_binary(product) do
+    p = String.downcase(product)
+
+    cond do
+      String.contains?(p, "droplet") -> "droplet"
+      String.contains?(p, "volume") -> "volume"
+      String.contains?(p, "snapshot") -> "snapshot"
+      String.contains?(p, "load balancer") -> "load_balancer"
+      String.contains?(p, "load_balancer") -> "load_balancer"
+      String.contains?(p, "floating ip") -> "floating_ip"
+      String.contains?(p, "spaces") -> "spaces_bucket"
+      String.contains?(p, "dns") -> "dns_zone"
+      String.contains?(p, "managed database") -> "managed_db"
+      String.contains?(p, "kubernetes") -> "k8s_cluster"
+      true -> nil
+    end
+  end
+
+  defp derive_kind(_, _), do: nil
+
+  defp parse_cents(nil), do: 0
+
+  defp parse_cents(value) when is_binary(value) do
+    cleaned = value |> String.replace(["$", ",", " "], "")
+
+    case Float.parse(cleaned) do
+      {f, _} -> round(f * 100)
+      :error -> 0
+    end
+  end
+
+  defp parse_decimal(nil), do: nil
+
+  defp parse_decimal(value) when is_binary(value) do
+    case Decimal.parse(value) do
+      {dec, _} -> dec
+      :error -> nil
+    end
+  end
+
+  # DO CSV uses "2026-04-01 00:00:00 +0000" (space separator, RFC822 offset).
+  # Also handle "2026-04-01T00:00:00Z" (ISO) and plain "YYYY-MM-DD".
+  defp parse_datetime(nil), do: nil
+  defp parse_datetime(""), do: nil
+
+  defp parse_datetime(str) when is_binary(str) do
+    cond do
+      String.contains?(str, "T") -> parse_iso_datetime(str)
+      String.contains?(str, " ") -> parse_space_datetime(str)
+      true -> parse_date_only(str)
+    end
+  end
+
+  defp parse_iso_datetime(str) do
+    case DateTime.from_iso8601(str) do
+      {:ok, dt, _} -> DateTime.truncate(dt, :second)
+      _ -> nil
+    end
+  end
+
+  # "2026-04-01 00:00:00 +0000" → ISO equivalent
+  defp parse_space_datetime(str) do
+    [date_part, rest] = String.split(str, " ", parts: 2)
+    [time_part | maybe_offset] = String.split(rest, " ", parts: 2)
+    iso = date_part <> "T" <> time_part <> normalize_offset(maybe_offset)
+    parse_iso_datetime(iso)
+  end
+
+  defp normalize_offset([]), do: "Z"
+  defp normalize_offset([off]) when is_binary(off), do: normalize_offset_str(off)
+
+  defp normalize_offset_str("+0000"), do: "Z"
+  defp normalize_offset_str("-0000"), do: "Z"
+
+  defp normalize_offset_str(<<sign::binary-1, hh::binary-2, mm::binary-2>>) when sign in ["+", "-"] do
+    sign <> hh <> ":" <> mm
+  end
+
+  defp normalize_offset_str(_), do: "Z"
+
+  defp parse_date_only(str) do
+    case Date.from_iso8601(str) do
+      {:ok, date} -> DateTime.new!(date, ~T[00:00:00], "Etc/UTC")
+      _ -> nil
+    end
+  end
+end
diff --git a/lib/arcadia_cloud_web/controllers/billing_controller.ex b/lib/arcadia_cloud_web/controllers/billing_controller.ex
new file mode 100644
index 0000000..012a891
--- /dev/null
+++ b/lib/arcadia_cloud_web/controllers/billing_controller.ex
@@ -0,0 +1,101 @@
+defmodule ArcadiaCloudWeb.BillingController do
+  @moduledoc """
+  Cost ingestion read endpoints. Phase 1 is operator-only (platform_admin).
+  Tenant-scoped per-deployment cost queries land in phase 3 with the
+  monthly_margin_rollup view.
+  """
+
+  use ArcadiaCloudWeb, :controller
+
+  alias ArcadiaCloud.Billing
+
+  def balance(conn, _params) do
+    with :ok <- require_platform_admin(conn) do
+      snap = Billing.latest_balance()
+      json(conn, %{balance: shape_balance(snap)})
+    end
+  end
+
+  def cost_lines(conn, params) do
+    with :ok <- require_platform_admin(conn) do
+      opts =
+        []
+        |> maybe_put(:period, parse_date(params["period"]))
+        |> maybe_put(:kind, params["kind"])
+        |> maybe_put(:limit, parse_int(params["limit"]) || 200)
+
+      lines =
+        Billing.list_cost_lines(opts)
+        |> Enum.map(&shape_line/1)
+
+      json(conn, %{cost_lines: lines, count: length(lines)})
+    end
+  end
+
+  # ---- helpers --------------------------------------------------------------
+
+  defp require_platform_admin(conn) do
+    identity = conn.assigns.current_identity
+
+    if is_list(identity.roles) and "platform_admin" in identity.roles do
+      :ok
+    else
+      conn
+      |> put_status(:forbidden)
+      |> json(%{error: "platform_admin_required"})
+      |> halt()
+    end
+  end
+
+  defp shape_balance(nil), do: nil
+
+  defp shape_balance(s) do
+    %{
+      provider: s.provider,
+      month_to_date_balance_cents: s.month_to_date_balance_cents,
+      account_balance_cents: s.account_balance_cents,
+      month_to_date_usage_cents: s.month_to_date_usage_cents,
+      generated_at: s.generated_at
+    }
+  end
+
+  defp shape_line(l) do
+    %{
+      id: l.id,
+      invoice_id: l.invoice_id,
+      resource_id: l.resource_id,
+      invoice_period: l.invoice_period,
+      kind: l.kind,
+      description: l.description,
+      qty: l.qty,
+      unit: l.unit,
+      amount_cents: l.amount_cents,
+      project_name: l.project_name,
+      category: l.category,
+      matched_at: l.matched_at
+    }
+  end
+
+  defp parse_date(nil), do: nil
+  defp parse_date(""), do: nil
+
+  defp parse_date(str) do
+    case Date.from_iso8601(str) do
+      {:ok, d} -> d
+      _ -> nil
+    end
+  end
+
+  defp parse_int(nil), do: nil
+  defp parse_int(""), do: nil
+
+  defp parse_int(str) when is_binary(str) do
+    case Integer.parse(str) do
+      {n, _} -> n
+      :error -> nil
+    end
+  end
+
+  defp maybe_put(opts, _key, nil), do: opts
+  defp maybe_put(opts, key, value), do: Keyword.put(opts, key, value)
+end
diff --git a/lib/arcadia_cloud_web/router.ex b/lib/arcadia_cloud_web/router.ex
index 8e7b792..e1aab13 100644
--- a/lib/arcadia_cloud_web/router.ex
+++ b/lib/arcadia_cloud_web/router.ex
@@ -19,5 +19,8 @@ defmodule ArcadiaCloudWeb.Router do
     pipe_through [:api, :authed]
 
     get "/inventory", InventoryController, :index
+
+    get "/billing/balance", BillingController, :balance
+    get "/billing/cost-lines", BillingController, :cost_lines
   end
 end
diff --git a/mix.exs b/mix.exs
index 1fadf0e..f16d91a 100644
--- a/mix.exs
+++ b/mix.exs
@@ -52,7 +52,8 @@ defmodule ArcadiaCloud.MixProject do
       {:guardian, "~> 2.3"},
       {:cors_plug, "~> 3.0"},
       {:oban, "~> 2.18"},
-      {:req, "~> 0.5"}
+      {:req, "~> 0.5"},
+      {:nimble_csv, "~> 1.2"}
     ]
   end
 
diff --git a/mix.lock b/mix.lock
index aabd0fe..3a5f347 100644
--- a/mix.lock
+++ b/mix.lock
@@ -15,6 +15,7 @@
   "jose": {:hex, :jose, "1.11.12", "06e62b467b61d3726cbc19e9b5489f7549c37993de846dfb3ee8259f9ed208b3", [:mix, :rebar3], [], "hexpm", "31e92b653e9210b696765cdd885437457de1add2a9011d92f8cf63e4641bab7b"},
   "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
   "mint": {:hex, :mint, "1.8.0", "b964eaf4416f2dee2ba88968d52239fca5621b0402b9c95f55a08eb9d74803e9", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "f3c572c11355eccf00f22275e9b42463bc17bd28db13be1e28f8e0bb4adbc849"},
+  "nimble_csv": {:hex, :nimble_csv, "1.3.0", "b7f998dc62b222bce9596e46f028c7a5af04cb5dde6df2ea197c583227c54971", [:mix], [], "hexpm", "41ccdc18f7c8f8bb06e84164fc51635321e80d5a3b450761c4997d620925d619"},
   "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},
   "oban": {:hex, :oban, "2.22.1", "9d2a38cec95070b31c1e274fae55f3925089f62159d8f3facabee0454d55b257", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:igniter, "~> 0.5", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.20", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 1.3", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "af2508c156c5b0ec30b21b0883babf7e2716af35ed5d264095896103fe3cea37"},
diff --git a/priv/repo/migrations/20260519130000_create_billing.exs b/priv/repo/migrations/20260519130000_create_billing.exs
new file mode 100644
index 0000000..f67f7da
--- /dev/null
+++ b/priv/repo/migrations/20260519130000_create_billing.exs
@@ -0,0 +1,69 @@
+defmodule ArcadiaCloud.Repo.Migrations.CreateBilling do
+  use Ecto.Migration
+
+  def change do
+    # Hourly snapshot of DO account balance + month-to-date usage.
+    # Drives the live-accrual side of the cost dashboard.
+    create table(:cloud_balance_snapshots, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+      add :provider, :string, null: false
+      add :month_to_date_balance_cents, :integer
+      add :account_balance_cents, :integer
+      add :month_to_date_usage_cents, :integer
+      add :generated_at, :utc_datetime, null: false
+      add :raw, :map
+
+      timestamps(type: :utc_datetime, updated_at: false)
+    end
+
+    create index(:cloud_balance_snapshots, [:provider, :generated_at])
+
+    # Monthly invoice summary headers (one row per provider invoice).
+    create table(:cloud_invoices, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+      add :provider, :string, null: false
+      add :provider_invoice_id, :string, null: false
+      add :invoice_period, :date, null: false
+      add :amount_cents, :integer
+      add :status, :string, default: "open", null: false
+      add :issued_at, :utc_datetime
+      add :csv_fetched_at, :utc_datetime
+      add :lines_ingested_at, :utc_datetime
+      add :raw, :map
+
+      timestamps(type: :utc_datetime)
+    end
+
+    create unique_index(:cloud_invoices, [:provider, :provider_invoice_id])
+    create index(:cloud_invoices, [:invoice_period])
+
+    # Per-line-item COGS. One row per CSV line of a provider invoice.
+    # Matched to cloud_resources where possible by (kind, name, region).
+    create table(:cloud_cost_lines, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+      add :invoice_id, references(:cloud_invoices, type: :binary_id, on_delete: :delete_all),
+        null: false
+      add :resource_id, references(:cloud_resources, type: :binary_id, on_delete: :nilify_all)
+      add :invoice_period, :date, null: false
+      add :kind, :string
+      add :description, :string
+      add :qty, :decimal
+      add :unit, :string
+      add :unit_cost_cents, :integer
+      add :amount_cents, :integer, null: false
+      add :start_at, :utc_datetime
+      add :end_at, :utc_datetime
+      add :project_name, :string
+      add :category, :string
+      add :matched_at, :utc_datetime
+      add :raw, :map
+
+      timestamps(type: :utc_datetime, updated_at: false)
+    end
+
+    create index(:cloud_cost_lines, [:invoice_id])
+    create index(:cloud_cost_lines, [:resource_id])
+    create index(:cloud_cost_lines, [:invoice_period])
+    create index(:cloud_cost_lines, [:kind])
+  end
+end