Revalue open foreign currency balances at month end with ECB rates and rate controls

id: fx-revaluation-month-end
namespace: company.team

description: |
  Month-end revaluation of open foreign currency balances: fetch the ECB reference rates,
  triangulate them to the functional currency, check them for gaps, staleness and implausible
  moves, revalue every open monetary item from its booked rate to the closing rate, book the
  unrealized gain or loss with an automatic reversal, show the exposure and its sensitivity,
  and lock the period.

inputs:
  - id: period
    type: STRING
    displayName: Period (YYYY-MM)
    defaults: "2026-08"
    validator: ^\d{4}-(0[1-9]|1[0-2])$

  - id: functional_currency
    type: STRING
    displayName: Functional currency
    description: Must be quoted by the ECB against the euro, or be EUR.
    defaults: USD

  - id: rates_source
    type: SELECT
    displayName: Rates source
    description: ECB downloads the reference rates from the ECB data API. EMBEDDED
      uses the ECB month-end fixings for July to September 2026, copied into the
      flow, for offline runs.
    values:
      - ECB
      - EMBEDDED
    defaults: ECB

  - id: max_rate_age_days
    type: INT
    displayName: Maximum rate age (days)
    description: The closing rate is the last fixing on or before period end. Older
      than this, it is stale.
    defaults: 4

  - id: max_move_pct
    type: FLOAT
    displayName: Rate move needing approval (%)
    description: A closing rate that moved more than this since the last close needs
      approved_by. A move above 50% is treated as a data error and blocks.
    defaults: 5.0

  - id: scenario
    type: SELECT
    displayName: Demo scenario
    description: ISSUES drops the INR rate and quotes JPY per 100 yen, a unit error,
      which must block the run.
    values:
      - CLEAN
      - ISSUES
    defaults: CLEAN

  - id: approved_by
    type: STRING
    displayName: Approved by
    defaults: ""

variables:
  lock_key: fx_revaluation
  pack_dir: "fx-revaluation/{{ inputs.period }}"

concurrency:
  limit: 1

triggers:
  - id: first_business_day
    type: io.kestra.plugin.core.trigger.Schedule
    description: The 1st of each month at 18:00 CET, after the ECB publishes the
      last fixing of the month. Disabled until the ledger is wired in.
    cron: "0 18 1 * *"
    timezone: Europe/Berlin
    disabled: true

