Add dbt models, docs, and CI skeleton

.github/workflows/dbt.yml

@@ -0,0 +1,49 @@
+name: dbt
+
+on:
+  pull_request:
+  push:
+    branches: [ main ]
+
+jobs:
+  dbt-build-test-docs:
+    runs-on: ubuntu-latest
+    env:
+      DBT_PROFILES_DIR: etl/dbt
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with: { python-version: "3.11" }
+
+      - name: Install dbt (Postgres-Beispiel)
+        run: |
+          python -m pip install -U pip
+          pip install dbt-postgres
+
+      - name: Install deps
+        working-directory: etl/dbt
+        run: dbt deps
+
+      - name: Seed (demo)
+        working-directory: etl/dbt
+        run: dbt seed --full-refresh --vars '{db_schema_raw: seed, db_schema_seed: seed}'
+
+      - name: Build (models + tests)
+        working-directory: etl/dbt
+        run: dbt build --fail-fast --vars '{db_schema_raw: seed}'
+
+      - name: Snapshots (SCD2)
+        working-directory: etl/dbt
+        run: dbt snapshot --vars '{db_schema_raw: seed}'
+
+      - name: Docs generate
+        working-directory: etl/dbt
+        run: dbt docs generate --vars '{db_schema_raw: seed}'
+
+      - name: Upload artifacts (dbt docs site)
+        uses: actions/upload-artifact@v4
+        with:
+          name: dbt-docs-site
+          path: etl/dbt/target

.gitignore

@@ -1 +1,6 @@
 node_modules
+
+# dbt artifacts
+target/
+logs/
+dbt_packages/

docs/dev/dwh_dbt.md

@@ -0,0 +1,25 @@
+# DWH dbt Runbook
+
+## Setup
+- Kopiere `profiles.example.yml` nach `profiles.yml` und passe ggf. `DBT_DUCKDB_PATH` an.
+- Setze `DBT_PROFILES_DIR` auf `etl/dbt` oder starte Befehle in diesem Verzeichnis.
+
+## Build & Tests
+```bash
+dbt deps
+dbt seed --full-refresh --vars '{db_schema_raw: seed, db_schema_seed: seed}'
+dbt run --vars '{db_schema_raw: seed}'
+dbt test --vars '{db_schema_raw: seed}'
+```
+
+## Snapshots
+```bash
+dbt snapshot --vars '{db_schema_raw: seed}'
+```
+
+## Docs
+```bash
+dbt docs generate --vars '{db_schema_raw: seed}'
+```
+
+Die generierte Seite liegt unter `etl/dbt/target/index.html`.

etl/dbt/dbt_project.yml

@@ -0,0 +1,27 @@
+name: infoterminal_dbt
+version: 1.0
+config-version: 2
+profile: infoterminal_dbt
+
+model-paths: ["models"]
+seed-paths: ["seeds"]
+macro-paths: ["macros"]
+snapshot-paths: ["snapshots"]
+
+models:
+  infoterminal_dbt:
+    +materialized: view
+
+seeds:
+  +schema: "{{ var('db_schema_seed','seed') }}"
+  +quote_columns: false
+  infoterminal_dbt:
+    demo_assets:
+      +alias: assets
+      +column_types:
+        ingested_at: timestamp
+    demo_prices:
+      +alias: prices
+      +column_types:
+        ts: date
+        ingested_at: timestamp

etl/dbt/macros/tests/expression_is_true.sql

@@ -0,0 +1,5 @@
+{% test expression_is_true(model, column_name, expression) %}
+select *
+from {{ model }}
+where not ({{ expression }})
+{% endtest %}

etl/dbt/macros/tests/non_negative.sql

@@ -0,0 +1,5 @@
+{% test non_negative(model, column_name) %}
+select *
+from {{ model }}
+where {{ column_name }} < 0
+{% endtest %}

etl/dbt/macros/tests/ratio_0_1.sql

@@ -0,0 +1,5 @@
+{% test ratio_0_1(model, column_name) %}
+select *
+from {{ model }}
+where {{ column_name }} < 0 or {{ column_name }} > 1
+{% endtest %}

etl/dbt/models/marts/core/dim_asset.sql

@@ -0,0 +1,14 @@
+with snapshot as (
+  select * from {{ ref('dim_asset_snapshot') }}
+),
+final as (
+  select
+    row_number() over(order by asset_id, dbt_valid_from) as asset_sk,
+    asset_id,
+    symbol,
+    type,
+    status,
+    case when dbt_valid_to is null then 1 else 0 end as is_current
+  from snapshot
+)
+select * from final

etl/dbt/models/marts/core/exposures.yml

@@ -0,0 +1,14 @@
+version: 2
+
+exposures:
+  - name: superset_prices_dashboard
+    type: dashboard
+    maturity: medium
+    url: "https://superset.example.com/superset/dashboard/price-overview"
+    description: "Preisübersichtsdashboard in Superset."
+    owner:
+      name: "Data Team"
+      email: "data@infoterminal.local"
+    depends_on:
+      - ref('fct_price')
+      - ref('dim_asset')

etl/dbt/models/marts/core/fct_price.sql

