Add final balances parser

xrpl/utils/__init__.py

@@ -9,7 +9,7 @@
     ripple_time_to_datetime,
     ripple_time_to_posix,
 )
-from xrpl.utils.txn_parser import get_balance_changes
+from xrpl.utils.txn_parser import get_balance_changes, get_final_balances
 from xrpl.utils.xrp_conversions import XRPRangeException, drops_to_xrp, xrp_to_drops
 
 __all__ = [
@@ -25,4 +25,5 @@
     "XRPLTimeRangeException",
     "create_cross_chain_payment",
     "get_balance_changes",
+    "get_final_balances",
 ]

xrpl/utils/txn_parser/__init__.py

@@ -1,5 +1,6 @@
 """Functions to parse a transaction."""
 
 from xrpl.utils.txn_parser.get_balance_changes import get_balance_changes
+from xrpl.utils.txn_parser.get_final_balances import get_final_balances
 
-__all__ = ["get_balance_changes"]
+__all__ = ["get_balance_changes", "get_final_balances"]

xrpl/utils/txn_parser/get_balance_changes.py

@@ -5,16 +5,16 @@
 
 from xrpl.models import TransactionMetadata
 from xrpl.utils.txn_parser.utils import (
-    BalanceChanges,
-    get_node_balance_changes,
+    ComputedBalances,
+    NormalizedNode,
+    get_node_balance,
     get_value,
     group_by_account,
     normalize_nodes,
 )
-from xrpl.utils.txn_parser.utils.nodes import NormalizedNode
 
 
-def get_balance_changes(metadata: TransactionMetadata) -> List[BalanceChanges]:
+def get_balance_changes(metadata: TransactionMetadata) -> List[ComputedBalances]:
     """
     Parse all balance changes from a transaction's metadata.
 
@@ -28,7 +28,7 @@ def get_balance_changes(metadata: TransactionMetadata) -> List[BalanceChanges]:
     quantities = [
         quantity
         for node in normalize_nodes(metadata)
-        for quantity in get_node_balance_changes(node, _compute_balance_change(node))
+        for quantity in get_node_balance(node, _compute_balance_change(node))
     ]
     return group_by_account(quantities)
 

xrpl/utils/txn_parser/get_final_balances.py

@@ -0,0 +1,59 @@
+"""Parse final balances of every account involved in the given transaction."""
+
+from decimal import Decimal
+from typing import List, Optional
+
+from xrpl.models import TransactionMetadata
+from xrpl.utils.txn_parser.utils import (
+    ComputedBalances,
+    NormalizedNode,
+    get_node_balance,
+    get_value,
+    group_by_account,
+    normalize_nodes,
+)
+
+
+def get_final_balances(metadata: TransactionMetadata) -> List[ComputedBalances]:
+    """
+    Parse all final balances from a transaction's metadata.
+
+    Args:
+        metadata: Transactions metadata.
+
+    Returns:
+        All final balances caused by a transaction.
+        The final balances are grouped by the affected account addresses.
+    """
+    quantities = [
+        quantity
+        for node in normalize_nodes(metadata)
+        for quantity in get_node_balance(node, _compute_final_balance(node))
+    ]
+    return group_by_account(quantities)
+
+
+def _compute_final_balance(node: NormalizedNode) -> Optional[Decimal]:
+    """
+    Get the final balance from a node.
+
+    Args:
+        node: The affected node.
+
+    Returns:
+        The final balance.
+    """
+    value: Optional[Decimal] = None
+    new_fields = node.get("NewFields")
+    final_fields = node.get("FinalFields")
+    if new_fields is not None:
+        balance = new_fields.get("Balance")
+        if balance is not None:
+            value = get_value(balance)
+    elif final_fields is not None:
+        balance = final_fields.get("Balance")
+        if balance is not None:
+            value = get_value(balance)
+    if value is None or value == Decimal(0):
+        return None
+    return value

xrpl/utils/txn_parser/utils/__init__.py

@@ -1,17 +1,18 @@
 """Utility functions for the transaction parser."""
 
 from xrpl.utils.txn_parser.utils.balance_parser import (
-    get_node_balance_changes,
+    get_node_balance,
     get_value,
     group_by_account,
 )
-from xrpl.utils.txn_parser.utils.nodes import normalize_nodes
-from xrpl.utils.txn_parser.utils.types import BalanceChanges
+from xrpl.utils.txn_parser.utils.nodes import NormalizedNode, normalize_nodes
+from xrpl.utils.txn_parser.utils.types import ComputedBalances
 
 __all__ = [
-    "get_node_balance_changes",
+    "get_node_balance",
     "get_value",
     "group_by_account",
+    "NormalizedNode",
     "normalize_nodes",
-    "BalanceChanges",
+    "ComputedBalances",
 ]

xrpl/utils/txn_parser/utils/balance_parser.py

@@ -4,14 +4,14 @@
 from typing import Dict, List, Optional, Union
 
 from xrpl.utils.txn_parser.utils.nodes import NormalizedNode
-from xrpl.utils.txn_parser.utils.types import Balance, BalanceChange, BalanceChanges
+from xrpl.utils.txn_parser.utils.types import Balance, ComputedBalance, ComputedBalances
 from xrpl.utils.xrp_conversions import drops_to_xrp
 
 
 def _get_xrp_quantity(
     node: NormalizedNode,
     value: Optional[Decimal],
-) -> Optional[BalanceChange]:
+) -> Optional[ComputedBalance]:
     if value is None:
         return None
     absolute_value = value.copy_abs()
@@ -24,7 +24,7 @@ def _get_xrp_quantity(
     if final_fields is not None:
         account = final_fields.get("Account")
         if account is not None:
-            return BalanceChange(
+            return ComputedBalance(
                 account=account,
                 balance=Balance(
                     currency="XRP",
@@ -35,7 +35,7 @@ def _get_xrp_quantity(
     if new_fields is not None:
         account = new_fields.get("Account")
         if account is not None:
-            return BalanceChange(
+            return ComputedBalance(
                 account=account,
                 balance=Balance(
                     currency="XRP",
@@ -45,15 +45,15 @@ def _get_xrp_quantity(
     return None
 
 
-def _flip_trustline_perspective(balance_change: BalanceChange) -> BalanceChange:
-    balance = balance_change["balance"]
+def _flip_trustline_perspective(computed_balance: ComputedBalance) -> ComputedBalance:
+    balance = computed_balance["balance"]
     negated_value = Decimal(balance["value"]).copy_negate()
     issuer = balance["issuer"]
-    return BalanceChange(
+    return ComputedBalance(
         account=issuer,
         balance=Balance(
             currency=balance["currency"],
-            issuer=balance_change["account"],
+            issuer=computed_balance["account"],
             value=f"{negated_value.normalize():f}",
         ),
     )
@@ -62,17 +62,16 @@ def _flip_trustline_perspective(balance_change: BalanceChange) -> BalanceChange:
 def _get_trustline_quantity(
     node: NormalizedNode,
     value: Optional[Decimal],
-) -> List[BalanceChange]:
+) -> List[ComputedBalance]:
     """
