uweziegenhagen.de

In this final article of this series I will add another data mart. In the models/marts subfolder we add fx_current_rates.sql with the following SQL code:

{{ config(materialized='view') }}
 
SELECT
    currency,
    valid_from,
    rate_vs_eur
FROM {{ REF('fx_rates_with_validity') }}
WHERE is_current = TRUE

This will be a new view on top of the „fx_rates_with_validity“ mart. As this view ‚materializes‘ in the Postgres DB, we can use HeidiSQL to retrieve data from it:

In this article we will continue our dbt journey. With dbt-core installed we can ask dbt to set up a new project for us.

Simply go to the root folder where your project folder shall be created and run dbt init projectname. dbt will then ask you for the database adapter and the connection settings as well as for the schema dbt is to build its objects in.

After these questions and answers dbt tells you that you may run dbt debug now to check if everything is fine, BUT: you need to switch first to the subdirectory you created otherwise there will be a meaningless error message:


Encountered an error:
Internal Error
Profile should not be None if loading profile completed


If run in the correct subdirectory dbt debug should not produce any errors. If it does, run the error message through the search engine of your choice.

C:\Users\UWe\Desktop\fx_project>dbt debug
16:52:20  Running with dbt=1.10.13
16:52:20  dbt version: 1.10.13
16:52:20  python version: 3.12.6
16:52:20  python path: C:\Python312\python.exe
16:52:20  os info: Windows-11-10.0.26200-SP0
16:52:20  Using profiles dir at C:\Users\UWe\.dbt
16:52:20  Using profiles.yml file at C:\Users\UWe\.dbt\profiles.yml
16:52:20  Using dbt_project.yml file at C:\Users\UWe\Desktop\fx_project\dbt_project.yml
16:52:20  adapter type: postgres
16:52:20  adapter version: 1.9.1
16:52:20  Configuration:
16:52:20    profiles.yml file [OK found and valid]
16:52:20    dbt_project.yml file [OK found and valid]
16:52:20  Required dependencies:
16:52:20   - git [OK found]

16:52:20  Connection:
16:52:20    host: localhost
16:52:20    port: 5432
16:52:20    user: postgres
16:52:20    database: FX
16:52:20    schema: dbt
16:52:20    connect_timeout: 10
16:52:20    role: None
16:52:20    search_path: None
16:52:20    keepalives_idle: 0
16:52:20    sslmode: None
16:52:20    sslcert: None
16:52:20    sslkey: None
16:52:20    sslrootcert: None
16:52:20    application_name: dbt
16:52:20    retries: 1
16:52:20  Registered adapter: postgres=1.9.1
16:52:20    Connection test: [OK connection ok]

16:52:20  All checks passed!

In the project folder you find a set of subfolders, remove the models/example subfolder, as it may cause error messages.

Next we will create some basic models, using YAML files.

In models, create sources.yml with the following content:

version: 2
sources:
  - name: raw
    schema: raw
    tables:
      - name: ecb_fx_rates

In models, create a subfolder staging and add stg_fx_rates.sql to this staging subfolder with the following content:

{{ config(materialized='view') }}
SELECT
    fx_date,
    currency,
    rate,
    created_at
FROM {{ SOURCE('raw', 'ecb_fx_rates') }}

This will simply select all the data we have in our raw DB schema.

Next comes is the interesting part. Each row gets a valid_from and valid_to date using a window function.

In models, create a subfolder marts and add fx_rates_with_validity.sql to this marts subfolder with the following content:

{{ config(materialized='table') }}
 
WITH base AS (
    SELECT
        fx_date,
        currency,
        rate
    FROM {{ REF('stg_fx_rates') }}
),
 
with_validity AS (
    SELECT
        currency,
        fx_date                        AS valid_from,
        -- valid_to is the day before the next rate, or null if it's the latest
        lead(fx_date) OVER (
            partition BY currency
            ORDER BY fx_date
        ) - INTERVAL '1 day'             AS valid_to,
        rate,
        -- flag the currently active rate
        CASE
            WHEN lead(fx_date) OVER (
                partition BY currency ORDER BY fx_date
            ) IS NULL THEN TRUE
            ELSE FALSE
        END                              AS is_current
    FROM base
)
 
