"""fpml_convert (single-file drop-in) — FpML interest-rate-swap XML <-> JSON. Generated from structile-fpml 0.6.3 (FpML 5.10) by scripts/build_artifacts.mjs — the exact projection the JS interpreter bundle and the C++ converter implement. Standard library only; drop this one file next to your code, no install: from fpml_convert_singlefile import fpml_to_json, json_to_fpml or run it: `python fpml_convert_singlefile.py in.xml -o out.json`. """ from __future__ import annotations import argparse import json import sys import xml.etree.ElementTree as ET FPML_VERSION = "5.10" _RULES = { "_comment": "Language-neutral FpML projection rule tables, the single source of truth shared by the JS bundle (injected by scripts/build.mjs as `const FPML_RULES`, consumed by src/40-support.js and src/60-binding.js) and the Python converter (python/fpml_convert/_rules.py). Widen the same way as before: against a real corpus file or the FpML 5.10 XSD, never speculatively. See src/40-support.js and src/60-binding.js headers for the rationale behind each table.", "supported": [ "dataDocument", "trade", "tradeHeader", "partyTradeIdentifier", "partyReference", "tradeId", "tradeDate", "party", "partyId", "partyName", "calculationAgent", "calculationAgentPartyReference", "swap", "swapStream", "payerPartyReference", "receiverPartyReference", "calculationPeriodDates", "effectiveDate", "terminationDate", "unadjustedDate", "adjustedStartDate", "adjustedEndDate", "dateAdjustments", "businessDayConvention", "businessCenters", "businessCenter", "businessCentersReference", "calculationPeriodDatesAdjustments", "calculationPeriodFrequency", "periodMultiplier", "period", "rollConvention", "firstPeriodStartDate", "firstRegularPeriodStartDate", "lastRegularPeriodEndDate", "firstPaymentDate", "stubCalculationPeriodAmount", "initialStub", "finalStub", "stubRate", "paymentDates", "calculationPeriodDatesReference", "paymentFrequency", "payRelativeTo", "paymentDatesAdjustments", "paymentDaysOffset", "adjustedPaymentDate", "paymentDate", "resetDates", "resetRelativeTo", "fixingDates", "dayType", "dateRelativeTo", "resetFrequency", "resetDatesAdjustments", "adjustedFixingDate", "calculationPeriodAmount", "calculation", "compoundingMethod", "notionalSchedule", "notionalStepSchedule", "notionalAmount", "initialValue", "currency", "step", "stepDate", "stepValue", "fixedRateSchedule", "fixedRate", "spreadSchedule", "floatingRateCalculation", "floatingRateDefinition", "floatingRate", "floatingRateIndex", "indexTenor", "finalRateRounding", "roundingDirection", "precision", "rateObservation", "observationWeight", "dayCountFraction", "calculationPeriod", "paymentCalculationPeriod", "cashflows", "cashflowsMatchParameters", "additionalPayment", "paymentAmount", "amount", "principalExchanges", "initialExchange", "intermediateExchange", "finalExchange", "principalExchange", "principalExchangeAmount", "principalExchangeDate", "adjustedPrincipalExchangeDate" ], "childOrder": { "additionalPayment": ["payerPartyReference", "receiverPartyReference", "paymentAmount", "paymentDate"], "businessCenters": ["businessCenter"], "calculation": ["notionalSchedule", "fixedRateSchedule", "floatingRateCalculation", "dayCountFraction", "compoundingMethod"], "calculationAgent": ["calculationAgentPartyReference"], "calculationPeriod": ["adjustedStartDate", "adjustedEndDate", "notionalAmount", "floatingRateDefinition", "fixedRate"], "calculationPeriodAmount": ["calculation"], "calculationPeriodDates": ["effectiveDate", "terminationDate", "calculationPeriodDatesAdjustments", "firstPeriodStartDate", "firstRegularPeriodStartDate", "lastRegularPeriodEndDate", "calculationPeriodFrequency"], "calculationPeriodDatesAdjustments": ["businessDayConvention", "businessCentersReference", "businessCenters"], "calculationPeriodFrequency": ["periodMultiplier", "period", "rollConvention"], "cashflows": ["cashflowsMatchParameters", "principalExchange", "paymentCalculationPeriod"], "dataDocument": ["trade", "party"], "dateAdjustments": ["businessDayConvention", "businessCenters", "businessCentersReference"], "effectiveDate": ["unadjustedDate", "dateAdjustments"], "finalRateRounding": ["roundingDirection", "precision"], "finalStub": ["floatingRate"], "firstPeriodStartDate": ["unadjustedDate", "dateAdjustments"], "fixedRateSchedule": ["initialValue", "step"], "fixingDates": ["periodMultiplier", "period", "dayType", "businessDayConvention", "businessCenters", "dateRelativeTo"], "floatingRate": ["floatingRateIndex", "indexTenor"], "floatingRateCalculation": ["floatingRateIndex", "indexTenor", "finalRateRounding", "spreadSchedule"], "floatingRateDefinition": ["rateObservation"], "indexTenor": ["periodMultiplier", "period"], "initialStub": ["floatingRate", "stubRate"], "notionalSchedule": ["notionalStepSchedule"], "notionalStepSchedule": ["initialValue", "step", "currency"], "party": ["partyId", "partyName"], "partyTradeIdentifier": ["partyReference", "tradeId"], "paymentAmount": ["currency", "amount"], "paymentCalculationPeriod": ["adjustedPaymentDate", "calculationPeriod"], "paymentDate": ["unadjustedDate", "dateAdjustments"], "paymentDates": ["calculationPeriodDatesReference", "paymentFrequency", "firstPaymentDate", "payRelativeTo", "paymentDaysOffset", "paymentDatesAdjustments"], "paymentDatesAdjustments": ["businessDayConvention", "businessCentersReference", "businessCenters"], "paymentDaysOffset": ["periodMultiplier", "period", "dayType"], "paymentFrequency": ["periodMultiplier", "period"], "principalExchange": ["adjustedPrincipalExchangeDate", "principalExchangeAmount"], "principalExchanges": ["initialExchange", "finalExchange", "intermediateExchange"], "rateObservation": ["adjustedFixingDate", "observationWeight"], "resetDates": ["calculationPeriodDatesReference", "resetRelativeTo", "fixingDates", "resetFrequency", "resetDatesAdjustments"], "resetDatesAdjustments": ["businessDayConvention", "businessCentersReference", "businessCenters"], "resetFrequency": ["periodMultiplier", "period"], "spreadSchedule": ["initialValue"], "step": ["stepDate", "stepValue"], "stubCalculationPeriodAmount": ["calculationPeriodDatesReference", "initialStub", "finalStub"], "swap": ["swapStream", "additionalPayment"], "swapStream": ["payerPartyReference", "receiverPartyReference", "calculationPeriodDates", "paymentDates", "resetDates", "calculationPeriodAmount", "stubCalculationPeriodAmount", "principalExchanges", "cashflows"], "terminationDate": ["unadjustedDate", "dateAdjustments"], "trade": ["tradeHeader", "swap", "calculationAgent", "fra"], "tradeHeader": ["partyTradeIdentifier", "tradeDate"] } } FPML_SUPPORTED = frozenset(_RULES["supported"]) FPML_CHILD_ORDER = dict(_RULES["childOrder"]) # --- projection (python/fpml_convert/_project.py) --------------------------- """FpML XML -> plain value projection — a faithful port of the JS bundle's ``src/20-ns.js`` + ``src/30-project.js`` + ``src/40-support.js`` + ``src/50-swap.js``. The rule tables come from the shared ``src/rules.json`` (see :mod:`fpml_convert._rules`); only this traversal logic is duplicated between the two implementations, and the committed JS snapshots (``test/__snapshots__/*.json``) are the cross-language contract. Projection rules (identical to the JS interpreter): * element local name -> key; a repeated tag under one parent -> a list, in document order. * attributes -> ``@``-prefixed keys: ``@id`` first, then ``@xsi:type``, then ``@href``, then any other plain attribute, all BEFORE the child keys. ``xmlns*`` and other ``xsi:*`` attributes are dropped. * a pure reference element (``href``, nothing else) -> the link scalar ``«href»``. One that also carries content stays a dict but still gets ``@href = «href»``. * ``xsi:type`` -> ``@xsi:type`` verbatim. Typed lexical forms are NEVER coerced — every leaf value stays the exact source string. * every element descended into is checked against ``FPML_SUPPORTED`` first; an unknown element is a loud error, not a best-effort guess. """ LINK_OPEN = "«" # « LINK_CLOSE = "»" # » LINK_ID_FIELD = "@id" _XSI_NS = "http://www.w3.org/2001/XMLSchema-instance" class FpmlUnsupportedError(Exception): """Raised for an element outside the supported IR-swap subset. A structural "I don't recognise this" — distinct from the soft rule violations Structile's ``__structileDiagnostic`` is for.""" def fpml_local_name(name) -> str: """``{uri}local`` -> ``local``; ``prefix:local`` -> ``local``; ``local`` -> ``local``. Reduces both DOM providers' spellings.""" t = "" if name is None else str(name) brace = t.rfind("}") if brace != -1: return t[brace + 1:] colon = t.rfind(":") return t[colon + 1:] if colon != -1 else t def fpml_element_local_name(el) -> str: return fpml_local_name(getattr(el, "tag", None)) def _attr_items(el): """``el.attrib`` items in source order (ElementTree preserves it since Python 3.8).""" return list(el.attrib.items()) def fpml_xsi_type(el): for raw, val in _attr_items(el): if raw == "xsi:type": return val if raw.startswith("{") and "}" in raw: uri = raw[1:raw.index("}")] if uri == _XSI_NS and fpml_local_name(raw) == "type": return val return None def fpml_plain_attributes(el): """Non-namespaced "business" attributes as an ordered ``dict``, with the structural ones removed: ``xmlns*``, ``xsi:*`` (``xsi:type`` / ``xsi:nil`` are surfaced separately), and ``id`` / ``href`` (projected explicitly as ``@id`` / ``@href``).""" out = {} for raw, val in _attr_items(el): if raw == "xmlns" or raw.startswith("xmlns:"): continue if raw in ("id", "href"): continue if raw.startswith("{"): uri = raw[1:raw.index("}")] if uri == _XSI_NS: continue elif raw.startswith("xsi:"): continue out[fpml_local_name(raw)] = val return out def fpml_is_xsi_nil(el) -> bool: for raw, val in _attr_items(el): is_nil_attr = raw == "xsi:nil" or ( raw.startswith("{") and "}" in raw and fpml_local_name(raw) == "nil" and raw[1:raw.index("}")] == _XSI_NS ) if is_nil_attr: return str(val).strip() == "true" return False def fpml_own_text(el) -> str: """A leaf element's own trimmed text, or ``""``. Only ever called on a childless element, so ``el.text`` IS its whole text.""" return (el.text or "").strip() def _wrap_ref(id_: str) -> str: return LINK_OPEN + id_ + LINK_CLOSE def fpml_assert_supported(el) -> None: name = fpml_element_local_name(el) if name not in FPML_SUPPORTED: raise FpmlUnsupportedError( "fpml converter (read-only IR-swap subset): element <" + name + "> " "is outside the supported subset. This is by design — the subset " "is widened only as real FpML example files are added to the test " "corpus." ) def fpml_project_element(el): fpml_assert_supported(el) id_ = el.get("id") href = el.get("href") xsi_type = fpml_xsi_type(el) is_nil = fpml_is_xsi_nil(el) plain = fpml_plain_attributes(el) kids = list(el) has_plain = len(plain) > 0 # Pure reference: href, nothing else of substance. if href is not None and not kids and not has_plain and xsi_type is None and fpml_own_text(el) == "": return _wrap_ref(href) # Childless: a scalar (or a small dict when it has attributes to keep). if not kids: text = fpml_own_text(el) bare = None if is_nil else (None if text == "" else text) if id_ is None and href is None and xsi_type is None and not has_plain: return bare leaf = {} if id_ is not None: leaf["@id"] = str(id_) if xsi_type is not None: leaf["@xsi:type"] = str(xsi_type) if href is not None: leaf["@href"] = _wrap_ref(href) for k, v in plain.items(): leaf["@" + k] = v if is_nil: leaf["@xsi:nil"] = "true" if text != "": leaf["#text"] = text return leaf # Container: attributes first (fixed order), then grouped children. out = {} if id_ is not None: out["@id"] = str(id_) if xsi_type is not None: out["@xsi:type"] = str(xsi_type) if href is not None: out["@href"] = _wrap_ref(href) for k, v in plain.items(): out["@" + k] = v if is_nil: out["@xsi:nil"] = "true" order = [] groups = {} for kid in kids: key = fpml_element_local_name(kid) if key not in groups: groups[key] = [] order.append(key) groups[key].append(kid) for key in order: g = groups[key] out[key] = fpml_project_element(g[0]) if len(g) == 1 else [fpml_project_element(x) for x in g] return out def _has_descendant(root, local_name: str) -> bool: return any(fpml_element_local_name(e) == local_name for e in root.iter()) def fpml_assert_is_ir_swap_data_document(root) -> None: """Root gate — the throw that tells interpreter selection FpML apart from every other candidate. Rejects anything that isn't an FpML interest-rate-swap ````.""" if root is None or fpml_element_local_name(root) != "dataDocument": got = fpml_element_local_name(root) if root is not None else "?" raise ValueError( "fpml converter: expected a root element, got <" + got + "> " "— this doesn't look like an FpML dataDocument." ) ver = root.get("fpmlVersion") if ver is not None and not str(ver).startswith("5-"): raise ValueError( 'fpml converter: fpmlVersion="' + str(ver) + '" is not an FpML 5.x document.' ) if not _has_descendant(root, "swap"): raise ValueError( "fpml converter: this contains no — only the " "interest-rate-swap product family is supported in this read-only build." ) def fpml_to_json(xml_text: str): """FpML interest-rate-swap XML -> the nested value Structile renders. Raises :class:`ValueError` if the document isn't an FpML IR-swap ``dataDocument``, or :class:`FpmlUnsupportedError` for an element outside the supported subset. """ root = ET.fromstring(xml_text) fpml_assert_is_ir_swap_data_document(root) return fpml_project_element(root) # --- serialize (python/fpml_convert/_serialize.py) ------------------------- """Value -> FpML XML — the inverse of :func:`fpml_convert._project.fpml_to_json`, a faithful port of the JS bundle's ``src/70-serialize.js`` (which uses the shared ``FPML_CHILD_ORDER`` from ``src/rules.json``). Round-trips at the VALUE level: ``fpml_to_json(json_to_fpml(fpml_to_json(x))) == fpml_to_json(x)``. Policies, matching the JS side exactly: * typed lexical forms are written back verbatim (no coercion — ``1000000`` vs ``1000000.00`` survives); * the confirmation-view default namespace + ``xsi:`` are re-declared on the root; ``xsi:schemaLocation`` is not reproduced; * children are re-emitted in ``FPML_CHILD_ORDER`` order, so a value whose keys were reordered still serialises as schema-ordered FpML. """ _FPML_CONFIRMATION_NS = "http://www.fpml.org/FpML-5/confirmation" _FPML_XSI_NS = "http://www.w3.org/2001/XMLSchema-instance" def _esc_attr(s) -> str: return (str(s).replace("&", "&").replace("<", "<") .replace(">", ">").replace('"', """)) def _esc_text(s) -> str: return str(s).replace("&", "&").replace("<", "<").replace(">", ">") def _is_ref(v) -> bool: return isinstance(v, str) and v.startswith(LINK_OPEN) and v.endswith(LINK_CLOSE) def _unwrap_ref(v: str) -> str: return v[len(LINK_OPEN):len(v) - len(LINK_CLOSE)] def _order_child_pairs(parent_local_name, pairs): order = FPML_CHILD_ORDER.get(parent_local_name) if not order: return pairs rank = {name: i for i, name in enumerate(order)} decorated = [ (rank[p[0]] if p[0] in rank else len(order) + i, i, p) for i, p in enumerate(pairs) ] decorated.sort(key=lambda t: (t[0], t[1])) return [t[2] for t in decorated] def _serialize_value(key, val, indent, out): if isinstance(val, list): for item in val: _serialize_value(key, item, indent, out) return if val is None: out.append(indent + "<" + key + "/>") return if not isinstance(val, dict): if _is_ref(val): out.append(indent + "<" + key + ' href="' + _esc_attr(_unwrap_ref(val)) + '"/>') else: out.append(indent + "<" + key + ">" + _esc_text(val) + "") return attrs = [] nil = False text = None child_pairs = [] for k in val: if k == "@xsi:nil": nil = str(val[k]) == "true" elif k == "@id": attrs.append(("id", val[k])) elif k == "@xsi:type": attrs.append(("xsi:type", val[k])) elif k == "@href": attrs.append(("href", _unwrap_ref(val[k]) if _is_ref(val[k]) else val[k])) elif k.startswith("@"): attrs.append((k[1:], val[k])) elif k == "#text": text = val[k] else: child_pairs.append((k, val[k])) attr_str = "".join(' %s="%s"' % (name, _esc_attr(v)) for name, v in attrs) if nil: attr_str += ' xsi:nil="true"' open_tag = "<" + key + attr_str if nil or (not child_pairs and text is None): out.append(indent + open_tag + "/>") return if not child_pairs: out.append(indent + open_tag + ">" + _esc_text(text) + "") return out.append(indent + open_tag + ">") ci = indent + " " if text is not None: out.append(ci + _esc_text(text)) # mixed content (not seen in the subset) for cp in _order_child_pairs(key, child_pairs): _serialize_value(cp[0], cp[1], ci, out) out.append(indent + "") def json_to_fpml(value) -> str: """The projected ``dataDocument`` value -> FpML text.""" if not isinstance(value, dict): raise ValueError("fpml json_to_fpml: expected the projected object") ver = value["@fpmlVersion"] if isinstance(value.get("@fpmlVersion"), str) else "5-10" out = [ '', '' % (_FPML_CONFIRMATION_NS, _FPML_XSI_NS, _esc_attr(ver)), ] child_pairs = [(k, value[k]) for k in value if not k.startswith("@")] for cp in _order_child_pairs("dataDocument", child_pairs): _serialize_value(cp[0], cp[1], " ", out) out.append("") return "\n".join(out) + "\n" __all__ = ["fpml_to_json", "json_to_fpml", "FpmlUnsupportedError", "FPML_VERSION"] def _main(argv=None) -> int: p = argparse.ArgumentParser( prog="fpml_convert_singlefile", description="FpML interest-rate-swap XML <-> JSON (structile projection).", ) p.add_argument("input", help="an FpML .xml or a projected .json, or - for stdin") p.add_argument("-o", "--output", help="write here instead of stdout") p.add_argument("--to", choices=("xml", "json"), help="force the output direction") p.add_argument("--indent", type=int, default=2, help="JSON indent (default 2)") p.add_argument("--version", action="version", version=f"fpml_convert_singlefile (FpML {FPML_VERSION})") args = p.parse_args(argv) text = sys.stdin.read() if args.input == "-" else open(args.input, encoding="utf-8").read() to = args.to or ("json" if text.lstrip().startswith("<") else "xml") if to == "json": out_text = json.dumps(fpml_to_json(text), indent=args.indent, ensure_ascii=False) + "\n" else: out_text = json_to_fpml(json.loads(text)) if args.output: with open(args.output, "w", encoding="utf-8") as fh: fh.write(out_text) else: sys.stdout.write(out_text) return 0 if __name__ == "__main__": sys.exit(_main())