diff --git a/backend/app/core/database.py b/backend/app/core/database.py
index a3540eea4..c58638f49 100644
--- a/backend/app/core/database.py
+++ b/backend/app/core/database.py
@@ -176,6 +176,7 @@ async def init_db():
         external_link,
         filament,
         filament_sku_settings,
+        finance,
         github_backup,
         group,
         kprofile_note,
@@ -671,6 +672,9 @@ async def run_migrations(conn):
     # Migration: Add is_favorite column to print_archives
     await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN is_favorite BOOLEAN DEFAULT 0")
 
+    # Migration: Add wallet_charge_skipped column to print_archives so deleted print charges stay deleted
+    await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN wallet_charge_skipped BOOLEAN DEFAULT 0")
+
     # Migration: Add content_hash column to print_archives for duplicate detection
     await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN content_hash VARCHAR(64)")
 
@@ -698,6 +702,31 @@ async def run_migrations(conn):
     # Migration: Add is_deleted column to maintenance_types for soft-deletes
     await _safe_execute(conn, "ALTER TABLE maintenance_types ADD COLUMN is_deleted BOOLEAN DEFAULT 0")
 
+    # Migration: Add cost_center columns expected by current finance model
+    await _safe_execute(conn, "ALTER TABLE cost_centers ADD COLUMN code VARCHAR(32)")
+    await _safe_execute(conn, "ALTER TABLE cost_centers ADD COLUMN is_private BOOLEAN DEFAULT 0")
+    await _safe_execute(
+        conn,
+        "ALTER TABLE cost_centers ADD COLUMN owner_user_id INTEGER REFERENCES users(id) ON DELETE SET NULL",
+    )
+    await _safe_execute(conn, "ALTER TABLE cost_centers ADD COLUMN total_budget FLOAT")
+    await _safe_execute(conn, "ALTER TABLE cost_centers ADD COLUMN monthly_budget FLOAT")
+    await _safe_execute(conn, "ALTER TABLE cost_centers ADD COLUMN created_at DATETIME")
+    await _safe_execute(conn, "ALTER TABLE cost_centers ADD COLUMN updated_at DATETIME")
+
+    # Backfill empty cost center codes on upgraded databases.
+    if is_sqlite():
+        await _safe_execute(
+            conn,
+            "UPDATE cost_centers SET code = lower(hex(randomblob(6))) WHERE code IS NULL OR trim(code) = ''",
+        )
+    else:
+        await _safe_execute(
+            conn,
+            "UPDATE cost_centers SET code = substr(md5(random()::text || clock_timestamp()::text), 1, 12) "
+            "WHERE code IS NULL OR btrim(code) = ''",
+        )
+
     # Migration: Add custom_interval_type column to printer_maintenance
     await _safe_execute(conn, "ALTER TABLE printer_maintenance ADD COLUMN custom_interval_type VARCHAR(20)")
 
@@ -716,6 +745,10 @@ async def run_migrations(conn):
     await _safe_execute(conn, "ALTER TABLE notification_providers ADD COLUMN daily_digest_enabled BOOLEAN DEFAULT 0")
     await _safe_execute(conn, "ALTER TABLE notification_providers ADD COLUMN daily_digest_time VARCHAR(5)")
 
+    # Migration: Add print_run_id to wallet_transactions so repeated prints of the same archive
+    # can be billed independently without mutating archive history.
+    await _safe_execute(conn, "ALTER TABLE wallet_transactions ADD COLUMN print_run_id VARCHAR(100)")
+
     # Migration: Add missing-spool-assignment print-start notification toggle
     try:
         async with conn.begin_nested():
@@ -760,6 +793,16 @@ async def run_migrations(conn):
         "CREATE UNIQUE INDEX IF NOT EXISTS uq_oidc_link_user_provider ON user_oidc_links (user_id, provider_id)",
     )
 
+    # Migration: Add unique indexes to prevent duplicate print-charge transactions
+    await _safe_execute(
+        conn,
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_wallet_transactions_print_run ON wallet_transactions (transaction_type, print_run_id)",
+    )
+    await _safe_execute(
+        conn,
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_wallet_transactions_archive ON wallet_transactions (transaction_type, print_archive_id)",
+    )
+
     # Migration: Create FTS5 virtual table for archive full-text search (SQLite only)
     # PostgreSQL uses tsvector + GIN index instead (set up in archives.py search route)
     if is_sqlite():