tasks:
  - id: period_control
    type: io.kestra.plugin.core.output.OutputValues
    values:
      lock: "{{ kv(vars.lock_key, errorOnMissing=false) ?? {'period': '',
        'closing_rates': {}} | toJson }}"
      last_closed: "{{ (kv(vars.lock_key, errorOnMissing=false) ?? {'period': ''}) |
        jq('.period') | first }}"
      previous_period: "{{ (inputs.period ~ '-01') | date('yyyy-MM-dd') | dateAdd(-1,
        'MONTHS') | date('yyyy-MM') }}"
      period_end: "{{ (inputs.period ~ '-01') | date('yyyy-MM-dd') | dateAdd(1,
        'MONTHS') | dateAdd(-1, 'DAYS') | date('yyyy-MM-dd') }}"
      reversal_date: "{{ (inputs.period ~ '-01') | date('yyyy-MM-dd') | dateAdd(1,
        'MONTHS') | date('yyyy-MM-dd') }}"
      rates_from: "{{ (inputs.period ~ '-01') | date('yyyy-MM-dd') | dateAdd(-15,
        'DAYS') | date('yyyy-MM-dd') }}"

  - id: period_gate
    type: io.kestra.plugin.core.flow.If
    condition: >-
      {{ outputs.period_control.values.last_closed != ''
         and (outputs.period_control.values.last_closed >= inputs.period
              or outputs.period_control.values.last_closed != outputs.period_control.values.previous_period) }}
    then:
      - id: period_refused
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          {{ outputs.period_control.values.last_closed >= inputs.period
             ? 'Period ' ~ inputs.period ~ ' is already closed (last closed: ' ~ outputs.period_control.values.last_closed ~ ').'
             : 'Period ' ~ outputs.period_control.values.previous_period ~ ' must be closed before ' ~ inputs.period ~ ' (last closed: ' ~ outputs.period_control.values.last_closed ~ ').' }}

  - id: fetch_rates
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.rates_source == 'ECB' }}"
    then:
      - id: ecb_rates
        type: io.kestra.plugin.core.http.Download
        description: Daily ECB reference rates against the euro, from two weeks before
          the previous month end to the period end.
        uri: >-
          https://data-api.ecb.europa.eu/service/data/EXR/D.USD+GBP+JPY+INR+CHF.EUR.SP00.A?startPeriod={{
          outputs.period_control.values.rates_from }}&endPeriod={{
          outputs.period_control.values.period_end }}&format=csvdata

  - id: revalue
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: >
      One DuckDB session: rates to the functional currency, rate controls, open
      items at period end (demo generator or your subledgers), revaluation,
      journals, exposure and sensitivity.
    inputFiles:
      ecb.csv: "{{ inputs.rates_source == 'ECB' ? outputs.ecb_rates.uri : 'not used' }}"
    outputFiles:
      - rates
      - items
      - by_account
      - exposure
      - journal
      - reversal
      - blockers
    fetchType: FETCH_ONE
    sql: |
      SET VARIABLE p_end = DATE '{{ outputs.period_control.values.period_end }}';
      SET VARIABLE prev_end = DATE '{{ inputs.period }}-01' - 1;
      SET VARIABLE fc = '{{ inputs.functional_currency }}';

      -- =======================================================================
      -- Rates: units of each currency per euro.
      -- =======================================================================
      CREATE TABLE eur_rates AS
      {% if inputs.rates_source == 'ECB' %}
      SELECT CURRENCY AS currency, TIME_PERIOD::DATE AS rate_date, OBS_VALUE::DOUBLE AS per_eur FROM read_csv('ecb.csv', all_varchar = true);
      {% else %}
      SELECT * FROM (VALUES
        ('USD', DATE '2026-07-31', 1.1485), ('GBP', DATE '2026-07-31', 0.85573), ('JPY', DATE '2026-07-31', 184.03), ('INR', DATE '2026-07-31', 109.5495), ('CHF', DATE '2026-07-31', 0.9304),
        ('USD', DATE '2026-08-31', 1.1596), ('GBP', DATE '2026-08-31', 0.85648), ('JPY', DATE '2026-08-31', 185.22), ('INR', DATE '2026-08-31', 110.361), ('CHF', DATE '2026-08-31', 0.9376),
        ('USD', DATE '2026-09-30', 1.1355), ('GBP', DATE '2026-09-30', 0.85463), ('JPY', DATE '2026-09-30', 178.27), ('INR', DATE '2026-09-30', 108.8205), ('CHF', DATE '2026-09-30', 0.9478)
      ) t(currency, rate_date, per_eur);
      {% endif %}
      -- The euro itself, on every fixing date.
      INSERT INTO eur_rates SELECT DISTINCT 'EUR', rate_date, 1.0 FROM eur_rates;
      {% if inputs.scenario == 'ISSUES' %}
      DELETE FROM eur_rates WHERE currency = 'INR' AND rate_date > getvariable('prev_end');
      UPDATE eur_rates SET per_eur = per_eur / 100 WHERE currency = 'JPY' AND rate_date > getvariable('prev_end');
      {% endif %}

      -- Last fixing on or before a date, converted to functional currency per unit:
      -- (functional per euro) / (currency per euro).
      CREATE MACRO closing_rate(d) AS TABLE
      WITH last AS (
        SELECT currency, max(rate_date) AS rate_date FROM eur_rates WHERE rate_date <= d GROUP BY currency
      ), r AS (SELECT e.* FROM eur_rates e JOIN last l USING (currency, rate_date))
      SELECT r.currency, r.rate_date, f.per_eur / r.per_eur AS rate
      FROM r JOIN eur_rates f ON f.currency = getvariable('fc') AND f.rate_date = r.rate_date;

      CREATE TABLE rates AS
      SELECT c.currency, c.rate_date, c.rate AS closing_rate,
             coalesce(l.locked_rate, p.rate) AS previous_rate,
             CASE WHEN l.locked_rate IS NOT NULL THEN 'last close' ELSE 'source' END AS previous_from,
             date_diff('day', c.rate_date, getvariable('p_end')) AS age_days,
             c.rate / coalesce(l.locked_rate, p.rate) - 1 AS move
      FROM closing_rate(getvariable('p_end')) c
      LEFT JOIN closing_rate(getvariable('prev_end')) p USING (currency)
      LEFT JOIN (SELECT key AS currency, value::DOUBLE AS locked_rate
                 FROM json_each('{{ outputs.period_control.values.lock | jq(".closing_rates") | first | toJson }}')) l USING (currency)
      WHERE c.currency <> getvariable('fc');

      -- =======================================================================
      -- Open items at period end, with the rate they were booked at. Replace with your
      -- AR, AP, bank and intercompany subledgers. Non-monetary items are not revalued.
      -- =======================================================================
      CREATE TABLE items AS
      WITH g AS (
        SELECT i, ['EUR', 'EUR', 'GBP', 'JPY', 'INR', 'CHF'][1 + i % 6] AS currency,
               DATE '2026-06-01' + ((i * 17) % 120)::INT AS booked_on
        FROM range(1, 241) t(i)
      ), base AS (
        SELECT 'ITM-' || lpad(i::VARCHAR, 4, '0') AS item, currency, booked_on,
               CASE WHEN i % 10 < 5 THEN 'AR' WHEN i % 10 < 8 THEN 'AP' WHEN i % 10 = 8 THEN 'PREPAID' ELSE 'AR' END AS kind,
               booked_on + (20 + (i * 13) % 70)::INT AS settled_on,
               round((CASE currency WHEN 'JPY' THEN 900000 WHEN 'INR' THEN 600000 ELSE 8000 END) * (0.5 + ((i * 7919) % 1000) / 500.0), 2) AS fc_amount
        FROM g
      )
      SELECT b.item, b.kind, b.currency, b.booked_on, b.settled_on,
             CASE WHEN b.kind = 'AP' THEN -b.fc_amount ELSE b.fc_amount END AS fc_amount,
             -- The rate the ERP booked the item at. The demo drifts it around a base rate.
             CASE b.currency WHEN 'EUR' THEN 1.155 WHEN 'GBP' THEN 1.352 WHEN 'JPY' THEN 0.00628 WHEN 'INR' THEN 0.01052 ELSE 1.236 END
               * (1 + ((hash(b.item) % 61)::INT - 30) / 1000.0) AS book_rate
      FROM base b
      UNION ALL
      SELECT 'BANK-' || currency, 'BANK', currency, DATE '2026-06-01', NULL, amount, rate
      FROM (VALUES ('EUR', 1250000.00, 1.1612), ('GBP', 310000.00, 1.3388), ('JPY', 48000000.00, 0.006254)) b(currency, amount, rate)
      UNION ALL
      -- A euro loan to a subsidiary. See "Things to know" about net investment loans.
      SELECT 'IC-LOAN-DE', 'IC_LOAN', 'EUR', DATE '2026-03-31', NULL, 2000000.00, 1.0807;

      CREATE TABLE accounts (kind VARCHAR, account VARCHAR, account_name VARCHAR, monetary BOOLEAN);
      INSERT INTO accounts VALUES ('AR', '1200', 'Trade receivables', true), ('AP', '2000', 'Trade payables', true),
                                  ('BANK', '1015', 'Foreign currency bank', true), ('IC_LOAN', '1800', 'Intercompany loan', true),
                                  ('PREPAID', '1300', 'Prepayments', false);

      CREATE TABLE open_items AS
      SELECT i.*, a.account, a.account_name, a.monetary, r.closing_rate,
             round(i.fc_amount * i.book_rate, 2)::DECIMAL(16, 2) AS book_value,
             round(i.fc_amount * r.closing_rate, 2)::DECIMAL(16, 2) AS closing_value,
             (round(i.fc_amount * r.closing_rate, 2)::DECIMAL(16, 2) - round(i.fc_amount * i.book_rate, 2)::DECIMAL(16, 2)) AS unrealized
      FROM items i JOIN accounts a USING (kind) LEFT JOIN rates r USING (currency)
      WHERE i.booked_on <= getvariable('p_end') AND (i.settled_on IS NULL OR i.settled_on > getvariable('p_end'))
        AND i.currency <> getvariable('fc');

      -- =======================================================================
      -- Controls on the rates themselves.
      -- =======================================================================
      CREATE TABLE blockers AS
      SELECT 'MISSING_RATE' AS code, currency AS ref, count(*) || ' open item(s) in ' || currency || ' and no rate on or before ' || getvariable('p_end') AS detail
        FROM open_items WHERE closing_rate IS NULL GROUP BY currency
      UNION ALL SELECT 'STALE_RATE', currency, 'Last fixing ' || rate_date || ', ' || age_days || ' days before period end'
        FROM rates WHERE age_days > {{ inputs.max_rate_age_days }}
      UNION ALL SELECT 'IMPLAUSIBLE_RATE', currency, 'Moved ' || round(move * 100, 1) || '% since the ' || previous_from || ' (' || round(previous_rate, 6) || ' -> ' || round(closing_rate, 6) || '). Check the quotation unit'
        FROM rates WHERE abs(move) > 0.5;

      CREATE TABLE big_moves AS
      SELECT currency, move FROM rates WHERE abs(move) > {{ inputs.max_move_pct }} / 100.0 AND abs(move) <= 0.5;

      -- =======================================================================
      -- Revaluation entry by account and currency, reversed on day one.
      -- =======================================================================
      CREATE TABLE by_account AS
      SELECT account, account_name, currency, count(*) AS items, sum(fc_amount) AS fc_balance, round(any_value(closing_rate), 6) AS closing_rate,
             sum(book_value) AS book_value, sum(closing_value) AS closing_value, sum(unrealized) AS unrealized
      FROM open_items WHERE monetary AND closing_rate IS NOT NULL GROUP BY ALL ORDER BY account, currency;

      CREATE TABLE journal AS
      SELECT row_number() OVER (ORDER BY account, currency) AS line, DATE '{{ outputs.period_control.values.period_end }}' AS posting_date, * FROM (
        SELECT account, currency, account_name || ' ' || currency || ' revaluation' AS memo,
               greatest(unrealized, 0)::DECIMAL(14, 2) AS debit, greatest(-unrealized, 0)::DECIMAL(14, 2) AS credit FROM by_account WHERE unrealized <> 0
        UNION ALL
        SELECT CASE WHEN sum(unrealized) >= 0 THEN '7810' ELSE '7820' END, currency,
               'Unrealized FX ' || CASE WHEN sum(unrealized) >= 0 THEN 'gain' ELSE 'loss' END || ' ' || currency,
               greatest(-sum(unrealized), 0)::DECIMAL(14, 2), greatest(sum(unrealized), 0)::DECIMAL(14, 2)
        FROM by_account GROUP BY currency HAVING sum(unrealized) <> 0
      );
      CREATE TABLE reversal AS
      SELECT line, DATE '{{ outputs.period_control.values.reversal_date }}' AS posting_date, account, currency, 'Reversal: ' || memo AS memo, credit AS debit, debit AS credit FROM journal;

      -- Net monetary exposure per currency and what a 10% move would do to it.
      CREATE TABLE exposure AS
      SELECT currency, sum(fc_balance) AS net_fc, round(sum(closing_value), 2) AS net_functional,
             round(sum(closing_value) * 0.10, 2)::DECIMAL(16, 2) AS impact_of_10pct_stronger, round(-sum(closing_value) * 0.10, 2)::DECIMAL(16, 2) AS impact_of_10pct_weaker
      FROM by_account GROUP BY currency ORDER BY abs(sum(closing_value)) DESC;

      COPY (SELECT currency, rate_date, round(closing_rate, 8) AS closing_rate, round(previous_rate, 8) AS previous_rate, previous_from, age_days, round(move * 100, 3) AS move_pct FROM rates ORDER BY currency) TO '{{ outputFiles.rates }}' (HEADER, DELIMITER ',');
      COPY (SELECT * FROM open_items ORDER BY currency, item) TO '{{ outputFiles.items }}' (HEADER, DELIMITER ',');
      COPY by_account TO '{{ outputFiles.by_account }}' (HEADER, DELIMITER ',');
      COPY exposure TO '{{ outputFiles.exposure }}' (HEADER, DELIMITER ',');
      COPY journal TO '{{ outputFiles.journal }}' (HEADER, DELIMITER ',');
      COPY reversal TO '{{ outputFiles.reversal }}' (HEADER, DELIMITER ',');
      COPY blockers TO '{{ outputFiles.blockers }}' (HEADER, DELIMITER ',');

      SELECT
        (SELECT string_agg(currency || ' ' || round(closing_rate, 6) || ' (' || rate_date || ', ' || CASE WHEN move >= 0 THEN '+' ELSE '' END || round(move * 100, 2) || '%)', ', ' ORDER BY currency) FROM rates) AS rates_detail,
        (SELECT json_group_object(currency, closing_rate)::VARCHAR FROM rates) AS closing_rates,
        (SELECT count(*) FROM open_items WHERE monetary) AS monetary_items,
        (SELECT count(*) FROM open_items WHERE NOT monetary) AS non_monetary_items,
        (SELECT coalesce(sum(unrealized), 0) FROM by_account) AS net_unrealized,
        (SELECT coalesce(string_agg(currency || ' ' || u, ', ' ORDER BY currency), '') FROM (SELECT currency, sum(unrealized) AS u FROM by_account GROUP BY currency)) AS unrealized_by_currency,
        (SELECT coalesce(string_agg(currency || ' ' || net_functional || ' (10% move: ' || impact_of_10pct_stronger || ')', ', '), '') FROM exposure) AS exposure_detail,
        (SELECT sum(debit) FROM journal) AS journal_debits,
        (SELECT sum(credit) FROM journal) AS journal_credits,
        (SELECT count(*) FROM big_moves) AS big_moves,
        (SELECT coalesce(string_agg(currency || ' ' || round(move * 100, 2) || '%', ', '), '') FROM big_moves) AS big_move_detail,
        (SELECT count(*) FROM blockers) AS blockers,
        (SELECT coalesce(string_agg(code || ' ' || ref || ': ' || detail, '; ' ORDER BY code, ref), '') FROM blockers) AS blocker_detail;

  - id: result
    type: io.kestra.plugin.core.output.OutputValues
    values:
      r: "{{ (outputs.revalue.outputs | last).row | toJson }}"

  - id: log_revaluation
    type: io.kestra.plugin.core.log.Log
    message: |
      FX revaluation {{ inputs.period }} into {{ inputs.functional_currency }}, rates from {{ inputs.rates_source }} ({{ inputs.scenario }}).
      {{ outputs.result.values.r | jq('"Closing rates: \(.rates_detail)."') | first }}
      {{ outputs.result.values.r | jq('"\(.monetary_items) monetary items revalued, \(.non_monetary_items) non-monetary items left at historical rate."') | first }}
      {{ outputs.result.values.r | jq('"Net unrealized \(.net_unrealized): \(.unrealized_by_currency). Journal debits \(.journal_debits), credits \(.journal_credits)."') | first }}
      {{ outputs.result.values.r | jq('"Exposure: \(.exposure_detail)."') | first }}
      Moves above {{ inputs.max_move_pct }}%: {{ outputs.result.values.r | jq('"\(.big_moves) \(.big_move_detail). Blockers: \(.blockers). \(.blocker_detail)"') | first }}

  - id: evidence
    type: io.kestra.plugin.core.namespace.UploadFiles
    namespace: "{{ flow.namespace }}"
    filesMap:
      "{{ render(vars.pack_dir) }}/rates.csv": "{{ outputs.revalue.outputFiles.rates }}"
      "{{ render(vars.pack_dir) }}/open-items.csv": "{{ outputs.revalue.outputFiles.items }}"
      "{{ render(vars.pack_dir) }}/revaluation-by-account.csv": "{{ outputs.revalue.outputFiles.by_account }}"
      "{{ render(vars.pack_dir) }}/exposure.csv": "{{ outputs.revalue.outputFiles.exposure }}"
      "{{ render(vars.pack_dir) }}/journal.csv": "{{ outputs.revalue.outputFiles.journal }}"
      "{{ render(vars.pack_dir) }}/reversal.csv": "{{ outputs.revalue.outputFiles.reversal }}"
      "{{ render(vars.pack_dir) }}/blockers.csv": "{{ outputs.revalue.outputFiles.blockers }}"

  - id: verdict
    type: io.kestra.plugin.core.output.OutputValues
    values:
      state: >-
        {%- set r = outputs.result.values.r -%} {%- if (r | jq('.blockers') |
        first) > 0 -%}BLOCKED {%- elseif (r | jq('.journal_debits') | first) !=
        (r | jq('.journal_credits') | first) -%}OUT_OF_BALANCE {%- elseif (r |
        jq('.big_moves') | first) > 0 and inputs.approved_by == ''
        -%}NEEDS_APPROVAL {%- else -%}READY{%- endif -%}

  - id: decide
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.verdict.values.state }}"
    cases:
      READY:
        - id: lock_period
          type: io.kestra.plugin.core.kv.Set
          description: The closing rates become the reference for the next close's move
            check.
          key: "{{ vars.lock_key }}"
          kvType: JSON
          value: |
            {"period": "{{ inputs.period }}", "closed_at": "{{ now() }}", "execution_id": "{{ execution.id }}",
             "functional_currency": "{{ inputs.functional_currency }}", "rates_source": "{{ inputs.rates_source }}",
             "closing_rates": {{ outputs.result.values.r | jq('.closing_rates') | first }},
             "net_unrealized": {{ outputs.result.values.r | jq('.net_unrealized') | first }},
             "reversal_date": "{{ outputs.period_control.values.reversal_date }}",
             "approved_by": "{{ inputs.approved_by }}"}

        - id: log_closed
          type: io.kestra.plugin.core.log.Log
          message: "Revaluation for {{ inputs.period }} closed: net {{
            outputs.result.values.r | jq('.net_unrealized') | first }},
            reversing on {{ outputs.period_control.values.reversal_date }}. Post
            journal.csv and reversal.csv from {{ render(vars.pack_dir) }}."

      NEEDS_APPROVAL:
        - id: approval_required
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: >-
            Rates moved more than {{ inputs.max_move_pct }}% since the last
            close: {{ outputs.result.values.r | jq('.big_move_detail') | first
            }}. Confirm them against a second source and run again with
            approved_by.

      OUT_OF_BALANCE:
        - id: out_of_balance
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: The revaluation journal does not balance. Do not post.

    defaults:
      - id: blocked
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Revaluation for {{ inputs.period }} not run: {{
          outputs.result.values.r | jq('.blocker_detail') | first }}"