Plugin

ECB euro reference rates

The daily euro reference rates of the European Central Bank for about 30 currencies since 1999, one feed per currency, plus the cross for your pair.

Data sourceNetwork: data-api.ecb.europa.euDesktop onlyv1.0.0

Get early access Plugin API documentation

About

Every TARGET business day the ECB publishes one reference rate per currency against the euro, around 16:00 CET. It is an official daily fixing, used in contracts, accounting and tax, and a neutral yardstick next to your broker's prices.

Before a run, the plugin downloads the rates from the ECB data portal and writes one feed per currency, named ecb_eur plus the currency code in lower case: ecb_eurusd, ecb_eurjpy, ecb_eurgbp. The value is units of that currency per 1 euro, as the ECB quotes it. When the run's instrument is a currency pair without the euro, such as GBPUSD, it also writes the cross ecb_gbpusd (US dollars per pound), computed day by day from the two euro rates.

Other plugins can rely on these names, for example to convert results to euro.

Configuration

Field	Type	Default	Description
`cache_days`Refetch after (days)	number	`1`	Days before the plugin downloads the rates again. Within that time a run reads the cached feeds and needs no network.
`currencies`Currencies (all, or codes such as USD,JPY,GBP)	string	`"all"`	Which currencies to fetch: all (every currency the ECB publishes, about 15 MB) or a list of codes such as USD,JPY,GBP for a smaller download. The two currencies of the run's pair are always added.

Use it in a strategy

pythonbars = ctx.bars(1)
if bars:
    fix = ctx.data("ecb_eurusd", bars[-1].ts)  # US dollars per euro, the last fix known
    if fix is not None and bars[-1].close > fix * 1.01:
        ctx.signal("EXIT")

The feeds are refreshed before a run whose strategy reads ctx.data, or when "Fetch external data (news and plugin feeds)" is ticked. A run on the server never fetches.

What it provides

`ecb_eur<ccy>`Feed	One feed per currency, the code in lower case (ecb_eurusd, ecb_eurjpy, ecb_eurchf): units of that currency per 1 euro, under each day from the day after the fix.
`ecb_<base><quote>`Feed	Only when the run's instrument is a currency pair without the euro (GBPUSD gives ecb_gbpusd): quote currency per 1 unit of the base currency, the ratio of the two euro rates, rounded to 6 significant digits.

Data source and terms

Publisher	European Central Bank, euro foreign exchange reference rates (ECB Data Portal, dataset EXR)
Terms	Read the terms
API key	None needed
Attribution	Source: ECB statistics
Rate limit	One download per refresh; the ECB publishes no limit. All currencies over the whole history is about 15 MB.

Hosts it may reach: data-api.ecb.europa.eu

When values are known

The ECB publishes the rates around 16:00 CET on the day they are for, so each rate is stored under the next calendar day and carried forward over weekends and TARGET holidays, for at most 7 days. A bar on Monday reads Friday's fix; a bar during the day of a fix never sees it, not even after 16:00.

Limits

A reference rate is a snapshot at about 14:10 CET for information; it is not a price you could have traded at, and it says nothing about the rest of the day.
Only currencies the ECB quotes, and only on TARGET business days. A currency the ECB stopped quoting (RUB since March 2022, BGN since Bulgaria joined the euro) ends 7 days after its last fix.
Crosses are computed from two euro rates, not quoted by the ECB, so they can differ from market prices in the last digits.
The data is downloaded to your machine and stays there; the ECB asks for the source line "Source: ECB statistics" wherever you show it.

Preview

json{
  "2024-05-03": 1.0712,
  "2024-05-04": 1.0712,
  "2024-05-05": 1.0712,
  "2024-05-06": 1.0712,
  "2024-05-07": 1.0768,
  "_meta": {
    "delay_rule": "next_day",
    "source": "Made-up sample, not vendor data",
    "terms": "https://www.ecb.europa.eu/stats/ecb_statistics/governance_and_quality_framework/html/usage_policy.en.html",
    "units": "units of the currency per 1 EUR"
  }
}

A made-up sample that shows the shape of the output. Not vendor data and not a result.

The code

main.py, the file the app installs.