SELECT * FROM with_validity
ORDER BY currency, valid_from

This will give uzs data as the following

It also makes sense now to add some data checks.

In models, create schema.yml with the following content:

version: 2

models:
  - name: stg_fx_rates
    columns:
      - name: rate_date
        tests: [not_null]
      - name: currency
        tests: [not_null]
      - name: rate_vs_eur
        tests: [not_null]

  - name: fx_rates_with_validity
    columns:
      - name: currency
        tests: [not_null]
      - name: valid_from
        tests: [not_null]
      - name: is_current
        tests: [not_null]

We can now run the whole setup:

dbt run
dbt test
dbt docs generate
dbt docs serve

Going forward

Since ECB publishes the latest rate every data at around 16:00 CET the Python job to fetch them should be run each day (apparently after 16:00 CET). Once the raw database is updated you can update the dbt models:

dbt run --select stg_fx_rates fx_rates_with_validity
dbt test --select stg_fx_rates fx_rates_with_validity

In the final part of the series we will add another data market for the latest rates.

Excel 365 aus Office365 hat eine Reihe neuer Funktionen, die ich auch noch nicht kannte. Hier eine Liste, die mir ChatGPT ausgespuckt hat.

Schauen wir uns das Forward Filling an. Je nach Version des SQL Servers funktioniert eine Version oder leider nicht. 🙂

Mit SQL Server 2022 kann man den folgenden Code nutzen, um einen weiteren CTE zu bauen:

Filled AS (
    SELECT
        ContNo,
        MonthEnd,
        Amount,
        last_value(Amount) IGNORE NULLS
        OVER (
            PARTITION BY ContNo
            ORDER BY MonthEnd
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS ForwardFilledAmount
    FROM Combined
)
SELECT * FROM Filled

Versionen von SQL Server vor 2022 unterstützen leider „ignore nulls“ nicht, hier kann man dann folgenden Code nutzen:

Filled AS (
    SELECT
        c.ContNo,
        c.MonthEnd,
        c.Amount,
        ca.ForwardFilledAmount
    FROM Combined c
    OUTER APPLY (
        SELECT TOP 1 Amount AS ForwardFilledAmount
        FROM Combined c2
        WHERE c2.ContNo = c.ContNo
          AND c2.MonthEnd <= c.MonthEnd
          AND c2.Amount IS NOT NULL
        ORDER BY c2.MonthEnd DESC
    ) ca
)
SELECT * FROM Filled

OUTER APPLY funktioniert dabei so ähnlich wie ein LEFT JOIN. Der Unterschied besteht darin, dass beim OUTER APPLY die rechte Seite von der linken Seite abhängt. In einem weiteren Artikel zum Thema werde ich noch ein paar Beispiele dazu zeigen. CROSS APPLY ist ähnlich, hier ist es aber kein Äquivalent zum LEFT JOIN, sondern zum INNER JOIN. Auch dazu mehr in einem der weiteren Artikel.

Wir kennen jetzt das Problem, nun können wir uns an die Lösung wagen! In diesem Blogbeitrag werden wir alles vorbereiten, was wir für das anschließende „Forward Filling“ der Daten brauchen.

Im ersten Schritt bestimmen wir für die einzelnen Verträge die Monatsendwerte von Start des jeweiligen Vertrags bis zu dessen Ende. Dazu nutze ich den den folgenden rekursiven CTE (Common Table Expression)

WITH MonthSequence AS (
    SELECT
        c.ContNo,
        EOMONTH(c.StartDate) AS MonthEnd,
        c.EndDate
    FROM Contracts c
    UNION ALL
    SELECT
        ms.ContNo,
        EOMONTH(DATEADD(MONTH, 1, ms.MonthEnd)),
        ms.EndDate
    FROM MonthSequence ms
    WHERE EOMONTH(DATEADD(MONTH, 1, ms.MonthEnd)) <= ms.EndDate
)

Das erzeugt uns die folgende Sequenz:

Als nächstes suchen wir uns die End-of-Month der Cashflows raus, die wir im nächsten Schritt mit den End-of-Month aus Schritt 1 kombinieren werden.

WITH CashflowBuckets AS (
    SELECT
        ContNo,
        EOMONTH(CashflowDate) AS MonthEnd,
        Amount
    FROM Cashflows
)
SELECT * FROM CashflowBuckets

Die Kombination ist dann recht einfach und gibt uns die Liste aller End-of-Months aus mit den dazugehörigen Cashflows.

WITH MonthSequence AS (
    SELECT
        c.ContNo,
        EOMONTH(c.StartDate) AS MonthEnd,
        c.EndDate
    FROM Contracts c
    UNION ALL
    SELECT
        ms.ContNo,
        EOMONTH(DATEADD(MONTH, 1, ms.MonthEnd)),
        ms.EndDate
    FROM MonthSequence ms
    WHERE EOMONTH(DATEADD(MONTH, 1, ms.MonthEnd)) <= ms.EndDate
)
,CashflowBuckets AS (
    SELECT
        ContNo,
        EOMONTH(CashflowDate) AS MonthEnd,
        Amount
    FROM Cashflows
)
,Combined AS (
   SELECT
        ms.ContNo,
        ms.MonthEnd,
        cb.Amount
    FROM MonthSequence ms
    LEFT JOIN CashflowBuckets cb
        ON ms.ContNo = cb.ContNo
        AND ms.MonthEnd = cb.MonthEnd
)
 
SELECT * FROM Combined

In gelb sind die Einträge markiert, die wir im finalen Schritt mit den jeweils letzten gültigen Cashflow-Werten befüllen müssen.

Ich halte mich schon recht erfahren im Umgang SQL, kürzlich bin ich aber an einer Ecke von SQL vorbeigekommen, die ich auch noch nicht kannte. Dabei handelt es sind um die „OUTER APPLY“ bzw. „CROSS APPLY“ Operatoren.

Um die Lösung herzuleiten betrachten wir zuerst das Problem:

Gegeben seien Kreditverträge mit verschiedenen Zahlplänen. Ein Vertrag zahlt vielleicht monatlich, einer zahlt quartalsweise. Ein Vertrag hat dabei ein Startdatum und ein Enddatum sowie verschiedene Cashflows.

Hier ein paar Testdaten mit den entsprechenden Tabellen, die dazugehörigen SQL-Statements folgen später.

Cashflows

Cashflows

Das Problem ist jetzt, wie kann man eine monatliche Übersicht pro Vertrag bekommen, bei der in den Monaten, wo es kein Cashflow gab, einfach der letzte Cashflow angezeigt wird? Das Endergebnis soll wie folgt aussehen:

Im nächsten Teil fangen wir dann mit den SQL Statements an.

Für ein kleines Projekt brauchte ich die Möglichkeit, flexible Zahlungspläne zu erzeugen. Der folgende Python-Code tut genau das und erzeugt auch gleich passende INSERT Statements für Postgres.

Das CREATE TABLE ist wie folgt:

CREATE TABLE paymentplan (
    contract INTEGER NOT NULL,
    paymentdate INTEGER NOT NULL,
    paymenttype INTEGER NOT NULL,
    amount NUMERIC(12, 2) NOT NULL
);

import pandas as pd
from datetime import datetime
from dateutil.relativedelta import relativedelta
 
def genPaymentPlan(contract, principal, paymentsperyear, interestrate, startdate, years):
    interest = interestrate/100*principal/paymentsperyear
    date = datetime.strptime(startdate, '%Y%m%d').date()
    output = pd.DataFrame(columns=['Contract', 'Paymentdate', 'Paymenttype', 'Amount'])
 
    # Add a row to the DataFrame
    start = {'Contract': contract, 'Paymentdate': startdate, 'Paymenttype': 1, 'Amount' : principal}
    output = pd.concat([output, pd.DataFrame([start])], ignore_index=True)  
 
    for payment in range(paymentsperyear*years):
        new_date = (date + relativedelta(months=12/paymentsperyear))
        date = new_date
        interest_line = {'Contract': contract, 'Paymentdate': new_date.strftime('%Y%m%d'),'Paymenttype': 2, 'Amount' : -1*interest}
        output = pd.concat([output, pd.DataFrame([interest_line])], ignore_index=True)  
 
    end = {'Contract': contract, 'Paymentdate': date.strftime('%Y%m%d'), 'Paymenttype': 3, 'Amount' : -principal}
    output = pd.concat([output, pd.DataFrame([end])], ignore_index=True)          
    return output
 
 
test = genPaymentPlan(123458, 2000, 2, 12, '20260101', 4)
#print(test)
 
 
 
# Generate INSERT statements
print("Generated SQL INSERT statements:\n")
 
for index, row in test.iterrows():
    contract = int(row['Contract'])
    date = row['Paymentdate']  # already string in format YYYY-MM-DD
    type =  row['Paymenttype'] 
    amount = float(row['Amount'])
 
 
    insert = f"INSERT INTO paymentplan (contract, paymentdate,paymenttype, amount) VALUES ({contract}, '{date}', {type} , {amount});"
    print(insert)

Hier ein einfacher Code-Schnipsel, um aus CSV-Dateien Excel-Dateien zu machen.

Es empfiehlt sich auch, die Engine explizit zu setzen, mit der die Excel-Datei geschrieben wird. Standardmäßig nutzt pandas openpyxl, xlsxwriter scheint nach meinen Tests aber um ca. 30% schneller zu sein.

import pandas as pd
from openpyxl import Workbook
from openpyxl.utils.dataframe import dataframe_to_rows
 
 
# output control
files = ['']
 
wb = Workbook()
wb.remove(wb['Sheet'])
 
for index, file in enumerate(files,0):
    temp = pd.read_csv(file + '.csv',sep='\t')
    ws1 = wb.create_sheet()
    ws1.title = file
    rows = dataframe_to_rows(temp, index=False, header=True)
 
    for r_idx, row in enumerate(rows, 1):
        for c_idx, value in enumerate(row, 1):
            ws1.cell(row=r_idx, column=c_idx, value=value)
 
wb.save('files45646.xlsx')
wb.close()

Aus aktuellem Anlass hier ein einfaches Beispiel, wie man Werte in pandas Dataframes ersetzen kann:

import pandas as pd
 
# initialize data of lists.
data = {'Hersteller': ['VW', 'BMW', 'VW', 'Porsche'],
        'Modell': ['Golf', '1er', 'Polo', '911']}
 
 
df = pd.DataFrame(data)
 
 
print(df,'\n')
 
 
df['Hersteller'] = df['Hersteller'].replace(
    {"VW": "Volkswagen", "Horch": "Audi"})
 
 
print(df)

Hier ein Code-Beispiel für die wichtigsten Funktionen von DuckDB.

import duckdb as ddb
import pandas as pd
 
con = ddb.connect(':memory:')
 
con_p = ddb.connect('my_database.db')
 
con_p.execute('CREATE OR REPLACE TABLE telefonnummern(fullname VARCHAR,phone VARCHAR);')
 
 
con_p.execute("INSERT INTO telefonnummern VALUES ('Max Mustermann', '0123-4567890')")
 
print(con_p.sql('SHOW ALL TABLES'))
 
print(con_p.sql('SELECT * FROM telefonnummern;'))
 
 
ddb_object = con_p.sql('SELECT * FROM telefonnummern;')
 
df = ddb_object.to_df()
 
ddb_tuple = ddb_object.fetchall()
 
print(df)
 
print(ddb_tuple)