dao_treasury.sorting package

Subpackages

Submodules

dao_treasury.sorting.factory module

final class dao_treasury.sorting.factory.SortRuleFactory[source]

Bases: Generic[TRule]

Builder for creating sort rule instances for a specific transaction group and network(s).

This factory supports two patterns:

  1. Decorating a function to register a dynamic matching rule.

  2. Calling match() to supply static match attributes.

Use the convenience functions like revenue(), expense(), etc., to obtain an instance of this factory preconfigured with the appropriate rule type.

Examples

>>> from dao_treasury.sorting.factory import revenue
>>> @revenue("Sales", networks=[1, 3])
... def match_large_sales(tx):
...     return tx.value_usd > 1000
__init__(txgroup, networks, rule_type)[source]

Initialize the sort rule factory.

Parameters:

  • txgroup (str) – Base name of the transaction group.

  • networks (Network | Iterable[Network]) – Single network ID or iterable of network IDs where the rule applies.

  • rule_type (TRule) – Sort rule class (e.g., RevenueSortRule) to instantiate.

Return type:

None

match(func=None, **match_values)[source]

Define static matching attributes for the sort rule.

Call this method with keyword matchers corresponding to rule attributes (e.g., hash, from_address, symbol) to create a rule matching based on these values.

Parameters:

  • func (None) – Must be None; a function match must use the decorator form.

  • **match_values (Any) – Attribute values for matching (e.g., hash=”0x123”, symbol=”DAI”).

Raises:
Return type:

None

See also

__call__()

Examples

>>> ignore("Dust").match(symbol="WETH", from_address="0xAAA")
property rule: TRule | None

Return the created sort rule instance, if any.

After decoration or a call to match(), this property holds the concrete SortRule instance.

Examples

>>> @other_income("Interest")
... def match_i(tx):
...     return tx.value_usd > 100
dao_treasury.sorting.factory.cost_of_revenue(txgroup, networks=1)[source]

Create a factory to register cost‐of‐revenue sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as cost of revenue.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[CostOfRevenueSortRule]

See also

revenue() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import cost_of_revenue
>>> @cost_of_revenue("Manufacturing")
... def match_manufacturing(tx):
...     return tx.from_address is not None and tx.amount_usd > 1000
dao_treasury.sorting.factory.expense(txgroup, networks=1)[source]

Create a factory to register expense sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as expense.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[ExpenseSortRule]

See also

other_expense() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import expense
>>> @expense("Office Supplies")
... def match_supplies(tx):
...     return tx.symbol == "USD" and tx.amount < 500
dao_treasury.sorting.factory.ignore(txgroup, networks=1)[source]

Create a factory to register ignore sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as ignored.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[IgnoreSortRule]

See also

SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import ignore
>>> @ignore("Dust")
... def match_dust(tx):
...     return abs(tx.value_usd) < 0.01
dao_treasury.sorting.factory.other_expense(txgroup, networks=1)[source]

Create a factory to register other‐expense sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as other expense.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[OtherExpenseSortRule]

See also

expense() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import other_expense
>>> @other_expense("Misc Fees")
... def match_misc(tx):
...     return tx.amount_usd < 0 and tx.symbol == "ETH"
dao_treasury.sorting.factory.other_income(txgroup, networks=1)[source]

Create a factory to register other‐income sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as other income.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[OtherIncomeSortRule]

See also

revenue() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import other_income
>>> @other_income("Interest")
... def match_interest(tx):
...     return tx.token_address == SOME_TOKEN and tx.amount > 0
dao_treasury.sorting.factory.revenue(txgroup, networks=1)[source]

Create a factory to register revenue sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as revenue.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[RevenueSortRule]

See also

cost_of_revenue() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import revenue
>>> @revenue("Token Sales")
... def match_sales(tx):
...     return tx.amount > 0 and tx.to_address is not None

dao_treasury.sorting.rule module

Module defining transaction sorting rules for the DAO treasury.

This module provides the _SortRule base class and subclasses for categorizing TreasuryTx entries based on their attributes or a custom function. When a rule is instantiated, it registers itself in the global SORT_RULES mapping under its class and configures which transaction attributes to match via _match_all.

Examples