python"""ECB euro reference rates: a data source on the plugin kit.

Before a run it downloads the euro foreign exchange reference rates the
European Central Bank publishes every TARGET business day, and writes one feed
per currency that a strategy reads:

    def on_market(self, ctx):
        bars = ctx.bars(1)
        if not bars:
            return
        fix = ctx.data("ecb_eurusd", bars[-1].ts)    # US dollars per euro
        if fix is not None and bars[-1].close > fix * 1.01:
            ctx.signal("EXIT")

Feed names: `ecb_eur<currency>` in lower case (`ecb_eurusd`, `ecb_eurjpy`),
the value being units of that currency per 1 euro, as the ECB quotes it. When
the run's instrument is a currency pair without the euro (GBPUSD), the cross
is written as well, `ecb_<pair>` (`ecb_gbpusd`, US dollars per pound), computed
from the two euro rates of each day.

When a value is known: the ECB publishes around 16:00 CET on the day the rate
is for, so each rate is stored under the next day and carried forward over
weekends and TARGET holidays, at most 7 days. A bar on Monday reads Friday's
fix; a bar on the day of a fix never sees it.

The download is skipped while the feeds are younger than cache_days, and a
failed download leaves the feeds that were there.
"""

import json
import re

from orionquant_engine.plugins.kit import feeds, net

HOSTS = ("data-api.ecb.europa.eu",)
#: The SDMX key is D.<currencies>.EUR.SP00.A: daily, against the euro, spot,
#: average (the reference rate). An empty currency part means every currency.
ECB_DATA = "https://data-api.ecb.europa.eu/service/data/EXR/D.{currencies}.EUR.SP00.A"
ATTRIBUTION = "Source: ECB statistics"
DELAY_RULE = "next_day"
#: Longest stretch a rate is carried: Easter (Good Friday to Easter Monday) is
#: four days without a fix; a week covers every TARGET closure. A currency the
#: ECB stopped quoting (RUB in 2022, BGN when Bulgaria joined the euro) ends a
#: week after its last fix instead of looking alive for years.
MAX_GAP_DAYS = 7
#: All currencies over the whole history is about 15 MB of CSV.
MAX_BYTES = 60_000_000

_CCY = re.compile(r"^[A-Z]{3}$")
_PAIR = re.compile(r"^([A-Z]{3})/?([A-Z]{3})$")

_API = {"api": None}


def _log(message):
    if _API["api"] is not None:
        _API["api"].log(message)


def _config(key, default):
    api = _API["api"]
    return api.get_config(key, default) if api is not None else default


def feed_name(currency):
    """The feed of one currency against the euro: ecb_eurusd."""
    return "ecb_eur" + currency.lower()


def wanted_currencies(setting):
    """The `currencies` setting as a sorted list, or None for every currency.

    Accepts commas or spaces between codes. Codes that are not three letters
    are logged and left out rather than sent to the ECB.
    """
    text = str(setting or "").strip().upper()
    if text in ("", "ALL"):
        return None
    out = set()
    for code in re.split(r"[\s,;]+", text):
        if not code:
            continue
        if _CCY.fullmatch(code) and code != "EUR":
            out.add(code)
        else:
            _log("{!r} is not a currency code the ECB quotes against the euro, left out".format(code))
    return sorted(out)


def run_pair(instrument):
    """(base, quote) when the run's instrument is a currency pair, else None.
    EURUSD and EUR/USD both work; US500 or XAUUSD.X do not count."""
    m = _PAIR.fullmatch(str(instrument or "").strip().upper())
    if not m or m.group(1) == m.group(2):
        return None
    return m.group(1), m.group(2)


def parse(rows):
    """The ECB's CSV rows to {currency: {observed day: rate}}. Rows without a
    value (the CSV leaves OBS_VALUE empty on some days) are skipped."""
    out = {}
    for row in rows:
        ccy = (row.get("CURRENCY") or "").strip().upper()
        day = (row.get("TIME_PERIOD") or "").strip()
        raw = (row.get("OBS_VALUE") or "").strip()
        if not _CCY.fullmatch(ccy) or not day or raw in ("", "NaN"):
            continue
        try:
            value = float(raw)
            key = feeds.day_key(day)
        except ValueError:
            continue
        if value > 0:
            out.setdefault(ccy, {})[key] = value
    return out


