//! **#4 ENTSO-E REST** (EU electricity) — `web-api.tp.entsoe.eu/api?securityToken=…`.
//! Pan-European generation/load/day-ahead-price, including Sweden's four bidding
//! zones **SE1–SE4**, mapped here to their ENTSO-E **EIC domain codes**. The
//! token is pluggable (env/config via [`AuthConfig`](super::AuthConfig)) — when
//! no token is available the client + this fixture parser still build and test.
//!
//! Offline contract: [`parse_entsoe_xml`] turns a canned `GL_MarketDocument` /
//! `Publication_MarketDocument` into per-`Point` rows (domain, bidding zone,
//! resolution, period start, position, quantity) → Arrow.
use anyhow::Result;
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
use quick_xml::events::Event;
use super::arrow_util::{batch, f64_col, i64_col, schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// Sweden's bidding zones → ENTSO-E EIC domain codes (and back).
pub const SE_EIC: [(&str, &str); 4] = [
("SE1", "10Y1001A1001A44P"),
("SE2", "10Y1001A1001A45N"),
("SE3", "10Y1001A1001A46L"),
("SE4", "10Y1001A1001A47J"),
];
/// EIC domain code → bidding-zone label (`"10Y1001A1001A46L"` → `"SE3"`).
pub fn zone_of(eic: &str) -> Option<&'static str> {
SE_EIC
.iter()
.find(|(_, code)| *code == eic)
.map(|(z, _)| *z)
}
/// EIC domain code for a bidding zone (`"SE3"` → `"10Y1001A1001A46L"`).
pub fn eic_of(zone: &str) -> Option<&'static str> {
SE_EIC
.iter()
.find(|(z, _)| z.eq_ignore_ascii_case(zone))
.map(|(_, c)| *c)
}
/// One time-series `Point`.
#[derive(Debug, Clone, PartialEq)]
pub struct EntsoePoint {
pub in_domain: Option<String>,
pub bidding_zone: Option<String>,
pub resolution: Option<String>,
pub period_start: Option<String>,
pub position: i64,
pub quantity: f64,
}
fn local_name(name: quick_xml::name::QName) -> String {
String::from_utf8_lossy(name.local_name().as_ref()).into_owned()
}
/// Parse an ENTSO-E market-document XML into time-series points.
pub fn parse_entsoe_xml(xml: &str) -> Result<Vec<EntsoePoint>> {
let mut reader = quick_xml::Reader::from_str(xml);
let mut points = Vec::new();
let mut cur = String::new();
let mut domain: Option<String> = None;
let mut resolution: Option<String> = None;
let mut period_start: Option<String> = None;
let mut position: Option<i64> = None;
let mut quantity: Option<f64> = None;
// ── Is this a document at all, or the provider saying no? ────────────────────
//
// ENTSO-E answers a request it will not serve with an IEC 62325
// `Acknowledgement_MarketDocument` carrying a `<Reason><code/><text/></Reason>`,
// NOT with an empty `Publication_MarketDocument`. Before this, the loop below simply
// found no `<Point>` elements in it and returned `Ok(vec![])`, and
// `EntsoeCollector::collect` turned that into the reason "ENTSO-E returned no
// TimeSeries points (window empty?)" — a refusal an operator could act on, reported
// as a shrug about an empty window.
//
// Found by committing the provider's REAL answer as a fixture
// (`tests/fixtures/provider/entsoe-acknowledgement-auth-failed.xml`, fetched
// unauthenticated 2026-09-06, `Reason` code 999 "Authentication failed."). No
// synthetic fixture in this crate could have found it, because every one of them was
// written to be a document we already expected to parse.
//
// Gated on the ROOT element, not on the mere presence of `<Reason>`: a real
// `Publication_MarketDocument` may carry a `Reason` inside a `TimeSeries`, and
// refusing that one would throw away good data.
let mut root: Option<String> = None;
let mut ack_code: Option<String> = None;
let mut ack_text: Option<String> = None;
loop {
match reader.read_event()? {
Event::Start(e) => {
cur = local_name(e.name());
if root.is_none() {
root = Some(cur.clone());
}
match cur.as_str() {
"TimeSeries" => domain = None,
"Point" => {
position = None;
quantity = None;
}
_ => {}
}
}
Event::Text(t) => {
let txt = t.unescape()?.trim().to_string();
if txt.is_empty() {
continue;
}
match cur.as_str() {
"code" if is_acknowledgement(&root) => ack_code = Some(txt),
"text" if is_acknowledgement(&root) => ack_text = Some(txt),
"in_Domain.mRID" => domain = Some(txt),
"resolution" => resolution = Some(txt),
// The Period's <timeInterval><start> sits closest to the Points;
// overwriting on every <start> leaves that one in effect.
"start" => period_start = Some(txt),
"position" => position = txt.parse::<i64>().ok(),
"quantity" => quantity = txt.parse::<f64>().ok(),
_ => {}
}
}
Event::End(e) => {
let name = local_name(e.name());
if name == "Point" {
if let (Some(pos), Some(q)) = (position, quantity) {
let bidding_zone = domain.as_deref().and_then(zone_of).map(str::to_string);
points.push(EntsoePoint {
in_domain: domain.clone(),
bidding_zone,
resolution: resolution.clone(),
period_start: period_start.clone(),
position: pos,
quantity: q,
});
}
}
// Leaf element closed — stop capturing text into the parent.
cur.clear();
}
Event::Eof => break,
_ => {}
}
}
if is_acknowledgement(&root) {
// The provider's own words, both of them: the CODE so an operator can look it up
// in ENTSO-E's reason-code table, and the TEXT so they do not have to.
let code = ack_code.as_deref().unwrap_or("(no code)");
let text = ack_text.as_deref().unwrap_or("(no reason text)");
return Err(anyhow::anyhow!(
"ENTSO-E refused the request: Acknowledgement_MarketDocument, Reason code \
{code} — {text}. This is NOT an empty result window; nothing was served."
));
}
Ok(points)
}
/// Is the document ENTSO-E's refusal envelope rather than a data document?
fn is_acknowledgement(root: &Option<String>) -> bool {
root.as_deref() == Some("Acknowledgement_MarketDocument")
}
/// ENTSO-E points → one Arrow `RecordBatch`.
pub fn points_batch(points: &[EntsoePoint]) -> Result<RecordBatch> {
let cols = vec![
str_col(
&points
.iter()
.map(|p| p.in_domain.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.bidding_zone.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.resolution.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.period_start.clone())
.collect::<Vec<_>>(),
),
i64_col(&points.iter().map(|p| Some(p.position)).collect::<Vec<_>>()),
f64_col(&points.iter().map(|p| Some(p.quantity)).collect::<Vec<_>>()),
];
batch(&entsoe_schema(), cols)
}
pub fn entsoe_schema() -> Schema {
schema(&[
("in_domain", DataType::Utf8, true),
("bidding_zone", DataType::Utf8, true),
("resolution", DataType::Utf8, true),
("period_start", DataType::Utf8, true),
("position", DataType::Int64, false),
("quantity", DataType::Float64, false),
])
}
/// **The human card every ENTSO-E collector shares.** The Transparency Platform is ONE
/// API behind three `documentType`s, so the authority, the token dance, the rate limit
/// and the licence are identical for [`EntsoeCollector`],
/// [`super::entsoe_generation::EntsoeGenerationCollector`] and
/// [`super::entsoe_day_ahead::EntsoeDayAheadCollector`]. Each of the three calls this
/// and appends the quirks that belong to ITS document, so the shared half is written
/// once rather than three times.
///
/// Every value below was read off ENTSO-E's own published material on 2026-09-04: the
/// knowledgebase articles *How to get security token?* and *API Rate Limit Part 1*, the
/// platform's *Legal Terms and Conditions* article, and the official Postman API
/// documentation the knowledgebase links to. Nothing here is inferred.
pub fn entsoe_human_doc() -> crate::human_doc::HumanDoc {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
HumanDoc::new(
"ENTSO-E — the European Network of Transmission System Operators for Electricity, \
which operates the Transparency Platform",
"the TSOs and other data providers that submit each item — ENTSO-E's own Terms of \
Use call them the \"Primary Owner of Data\" and ENTSO-E publishes on their behalf",
"https://transparencyplatform.zendesk.com/hc/en-us/articles/40921911218961-Legal-Terms-and-Conditions",
"https://documenter.getpostman.com/view/7009892/2s93JtP3F6",
)
.with_contact(Contact::Email("transparency@entsoe.eu".into()))
.with_registration(Registration::ByRequest {
how: "register an account at https://transparency.entsoe.eu/ and verify it by email, \
then send an email to transparency@entsoe.eu with \"RESTful API access\" in the \
SUBJECT LINE and the registered address in the body. Access is granted within 3 \
working days; after that, My Account ▸ generate a token yields the ENTSOE_TOKEN \
this collector reads."
.into(),
})
.with_rate_limit(RateLimit::Documented(
"400 requests/minute, counted per TOKEN (not per IP)".into(),
))
.with_quirk(
"The 400/min cap is per SECURITY TOKEN, not per IP — every node of a distributed \
deployment sharing one token is aggregated into the same budget, which is how a \
Kubernetes fleet bans itself. Over the cap the API answers HTTP 429 and \
temporarily bans the token; ENTSO-E's own guidance is to throttle to ~6–7 \
requests/second on average with burst handling, and the ban clears automatically \
after roughly 10 minutes.",
)
.with_quirk(
"\"Free re-use\" is narrower than the CC-BY-4.0 tag on `CollectorMeta` suggests: \
clause 2.5 of the Terms of Use applies CC-BY-4.0 only to the items on ENTSO-E's \
published Open Data list. Anything outside that list still needs the Primary Owner \
of Data's agreement before it is redistributed — check the list before you publish \
a derived table.",
)
.with_quirk(
"The token rides the QUERY STRING as `securityToken`, not a header (the collector \
wires it through `AuthConfig`'s QueryKey). A request that is otherwise perfectly \
well formed answers 401 when it is missing, which is why `requirements()` names \
ENTSOE_TOKEN and the pane reads NeedsConfig instead of offering a ▶ Run that \
cannot work.",
)
}
/// The ENTSO-E collector.
#[derive(Default)]
pub struct EntsoeCollector;
#[async_trait::async_trait]
impl Collector for EntsoeCollector {
/// **ENTSO-E needs a free security token, and it has a NAME.** Without
/// `ENTSOE_TOKEN` the Transparency Platform answers 401 to a request that was
/// otherwise well formed — a failure the UI can and must predict, so this
/// collector reads `NeedsConfig` rather than offering a ▶ Run that cannot work.
fn requirements(&self) -> crate::requires::Requirements {
crate::requires::Requirements::open(self.source_target().source_endpoint).with_secrets(
vec![crate::requires::SecretReq::new(
"ENTSOE_TOKEN",
"the free ENTSO-E Transparency Platform security token (register at \
transparency.entsoe.eu, then Account Settings ▸ Web Api Security Token)",
)],
)
}
/// **The operator-facing card** — the shared ENTSO-E half (authority, token dance,
/// 400/min per token, the Open Data list) plus the three things that bite *this*
/// document type. Everything specific below is read off the parser in this file, not
/// off the source's website.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
Some(
entsoe_human_doc()
.with_quirk(
"This parser reads `<quantity>`, so it serves the LOAD and generation \
documents. Point it at `documentType=A44` (day-ahead prices) and it lands \
zero rows, because a price document carries `<price.amount>` instead — use \
the `entsoe_day_ahead` collector for prices.",
)
.with_quirk(
"Only SE1–SE4 have an EIC mapping here (`SE_EIC`). Every other `in_Domain` \
still lands, with the raw EIC in `in_domain` and `bidding_zone` NULL — join \
on `in_domain` when you collect outside Sweden, or the zone column silently \
drops the rows.",
)
.with_quirk(
"A row carries `period_start`, `resolution` and a 1-based `position`; the \
collector does NOT expand them into a timestamp. The instant a value refers \
to is `period_start + (position − 1) × resolution`, and `period_start` is \
the Period's `<timeInterval><start>` — the innermost one, because the parser \
overwrites on every `<start>` it sees.",
),
)
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "entsoe".into(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["xml".into()],
license: "CC-BY 4.0 (free token, redistributable)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::Token,
join_keys: vec!["elområde".into(), "period_start".into()],
description: "ENTSO-E Transparency Platform — EU electricity generation/load/price, \
SE1–SE4 via EIC domain codes."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::EntsoE,
"https://web-api.tp.entsoe.eu/api",
"entsoe_prices",
)
}
/// **One point per (domain, period, position).** The derived key named `elområde`,
/// which is the Swedish word for the axis — this table spells it `in_domain` (the
/// EIC code) — and it stopped at the PERIOD, so every position inside a period
/// would have collapsed into one row.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["in_domain", "period_start", "position"],
"one load point per (EIC domain, period, position) — `elområde` is the \
join-spine word for this axis; the column is `in_domain`",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(entsoe_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(6); // ENTSO-E ≤400 req/min
let auth = cfg.auth.resolve_async().await?; // securityToken via QueryKey auth
// Build the request URL from params (documentType/processType/in_Domain/
// periodStart/periodEnd). A bidding-zone shorthand is mapped to its EIC.
let mut query: Vec<(String, String)> = Vec::new();
if let Some(obj) = cfg.params.as_object() {
for (k, v) in obj {
if let Some(s) = v.as_str() {
let val = if k == "in_Domain" || k == "out_Domain" {
eic_of(s)
.map(str::to_string)
.unwrap_or_else(|| s.to_string())
} else {
s.to_string()
};
query.push((k.clone(), val));
}
}
}
let qs = query
.iter()
.map(|(k, v)| format!("{}={}", k, super::http::percent_encode(v)))
.collect::<Vec<_>>()
.join("&");
let url = if qs.is_empty() {
cfg.base_url.clone()
} else {
format!("{}?{}", cfg.base_url, qs)
};
let xml = http.get_text(&url, auth.as_ref()).await?;
let points = parse_entsoe_xml(&xml)?;
if points.is_empty() {
return Ok(CollectStats {
reason: Some("ENTSO-E returned no TimeSeries points (window empty?)".into()),
..Default::default()
});
}
let n = points.len() as u64;
sink.push(points_batch(&points)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}