Trading Workflow Architecture For The FCA's 2028 Transaction Reports: Build, Check And Approve The Draft Schema In Python

"""Build a transaction report in the FCA's draft 2028 format (auth.016, DATTRA draft V1.0.0).

Sample data only: the LEIs, trade references and prices below are made up.
Standard library only. Tested on Python 3.14.4 and 3.11.14 with mypy --strict.
"""

from __future__ import annotations

import re
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path

# Target namespace of auth.016.001.03_FCA_DATTRA_1.0.0.xsd (published 2 October 2026).
NS = "urn:fca:org:uk:tech:xsd:auth.016.200.03"
PACKAGE_ID = re.compile(r"[A-Z0-9]{1,35}")  # Field 37 pattern in the draft XSD


@dataclass(frozen=True)
class Party:
    lei: str
    is_client: bool  # Field 8 / 17, "Client indicator", mandatory from 3 April 2028


@dataclass(frozen=True)
class Trade:
    tx_id: str
    executing_lei: str
    buyer: Party
    seller: Party
    traded_at: datetime  # must carry a time zone
    capacity: str  # DEAL, MTCH or AOTC
    quantity: Decimal
    price: Decimal
    currency: str
    venue_mic: str
    isin: str
    algo_id: str
    package_id: str | None = None
    package_price: Decimal | None = None


def utc_z(ts: datetime) -> str:
    """YYYY-MM-DDThh:mm:ss.ssssssZ. datetime.isoformat() gives +00:00, which the draft XSD rejects."""
    if ts.tzinfo is None:
        raise ValueError("naive datetime: refusing to guess the time zone")
    return ts.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")


def money(value: Decimal) -> str:
    return format(value, "f")


def _el(parent: ET.Element, tag: str, text: str | None = None, **attrs: str) -> ET.Element:
    el = ET.SubElement(parent, f"{{{NS}}}{tag}", attrs)
    if text is not None:
        el.text = text
    return el


def _party(parent: ET.Element, tag: str, party: Party) -> None:
    node = _el(parent, tag)
    _el(node, "ClntInd", "true" if party.is_client else "false")
    _el(_el(_el(node, "AcctOwnr"), "Id"), "LEI", party.lei)


def _amount(parent: ET.Element, value: Decimal, currency: str) -> None:
    """AmountAndDirection61: Amt has minInclusive 0, so a negative price is Amt plus Sgn=false."""
    mntry = _el(_el(parent, "Pric"), "MntryVal")
    _el(mntry, "Amt", money(abs(value)), Ccy=currency)
    if value < 0:
        _el(mntry, "Sgn", "false")


def add_new_report(doc_root: ET.Element, trade: Trade) -> None:
    if trade.capacity not in {"DEAL", "MTCH", "AOTC"}:
        raise ValueError(f"{trade.tx_id}: unknown trading capacity {trade.capacity}")
    if trade.package_id is not None and not PACKAGE_ID.fullmatch(trade.package_id):
        raise ValueError(f"{trade.tx_id}: package id must be 1-35 capital letters or digits")
    new = _el(_el(doc_root, "Tx"), "New")
    _el(new, "TxId", trade.tx_id)
    _el(new, "ExctgPty", trade.executing_lei)
    _el(new, "ExctgNttyTxRptngPtyInd", "true")
    _el(new, "SubmitgPty", trade.executing_lei)
    _party(new, "Buyr", trade.buyer)
    _party(new, "Sellr", trade.seller)
    _el(new, "OrdrTrnsmssn")  # mandatory element, both children optional
    tx = _el(new, "Tx")
    _el(tx, "TradDt", utc_z(trade.traded_at))
    _el(tx, "TradgCpcty", trade.capacity)
    _el(_el(tx, "Qty"), "Unit", money(trade.quantity))
    _amount(_el(tx, "Pric"), trade.price, trade.currency)
    _el(tx, "TradVn", trade.venue_mic)
    if trade.package_id is not None:
        _el(tx, "PckgId", trade.package_id)
        if trade.package_price is not None:
            _amount(_el(tx, "PckgTxPric"), trade.package_price, trade.currency)
    _el(_el(new, "FinInstrm"), "Id", trade.isin)
    _el(_el(new, "InvstmtDcsnPrsn"), "Algo", trade.algo_id)
    _el(_el(new, "ExctgPrsn"), "Algo", trade.algo_id)