-    Computes the complete list of every balance that changed in the ledger
-    as a result of the given transaction.
+    Computes the complete list of every balance affected by the transaction.
 
     Args:
         node: The affected node.
         value: The currency amount value.
 
     Returns:
-        A list of balance changes.
+        A list of computed balances.
     """
     if value is None:
         return []
@@ -93,7 +92,7 @@ def _get_trustline_quantity(
             and balance_currency is not None
             and high_limit_issuer is not None
         ):
-            result = BalanceChange(
+            result = ComputedBalance(
                 account=low_limit_issuer,
                 balance=Balance(
                     currency=balance_currency,
@@ -105,16 +104,16 @@ def _get_trustline_quantity(
     return []
 
 
-def _group_balance_changes(
-    balance_changes: List[BalanceChange],
-) -> Dict[str, List[BalanceChange]]:
-    grouped_balance_changes: Dict[str, List[BalanceChange]] = {}
-    for change in balance_changes:
-        account = change["account"]
-        if account not in grouped_balance_changes:
-            grouped_balance_changes[account] = []
-        grouped_balance_changes[account].append(change)
-    return grouped_balance_changes
+def _group_balance(
+    computed_balances: List[ComputedBalance],
+) -> Dict[str, List[ComputedBalance]]:
+    grouped_balances: Dict[str, List[ComputedBalance]] = {}
+    for balance in computed_balances:
+        account = balance["account"]
+        if account not in grouped_balances:
+            grouped_balances[account] = []
+        grouped_balances[account].append(balance)
+    return grouped_balances
 
 
 def get_value(balance: Union[Dict[str, str], str]) -> Decimal:
@@ -132,19 +131,19 @@ def get_value(balance: Union[Dict[str, str], str]) -> Decimal:
     return Decimal(balance["value"])
 
 
-def get_node_balance_changes(
+def get_node_balance(
     node: NormalizedNode,
     value: Optional[Decimal],
-) -> List[BalanceChange]:
+) -> List[ComputedBalance]:
     """
-    Retrieve the balance changes from a node.
+    Retrieve the balance from a node.
 
     Args:
         node: The affected node.
         value: The currency amount's value
 
     Returns:
-        A list of balance changes.
+        A list of balances.
     """
     if node["LedgerEntryType"] == "AccountRoot":
         xrp_quantity = _get_xrp_quantity(node, value)
@@ -158,25 +157,25 @@ def get_node_balance_changes(
 
 
 def group_by_account(
-    balance_changes: List[BalanceChange],
-) -> List[BalanceChanges]:
+    computed_balance: List[ComputedBalance],
+) -> List[ComputedBalances]:
     """
-    Groups the balance changes in one list for each account.
+    Groups the computed balances in one list for each account.
 
     Args:
-        balance_changes: All balance changes cause by a transaction.
+        computed_balance: All computed balances cause by a transaction.
 
     Returns:
-        The grouped balance changes.
+        The grouped computed balances.
     """
-    grouped = _group_balance_changes(balance_changes)
+    grouped = _group_balance(computed_balance)
     result = []
     for account, account_balances in grouped.items():
         balances: List[Balance] = []
         for balance in account_balances:
             balances.append(balance["balance"])
         result.append(
-            BalanceChanges(
+            ComputedBalances(
                 account=account,
                 balances=balances,
             )

xrpl/utils/txn_parser/utils/types.py

@@ -25,17 +25,17 @@ class Balance(OptionalIssuer):
     """The amount of the currency."""
 
 
-class BalanceChange(TypedDict):
-    """A single balance change."""
+class ComputedBalance(TypedDict):
+    """A single computed balance."""
 
     account: str
     """The affected account."""
     balance: Balance
-    """The balance change."""
+    """The balance."""
 
 
-class BalanceChanges(TypedDict):
-    """A model representing an account's balance changes."""
+class ComputedBalances(TypedDict):
+    """A model representing an account's computed balances."""
 
     account: str
     balances: List[Balance]