@@ -2078,6 +2121,61 @@ async def run_migrations(conn):
     # Migration: Auto-print G-code injection (#422)
     await _safe_execute(conn, "ALTER TABLE print_queue ADD COLUMN gcode_injection BOOLEAN DEFAULT FALSE NOT NULL")
 
+    # Migration: Store estimated print cost for budget checks before queued jobs start
+    await _safe_execute(conn, "ALTER TABLE print_queue ADD COLUMN estimated_cost FLOAT")
+
+    # Migration: Persist active budget reservations for accepted background dispatch jobs.
+    if is_sqlite():
+        await _safe_execute(
+            conn,
+            """
+            CREATE TABLE IF NOT EXISTS budget_reservations (
+                id INTEGER PRIMARY KEY,
+                cost_center_id INTEGER NOT NULL REFERENCES cost_centers(id) ON DELETE CASCADE,
+                amount FLOAT NOT NULL,
+                status VARCHAR(20),
+                source_type VARCHAR(50),
+                source_id INTEGER,
+                print_archive_id INTEGER REFERENCES print_archives(id) ON DELETE SET NULL,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                released_at DATETIME
+            )
+            """,
+        )
+    else:
+        await _safe_execute(
+            conn,
+            """
+            CREATE TABLE IF NOT EXISTS budget_reservations (
+                id SERIAL PRIMARY KEY,
+                cost_center_id INTEGER NOT NULL REFERENCES cost_centers(id) ON DELETE CASCADE,
+                amount FLOAT NOT NULL,
+                status VARCHAR(20),
+                source_type VARCHAR(50),
+                source_id INTEGER,
+                print_archive_id INTEGER REFERENCES print_archives(id) ON DELETE SET NULL,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                released_at TIMESTAMP
+            )
+            """,
+        )
+    await _safe_execute(
+        conn, "CREATE INDEX IF NOT EXISTS ix_budget_reservations_cost_center_id ON budget_reservations (cost_center_id)"
+    )
+    await _safe_execute(
+        conn, "CREATE INDEX IF NOT EXISTS ix_budget_reservations_status ON budget_reservations (status)"
+    )
+    await _safe_execute(
+        conn, "CREATE INDEX IF NOT EXISTS ix_budget_reservations_source_type ON budget_reservations (source_type)"
+    )
+    await _safe_execute(
+        conn, "CREATE INDEX IF NOT EXISTS ix_budget_reservations_source_id ON budget_reservations (source_id)"
+    )
+    await _safe_execute(
+        conn,
+        "CREATE INDEX IF NOT EXISTS ix_budget_reservations_print_archive_id ON budget_reservations (print_archive_id)",
+    )
+
     # Migration: Add backup_spools and backup_archives columns to github_backup_config
     await _safe_execute(conn, "ALTER TABLE github_backup_config ADD COLUMN backup_spools BOOLEAN DEFAULT 0")
     await _safe_execute(conn, "ALTER TABLE github_backup_config ADD COLUMN backup_archives BOOLEAN DEFAULT 0")
@@ -3216,6 +3314,18 @@ async def seed_default_groups():
         "library:read": "library:read_own",
     }
 
+    FINANCE_PERMISSION_MIGRATION = {
+        "finance:read_own": "cost_centers:read_own",
+        "finance:read_all": "cost_centers:read_all",
+        "finance:transactions:create": "cost_centers:modify",
+        "finance:create_transactions": "cost_centers:modify",
+        "finance:createTransactions:create": "cost_centers:modify",
+        "finance:cost_centers:create": "cost_centers:create",
+        "finance:cost_centers:update": "cost_centers:modify",
+        "finance:cost_centers:assign_users": "cost_centers:modify",
+        "finance:budgets:update": "cost_centers:modify",
+    }
+
     async with async_session() as session:
         # Get existing groups
         result = await session.execute(select(Group))
@@ -3256,6 +3366,16 @@ async def seed_default_groups():
                                 "Migrated permission '%s' to '%s' in group '%s'", old_perm, new_perm, group_name
                             )
 
+                    for old_perm, new_perm in FINANCE_PERMISSION_MIGRATION.items():
+                        if old_perm in new_permissions:
+                            new_permissions.remove(old_perm)
+                            if new_perm not in new_permissions:
+                                new_permissions.append(new_perm)
+                            updated = True
+                            logger.info(
+                                "Migrated permission '%s' to '%s' in group '%s'", old_perm, new_perm, group_name
+                            )
+
                     # For Administrators, also ensure they get *_all permissions if they have any new *_own
                     if group_name == "Administrators":
                         for _own_perm, all_perm in [
@@ -3514,3 +3634,89 @@ async def seed_color_catalog():
             )
         await session.commit()
         logger.info("Seeded %d default color catalog entries", len(DEFAULT_COLOR_CATALOG))
