Riksbank SWEA

-- models/raw/exchange_rates.sql
-- @kind: append
-- @incremental: date
-- @incremental_initial: 2026-04-01

SELECT series, date, value
FROM riksbank('SEKEURPMI,SEKUSDPMI,SEKGBPPMI,SEKNOKPMI,SEKDKKPMI,SEKCHFPMI,SEKJPYPMI')

-- models/staging/exchange_rates.sql
-- @kind: table
-- @constraint: not_null(date)

SELECT
    date::DATE AS date,
    MAX(CASE WHEN series = 'SEKEURPMI' THEN value END)::DECIMAL AS eur,
    MAX(CASE WHEN series = 'SEKUSDPMI' THEN value END)::DECIMAL AS usd,
    MAX(CASE WHEN series = 'SEKGBPPMI' THEN value END)::DECIMAL AS gbp,
    MAX(CASE WHEN series = 'SEKNOKPMI' THEN value END)::DECIMAL AS nok,
    MAX(CASE WHEN series = 'SEKDKKPMI' THEN value END)::DECIMAL AS dkk,
    MAX(CASE WHEN series = 'SEKCHFPMI' THEN value END)::DECIMAL AS chf,
    MAX(CASE WHEN series = 'SEKJPYPMI' THEN value END)::DECIMAL AS jpy
FROM raw.exchange_rates
GROUP BY date
ORDER BY date

ondatrasql run

ondatrasql run raw.exchange_rates

-- Riksbank policy rate
FROM riksbank('SECBREPOEFF')

-- STIBOR rates
FROM riksbank('SEDPT_NSTIBORDELAYC,SEDP3MSTIBORDELAYC,SEDP6MSTIBORDELAYC')

-- Reference and discount rates
FROM riksbank('SECBREFEFF,SECBDISCEFF')

-- Swedish government bonds (2, 5, 10 year)
FROM riksbank('SEGVB2YC,SEGVB5YC,SEGVB10YC')

-- Swedish treasury bills
FROM riksbank('SETB1MBENCHC,SETB3MBENCH,SETB6MBENCH')

-- International (US, Germany, Japan 10-year)
FROM riksbank('USGVB10Y,DEGVB10Y,JPGVB10Y')

SELECT series, date, value
FROM riksbank('SEKEURPMI,SEKUSDPMI')

-- @incremental_initial: 2020-01-01

-- models/mart/revenue_sek.sql
-- @kind: table

SELECT
    r.date,
    r.amount,
    r.currency,
    r.amount * CASE r.currency
        WHEN 'EUR' THEN fx.eur
        WHEN 'USD' THEN fx.usd
        WHEN 'GBP' THEN fx.gbp
    END AS amount_sek
FROM staging.revenue r
JOIN staging.exchange_rates fx ON r.date = fx.date

-- models/staging/exchange_rates.sql
-- @kind: table

PIVOT raw.exchange_rates
ON series
USING MAX(value)
GROUP BY date::DATE AS date
ORDER BY date

RESERVED_COLUMNS = ()

API = {
    "base_url": "https://api.riksbank.se/swea/v1",
    "retry": 3,
    "backoff": 2,
    "rate_limit": {"requests": 1, "per": "1s"},
    "fetch": {
        "args": ["series"],
        "supported_kinds": ["append"],
    },
}

def fetch(series, page, is_backfill=True, last_value="", initial_value=""):
    if is_backfill:
        start_date = _to_iso_date(initial_value)
    else:
        start_date = _next_day(_to_iso_date(last_value))

    series_list = [s.strip() for s in series.split(",")]
    series_idx = 0
    date_cursor = start_date
    # Pin yesterday at pipeline start — stable across midnight
    yesterday = _yesterday()

    if page.cursor != None:
        series_idx = page.cursor["idx"]
        date_cursor = page.cursor["from"]
        yesterday = page.cursor["until"]

    if date_cursor > yesterday:
        abort()

    if series_idx >= len(series_list):
        return {"rows": [], "next": None}

    current_series = series_list[series_idx]
    end_date = min(_add_days(date_cursor, 365), yesterday)

    resp = http.get("/Observations/" + current_series + "/" + date_cursor + "/" + end_date)

    rows = []
    if not resp.ok:
        fail(current_series + " (" + date_cursor + " to " + end_date + "): " + resp.text)

    # 200 = data, 204 = no observations in range (weekends, holidays, future)
    if resp.status_code == 200:
        for obs in resp.json:
            rows.append({
                "series": current_series,
                "date": obs["date"],
                "value": obs["value"],
            })

    # Advance: more dates for this series, or next series
    if end_date < yesterday:
        next_cursor = {"idx": series_idx, "from": _next_day(end_date), "until": yesterday}
    elif series_idx + 1 < len(series_list):
        next_cursor = {"idx": series_idx + 1, "from": start_date, "until": yesterday}
    else:
        next_cursor = None

    return {"rows": rows, "next": next_cursor}


def _to_iso_date(d):
    """Normalize to YYYY-MM-DD. Handles YYYYMMDD, YYYY-MM-DD, and YYYY-MM-DDTHH:MM:SSZ."""
    if len(d) >= 8 and d[4] != "-":
        return d[:4] + "-" + d[4:6] + "-" + d[6:8]
    return d[:10]


def _yesterday():
    return (time.now() - time.parse_duration("24h")).format("2006-01-02")


def _next_day(d):
    return (time.parse_time(d + "T00:00:00Z") + time.parse_duration("24h")).format("2006-01-02")


def _add_days(d, n):
    return (time.parse_time(d + "T00:00:00Z") + time.parse_duration(str(n * 24) + "h")).format("2006-01-02")