# Define a revenue rule for sales (assuming you only transact in DAI for sales) >>> from dao_treasury.sorting.rule import RevenueSortRule, SORT_RULES >>> RevenueSortRule( … txgroup=’Sale’, … token_address=’0x6B175474E89094d879c81e570a000000000000’, … symbol=’DAI’ … ) # Inspect rules registered for RevenueSortRule >>> len(SORT_RULES[RevenueSortRule]) 1

# Iterate over all ExpenseSortRule instances >>> from dao_treasury.sorting.rule import ExpenseSortRule >>> for rule in SORT_RULES[ExpenseSortRule]: … print(rule.txgroup)

See also

SORT_RULES _SortRule

class dao_treasury.sorting.rule.CostOfRevenueSortRule[source]

Bases: _OutboundSortRule

Rule to categorize outbound transactions as cost of revenue.

Prepends ‘Cost of Revenue:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.rule.ExpenseSortRule[source]

Bases: _OutboundSortRule

Rule to categorize outbound transactions as expenses.

Prepends ‘Expenses:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.rule.IgnoreSortRule[source]

Bases: _SortRule

Rule to ignore certain transactions.

Prepends ‘Ignore:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.rule.OtherExpenseSortRule[source]

Bases: _OutboundSortRule

Rule to categorize outbound transactions as other expenses.

Prepends ‘Other Expenses:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.rule.OtherIncomeSortRule[source]

Bases: _InboundSortRule

Rule to categorize inbound transactions as other income.

Prepends ‘Other Income:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.rule.RevenueSortRule[source]

Bases: _InboundSortRule

Rule to categorize inbound transactions as revenue.

Prepends ‘Revenue:’ to the txgroup name before registration.

Examples

>>> RevenueSortRule(txgroup='Sale', to_address='0xabc...', symbol='DAI')
# results in a rule with txgroup 'Revenue:Sale'
__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

dao_treasury.sorting.rule.SORT_RULES: DefaultDict[Type[RevenueSortRule | CostOfRevenueSortRule | ExpenseSortRule | OtherIncomeSortRule | OtherExpenseSortRule | IgnoreSortRule], List[RevenueSortRule | CostOfRevenueSortRule | ExpenseSortRule | OtherIncomeSortRule | OtherExpenseSortRule | IgnoreSortRule]] = {<class 'dao_treasury.sorting.rule.IgnoreSortRule'>: [IgnoreSortRule(txgroup='Ignore:LlamaPay', hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=<function is_llamapay_stream_replenishment>)]}

Mapping from sort rule classes to lists of instantiated rules, in creation order per class.

Each key is a subclass of SortRule and the corresponding value is the list of rule instances of that class.

Examples

>>> from dao_treasury.sorting.rule import RevenueSortRule, SORT_RULES
>>> RevenueSortRule(txgroup='Interest', symbol='DAI')
>>> SORT_RULES[RevenueSortRule][0].txgroup
'Revenue:Interest'

Module contents

This module provides the core logic for sorting DAO Treasury transactions into transaction groups (categories).

Sorting enables comprehensive financial reporting and categorization tailored for on-chain organizations. Transactions are matched against either statically defined rules or more advanced dynamic rules based on user-defined matching functions.

Sorting works by attempting matches in this order:

  1. Check if the transaction is an internal transfer (within treasury wallets).

  2. Check if the transaction is “Out of Range” (neither sender nor receiver was a treasury wallet at the time of the tx).

  3. Match by transaction hash using registered HashMatchers.

  4. Match by sender address using registered FromAddressMatchers.

  5. Match by recipient address using registered ToAddressMatchers.

  6. Assign “Must Sort Inbound” or “Must Sort Outbound” groups if part of treasury.

  7. Raise an error if no match is found (unexpected case).