+
+
+async def repair_wallet_ledger():
+    """Repair wallet ledger balance_after values.
+
+    This is called during database initialization to ensure all balance_after
+    values are correct after code changes. Fixes old balance_after values to:
+    - Personal transactions: per-user running balance
+    - Cost-center transactions: global running balance for the entire cost center
+    Also updates UserWallet.balance to match final personal transaction balance.
+    """
+    from sqlalchemy import select
+
+    from backend.app.models.finance import WalletTransaction
+
+    async with async_session() as session:
+        # Check if there are any transactions first
+        result = await session.execute(select(WalletTransaction).limit(1))
+        if not result.scalar_one_or_none():
+            logger.info("No wallet transactions found, skipping ledger rebuild")
+            return
+
+        updated_count = await repair_wallet_ledger_internal(session)
+        await session.commit()
+
+        if updated_count > 0:
+            logger.info("Repaired wallet ledger: updated %d transactions/wallets", updated_count)
+
+
+async def repair_wallet_ledger_internal(session: AsyncSession):
+    """Internal helper that repairs wallet ledger using an existing session.
+
+    Used by API endpoints that need to rebuild the ledger within their own transaction.
+    """
+    from sqlalchemy import select
+
+    from backend.app.models.finance import UserWallet, WalletTransaction
+
+    # Get ALL transactions sorted by timestamp
+    result = await session.execute(
+        select(WalletTransaction).order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc())
+    )
+    all_transactions = result.scalars().all()
+
+    if not all_transactions:
+        return 0
+
+    # Build running balances per (user, cost_center_id) pair
+    cc_running_balances: dict[int, float] = {}  # cost_center_id -> running balance
+    user_personal_balances: dict[int, float] = {}  # user_id -> personal running balance
+
+    tx_updates: list[tuple[WalletTransaction, float]] = []
+
+    for tx in all_transactions:
+        if tx.cost_center_id is None:
+            # Personal transaction: per-user running balance
+            current = user_personal_balances.get(tx.user_id, 0.0)
+            new_balance = current + float(tx.amount)
+            user_personal_balances[tx.user_id] = new_balance
+            tx_updates.append((tx, new_balance))
+        else:
+            # Cost-center transaction: global running balance for this cost center
+            current = cc_running_balances.get(tx.cost_center_id, 0.0)
+            new_balance = current + float(tx.amount)
+            cc_running_balances[tx.cost_center_id] = new_balance
+            tx_updates.append((tx, new_balance))
+
+    # Update all transactions with the new balance_after values
+    updated_count = 0
+    for tx, new_balance in tx_updates:
+        if tx.balance_after != new_balance:
+            tx.balance_after = new_balance
+            session.add(tx)
+            updated_count += 1
+
+    # Update UserWallet balances to match final personal balances
+    for user_id, balance in user_personal_balances.items():
+        wallet_result = await session.execute(select(UserWallet).where(UserWallet.user_id == user_id))
+        wallet = wallet_result.scalar_one_or_none()
+        if wallet and wallet.balance != balance:
+            wallet.balance = balance
+            session.add(wallet)
+            updated_count += 1
+
+    await session.flush()
+    return updated_count
diff --git a/backend/app/core/permissions.py b/backend/app/core/permissions.py
index 01406ff1f..91b0ecd8f 100644
--- a/backend/app/core/permissions.py
+++ b/backend/app/core/permissions.py
@@ -139,6 +139,12 @@ class Permission(StrEnum):
     STATS_READ = "stats:read"
     STATS_FILTER_BY_USER = "stats:filter_by_user"
 
+    # Cost Centers
+    COST_CENTERS_READ_OWN = "cost_centers:read_own"
+    COST_CENTERS_READ_ALL = "cost_centers:read_all"
+    COST_CENTERS_MODIFY = "cost_centers:modify"
+    COST_CENTERS_CREATE = "cost_centers:create"
+
     # System Info
     SYSTEM_READ = "system:read"
 
@@ -300,6 +306,12 @@ PERMISSION_CATEGORIES = {
         Permission.STATS_READ,
         Permission.STATS_FILTER_BY_USER,
     ],
