Financial Health from Bundesanzeiger Data: Train & Deploy a Simple ML Model Locally (with Python)
From data retrieval to deployment: learn step by step how to build a local machine learning model for financial health scoring from Bundesanzeiger annual reports — using Python, scikit-learn, and the handelsregister.ai API.
Annual financial statements published in Germany's Bundesanzeiger (Federal Gazette) provide hard facts about a company's assets, finances, and earnings. For developers, data scientists, and fintech teams looking to build automated scorings on top of this data, they're an excellent foundation. In an earlier article, we described which financial ratios are typically used for financial and risk scoring — but that stayed conceptual. The gap between "theory of ratio analysis" and "actually running model" remained open.
This tutorial closes that gap. You'll build a simple, reproducible machine learning model that estimates a "financial health probability" from Bundesanzeiger data. You train it locally on your machine and run it locally — no cloud training, no external hosting, no black box.
We pull the data through the handelsregister.ai API. The advantage: instead of unstructured PDFs, you get clean, machine-readable JSON responses that you can feed directly into your pipeline. The underlying data comes from the official sources handelsregister.de and bundesanzeiger.de.
Note: This tutorial serves exclusively as a technical learning example. The resulting model is neither an official rating nor investment or credit advice. It works with so-called "weak labels" — simplified rules that don't replace a real rating.
What You'll Have at the End
By the end of this article, you'll have a complete, runnable Python project on your machine. It consists of four building blocks that build on each other:
First, a data fetching script that downloads Bundesanzeiger financial data (balance sheet, P&L, and KPIs) via the handelsregister.ai API as JSON and stores it locally. Then a feature engineering script that transforms these raw JSON files into a tabular training dataset — with one row per company and fiscal year, enriched with derived ratios like equity ratio, ROA, or personnel intensity. Building on that, a training script that trains a logistic regression on this dataset and saves it as a single file. And finally, a scoring CLI that lets you query any company at any time and calculate a health score in real time — optionally also as a local REST service with FastAPI.
Prerequisites
To follow this tutorial, you need Python 3.10 or newer and an API key from handelsregister.ai (which you can set as the environment variable HANDELSREGISTER_API_KEY or hardcode directly in your script). A basic understanding of balance sheet and P&L ratios is also helpful — the scoring article on the blog gives you a good starting point.
Important API basics for practice:
The handelsregister.ai API is a REST interface with the base URL https://handelsregister.ai/api. Authentication is preferably done via the HTTP header x-api-key: YOUR_API_KEY (alternatively as a query parameter api_key, which is considered a legacy method). For financial data, you activate individual features as query parameters: feature=financial_kpi, feature=balance_sheet_accounts, and feature=profit_and_loss_account. Each activated feature consumes credits. The rate limit is 60 requests per minute, so you should add brief pacing for sequential requests — which is exactly what we do in the code below.
Data Overview: What Does the API Return?
Before we dive into the code, it's worth understanding how the financial data is structured. For our model, we use three API features, each providing a different perspective on a company's financial situation:
The financial_kpi feature delivers aggregated annual metrics such as revenue, net income/loss, total assets, and employee count. It's the most compact entry point into the financial data and contains values that handelsregister.ai has already extracted from the annual reports. The balance_sheet_accounts feature contains the full balance sheet as a hierarchical tree structure — assets and liabilities with all sub-items (fixed assets, current assets, equity, liabilities, etc.). And finally, profit_and_loss_account gives us the income statement in the same hierarchical structure: revenue, material costs, personnel expenses, down to the net result.
The data is grouped by fiscal year. A typical entry looks roughly like this (simplified):
{
"entity_id": "…",
"balance_sheet_accounts": [
{
"year": 2024,
"balance_sheet_accounts": [
{"name": {"de": "Aktivseite"}, "value": 24966659.45, "children": [...]},
{"name": {"de": "Passivseite"}, "value": 24966659.45, "children": [...]}
]
}
],
"financial_kpi": [
{
"year": 2024,
"active_total": 24966659.45,
"revenue": 9502936.17,
"net_income": -14369731.62,
"employees": 111
}
]
}
This example is intentionally abbreviated and anonymized. In practice, each balance sheet node contains further children nodes that break down the sub-items — and it's precisely this tree structure that we use later to extract equity, liabilities, provisions, and other line items.
Our Goal: Defining "Financial Health" as a Model Problem
Before we compute features and train a model, we need to answer a fundamental question: What exactly should the model predict — and where do the labels come from?
In an ideal scenario, you'd have actual default data (e.g., "Company X filed for insolvency in year Y") or externally assigned ratings as your target variable. In practice, however, such labels are often unavailable — especially when you're starting out as an individual developer or small team.
The solution is called weak supervision: instead of perfect labels, you define a simple, rule-based heuristic that separates "financially critical" from "unremarkable." The model then learns to generalize this decision from the financial ratios — ideally more robustly than the single rule alone.
Specifically, we define the label as follows: for each company and year t, we compute the features and then look at the state in t+1 (the following year). The label distress_next_year is set to 1 if the following year shows either a negative equity ratio (meaning the company would be technically insolvent on paper) or an equity ratio below 5% combined with a loss (negative ROA). Everything else is labeled as 0 (no distress).
This definition is deliberately kept simple. It catches the most severe financial distress situations but is no substitute for a professional rating. The point is different: it gives us a reproducible, automatable ground truth on which we can build an initial model that you then iteratively refine with better labels or manually curated examples.
Step 1: Project Setup
Start by creating a straightforward project structure. You don't need any complex architecture — four Python files in a src/ folder and a data/ directory for input and output are perfectly sufficient:
financial-health-model/
data/
companies.csv ← your company list as input
raw/ ← downloaded JSON files go here
dataset.parquet ← the finished training dataset
models/ ← trained model + metadata
src/
fetch_data.py ← data retrieval from the API
build_dataset.py ← feature engineering + labeling
train_model.py ← model training + evaluation
score_company.py ← live scoring of individual companies
Create a virtual environment and install the required packages:
python -m venv .venv
source .venv/bin/activate # macOS/Linux
# .venv\Scripts\activate # Windows PowerShell
pip install -U pip
pip install requests pandas numpy scikit-learn joblib pyarrow
Set your API key as an environment variable. This is the recommended approach — analogous to the SDK examples on the blog:
export HANDELSREGISTER_API_KEY="YOUR_API_KEY"
Step 2: Fetch Data and Store Locally
The first building block of our pipeline reads a company list from a CSV file and retrieves the three financial features from the handelsregister.ai API for each entry. For an initial model, a few hundred to a few thousand companies are often enough — depending on how many of them actually have annual reports published in the Bundesanzeiger.
The Input File: data/companies.csv
The CSV file supports two formats. The simplest is a single column q that gets passed directly as a search query to the API:
q
Muster GmbH Berlin
Beispiel AG Hamburg
Beispiel GmbH München
Alternatively, you can use two columns company_name and city, which the script automatically combines into a query:
company_name,city
Muster GmbH,Berlin
Beispiel AG,Hamburg
Beispiel GmbH,München
The Fetch Script: src/fetch_data.py
The following script is deliberately commented in detail so you can follow every step. The core logic is simple: read CSV, make one API call per row, reduce the response to the financial data, and save as JSON.
Two design decisions are important here: first, we deliberately save only the financial portion of the API response (KPIs, balance sheet, P&L) plus the entity_id. Company names and addresses are not written to the raw data — this way you can share the training material without exposing identifying information. Second, we build in simple rate limiting: the API allows 60 requests per minute, so we enforce a minimum of about 1 second between consecutive requests.
import csv
import json
import os
import re
import time
from pathlib import Path
from typing import Any, Dict, List, Optional
import requests
# ──────────────────────────────────────────────
# Configuration
# ──────────────────────────────────────────────
API_BASE = "https://handelsregister.ai/api"
FETCH_ENDPOINT = f"{API_BASE}/v1/fetch-organization"
# The three financial features we need for our model.
# Each feature consumes credits — so we only fetch what
# we actually process.
FEATURES = ["financial_kpi", "balance_sheet_accounts", "profit_and_loss_account"]
def _get_api_key() -> str:
"""Reads the API key from the environment variable."""
api_key = os.getenv("HANDELSREGISTER_API_KEY")
if not api_key:
raise RuntimeError(
"Please set the environment variable HANDELSREGISTER_API_KEY "
"(or adjust the code accordingly)."
)
return api_key
def _slugify(text: str) -> str:
"""
Converts arbitrary text into a filesystem-friendly filename
(lowercase, hyphens instead of special characters).
"""
text = text.strip().lower()
text = re.sub(r"[^a-z0-9äöüß]+", "-", text, flags=re.IGNORECASE)
text = re.sub(r"-+", "-", text).strip("-")
return text[:80] or "company"
def _read_queries(csv_path: Path) -> List[str]:
"""
Reads the company list from the CSV file and returns
a list of search queries.
Supports two CSV formats:
- Variant A: A single column 'q' with the ready-made query.
- Variant B: Columns 'company_name' (or 'name') and optionally
'city' (or 'location'/'ort'), which are concatenated.
"""
with csv_path.open("r", encoding="utf-8") as f:
reader = csv.DictReader(f)
fieldnames = reader.fieldnames or []
fieldnames_lower = [c.lower() for c in fieldnames]
queries: List[str] = []
for row in reader:
# Variant A: column "q"
if "q" in fieldnames_lower:
q = row.get(fieldnames[fieldnames_lower.index("q")], "").strip()
if q:
queries.append(q)
continue
# Variant B: "company_name" + "city"
name_key = None
city_key = None
for cand in ["company_name", "name"]:
if cand in fieldnames_lower:
name_key = fieldnames[fieldnames_lower.index(cand)]
break
for cand in ["city", "location", "ort"]:
if cand in fieldnames_lower:
city_key = fieldnames[fieldnames_lower.index(cand)]
break
if not name_key:
raise ValueError(
"CSV must contain either 'q' or 'company_name'/'name'."
)
name = (row.get(name_key) or "").strip()
city = (row.get(city_key) or "").strip() if city_key else ""
q = f"{name} {city}".strip()
if q:
queries.append(q)
return queries
def fetch_organization(
session: requests.Session,
api_key: str,
q: str,
features: Optional[List[str]] = None,
ai_search: Optional[str] = None,
timeout_s: int = 30,
) -> Dict[str, Any]:
"""
Fetches a company profile with the specified features
from the handelsregister.ai API.
Authentication is done via the x-api-key header (recommended).
Features are passed as repeated query parameters
(e.g. ?feature=financial_kpi&feature=balance_sheet_accounts).
The requests library automatically renders Python lists correctly
as multiple parameters.
"""
headers = {"x-api-key": api_key}
params: Dict[str, Any] = {"q": q}
if features:
params["feature"] = features
if ai_search:
params["ai_search"] = ai_search
resp = session.get(
FETCH_ENDPOINT, headers=headers, params=params, timeout=timeout_s
)
resp.raise_for_status()
return resp.json()
def sanitize_financial_payload(data: Dict[str, Any]) -> Dict[str, Any]:
"""
Reduces the API response to the financial portion only.
This is important for two reasons:
1. Privacy: We don't store names/addresses in training artifacts.
2. Focus: Less data = faster to load and process.
"""
return {
"entity_id": data.get("entity_id"),
"financial_kpi": data.get("financial_kpi") or [],
"balance_sheet_accounts": data.get("balance_sheet_accounts") or [],
"profit_and_loss_account": data.get("profit_and_loss_account") or [],
"meta": data.get("meta") or {},
}
def main() -> None:
base_dir = Path(__file__).resolve().parents[1]
csv_path = base_dir / "data" / "companies.csv"
out_dir = base_dir / "data" / "raw"
out_dir.mkdir(parents=True, exist_ok=True)
api_key = _get_api_key()
queries = _read_queries(csv_path)
session = requests.Session()
# Simple pacing: 60 requests/min ≈ 1 request/second.
# We add a small buffer and wait 1.05s between calls.
min_interval_s = 1.05
last_call = 0.0
for i, q in enumerate(queries, start=1):
now = time.time()
elapsed = now - last_call
if elapsed < min_interval_s:
time.sleep(min_interval_s - elapsed)
try:
data = fetch_organization(
session=session,
api_key=api_key,
q=q,
features=FEATURES,
ai_search=None, # or "on-default" for fuzzy matching
)
last_call = time.time()
except requests.HTTPError as e:
status = getattr(e.response, "status_code", None)
print(f"[{i}/{len(queries)}] ERROR ({status}) for query: {q!r}")
continue
except Exception as e:
print(f"[{i}/{len(queries)}] ERROR for query: {q!r} -> {e}")
continue
sanitized = sanitize_financial_payload(data)
entity_id = sanitized.get("entity_id") or f"unknown-{i}"
filename = f"{entity_id}_{_slugify(q)}.json"
out_path = out_dir / filename
with out_path.open("w", encoding="utf-8") as f:
json.dump(sanitized, f, ensure_ascii=False)
print(f"[{i}/{len(queries)}] OK -> {out_path.name}")
print("Done.")
if __name__ == "__main__":
main()
After the run, you'll find one JSON file per company in the data/raw/ folder containing the entity_id, KPIs, balance sheet, and P&L — ready for the next step.
Step 3: Build the Training Dataset (Features + Labels)
The heart of any ML project is feature engineering. Here we transform the raw JSON structures into a flat table where each row represents a tuple (entity_id, year) — with all computed ratios and the corresponding label.
Which Features Do We Compute?
The features we derive are the classics of financial statement analysis — many of which we already described in the conceptual scoring article. Here's an overview of what our script computes:
Capital structure: The equity ratio (equity ÷ total assets) is the single most important indicator of financial stability. It's complemented by the debt ratio (liabilities ÷ total assets) and the current assets ratio, which shows how liquid the asset base is.
Profitability: Return on Assets (ROA = net income ÷ total assets) measures how efficiently the company uses its total assets. Net margin (net income ÷ revenue) shows what remains as profit from each euro of revenue.
Cost structure: Material cost ratio and personnel cost ratio (each relative to revenue) reveal the cost structure. If these ratios are very high, there's little room for profits.
Size and efficiency: The log-transformed total assets and employee count capture company size (log-transformed because financial values are typically right-skewed). Revenue per employee is a common efficiency indicator.
Growth dynamics: Year-over-year changes in total assets, revenue, and net income reveal trends. Declining revenue combined with rising liabilities is a classic warning sign.
The Feature Engineering Script: src/build_dataset.py
The script iterates over all JSON files in data/raw/, extracts the relevant line items from the tree structure of the balance sheet and P&L for each year, computes the ratios, and generates the label. The central challenge: the balance sheet and P&L data is hierarchically structured. To find equity, for example, we need to traverse the tree and identify the correct node by its German name.
One HGB-specific detail (German GAAP) that's easily overlooked: when a company is technically insolvent (negative equity), this is not simply shown as a negative equity item on the liabilities side in HGB financial statements. Instead, a line item "Nicht durch Eigenkapital gedeckter Fehlbetrag" (shortfall not covered by equity) appears on the assets side. Our code explicitly catches this case and converts the value into negative economic equity — without this step, we might not recognize insolvent companies as such.
import json
import math
import re
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional
import numpy as np
import pandas as pd
# ──────────────────────────────────────────────
# Helper functions for tree navigation
# ──────────────────────────────────────────────
def _norm(s: Optional[str]) -> str:
"""Normalizes a string for comparisons (lowercase, reduce whitespace)."""
if not s:
return ""
s = s.strip().lower()
s = re.sub(r"\s+", " ", s)
return s
def iter_nodes(node: Dict[str, Any]) -> Iterable[Dict[str, Any]]:
"""Recursively traverses a balance sheet / P&L node and all its children."""
yield node
for child in node.get("children") or []:
yield from iter_nodes(child)
def iter_roots(roots: List[Dict[str, Any]]) -> Iterable[Dict[str, Any]]:
"""Traverses a list of root nodes (e.g. assets + liabilities)."""
for r in roots:
yield from iter_nodes(r)
def find_value(
roots: List[Dict[str, Any]],
*,
exact_de: Optional[str] = None,
contains: Optional[str] = None,
) -> Optional[float]:
"""
Searches the tree structure for a specific line item.
- exact_de: Exact match against the German name
(e.g. 'Eigenkapital')
- contains: Substring search in the name and/or 'in_report' field
(e.g. 'Kassenbestand' also finds 'Kassenbestand, Bundesbankguthaben...')
Returns the 'value' of the first match, or None.
"""
target_exact = _norm(exact_de) if exact_de else None
target_contains = _norm(contains) if contains else None
for n in iter_roots(roots):
name = n.get("name") or {}
de = name.get("de")
in_report = name.get("in_report")
hay = _norm(f"{de or ''} {in_report or ''}")
if target_exact and _norm(de) == target_exact:
return n.get("value")
if target_contains and target_contains in hay:
return n.get("value")
return None
def safe_div(num: Optional[float], den: Optional[float]) -> Optional[float]:
"""Division with None and zero safety. Returns None instead of crashing."""
if num is None or den is None:
return None
den = float(den)
if den == 0.0:
return None
return float(num) / den
# ──────────────────────────────────────────────
# Extraction: Balance Sheet
# ──────────────────────────────────────────────
def extract_balance_sheet_year(year_entry: Dict[str, Any]) -> Dict[str, Any]:
"""
Extracts relevant balance sheet items for a single fiscal year.
Special attention: the HGB edge case for negative equity.
Under German GAAP (HGB), a 'Nicht durch Eigenkapital gedeckter
Fehlbetrag' (shortfall not covered by equity) is shown on the
assets side instead of negative equity on the liabilities side.
We catch this here and compute an 'economic equity' that can
be negative.
"""
roots = year_entry.get("balance_sheet_accounts") or []
assets_total = (
find_value(roots, exact_de="Aktivseite")
or find_value(roots, contains="Aktiva")
)
fixed_assets = find_value(roots, exact_de="Anlagevermögen")
current_assets = find_value(roots, exact_de="Umlaufvermögen")
inventories = find_value(roots, exact_de="Vorräte")
receivables = find_value(roots, contains="Forderungen")
cash = find_value(roots, contains="Kassenbestand")
equity_reported = find_value(roots, exact_de="Eigenkapital")
liabilities = find_value(roots, exact_de="Verbindlichkeiten")
provisions = find_value(roots, exact_de="Rückstellungen")
# HGB edge case: negative equity
neg_equity_abs = find_value(
roots, contains="nicht durch eigenkapital gedeckter fehlbetrag"
)
equity_economic = (
-neg_equity_abs
if (neg_equity_abs is not None and neg_equity_abs > 0)
else equity_reported
)
return {
"assets_total_bs": assets_total,
"fixed_assets": fixed_assets,
"current_assets": current_assets,
"inventories": inventories,
"receivables": receivables,
"cash": cash,
"equity_economic": equity_economic,
"liabilities": liabilities,
"provisions": provisions,
"neg_equity_abs": neg_equity_abs,
}
# ──────────────────────────────────────────────
# Extraction: P&L (Income Statement)
# ──────────────────────────────────────────────
def extract_profit_loss_year(year_entry: Dict[str, Any]) -> Dict[str, Any]:
"""
Extracts relevant P&L items for a single fiscal year.
Note: The key for the P&L nodes within a year entry may be
'profit_and_loss_accounts' (plural). Check the actual JSON
structure if you get unexpected results.
"""
# The API returns the P&L nodes within a year entry under the
# key 'profit_and_loss_accounts' (plural).
roots = year_entry.get("profit_and_loss_accounts") or []
revenue = find_value(roots, exact_de="Umsatzerlöse")
material = find_value(roots, exact_de="Materialaufwand")
personnel = find_value(roots, exact_de="Personalaufwand")
net_income = (
find_value(roots, contains="Jahresüberschuss")
or find_value(roots, contains="Jahresfehlbetrag")
or find_value(roots, contains="Ergebnis nach Steuern")
)
def abs_or_none(x: Any) -> Optional[float]:
return abs(float(x)) if isinstance(x, (int, float)) else None
return {
"revenue_pl": abs_or_none(revenue),
"material_expenses_pl": abs_or_none(material),
"personnel_expenses_pl": abs_or_none(personnel),
"net_income_pl": (
float(net_income) if isinstance(net_income, (int, float)) else None
),
}
# ──────────────────────────────────────────────
# Feature computation: bringing it all together
# ──────────────────────────────────────────────
def build_features_for_entity(entity: Dict[str, Any]) -> Dict[int, Dict[str, Any]]:
"""
Computes ML features for a company across all available years.
The function combines three data sources:
1. financial_kpi (pre-extracted key metrics)
2. balance_sheet_accounts (detailed balance sheet)
3. profit_and_loss_account (detailed P&L)
Where possible, values from the KPIs are preferred (since they're
already validated and normalized). If unavailable there, we fall
back to the detailed balance sheet / P&L data.
"""
kpi_by_year = {
int(d["year"]): d
for d in (entity.get("financial_kpi") or [])
if d.get("year") is not None
}
bs_by_year = {
int(d["year"]): extract_balance_sheet_year(d)
for d in (entity.get("balance_sheet_accounts") or [])
if d.get("year") is not None
}
pl_by_year = {
int(d["year"]): extract_profit_loss_year(d)
for d in (entity.get("profit_and_loss_account") or [])
if d.get("year") is not None
}
years = sorted(set(kpi_by_year) | set(bs_by_year) | set(pl_by_year))
out: Dict[int, Dict[str, Any]] = {}
prev: Optional[Dict[str, Any]] = None
for y in years:
kpi = kpi_by_year.get(y, {})
bs = bs_by_year.get(y, {})
pl = pl_by_year.get(y, {})
# ── Base metrics (KPIs as primary source, balance sheet/P&L as fallback) ──
total_assets = kpi.get("active_total") or bs.get("assets_total_bs")
revenue = kpi.get("revenue") or pl.get("revenue_pl")
net_income = kpi.get("net_income")
if net_income is None:
net_income = pl.get("net_income_pl")
# Expense items: we take the absolute value since sign conventions
# can vary depending on the reporting format.
material_expenses = kpi.get("material_expenses") or pl.get("material_expenses_pl")
material_expenses = (
abs(float(material_expenses))
if isinstance(material_expenses, (int, float))
else None
)
personnel_expenses = kpi.get("personnel_expenses") or pl.get("personnel_expenses_pl")
personnel_expenses = (
abs(float(personnel_expenses))
if isinstance(personnel_expenses, (int, float))
else None
)
employees = kpi.get("employees")
employees = (
int(employees)
if isinstance(employees, (int, float)) and not math.isnan(employees)
else None
)
equity = bs.get("equity_economic")
liabilities = bs.get("liabilities")
current_assets = bs.get("current_assets")
cash = bs.get("cash")
inventories = bs.get("inventories")
receivables = bs.get("receivables")
# ── Derived ratios ──
equity_ratio = safe_div(equity, total_assets)
liabilities_ratio = safe_div(liabilities, total_assets)
current_assets_ratio = safe_div(current_assets, total_assets)
cash_assets_ratio = safe_div(cash, total_assets)
roa = safe_div(net_income, total_assets)
net_margin = safe_div(net_income, revenue)
inventory_share = safe_div(inventories, current_assets)
receivables_share = safe_div(receivables, current_assets)
material_ratio = safe_div(material_expenses, revenue)
personnel_ratio = safe_div(personnel_expenses, revenue)
revenue_per_employee = safe_div(revenue, employees)
personnel_per_employee = safe_div(personnel_expenses, employees)
# ── Log-transformed size features ──
# Financial data is typically right-skewed.
# log1p(x) = ln(1+x) compresses the distribution and makes
# it more tractable for linear models.
log_assets = (
math.log1p(total_assets)
if isinstance(total_assets, (int, float)) and total_assets > 0
else None
)
log_employees = (
math.log1p(employees)
if isinstance(employees, int) and employees > 0
else None
)
log_rev_per_emp = (
math.log1p(revenue_per_employee)
if isinstance(revenue_per_employee, (int, float))
and revenue_per_employee > 0
else None
)
# ── Year-over-year growth ──
# Growth rates show trends and are often more informative
# than absolute values.
asset_growth = None
revenue_growth = None
income_growth = None
if prev is not None:
if total_assets is not None and prev["total_assets"]:
asset_growth = safe_div(
total_assets - prev["total_assets"], prev["total_assets"]
)
if revenue is not None and prev["revenue"]:
revenue_growth = safe_div(
revenue - prev["revenue"], prev["revenue"]
)
# For net income, we divide by |prior year value| so that
# a sign change doesn't produce exploding values.
if net_income is not None and prev["net_income"] not in (None, 0):
income_growth = safe_div(
net_income - prev["net_income"], abs(prev["net_income"])
)
out[y] = {
"year": y,
"total_assets": total_assets,
"revenue": revenue,
"net_income": net_income,
"employees": employees,
"equity_ratio": equity_ratio,
"liabilities_ratio": liabilities_ratio,
"current_assets_ratio": current_assets_ratio,
"cash_assets_ratio": cash_assets_ratio,
"roa": roa,
"net_margin": net_margin,
"inventory_share": inventory_share,
"receivables_share": receivables_share,
"material_ratio": material_ratio,
"personnel_ratio": personnel_ratio,
"log_assets": log_assets,
"log_employees": log_employees,
"log_revenue_per_employee": log_rev_per_emp,
"personnel_per_employee": personnel_per_employee,
"asset_growth": asset_growth,
"revenue_growth": revenue_growth,
"income_growth": income_growth,
}
prev = {
"total_assets": total_assets,
"revenue": revenue,
"net_income": net_income,
}
return out
# ──────────────────────────────────────────────
# Label function: Weak Supervision
# ──────────────────────────────────────────────
def distress_rule(next_year_features: Dict[str, Any]) -> Optional[int]:
"""
Rule-based label for financial distress in the following year.
Conditions:
- Equity ratio < 0 → technical insolvency → distress = 1
- Equity ratio < 5% AND ROA < 0 → critically thin capitalization
with simultaneous losses → distress = 1
- Otherwise → distress = 0
Returns None if the data is insufficient for a decision
(e.g. missing equity ratio or ROA).
"""
eqr = next_year_features.get("equity_ratio")
roa = next_year_features.get("roa")
if eqr is None or roa is None:
return None
if eqr < 0:
return 1
if eqr < 0.05 and roa < 0:
return 1
return 0
FEATURE_COLUMNS = [
"equity_ratio",
"liabilities_ratio",
"current_assets_ratio",
"cash_assets_ratio",
"roa",
"net_margin",
"inventory_share",
"receivables_share",
"material_ratio",
"personnel_ratio",
"log_assets",
"log_employees",
"log_revenue_per_employee",
"personnel_per_employee",
"asset_growth",
"revenue_growth",
"income_growth",
]
def main() -> None:
base_dir = Path(__file__).resolve().parents[1]
raw_dir = base_dir / "data" / "raw"
out_path = base_dir / "data" / "dataset.parquet"
out_csv = base_dir / "data" / "dataset.csv"
rows: List[Dict[str, Any]] = []
for json_path in raw_dir.glob("*.json"):
with json_path.open("r", encoding="utf-8") as f:
entity = json.load(f)
entity_id = entity.get("entity_id")
if not entity_id:
continue
feat_by_year = build_features_for_entity(entity)
years = sorted(feat_by_year.keys())
# For the label we need the following year.
# We iterate up to the second-to-last year and look
# at year+1 for the label decision.
for idx, y in enumerate(years[:-1]):
x = feat_by_year[y]
y_next = years[idx + 1]
label = distress_rule(feat_by_year[y_next])
if label is None:
continue
row = {
"entity_id": entity_id,
"year": y,
"label_distress_next_year": int(label),
}
for col in FEATURE_COLUMNS:
row[col] = x.get(col)
rows.append(row)
df = pd.DataFrame(rows)
if df.empty:
raise RuntimeError(
"No training dataset generated. "
"Check raw data / CSV list / feature availability."
)
# Remove columns that are entirely empty
all_nan_cols = [c for c in FEATURE_COLUMNS if c in df.columns and df[c].isna().all()]
if all_nan_cols:
df = df.drop(columns=all_nan_cols)
print(f"Note: Columns entirely empty, removed: {all_nan_cols}")
print("Samples:", len(df))
print("Entities:", df["entity_id"].nunique())
print(
"Label distribution:\n",
df["label_distress_next_year"].value_counts(normalize=True),
)
df.to_parquet(out_path, index=False)
df.to_csv(out_csv, index=False)
print("Saved:", out_path)
print("Saved:", out_csv)
if __name__ == "__main__":
main()
When you run this script, you'll get a dataset as both a Parquet and CSV file. The output immediately shows you how many samples and entities you have and what the label distribution looks like. With heavily imbalanced labels (e.g., only 3% distress), that's normal — companies in financial trouble are fortunately the exception.
Step 4: Train the Model
Now it gets exciting. We train a logistic regression — deliberately simple, deliberately interpretable. Logistic regression has several advantages as a starting model: it's fast to train, produces calibrated probabilities, and you can directly read from the coefficients which features push in which direction. For deployment, you only need a single joblib file.
An important technical detail about the train/test split: we use a Group Shuffle Split where the grouping is by entity_id. This means all years of the same company end up either entirely in the training set or entirely in the test set. Without this grouping, the model would learn to recognize company-specific patterns (e.g., "total assets of X with equity ratio Y belongs to company Z"), which would artificially inflate test metrics. With the group split, we truly test on companies the model has never seen before.
We also use class_weight="balanced" so that the logistic regression gives more weight to the (typically underrepresented) distress class. Without this parameter, the model would simply always predict "no distress" on heavily imbalanced data and still achieve high accuracy — which would of course be useless.
The Training Script: src/train_model.py
import json
from pathlib import Path
from typing import List, Tuple
import joblib
import numpy as np
import pandas as pd
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
average_precision_score,
classification_report,
confusion_matrix,
roc_auc_score,
)
from sklearn.model_selection import GroupShuffleSplit
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
def find_working_split(
X: pd.DataFrame,
y: pd.Series,
groups: pd.Series,
test_size: float = 0.2,
max_tries: int = 50,
) -> Tuple[np.ndarray, np.ndarray]:
"""
Creates a group-based train/test split.
Why GroupShuffleSplit? We want to ensure that no company appears
in both train and test sets simultaneously (avoiding data leakage).
The loop tries different random seeds until a split is found
that contains both classes in train AND test. With small datasets
containing few positive labels, a random split might place all
positive cases in one set — we catch that here.
"""
for seed in range(max_tries):
splitter = GroupShuffleSplit(
n_splits=1, test_size=test_size, random_state=seed
)
train_idx, test_idx = next(splitter.split(X, y, groups=groups))
y_train = y.iloc[train_idx]
y_test = y.iloc[test_idx]
if y_train.nunique() == 2 and y_test.nunique() == 2:
return train_idx, test_idx
raise RuntimeError(
"Could not find a split containing both classes in train & test. "
"You need more data or a different label definition."
)
def main() -> None:
base_dir = Path(__file__).resolve().parents[1]
data_path = base_dir / "data" / "dataset.parquet"
model_dir = base_dir / "models"
model_dir.mkdir(parents=True, exist_ok=True)
df = pd.read_parquet(data_path)
target = "label_distress_next_year"
group_col = "entity_id"
# All columns except target, group ID, and year are features
feature_cols: List[str] = [
c for c in df.columns if c not in [target, group_col, "year"]
]
X = df[feature_cols]
y = df[target].astype(int)
groups = df[group_col].astype(str)
train_idx, test_idx = find_working_split(X, y, groups, test_size=0.2)
X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
# The pipeline runs three steps in sequence:
# 1. Imputer: Replace missing values with the median
# (more robust than mean for skewed distributions)
# 2. Scaler: Normalize features to mean 0 and std 1
# (important for logistic regression)
# 3. Model: Logistic regression with balanced class weighting
pipe = Pipeline(
steps=[
("imputer", SimpleImputer(strategy="median")),
("scaler", StandardScaler()),
("model", LogisticRegression(max_iter=2000, class_weight="balanced")),
]
)
pipe.fit(X_train, y_train)
# ── Evaluation ──
proba = pipe.predict_proba(X_test)[:, 1]
pred = (proba >= 0.5).astype(int)
print("ROC-AUC:", roc_auc_score(y_test, proba))
print("Average Precision:", average_precision_score(y_test, proba))
print("Confusion Matrix:\n", confusion_matrix(y_test, pred))
print(classification_report(y_test, pred, digits=3))
# ── Save model and metadata ──
model_path = model_dir / "financial_health_model.joblib"
joblib.dump(pipe, model_path)
meta = {
"feature_cols": feature_cols,
"target": target,
"label_definition": (
"distress_next_year = (equity_ratio<0) OR "
"(equity_ratio<0.05 AND roa<0) in the following year"
),
"threshold": 0.5,
}
meta_path = model_dir / "financial_health_model.meta.json"
meta_path.write_text(
json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8"
)
print("Model saved:", model_path)
print("Meta saved:", meta_path)
if __name__ == "__main__":
main()
Alongside the model, we save a meta file in JSON format that records which feature columns the model expects and how the label is defined. This might sound like a small detail, but it becomes important as soon as you update the model: it ensures that training and inference always work with the same feature definition.
Step 5: Score a Company Locally (CLI)
Now we build the final piece: a CLI tool that queries a company via the API in real time, computes the features, and outputs the trained score. The tool imports the feature logic directly from build_dataset.py — ensuring that training and inference use identical computations.
A small bonus feature: since we use logistic regression, we can provide a simple feature explanation. The idea: we look at which features (after standardization and weighting by the coefficients) contribute most strongly to the score. This immediately gives you an intuitive answer to the question "Why does this company have this score?" — with more complex models, you'd need SHAP or similar methods for that.
The Scoring Script: src/score_company.py
import json
import os
import sys
from typing import Any, Dict, List, Optional, Tuple
import joblib
import numpy as np
import pandas as pd
import requests
# We import the feature logic 1:1 from build_dataset.py
# to ensure training and inference are identical.
from build_dataset import build_features_for_entity, FEATURE_COLUMNS
API_BASE = "https://handelsregister.ai/api"
FETCH_ENDPOINT = f"{API_BASE}/v1/fetch-organization"
FEATURES = ["financial_kpi", "balance_sheet_accounts", "profit_and_loss_account"]
def get_api_key() -> str:
api_key = os.getenv("HANDELSREGISTER_API_KEY")
if not api_key:
raise RuntimeError("Please set HANDELSREGISTER_API_KEY.")
return api_key
def fetch_financial_data(q: str) -> Dict[str, Any]:
"""
Fetches a company's financial data and reduces the
response to the relevant portion (same as during training).
"""
headers = {"x-api-key": get_api_key()}
params = {"q": q, "feature": FEATURES}
resp = requests.get(FETCH_ENDPOINT, headers=headers, params=params, timeout=30)
resp.raise_for_status()
data = resp.json()
return {
"entity_id": data.get("entity_id"),
"financial_kpi": data.get("financial_kpi") or [],
"balance_sheet_accounts": data.get("balance_sheet_accounts") or [],
"profit_and_loss_account": data.get("profit_and_loss_account") or [],
}
def explain_linear_model(
pipeline, x_row: pd.DataFrame, feature_cols: List[str], top_k: int = 8
) -> List[Dict[str, Any]]:
"""
Generates a simple, local explanation for the prediction.
For logistic regression, the explanation is particularly intuitive:
each feature's contribution to the log-odds score is
coef[i] * z[i], where z[i] is the standardized feature value.
Positive contributions increase the predicted distress probability,
negative ones decrease it. We sort by absolute contribution and
show the top_k most influential features.
"""
imputer = pipeline.named_steps["imputer"]
scaler = pipeline.named_steps["scaler"]
model = pipeline.named_steps["model"]
x_imp = imputer.transform(x_row[feature_cols])
x_z = scaler.transform(x_imp)
coefs = model.coef_.reshape(-1)
contrib = x_z.reshape(-1) * coefs
order = np.argsort(np.abs(contrib))[::-1][:top_k]
out = []
for idx in order:
out.append(
{
"feature": feature_cols[idx],
"contribution_log_odds": float(contrib[idx]),
"direction": (
"increases risk" if contrib[idx] > 0 else "decreases risk"
),
"value": (
None
if pd.isna(
x_row.iloc[0, x_row.columns.get_loc(feature_cols[idx])]
)
else float(x_row.iloc[0][feature_cols[idx]])
),
}
)
return out
def risk_to_health_score(prob_distress: float) -> float:
"""Converts the distress probability into a 0-100 score."""
return round((1.0 - prob_distress) * 100.0, 1)
def bucket(score: float) -> str:
"""Maps the score to a human-readable category."""
if score >= 70:
return "healthy"
if score >= 40:
return "watch"
return "critical"
def main() -> None:
if len(sys.argv) < 2:
print('Usage: python src/score_company.py "Company Name City"')
sys.exit(1)
q = " ".join(sys.argv[1:]).strip()
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
model_path = os.path.join(base_dir, "models", "financial_health_model.joblib")
meta_path = os.path.join(base_dir, "models", "financial_health_model.meta.json")
pipe = joblib.load(model_path)
meta = json.loads(open(meta_path, "r", encoding="utf-8").read())
feature_cols = meta["feature_cols"]
entity = fetch_financial_data(q)
feat_by_year = build_features_for_entity(entity)
if not feat_by_year:
raise RuntimeError("No fiscal years found. Check query/features.")
latest_year = max(feat_by_year.keys())
x = feat_by_year[latest_year]
row = {c: x.get(c) for c in feature_cols}
x_row = pd.DataFrame([row])
prob = float(pipe.predict_proba(x_row)[:, 1][0])
score = risk_to_health_score(prob)
explanation = explain_linear_model(pipe, x_row, feature_cols, top_k=8)
result = {
"query": q,
"entity_id": entity.get("entity_id"),
"year": latest_year,
"prob_distress_next_year": round(prob, 4),
"financial_health_score_0_100": score,
"bucket": bucket(score),
"top_drivers": explanation,
"disclaimer": (
"This is only a technical example model (weak labels). "
"Not an official rating, not financial advice."
),
}
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
Running the Full Pipeline
The complete pipeline can be started in four commands:
python src/fetch_data.py # Fetch data
python src/build_dataset.py # Generate features + labels
python src/train_model.py # Train model
python src/score_company.py "Beispiel GmbH Berlin" # Live scoring
The output of the scoring script looks roughly like this (values are illustrative):
{
"query": "Beispiel GmbH Berlin",
"entity_id": "abc123...",
"year": 2023,
"prob_distress_next_year": 0.1842,
"financial_health_score_0_100": 81.6,
"bucket": "healthy",
"top_drivers": [
{"feature": "equity_ratio", "contribution_log_odds": -1.23, "direction": "decreases risk", "value": 0.42},
{"feature": "roa", "contribution_log_odds": -0.87, "direction": "decreases risk", "value": 0.05}
],
"disclaimer": "This is only a technical example model (weak labels). Not an official rating, not financial advice."
}
Optional: Local REST Service (FastAPI)
If you want to access the score not just from the command line but integrate it into an internal tool or dashboard, you can serve it as a local HTTP endpoint with minimal effort. FastAPI is excellent for this — it's fast, type-safe, and automatically generates interactive API documentation at /docs.
pip install fastapi uvicorn
Create the file src/app.py:
import json
import os
from typing import Any, Dict
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException
import requests
from build_dataset import build_features_for_entity
API_BASE = "https://handelsregister.ai/api"
FETCH_ENDPOINT = f"{API_BASE}/v1/fetch-organization"
FEATURES = ["financial_kpi", "balance_sheet_accounts", "profit_and_loss_account"]
app = FastAPI(title="Financial Health Scoring (local)")
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
MODEL_PATH = os.path.join(BASE_DIR, "models", "financial_health_model.joblib")
META_PATH = os.path.join(BASE_DIR, "models", "financial_health_model.meta.json")
# Load model and metadata once at startup
pipe = joblib.load(MODEL_PATH)
meta = json.loads(open(META_PATH, "r", encoding="utf-8").read())
feature_cols = meta["feature_cols"]
def api_key() -> str:
k = os.getenv("HANDELSREGISTER_API_KEY")
if not k:
raise RuntimeError("HANDELSREGISTER_API_KEY is missing.")
return k
def fetch_financial(q: str) -> Dict[str, Any]:
headers = {"x-api-key": api_key()}
params = {"q": q, "feature": FEATURES}
r = requests.get(FETCH_ENDPOINT, headers=headers, params=params, timeout=30)
if r.status_code != 200:
raise HTTPException(
status_code=502, detail=f"Upstream error: {r.status_code}"
)
data = r.json()
return {
"entity_id": data.get("entity_id"),
"financial_kpi": data.get("financial_kpi") or [],
"balance_sheet_accounts": data.get("balance_sheet_accounts") or [],
"profit_and_loss_account": data.get("profit_and_loss_account") or [],
}
@app.get("/health")
def health():
return {"ok": True}
@app.post("/score")
def score(payload: Dict[str, Any]):
q = (payload.get("q") or "").strip()
if not q:
raise HTTPException(
status_code=400, detail="payload must include non-empty 'q'"
)
entity = fetch_financial(q)
feat_by_year = build_features_for_entity(entity)
if not feat_by_year:
raise HTTPException(status_code=404, detail="no financial years found")
latest_year = max(feat_by_year.keys())
x = feat_by_year[latest_year]
row = {c: x.get(c) for c in feature_cols}
x_row = pd.DataFrame([row])
prob = float(pipe.predict_proba(x_row)[:, 1][0])
score_0_100 = round((1.0 - prob) * 100.0, 1)
return {
"q": q,
"entity_id": entity.get("entity_id"),
"year": latest_year,
"prob_distress_next_year": round(prob, 4),
"financial_health_score_0_100": score_0_100,
}
Start:
uvicorn src.app:app --reload --port 8000
Then access the interactive documentation at http://localhost:8000/docs and retrieve scores via POST requests.
What You Can Improve Next
This tutorial gives you a working foundation. If you want to use it in production or build on it further, there are several natural next steps:
Better labels are the biggest lever. Weak labels are a pragmatic start, but a model can't get better than its labels. If you have access to actual default data (e.g., from insolvency announcements or via the handelsregister.ai API with the insolvency_publications feature), you can define significantly more valid target variables. Manual curation — reviewing and correcting labels for a few hundred cases — also noticeably improves model quality.
More and richer features offer further potential. For instance, you could compute trends over 3–5 years (e.g., average equity ratio over the last three years, or the standard deviation of ROA as a volatility measure). Industry as a categorical feature can also help, since "normal" ratio values vary heavily by sector.
Model calibration matters if you want to interpret the output probabilities as actual probabilities. Methods like Platt Scaling or Isotonic Regression ensure that an output of 0.8 truly means "80% distress probability."
Alternative models: Logistic regression is an excellent starting point, but gradient-boosted trees (e.g., LightGBM or XGBoost) can capture non-linear relationships more effectively. The trade-off: somewhat less interpretability, but often better predictive performance.
Conclusion
You now have a complete end-to-end pipeline: from structured retrieval of Bundesanzeiger financial data via the handelsregister.ai API (JSON instead of PDF), through feature engineering and model training, to local deployment as a CLI or REST service. Everything runs on your machine, with no external dependencies.
If you want to use this at larger scale (more companies, regular updates), the next step is typically: batch enrichment with repeatable training and monitoring — exactly the direction that many of the technical blog posts already cover.