See the complete [sort rules documentation](https://bobthebuidler.github.io/dao-treasury/sort_rules.html) for detailed explanations and examples on defining and registering sort rules.

See also

dao_treasury.sorting.sort_basic() dao_treasury.sorting.sort_basic_entity() dao_treasury.sorting.sort_advanced() dao_treasury.sorting.HashMatcher dao_treasury.sorting.FromAddressMatcher dao_treasury.sorting.ToAddressMatcher

class dao_treasury.sorting.CostOfRevenueSortRule[source]

Bases: _OutboundSortRule

Rule to categorize outbound transactions as cost of revenue.

Prepends ‘Cost of Revenue:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.ExpenseSortRule[source]

Bases: _OutboundSortRule

Rule to categorize outbound transactions as expenses.

Prepends ‘Expenses:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

final class dao_treasury.sorting.FromAddressMatcher[source]

Bases: _AddressMatcher

Final matcher that categorizes by transaction from_address.

Examples

>>> from dao_treasury.sorting._matchers import FromAddressMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> address = "0xAbC1230000000000000000000000000000000000"
>>> fam = FromAddressMatcher(TxGroupDbid(7), [address])
>>> FromAddressMatcher.match(address)
TxGroupDbid(7)
__init__(txgroup, addresses)

Create an address matcher with checksum validation.

Converts inputs to checksummed addresses and ensures that each address is only registered once. Duplicate addresses in the input iterable will log a warning, but only the first occurrence is used.

Parameters:
Raises:

ValueError – If addresses is empty, or if any address already has an existing matcher.

Return type:

None

Examples

>>> from dao_treasury.sorting._matchers import _AddressMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> addr = "0xAbC1230000000000000000000000000000000000"
>>> # duplicate in list triggers warning but does not raise
>>> am = _AddressMatcher(TxGroupDbid(5), [addr, addr])
>>> addr in am
True

See also

_validate_hexstr()

classmethod match(string)

Return the TxGroupDbid for a matching instance or None if no match.

The lookup first checks the internal cache, then iterates through all instances and tests membership with __contains__. On first hit, the result is cached for future calls.

Parameters:

string (str) – Input string to match (e.g., address or hash).

Return type:

TxGroupDbid | None

Examples

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hash_str = "0xdeadbeef" + "00"*28
>>> hmatch = HashMatcher(TxGroupDbid(2), [hash_str])
>>> HashMatcher.match(hash_str)
TxGroupDbid(2)
>>> HashMatcher.match("0xother")
None

See also

__cache__

expected_length: ClassVar[int] = 42
property values: Set[HexStr]

Set of all validated strings used for matching.

Returns:

The original set of strings passed at initialization.

Example

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hex_str = "0x" + "f"*64
>>> matcher = HashMatcher(TxGroupDbid(4), [hex_str])
>>> matcher.values
{'0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff'}

See also

match()

final class dao_treasury.sorting.HashMatcher[source]

Bases: _HexStringMatcher

Final matcher that categorizes by transaction hash.

Matches full 66-character hex transaction hashes.

Examples

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hash_str = '0x' + 'f' * 64
>>> hm = HashMatcher(TxGroupDbid(9), [hash_str])
>>> HashMatcher.match(hash_str)
TxGroupDbid(9)
>>> repr(hm)
"HashMatcher(txgroup='Root:Group', hashes=['0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff'])"
__init__(txgroup, hashes)[source]

Initialize hash matcher ensuring unique transaction hashes.

Validates and normalizes hashes to fixed length, and ensures that each hash is only registered once. Duplicate hashes in the input iterable will log a warning, but only the first occurrence is used.

Parameters:

  • txgroup (TxGroupDbid) – Identifier of the transaction group.

  • hashes (Iterable[HexStr]) – Iterable of hex string hashes.

Raises:

ValueError – If hashes is empty, or if any hash already has an existing matcher.

Return type:

None

Examples

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hash_str = '0x' + 'f' * 64
>>> # duplicate in list logs warning but does not raise
>>> hm = HashMatcher(TxGroupDbid(9), [hash_str, hash_str])
>>> HashMatcher.match(hash_str)
TxGroupDbid(9)

See also

_validate_hexstr()

classmethod match(string)

Return the TxGroupDbid for a matching instance or None if no match.

The lookup first checks the internal cache, then iterates through all instances and tests membership with __contains__. On first hit, the result is cached for future calls.

Parameters:

string (str) – Input string to match (e.g., address or hash).

Return type:

TxGroupDbid | None

Examples

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hash_str = "0xdeadbeef" + "00"*28
>>> hmatch = HashMatcher(TxGroupDbid(2), [hash_str])
>>> HashMatcher.match(hash_str)
TxGroupDbid(2)
>>> HashMatcher.match("0xother")
None

See also

__cache__

expected_length: ClassVar[int] = 66
property values: Set[HexStr]

Set of all validated strings used for matching.

Returns:

The original set of strings passed at initialization.

Example

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hex_str = "0x" + "f"*64
>>> matcher = HashMatcher(TxGroupDbid(4), [hex_str])
>>> matcher.values
{'0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff'}

See also

match()

class dao_treasury.sorting.IgnoreSortRule[source]

Bases: _SortRule

Rule to ignore certain transactions.

Prepends ‘Ignore:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.OtherExpenseSortRule[source]

Bases: _OutboundSortRule

Rule to categorize outbound transactions as other expenses.

Prepends ‘Other Expenses:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.OtherIncomeSortRule[source]

Bases: _InboundSortRule

Rule to categorize inbound transactions as other income.

Prepends ‘Other Income:’ to the txgroup name before registration.

__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

class dao_treasury.sorting.RevenueSortRule[source]

Bases: _InboundSortRule

Rule to categorize inbound transactions as revenue.

Prepends ‘Revenue:’ to the txgroup name before registration.

Examples

>>> RevenueSortRule(txgroup='Sale', to_address='0xabc...', symbol='DAI')
# results in a rule with txgroup 'Revenue:Sale'
__init__(*, txgroup, hash=None, from_address=None, from_nickname=None, to_address=None, to_nickname=None, token_address=None, symbol=None, log_index=None, func=None)
Parameters:
Return type:

None

async match(tx)

Determine if the given transaction matches this rule.

Parameters:

tx (TreasuryTx) – A TreasuryTx entity to test against this rule.

Returns:

True if the transaction matches the rule criteria; otherwise False.

Return type:

bool

Examples

# match by symbol and recipient >>> rule = _SortRule(txgroup=’Foo’, symbol=’DAI’, to_address=’0xabc…’) >>> await rule.match(tx) # where tx.symbol == ‘DAI’ and tx.to_address == ‘0xabc…’ True

See also

_match_all

from_address: EthAddress | None = None

Source wallet address to match.

from_nickname: str | None = None

Sender nickname (alias) to match.

func: SortFunction | None = None

Custom matching function that takes a TreasuryTx and returns a bool or an awaitable that returns a bool.

hash: HexStr | None = None

Exact transaction hash to match.

log_index: int | None = None

Log index within the transaction receipt to match.

symbol: str | None = None

Token symbol to match.

to_address: EthAddress | None = None

Recipient wallet address to match.

to_nickname: str | None = None

Recipient nickname (alias) to match.

token_address: EthAddress | None = None

Token contract address to match.

txgroup: TxGroupName

Name of the transaction group to assign upon match.

property txgroup_dbid: TxGroupDbid

Compute the database ID for this rule’s txgroup.

Splits the txgroup string on ‘:’ and resolves or creates the hierarchical TxGroup entries in the database, returning the final group ID.

See also

TxGroup.

final class dao_treasury.sorting.SortRuleFactory[source]

Bases: Generic[TRule]

Builder for creating sort rule instances for a specific transaction group and network(s).

This factory supports two patterns:

  1. Decorating a function to register a dynamic matching rule.

  2. Calling match() to supply static match attributes.

Use the convenience functions like revenue(), expense(), etc., to obtain an instance of this factory preconfigured with the appropriate rule type.

Examples

>>> from dao_treasury.sorting.factory import revenue
>>> @revenue("Sales", networks=[1, 3])
... def match_large_sales(tx):
...     return tx.value_usd > 1000
__init__(txgroup, networks, rule_type)[source]

Initialize the sort rule factory.

Parameters:

  • txgroup (str) – Base name of the transaction group.

  • networks (Network | Iterable[Network]) – Single network ID or iterable of network IDs where the rule applies.

  • rule_type (TRule) – Sort rule class (e.g., RevenueSortRule) to instantiate.

Return type:

None

match(func=None, **match_values)[source]

Define static matching attributes for the sort rule.

Call this method with keyword matchers corresponding to rule attributes (e.g., hash, from_address, symbol) to create a rule matching based on these values.

Parameters:

  • func (None) – Must be None; a function match must use the decorator form.

  • **match_values (Any) – Attribute values for matching (e.g., hash=”0x123”, symbol=”DAI”).

Raises:
Return type:

None

See also

__call__()

Examples

>>> ignore("Dust").match(symbol="WETH", from_address="0xAAA")
property rule: TRule | None

Return the created sort rule instance, if any.

After decoration or a call to match(), this property holds the concrete SortRule instance.

Examples

>>> @other_income("Interest")
... def match_i(tx):
...     return tx.value_usd > 100
final class dao_treasury.sorting.ToAddressMatcher[source]

Bases: _AddressMatcher

Final matcher that categorizes by transaction to_address.

Examples

>>> from dao_treasury.sorting._matchers import ToAddressMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> address = "0xDef4560000000000000000000000000000000000"
>>> tam = ToAddressMatcher(TxGroupDbid(8), [address])
>>> ToAddressMatcher.match(address)
TxGroupDbid(8)
__init__(txgroup, addresses)

Create an address matcher with checksum validation.

Converts inputs to checksummed addresses and ensures that each address is only registered once. Duplicate addresses in the input iterable will log a warning, but only the first occurrence is used.

Parameters:
Raises:

ValueError – If addresses is empty, or if any address already has an existing matcher.

Return type:

None

Examples

>>> from dao_treasury.sorting._matchers import _AddressMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> addr = "0xAbC1230000000000000000000000000000000000"
>>> # duplicate in list triggers warning but does not raise
>>> am = _AddressMatcher(TxGroupDbid(5), [addr, addr])
>>> addr in am
True

See also

_validate_hexstr()

classmethod match(string)

Return the TxGroupDbid for a matching instance or None if no match.

The lookup first checks the internal cache, then iterates through all instances and tests membership with __contains__. On first hit, the result is cached for future calls.

Parameters:

string (str) – Input string to match (e.g., address or hash).

Return type:

TxGroupDbid | None

Examples

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hash_str = "0xdeadbeef" + "00"*28
>>> hmatch = HashMatcher(TxGroupDbid(2), [hash_str])
>>> HashMatcher.match(hash_str)
TxGroupDbid(2)
>>> HashMatcher.match("0xother")
None

See also

__cache__

expected_length: ClassVar[int] = 42
property values: Set[HexStr]

Set of all validated strings used for matching.

Returns:

The original set of strings passed at initialization.

Example

>>> from dao_treasury.sorting._matchers import HashMatcher
>>> from dao_treasury.types import TxGroupDbid
>>> hex_str = "0x" + "f"*64
>>> matcher = HashMatcher(TxGroupDbid(4), [hex_str])
>>> matcher.values
{'0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff'}

See also

match()

dao_treasury.sorting.cost_of_revenue(txgroup, networks=1)[source]

Create a factory to register cost‐of‐revenue sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as cost of revenue.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[CostOfRevenueSortRule]

See also

revenue() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import cost_of_revenue
>>> @cost_of_revenue("Manufacturing")
... def match_manufacturing(tx):
...     return tx.from_address is not None and tx.amount_usd > 1000
dao_treasury.sorting.expense(txgroup, networks=1)[source]

Create a factory to register expense sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as expense.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[ExpenseSortRule]

See also

other_expense() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import expense
>>> @expense("Office Supplies")
... def match_supplies(tx):
...     return tx.symbol == "USD" and tx.amount < 500
dao_treasury.sorting.ignore(txgroup, networks=1)[source]

Create a factory to register ignore sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as ignored.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[IgnoreSortRule]

See also

SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import ignore
>>> @ignore("Dust")
... def match_dust(tx):
...     return abs(tx.value_usd) < 0.01
dao_treasury.sorting.other_expense(txgroup, networks=1)[source]

Create a factory to register other‐expense sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as other expense.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[OtherExpenseSortRule]

See also

expense() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import other_expense
>>> @other_expense("Misc Fees")
... def match_misc(tx):
...     return tx.amount_usd < 0 and tx.symbol == "ETH"
dao_treasury.sorting.other_income(txgroup, networks=1)[source]

Create a factory to register other‐income sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as other income.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[OtherIncomeSortRule]

See also

revenue() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import other_income
>>> @other_income("Interest")
... def match_interest(tx):
...     return tx.token_address == SOME_TOKEN and tx.amount > 0
dao_treasury.sorting.revenue(txgroup, networks=1)[source]

Create a factory to register revenue sort rules for a given transaction group.

Parameters:

  • txgroup (str) – Base name of the transaction group to categorize as revenue.

  • networks (Network | Iterable[Network]) – Network ID or iterable of network IDs on which this rule applies.

Return type:

SortRuleFactory[RevenueSortRule]

See also

cost_of_revenue() SortRuleFactory

Examples

>>> from dao_treasury.sorting.factory import revenue
>>> @revenue("Token Sales")
... def match_sales(tx):
...     return tx.amount > 0 and tx.to_address is not None