+    "Finance": [
+        Permission.COST_CENTERS_READ_OWN,
+        Permission.COST_CENTERS_READ_ALL,
+        Permission.COST_CENTERS_MODIFY,
+        Permission.COST_CENTERS_CREATE,
+    ],
     "System": [
         Permission.SYSTEM_READ,
     ],
@@ -450,6 +462,8 @@ DEFAULT_GROUPS = {
             Permission.PRINTER_SENSOR_HISTORY_READ.value,
             Permission.STATS_READ.value,
             Permission.SYSTEM_READ.value,
+            # Finance - own visibility
+            Permission.COST_CENTERS_READ_OWN.value,
             # Settings - read only
             Permission.SETTINGS_READ.value,
             # WebSocket
diff --git a/backend/app/schemas/archive.py b/backend/app/schemas/archive.py
index 7f1ca2cfc..a4dd64d35 100644
--- a/backend/app/schemas/archive.py
+++ b/backend/app/schemas/archive.py
@@ -240,4 +240,6 @@ class ReprintRequest(BaseModel):
     layer_inspect: bool = False
     timelapse: bool = False
     use_ams: bool = True  # Not exposed in UI, but needed for API
+    cost_center_id: int | None = None
+    estimated_cost: float | None = None
     nozzle_offset_cali: bool = True  # Dual-nozzle printers only — MQTT-gated (#1682)
diff --git a/backend/app/schemas/finance.py b/backend/app/schemas/finance.py
new file mode 100644
index 000000000..c7e6d7440
--- /dev/null
+++ b/backend/app/schemas/finance.py
@@ -0,0 +1,118 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class WalletBalanceResponse(BaseModel):
+    user_id: int
+    balance: float
+    currency: str
+    updated_at: datetime | None = None
+
+
+class WalletTransactionResponse(BaseModel):
+    id: int
+    user_id: int
+    cost_center_id: int | None = None
+    transaction_type: Literal["print_charge", "deposit", "withdraw", "manual_adjustment"]
+    amount: float
+    balance_after: float | None = None
+    description: str | None = None
+    created_by_user_id: int | None = None
+    print_run_id: str | None = None
+    print_archive_id: int | None = None
+    print_queue_id: int | None = None
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+
+class WalletTransactionListResponse(BaseModel):
+    items: list[WalletTransactionResponse]
+    total: int
+    limit: int
+    offset: int
+
+
+class CostCenterSummaryResponse(BaseModel):
+    id: int
+    name: str
+    is_private: bool
+    owner_user_id: int | None = None
+    is_active: bool
+    total_balance: float = 0.0
+    total_budget: float | None = None
+    monthly_budget: float | None = None
+    budget_mode: str = "none"
+    budget_limit: float | None = None
+    budget_used: float | None = None
+    budget_available: float | None = None
+    can_print: bool = True
+
+    class Config:
+        from_attributes = True
+
+
+class WalletAdjustmentRequest(BaseModel):
+    amount: float = Field(..., gt=0)
+    description: str | None = None
+    cost_center_id: int | None = None
+
+
+class WalletAdjustmentResponse(BaseModel):
+    transaction: WalletTransactionResponse
+    balance: WalletBalanceResponse
+
+
+class TransactionEditRequest(BaseModel):
+    user_id: int | None = None
+    cost_center_id: int | None = None
+    amount: float | None = None
+    description: str | None = None
+
+
+class ManualPrintRequest(BaseModel):
+    user_id: int
+    cost_center_id: int
+    amount: float
+    description: str | None = None
+    created_at: datetime | None = None
+
+
+class CostCenterCreateRequest(BaseModel):
+    name: str = Field(..., min_length=1, max_length=150)
+    total_budget: float | None = None
+    monthly_budget: float | None = None
+    is_active: bool = True
+
+
+class CostCenterUpdateRequest(BaseModel):
+    name: str | None = Field(default=None, min_length=1, max_length=150)
+    is_active: bool | None = None
+
+
+class CostCenterBudgetUpdateRequest(BaseModel):
+    total_budget: float | None = None
+    monthly_budget: float | None = None
+
+
+class CostCenterMemberRequest(BaseModel):
+    user_id: int
+    can_print: bool = True
+
+
+class CostCenterMemberResponse(BaseModel):
+    id: int
+    cost_center_id: int
+    user_id: int
+    can_print: bool
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+
+class CostCenterDetailResponse(CostCenterSummaryResponse):
+    members: list[CostCenterMemberResponse] = []
diff --git a/backend/app/schemas/library.py b/backend/app/schemas/library.py
index bfd2ad96f..82538e704 100644
--- a/backend/app/schemas/library.py
+++ b/backend/app/schemas/library.py
@@ -294,6 +294,8 @@ class FilePrintRequest(BaseModel):
     layer_inspect: bool = False
     timelapse: bool = False
     use_ams: bool = True
+    cost_center_id: int | None = None
+    estimated_cost: float | None = None
     nozzle_offset_cali: bool = True  # Dual-nozzle printers only — MQTT-gated (#1682)
     # Project to associate the resulting archive with
     project_id: int | None = None
diff --git a/backend/app/schemas/print_queue.py b/backend/app/schemas/print_queue.py
index b37136306..4388ad81d 100644
--- a/backend/app/schemas/print_queue.py
+++ b/backend/app/schemas/print_queue.py
@@ -58,6 +58,8 @@ class PrintQueueItemCreate(BaseModel):
     batch_id: int | None = None
     # Project to associate the resulting archive with
     project_id: int | None = None
+    cost_center_id: int | None = None
+    estimated_cost: float | None = None
 
 
 class PrintQueueItemUpdate(BaseModel):
@@ -82,6 +84,8 @@ class PrintQueueItemUpdate(BaseModel):
     nozzle_offset_cali: bool | None = None
     # Auto-print G-code injection
     gcode_injection: bool | None = None
+    cost_center_id: int | None = None
+    estimated_cost: float | None = None
     # H2C dual-nozzle-rack slicer pick (#1780). list[int] per-filament
     # physical nozzle position IDs from BambuStudio's project_file MQTT
     # body; sent back to the printer verbatim on dispatch.
@@ -98,6 +102,8 @@ class PrintQueueItemResponse(BaseModel):
     waiting_reason: str | None = None  # Why a model-based job hasn't started yet
     archive_id: int | None  # None if library_file_id is set (archive created at print start)
     library_file_id: int | None  # For queue items from library files
+    cost_center_id: int | None = None
+    estimated_cost: float | None = None
     position: int
     scheduled_time: UTCDatetime
     require_previous_success: bool
@@ -205,6 +211,8 @@ class PrintQueueBulkUpdate(BaseModel):
     nozzle_offset_cali: bool | None = None
     # Auto-print G-code injection
     gcode_injection: bool | None = None
+    cost_center_id: int | None = None
+    estimated_cost: float | None = None
 
 
 class PrintQueueBulkUpdateResponse(BaseModel):
diff --git a/backend/app/schemas/settings.py b/backend/app/schemas/settings.py
index e046a009e..ca4f01ae0 100644
--- a/backend/app/schemas/settings.py
+++ b/backend/app/schemas/settings.py
@@ -298,6 +298,26 @@ class AppSettings(BaseModel):
         default=5, ge=1, le=60, description="Minutes between staggered printer groups"
     )
 
