stats: new public stats schema (source/series/obs/region/fetch_run + raw CBS, Eurostat, FRED, BIS tables) and the stats collector (scripts/stats-fetch.ts); preflight checks the stats tables

2026-10-25 20:27:04 +00:00
parent 902b1cadd5
commit 8ef291f586
14 changed files with 33128 additions and 2 deletions
--- a/scripts/stats/cbs.ts
+++ b/scripts/stats/cbs.ts
@@ -0,0 +1,240 @@
+// A1 CBS StatLine via the OData v4 API (long format: one row per measure × dimensions × period).
+// Raw rows go to stats.cbs_obs (+ code lists in stats.cbs_code); the chart series in
+// CBS_SERIES are then normalised into stats.series / stats.obs.
+// Nightly: re-pull a table only when its `Modified` changed (or with --force).
+import {
+	GOUDA_RING,
+	getJson,
+	log,
+	parsePeriod,
+	run,
+	sql,
+	writeSeries,
+	type ObsRow,
+	type SeriesRow
+} from './common';
+import { CBS_SERIES } from './cbs-series';
+
+const BASE = 'https://datasets.cbs.nl/odata/v1/CBS';
+const NL = ['NL01', 'NL00'];
+const regionIn = (dim: string, extra: string[] = []) =>
+	'(' + [...NL, ...GOUDA_RING, 'CR28', ...extra].map((c) => `${dim} eq '${c}'`).join(' or ') + ')';
+const kwbFilter =
+	`(startswith(WijkenEnBuurten,'WK0513') or startswith(WijkenEnBuurten,'BU0513') or ` +
+	[...NL, ...GOUDA_RING].map((c) => `WijkenEnBuurten eq '${c}'`).join(' or ') +
+	')';
+
+export type CbsTable = { id: string; theme: string; filter?: string; note?: string };
+
+/** Tables pulled in build step 1. Big tables are filtered to NL + Gouda + its ring. */
+export const CBS_TABLES: CbsTable[] = [
+	// energy
+	{ id: '85592NED', theme: 'energy' },
+	{ id: '84672NED', theme: 'energy', note: 'discontinued 2018–2023 tariffs (history)' },
+	{ id: '81309NED', theme: 'energy', note: 'discontinued 2007–2022 prices (history)' },
+	{ id: '83140NED', theme: 'energy' },
+	{ id: '82610NED', theme: 'energy' },
+	{ id: '85523NED', theme: 'energy' },
+	{ id: '85005NED', theme: 'energy' },
+	{ id: '85237NED', theme: 'energy' },
+	// economy
+	{ id: '86141NED', theme: 'economy' },
+	{ id: '83136NED', theme: 'economy' },
+	{ id: '86144NED', theme: 'economy', filter: "Bestedingscategorieen eq 'T001112'" },
+	{ id: '85865NED', theme: 'economy' },
+	{ id: '85773NED', theme: 'economy' },
+	// budget
+	{ id: '84120NED', theme: 'budget' },
+	{ id: '84117NED', theme: 'budget' },
+	{ id: '85590NED', theme: 'budget' },
+	{ id: '80509ned', theme: 'budget' },
+	{ id: '83642NED', theme: 'budget' },
+	{ id: '83641NED', theme: 'budget' },
+	// gouda
+	{ id: '37230ned', theme: 'gouda', filter: regionIn('RegioS') },
+	{ id: '37259ned', theme: 'gouda', filter: regionIn('RegioS') },
+	{ id: '70072ned', theme: 'gouda', filter: regionIn('RegioS') },
+	{ id: '83625NED', theme: 'gouda' },
+	{ id: '85819NED', theme: 'gouda' },
+	{ id: '86079NED', theme: 'gouda', filter: regionIn('RegioS') },
+	{ id: '83162NED', theme: 'gouda' },
+	{ id: '70675ned', theme: 'gouda' },
+	// Kerncijfers wijken en buurten, one table per year (Gouda wijken/buurten + ring + NL)
+	{ id: '86358NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2026' },
+	{ id: '86165NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2025' },
+	{ id: '85984NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2024' },
+	{ id: '85618NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2023' },
+	{ id: '85318NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2022' },
+	{ id: '85039NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2021' },
+	{ id: '84799NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2020' },
+	{ id: '84583NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2019' }
+];
+
+type Props = { Title: string; Modified: string; Status: string; ObservationCount?: number };
+type Page<T> = { value: T[]; '@odata.nextLink'?: string };
+
+async function all<T>(url: string): Promise<T[]> {
+	const out: T[] = [];
+	let next: string | undefined = url;
+	while (next) {
+		const page: Page<T> = await getJson<Page<T>>(next);
+		out.push(...page.value);
+		next = page['@odata.nextLink'];
+	}
+	return out;
+}
+
+async function pullTable(t: CbsTable, force: boolean, note: (w: string) => void): Promise<number> {
+	const props = await getJson<Props>(`${BASE}/${t.id}/Properties`);
+	const [known] = await sql<{ modified: Date | null; filter: string | null }[]>`
+		select modified, filter from stats.cbs_table where table_id = ${t.id}`;
+	if (props.Status && props.Status !== 'Regulier' && !t.note?.includes('discontinued'))
+		note(
+			`${t.id} status is ${props.Status}: check the CBS catalog for its successor (base-year swap)`
+		);
+	if (
+		!force &&
+		known?.modified &&
+		new Date(props.Modified).getTime() === known.modified.getTime() &&
+		(known.filter ?? null) === (t.filter ?? null)
+	) {
+		log(`cbs ${t.id}: unchanged since ${props.Modified}`);
+		return 0;
+	}
+	log(`cbs ${t.id}: pulling "${props.Title.trim()}"${t.filter ? ' (filtered)' : ''}`);
+	const dims = await getJson<Page<{ Identifier: string; Kind: string; CodesUrl: string }>>(
+		`${BASE}/${t.id}/Dimensions`
+	);
+	const codes: Record<string, unknown>[] = [];
+	const measures = await all<{
+		Identifier: string;
+		Title: string;
+		Unit?: string;
+		Decimals?: number;
+		MeasureGroupId?: string;
+		Index?: number;
+	}>(`${BASE}/${t.id}/MeasureCodes`);
+	for (const m of measures)
+		codes.push({
+			table_id: t.id,
+			dimension: 'Measure',
+			code: m.Identifier.trim(),
+			title: (m.Title ?? '').trim() || m.Identifier,
+			unit: m.Unit?.trim() || null,
+			decimals: m.Decimals ?? null,
+			group_id: m.MeasureGroupId ?? null,
+			sort_index: m.Index ?? null
+		});
+	const dimIds: string[] = [];
+	for (const d of dims.value) {
+		dimIds.push(d.Identifier);
+		const list = await all<{
+			Identifier: string;
+			Title: string;
+			DimensionGroupId?: string;
+			Index?: number;
+		}>(d.CodesUrl);
+		for (const c of list)
+			codes.push({
+				table_id: t.id,
+				dimension: d.Identifier,
+				code: c.Identifier.trim(),
+				title: (c.Title ?? '').trim() || c.Identifier,
+				unit: null,
+				decimals: null,
+				group_id: c.DimensionGroupId ?? null,
+				sort_index: c.Index ?? null
+			});
+	}
+	const timeDim = dims.value.find((d) => d.Kind === 'TimeDimension')?.Identifier;
+	const url =
+		`${BASE}/${t.id}/Observations` + (t.filter ? `?$filter=${encodeURIComponent(t.filter)}` : '');
+	const raw = await all<Record<string, unknown>>(url);
+	const rows = raw.map((o) => {
+		const d: Record<string, string> = {};
+		for (const id of dimIds) if (id !== timeDim && o[id] != null) d[id] = String(o[id]).trim();
+		return {
+			table_id: t.id,
+			obs_id: Number(o.Id),
+			measure: String(o.Measure).trim(),
+			dims: sql.json(d),
+			period: timeDim ? String(o[timeDim]).trim() : '',
+			value: typeof o.Value === 'number' ? o.Value : null,
+			value_attribute:
+				o.ValueAttribute && o.ValueAttribute !== 'None' ? String(o.ValueAttribute) : null
+		};
+	});
+	await sql.begin(async (tx) => {
+		await tx`delete from stats.cbs_obs where table_id = ${t.id}`;
+		await tx`delete from stats.cbs_code where table_id = ${t.id}`;
+		for (let i = 0; i < codes.length; i += 5000)
+			await tx`insert into stats.cbs_code ${tx(codes.slice(i, i + 5000))} on conflict do nothing`;
+		for (let i = 0; i < rows.length; i += 5000)
+			await tx`insert into stats.cbs_obs ${tx(rows.slice(i, i + 5000))}`;
+		await tx`insert into stats.cbs_table (table_id, title, modified, status, filter, obs_count, fetched_at)
+			values (${t.id}, ${props.Title.trim()}, ${props.Modified}, ${props.Status ?? null}, ${t.filter ?? null}, ${rows.length}, now())
+			on conflict (table_id) do update set title = excluded.title, modified = excluded.modified,
+			status = excluded.status, filter = excluded.filter, obs_count = excluded.obs_count, fetched_at = now()`;
+	});
+	log(`cbs ${t.id}: ${rows.length} observations, ${codes.length} codes`);
+	return rows.length;
+}
+
+/** Normalise CBS_SERIES from the raw rows into stats.series / stats.obs. */
+export async function normaliseCbs(): Promise<number> {
+	const series: SeriesRow[] = [];
+	const obs: ObsRow[] = [];
+	for (const def of CBS_SERIES) {
+		const regions = def.regions ?? [null];
+		for (const region of regions) {
+			const dims = { ...(def.dims ?? {}) };
+			if (region && def.regionDim) dims[def.regionDim] = region.code;
+			const rows = await sql<
+				{ period: string; value: number | null; value_attribute: string | null }[]
+			>`
+				select period, value, value_attribute from stats.cbs_obs
+				where table_id = ${def.table} and measure = ${def.measure} and dims @> ${sql.json(dims)}`;
+			const id = region ? `${def.id}:${region.code}` : def.id;
+			let n = 0;
+			const seen = new Set<string>();
+			for (const r of rows) {
+				const p = parsePeriod(r.period);
+				if (!p || p.freq !== def.freq || seen.has(p.date)) continue;
+				seen.add(p.date);
+				const value = r.value == null ? null : r.value * (def.scale ?? 1);
+				obs.push({ series_id: id, period: p.date, value, status: r.value_attribute });
+				n++;
+			}
+			if (!n) continue;
+			series.push({
+				series_id: id,
+				source_id: 'cbs',
+				source_code: `${def.table}/${def.measure}/${JSON.stringify(dims)}`,
+				title: def.title,
+				unit: def.unit,
+				freq: def.freq,
+				region_code: region?.region ?? def.region ?? 'NLD',
+				theme: def.theme,
+				meta: { table: def.table, measure: def.measure, dims }
+			});
+		}
+	}
+	return writeSeries(series, obs);
+}
+
+export async function cbs(opts: { force: boolean; only?: string; job: string }) {
+	let total = 0;
+	await run('cbs', opts.job, async (note) => {
+		for (const t of CBS_TABLES) {
+			if (opts.only && !opts.only.split(',').includes(t.id)) continue;
+			try {
+				total += await pullTable(t, opts.force, note);
+			} catch (error) {
+				note(`${t.id}: ${error instanceof Error ? error.message : String(error)}`);
+			}
+		}
+		const normalised = await normaliseCbs();
+		log(`cbs: ${normalised} normalised observations`);
+		return total;
+	});
+}