@@ -0,0 +1,12 @@
+with prices as (
+  select * from {{ ref('stg_prices') }}
+),
+assets as (
+  select * from {{ ref('dim_asset') }} where is_current = 1
+)
+select
+  a.asset_sk,
+  p.ts,
+  p.close
+from prices p
+join assets a on p.asset_id = a.asset_id

etl/dbt/models/marts/core/schema.yml

@@ -0,0 +1,40 @@
+version: 2
+
+models:
+  - name: dim_asset
+    description: "Stammdaten für Assets (ISIN, Symbol, Typ, Status, Zeitgültigkeit per SCD2)."
+    columns:
+      - name: asset_sk
+        description: "Surrogat-Schlüssel (SCD2)."
+        tests: [not_null, unique]
+      - name: asset_id
+        description: "Natürlicher Schlüssel aus Staging."
+        tests: [not_null]
+      - name: symbol
+        description: "Kurzbezeichnung des Assets."
+      - name: type
+        description: "Asset-Typ."
+        tests:
+          - accepted_values:
+              values: ['equity','etf','crypto','fx','commodity','index']
+      - name: is_current
+        description: "Flag für aktuell gültige SCD2-Zeile."
+        tests:
+          - accepted_values: {values: [0,1]}
+
+  - name: fct_price
+    description: "Faktentabelle für Preise (täglich/minütlich)."
+    columns:
+      - name: asset_sk
+        tests:
+          - not_null
+          - relationships:
+              to: ref('dim_asset')
+              field: asset_sk
+      - name: ts
+        tests: [not_null]
+      - name: close
+        tests:
+          - not_null
+          - non_negative:
+              column_name: close

etl/dbt/models/staging/sources.yml

@@ -0,0 +1,40 @@
+version: 2
+
+sources:
+  - name: raw_finance
+    description: "Rohdaten aus Finanz-Pipelines (OpenBB, ETL)."
+    database: "{{ var('db_database', target.database) }}"
+    schema: "{{ var('db_schema_raw', 'raw_finance') }}"
+    tables:
+      - name: assets
+        description: "Roh-Assets (Symbols, ISIN, Typ)."
+        loaded_at_field: ingested_at
+        freshness:
+          warn_after: {count: 24, period: hour}
+          error_after: {count: 72, period: hour}
+        columns:
+          - name: asset_id
+            tests: [not_null]
+          - name: symbol
+            tests: [not_null]
+          - name: type
+            tests:
+              - accepted_values:
+                  values: ['equity','etf','crypto','fx','commodity','index']
+      - name: prices
+        description: "Preiszeitreihen (OHLC, Volume)."
+        loaded_at_field: ingested_at
+        columns:
+          - name: asset_id
+            tests:
+              - not_null
+              - relationships:
+                  to: ref('dim_asset')
+                  field: asset_id
+          - name: ts
+            tests: [not_null]
+          - name: close
+            tests:
+              - not_null
+              - expression_is_true:
+                  expression: "close >= 0"

etl/dbt/models/staging/stg_assets.sql

@@ -0,0 +1,7 @@
+select
+  asset_id,
+  symbol,
+  type,
+  status,
+  ingested_at
+from {{ source('raw_finance', 'assets') }}

etl/dbt/models/staging/stg_prices.sql

@@ -0,0 +1,6 @@
+select
+  asset_id,
+  ts,
+  close,
+  ingested_at
+from {{ source('raw_finance', 'prices') }}

etl/dbt/packages.yml

@@ -0,0 +1 @@
+packages: []

etl/dbt/profiles.example.yml

@@ -0,0 +1,6 @@
+infoterminal_dbt:
+  target: dev
+  outputs:
+    dev:
+      type: duckdb
+      path: "{{ env_var('DBT_DUCKDB_PATH', 'dbt.duckdb') }}"

etl/dbt/seeds/demo_assets.csv

@@ -0,0 +1,3 @@
+asset_id,symbol,type,status,ingested_at
+A1,ACME,equity,active,2025-01-01T00:00:00Z
+A2,FOO,etf,active,2025-01-01T00:00:00Z

etl/dbt/seeds/demo_prices.csv

@@ -0,0 +1,4 @@
+asset_id,ts,close,ingested_at
+A1,2025-02-01,101.5,2025-02-01T00:00:00Z
+A1,2025-02-02,103.0,2025-02-02T00:00:00Z
+A2,2025-02-01,50.1,2025-02-01T00:00:00Z

etl/dbt/snapshots/dim_asset_snapshot.sql

@@ -0,0 +1,22 @@
+{% snapshot dim_asset_snapshot %}
+
+{{
+  config(
+    target_database = target.database,
+    target_schema = var('db_schema_snapshots','snapshots'),
+    unique_key = 'asset_id',
+    strategy = 'check',
+    check_cols = ['symbol','type','status'],
+    invalidate_hard_deletes = true
+  )
+}}
+
+select
+  asset_id,
+  symbol,
+  type,
+  status,
+  current_timestamp as snapshot_ts
+from {{ ref('stg_assets') }}
+
+{% endsnapshot %}