def build_document(trades: list[Trade]) -> ET.ElementTree[ET.Element]:
    ET.register_namespace("", NS)
    document = ET.Element(f"{{{NS}}}Document")
    report = _el(document, "FinInstrmRptgTxRpt")
    for trade in trades:
        add_new_report(report, trade)
    tree: ET.ElementTree[ET.Element] = ET.ElementTree(document)
    ET.indent(tree)
    return tree


FIRM = "213800SAMPLEFIRM0001"  # sample LEI pattern, not a real entity
CLIENT = Party("5493000SAMPLECLNT042", is_client=True)
STREET = Party("549300SAMPLECOUNTE07", is_client=False)
WHEN = datetime(2028, 4, 3, 9, 30, 45, 124000, tzinfo=timezone.utc)


def spread_leg(tx_id: str, buyer: Party, seller: Party, price: Decimal, isin: str) -> Trade:
    return Trade(tx_id=tx_id, executing_lei=FIRM, buyer=buyer, seller=seller, traded_at=WHEN,
                 capacity="DEAL", quantity=Decimal("10"), price=price, currency="GBP",
                 venue_mic="XLON", isin=isin, algo_id="EXECALGO1",
                 package_id="SPRD00001", package_price=Decimal("-150.00"))


def sample_trades() -> list[Trade]:
    """A two-leg spread (package SPRD00001, as in CP26/34 Appendix 2) and one client trade."""
    us = Party(FIRM, is_client=False)
    return [
        spread_leg("SPRD00001LEG1", STREET, us, Decimal("3600"), "GB00B03MLX29"),
        spread_leg("SPRD00001LEG2", us, STREET, Decimal("3750"), "GB0002634946"),
        Trade(tx_id="AGENCY000042", executing_lei=FIRM, buyer=CLIENT, seller=STREET, traded_at=WHEN,
              capacity="AOTC", quantity=Decimal("2500"), price=Decimal("12.345"), currency="GBP",
              venue_mic="XLON", isin="GB0007980591", algo_id="EXECALGO1"),
    ]


if __name__ == "__main__":
    out = Path("report.xml")
    build_document(sample_trades()).write(out, encoding="UTF-8", xml_declaration=True)
    print(f"wrote {out} with {len(sample_trades())} new transaction reports")

Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict. Standard library only. Sample LEIs and prices.

"""Pre-submission checks for draft 2028 transaction reports.

Stage 1 validates against the FCA's draft XSD (auth.016.001.03_FCA_DATTRA_1.0.0.xsd, 2 October 2026).
Stage 2 applies a subset of the draft content rules (validation rules V1.3 DRAFT, 2 October 2026).
Stage 3 needs a named person to approve the exact file before it can be released, and every step is
written to a hash-chained audit log. Nothing here submits to the FCA.

Requires xmlschema 4.3.2. Tested on Python 3.14.4 and 3.11.14 with mypy --strict. Sample data only.
"""

from __future__ import annotations

import hashlib
import json
import sys
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import date, datetime, timezone
from pathlib import Path

import xmlschema

NS = {"t": "urn:fca:org:uk:tech:xsd:auth.016.200.03"}
REGIME_START = date(2028, 4, 3)
# Rule TBC / CON-391: special currency codes not allowed for the package transaction currency
SPECIAL_CCY = {"XAG", "XAU", "XBA", "XBB", "XBC", "XBD", "XDR", "XEU", "XFU", "XPD", "XPT", "XXX"}


@dataclass(frozen=True)
class Finding:
    tx_id: str
    code: str  # FCA error code, or GUIDANCE for CP26/34 Appendix 2
    message: str


@dataclass(frozen=True)
class Review:
    file: str
    sha256: str
    reports: int
    schema_errors: list[str]
    findings: list[Finding]

    @property
    def clean(self) -> bool:
        return not self.schema_errors and not self.findings