def source_terms():
    """The ECB's terms link from this plugin's docs.json, where the source is
    written down once. The link is on www.ecb.europa.eu, a host the plugin
    never fetches from, so it is not repeated here as a URL."""
    try:
        with open(__plugin_dir__ + "/docs.json", encoding="utf-8") as fh:  # noqa: F821  # set by the loader
            return json.load(fh)["sources"][0]["terms"]
    except (OSError, ValueError, KeyError, IndexError, NameError):
        return "ECB statistics reuse policy (see the plugin's docs)"


def _meta(**extra):
    meta = {
        "source": "European Central Bank, euro foreign exchange reference rates (EXR)",
        "terms": source_terms(),
        "delay_rule": DELAY_RULE,
        "attribution": ATTRIBUTION,
        "units": "units of the currency per 1 EUR",
    }
    meta.update(extra)
    return meta


def write_cross(base, quote):
    """Write ecb_<base><quote> (quote currency per 1 base) from the two euro
    feeds already on disk, day by day. Both feeds are keyed by the day each
    rate became known, so their ratio on a day is known that day as well.
    Returns the number of days written, 0 when an input is missing."""
    legs = {}
    for ccy in (base, quote):
        if ccy == "EUR":
            continue
        legs[ccy] = feeds.read_feed(feed_name(ccy))
        if not legs[ccy]:
            _log("no ECB feed for {}, so no cross {}{}".format(ccy, base, quote))
            return 0
    days = set.intersection(*(set(v) for v in legs.values()))
    rows = {}
    for day in days:
        per_eur_base = 1.0 if base == "EUR" else legs[base][day]
        per_eur_quote = 1.0 if quote == "EUR" else legs[quote][day]
        rows[day] = float("{:.6g}".format(per_eur_quote / per_eur_base))
    if not rows:
        return 0
    feeds.write_feed("ecb_" + (base + quote).lower(), rows, meta=_meta(
        units="{} per 1 {}".format(quote, base), cross_of=[feed_name(c) for c in legs]))
    return len(rows)


def _cache_is_fresh(currencies, pair, max_age):
    """Whether every feed this fetch would write is younger than max_age and
    was written for the same currency list."""
    signature = "all" if currencies is None else ",".join(currencies)
    probe = feed_name(currencies[0] if currencies else "USD")
    meta = feeds.read_meta(probe) or {}
    if meta.get("request") != signature or not feeds.fresh(probe, max_age):
        return False
    needed = list(currencies or []) + [c for c in (pair or ()) if c != "EUR"]
    return all(feeds.exists(feed_name(c)) for c in needed)


def data_fetch(context):
    """Lifecycle hook: refresh the feeds before the backtest runs."""
    currencies = wanted_currencies(_config("currencies", "all"))
    if currencies == []:
        _log("no valid currency in the currencies setting, nothing fetched")
        return
    pair = run_pair(getattr(context, "instrument", ""))
    max_age = float(_config("cache_days", 1)) * 86400
    cross_name = "ecb_" + "".join(pair).lower() if pair and "EUR" not in pair else None

    if _cache_is_fresh(currencies, pair, max_age):
        if cross_name and not feeds.exists(cross_name):
            write_cross(*pair)
        return

    request = None if currencies is None else sorted(set(currencies) | {c for c in (pair or ()) if c != "EUR"})
    url = ECB_DATA.format(currencies="" if request is None else "+".join(request))
    try:
        rows = net.get_csv(url, hosts=HOSTS, params={"format": "csvdata", "detail": "dataonly"},
                           timeout=120.0, max_bytes=MAX_BYTES)
    except net.NetError as exc:
        _log("fetch failed, the cached feeds stay: {}".format(exc))
        return
    observed = parse(rows)
    if not observed:
        _log("the ECB returned no rates, the cached feeds stay")
        return

    signature = "all" if currencies is None else ",".join(currencies)
    until = feeds.today()
    for ccy in sorted(observed):
        known = feeds.shift(observed[ccy], DELAY_RULE)
        feeds.write_feed(feed_name(ccy), feeds.fill_forward(known, until=until, max_gap_days=MAX_GAP_DAYS),
                         meta=_meta(currency=ccy, request=signature))
    for ccy in request or ():
        if ccy not in observed:
            _log("the ECB has no reference rate for {}".format(ccy))
    if cross_name:
        write_cross(*pair)
    _log("cached ECB reference rates for {} currencies".format(len(observed)))


def register(api):
    _API["api"] = api
    api.register_data_fetch(data_fetch)