+    # Finance budget window settings
+    billing_enabled: bool = Field(
+        default=False,
+        description="Enable cost-center billing enforcement for print and queue operations",
+    )
+    printer_kill_switch_enabled: bool = Field(
+        default=False,
+        description="Immediately stop printer jobs that start without Bambuddy authorization",
+    )
+    finance_budget_reset_day: int = Field(
+        default=1,
+        ge=1,
+        le=31,
+        description="Day of month when monthly finance budget window resets (1-31, clamped for short months)",
+    )
+    finance_budget_reset_timezone: str = Field(
+        default="UTC",
+        description="IANA timezone for finance monthly budget reset calculation (e.g., Europe/Berlin)",
+    )
+
     # Plate-clear confirmation for queue scheduling
     require_plate_clear: bool = Field(
         default=False,
@@ -477,6 +497,10 @@ class AppSettingsUpdate(BaseModel):
     default_nozzle_offset_cali: bool | None = None
     stagger_group_size: int | None = Field(default=None, ge=1, le=50)
     stagger_interval_minutes: int | None = Field(default=None, ge=1, le=60)
+    billing_enabled: bool | None = None
+    printer_kill_switch_enabled: bool | None = None
+    finance_budget_reset_day: int | None = Field(default=None, ge=1, le=31)
+    finance_budget_reset_timezone: str | None = None
     require_plate_clear: bool | None = None
     queue_shortest_first: bool | None = None
     nozzle_temp_presets: str | None = None