def sha256_of(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def isin_ok(isin: str) -> bool:
    """ISO 6166 check digit (rule 130, CON-400): letters to numbers, then Luhn from the right."""
    if len(isin) != 12 or not isin[:2].isalpha() or not isin[-1].isdigit():
        return False
    digits = "".join(str(int(ch, 36)) for ch in isin[:11])
    total = 0
    for i, ch in enumerate(reversed(digits)):
        d = int(ch)
        if i % 2 == 0:
            d *= 2
            if d > 9:
                d -= 9
        total += d
    return (10 - total % 10) % 10 == int(isin[-1])


def _text(node: ET.Element, path: str) -> str | None:
    found = node.find(path, NS)
    return found.text if found is not None else None


def schema_errors(xsd: Path, report: Path) -> list[str]:
    schema = xmlschema.XMLSchema(str(xsd))
    return [f"{err.path}: {err.reason}" for err in schema.iter_errors(str(report))]


def content_findings(report: Path, as_of: datetime) -> list[Finding]:
    root = ET.parse(report).getroot()
    findings: list[Finding] = []
    package_prices: dict[str, set[str]] = {}
    for new in root.iterfind("t:FinInstrmRptgTxRpt/t:Tx/t:New", NS):
        tx_id = _text(new, "t:TxId") or "(missing TxId)"
        executing = _text(new, "t:ExctgPty")
        traded_raw = _text(new, "t:Tx/t:TradDt") or ""
        try:
            traded = datetime.fromisoformat(traded_raw)
        except ValueError:
            findings.append(Finding(tx_id, "SCHEMA", f"unreadable trading date time {traded_raw!r}"))
            continue
        in_regime = traded.date() >= REGIME_START

        def add(code: str, message: str) -> None:
            findings.append(Finding(tx_id, code, message))

        if traded > as_of:
            add("CON-270", "Trading date time is in the future")
        if in_regime and new.find("t:Buyr/t:ClntInd", NS) is None:
            add("CON-080", "Client indicator for the buyer is missing")
        if in_regime and new.find("t:Sellr/t:ClntInd", NS) is None:
            add("CON-170", "Client indicator for the seller is missing")
        sides = {_text(new, "t:Buyr/t:AcctOwnr/t:Id/t:LEI"), _text(new, "t:Sellr/t:AcctOwnr/t:Id/t:LEI")}
        capacity = _text(new, "t:Tx/t:TradgCpcty")
        if capacity == "DEAL" and executing not in sides:
            add("CON-280", "When using 'DEAL' either Buyer or Seller should be identical with the "
                           "executing entity identification code")
        if capacity in {"MTCH", "AOTC"} and executing in sides:
            add("CON-281", "When using 'MTCH' or 'AOTC' neither Buyer or Seller should be identical "
                           "with the executing entity identification code")
        package = _text(new, "t:Tx/t:PckgId")
        price_node = new.find("t:Tx/t:PckgTxPric", NS)
        if package is not None and price_node is None and in_regime:
            add("CON-380", "The package transaction price is missing")
        if package is not None and price_node is not None:
            amt = price_node.find(".//t:Amt", NS)
            if amt is not None and amt.get("Ccy") in SPECIAL_CCY:
                add("CON-391", "Currency code is not valid for the trade date")
            signature = ET.tostring(price_node, encoding="unicode")
            package_prices.setdefault(package, set()).add(signature)
        isin = _text(new, "t:FinInstrm/t:Id")
        if isin is not None and not isin_ok(isin):
            add("CON-400", "The ISIN code is invalid")
    for package, prices in package_prices.items():
        if len(prices) > 1:
            findings.append(Finding(package, "GUIDANCE",
                                    "Field 38 differs between legs of the same package (CP26/34 Appendix 2)"))
    return findings


def review(xsd: Path, report: Path, as_of: datetime) -> Review:
    root = ET.parse(report).getroot()
    count = len(root.findall("t:FinInstrmRptgTxRpt/t:Tx/t:New", NS))
    return Review(str(report), sha256_of(report), count, schema_errors(xsd, report),
                  content_findings(report, as_of))


def append_audit(log: Path, event: dict[str, object]) -> str:
    """Append-only JSONL. Each entry carries the hash of the previous one, so edits are detectable."""
    prev = "0" * 64
    if log.exists() and log.stat().st_size:
        prev = str(json.loads(log.read_text().splitlines()[-1])["entry_hash"])
    body = {**event, "prev_hash": prev, "at": datetime.now(timezone.utc).isoformat()}
    entry_hash = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()
    with log.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps({**body, "entry_hash": entry_hash}) + "\n")
    return entry_hash


def approve(result: Review, approver: str, log: Path) -> None:
    """A named person approves this exact file. Files with errors cannot be approved."""
    if not result.clean:
        raise PermissionError(f"{result.file}: fix {len(result.schema_errors)} schema errors and "
                              f"{len(result.findings)} findings before approval")
    append_audit(log, {"event": "approved", "file": result.file, "sha256": result.sha256,
                       "reports": result.reports, "approver": approver})


def release_for_submission(report: Path, log: Path) -> Path:
    """Hand the file to your submission gateway only if this exact content was approved."""
    digest = sha256_of(report)
    events = [json.loads(line) for line in log.read_text().splitlines()] if log.exists() else []
    if not any(e["event"] == "approved" and e["sha256"] == digest for e in events):
        append_audit(log, {"event": "release_refused", "file": str(report), "sha256": digest})
        raise PermissionError(f"{report}: no approval recorded for sha256 {digest[:12]}")
    append_audit(log, {"event": "released", "file": str(report), "sha256": digest})
    return report


def print_review(result: Review) -> None:
    print(f"{result.file}: {result.reports} reports, sha256 {result.sha256[:12]}")
    for err in result.schema_errors:
        print(f"  SCHEMA  {err}")
    for f in result.findings:
        print(f"  {f.code:8} {f.tx_id}: {f.message}")
    if result.schema_errors:
        print("  -> the FCA rejects the whole file on a schema error")
    print("  -> clean" if result.clean else "  -> not ready for approval")


if __name__ == "__main__":
    from build_report import build_document, sample_trades

    xsd = Path(sys.argv[1] if len(sys.argv) > 1 else "fca_draft/auth.016.001.03_FCA_DATTRA_1.0.0.xsd")
    as_of = datetime(2028, 4, 4, 7, 0, tzinfo=timezone.utc)  # simulate a run after the switch-over
    log = Path("audit.jsonl")
    log.unlink(missing_ok=True)

    good = Path("report.xml")
    build_document(sample_trades()).write(good, encoding="UTF-8", xml_declaration=True)
    clean = review(xsd, good, as_of)
    print_review(clean)

    # A copy with three mistakes that are easy to make when moving to the new format
    tree = ET.parse(good)
    news = tree.getroot().findall("t:FinInstrmRptgTxRpt/t:Tx/t:New", NS)
    when = news[0].find("t:Tx/t:TradDt", NS)
    assert when is not None and when.text is not None
    when.text = datetime.fromisoformat(when.text).isoformat()  # +00:00 instead of Z
    tx2 = news[1].find("t:Tx", NS)
    pkg_price = news[1].find("t:Tx/t:PckgTxPric", NS)
    assert tx2 is not None and pkg_price is not None
    tx2.remove(pkg_price)  # leg 2 loses its package price
    isin = news[2].find("t:FinInstrm/t:Id", NS)
    assert isin is not None
    isin.text = "GB0007980592"  # wrong check digit
    bad = Path("report_bad.xml")
    tree.write(bad, encoding="UTF-8", xml_declaration=True)
    print_review(review(xsd, bad, as_of))

    approve(clean, approver="ops.lead@example.co.uk", log=log)
    print("released:", release_for_submission(good, log))
    good.write_text(good.read_text().replace("12.345", "12.346"))  # edited after approval
    try:
        release_for_submission(good, log)
    except PermissionError as exc:
        print("refused:", exc)
    print(f"audit log has {len(log.read_text().splitlines())} chained entries")

Requires xmlschema 4.3.2 and the draft XSD from the FCA's zip file. Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict.