dao_treasury package

Subpackages

Submodules

dao_treasury.ENVIRONMENT_VARIABLES module

Environment variable configuration for DAO Treasury.

Defines and loads environment variables (with types and defaults) used for system configuration, such as SQL debugging. Uses typed_envs for convenience and safety.

Key Responsibilities:

  • Define and load environment variables for the system.

  • Provide type-safe access to configuration options.

This is the single source of truth for environment-based settings.

dao_treasury.constants module

Core constants for DAO Treasury.

All constants are marked with Final, ensuring immutability and allowing mypyc to compile them as extremely fast C-level constants for maximum performance. Defines chain IDs, zero address, and key contract addresses (e.g., Disperse.app) used throughout the system for transaction processing, nickname assignment, and analytics.

Key Responsibilities:

  • Provide canonical addresses and chain IDs.

  • Support nickname setup and transaction categorization.

  • Guarantee fast, immutable constants at runtime.

This is the single source of truth for system-wide constants.

dao_treasury.constants.DISPERSE_APP: Final = ('0xD152f549545093347A162Dce210e7293f1452150', '0xd15fE25eD0Dba12fE05e7029C88b10C25e8880E3')

If your treasury sends funds to disperse.app, we create additional txs in the db so each individual send can be accounted for.

dao_treasury.db module

Database models and utilities for DAO treasury reporting.

This module defines Pony ORM entities for:

  • Blockchain networks (Chain)

  • On-chain addresses (Address)

  • ERC-20 tokens and native coin placeholder (Token)

  • Hierarchical transaction grouping (TxGroup)

  • Treasury transaction records (TreasuryTx)

  • Streams and StreamedFunds for streaming payments

It also provides helper functions for inserting ledger entries, resolving integrity conflicts, caching transaction receipts, and creating SQL views for reporting.

final exception dao_treasury.db.BadToken[source]

Bases: ValueError

Raised when a token contract returns invalid metadata.

This exception is thrown if the token name or symbol is empty or cannot be decoded.

Examples

>>> raise BadToken("symbol for 0x0 is ''")
__init__(*args, **kwargs)
__new__(**kwargs)
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

args
final class dao_treasury.db.Address[source]

Bases: Entity

Pony ORM entity representing an on-chain address.

Records both contract and externally owned addresses for tracing funds.

Examples

>>> Address.get_dbid("0x0000000000000000000000000000000000000000")
1

See also

get_or_insert()

__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
before_delete()
before_insert()
before_update()
delete()
find_updated_attributes()
flush()
static get_dbid(address)[source]

Get the DB ID for an address, inserting if necessary.

Parameters:

address (HexAddress) – Hex string of the address (any case, any prefix).

Return type:

int

Examples

>>> Address.get_dbid("0x0000000000000000000000000000000000000000")
1
static get_or_insert(address)[source]

Insert or fetch an Address for address.

If the address has on-chain code, attempts to label it using the verified contract name or fallback label.

Parameters:

address (HexAddress) – Hex address string.

Return type:

Address

Examples

>>> addr = Address.get_or_insert("0x0000000000000000000000000000000000000000")
>>> addr.is_contract
False
get_pk()
load(*attrs)
set(**kwargs)
static set_nickname(address, nickname)[source]
Parameters:
Return type:

None

static set_nicknames(nicknames)[source]
Parameters:

nicknames (Dict[HexAddress, str])

Return type:

None

to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
address

Checksum string of the on-chain address.

address_id

Auto-incremented primary key for the addresses table.

chain

Reference to the chain on which this address resides.

property contract: Contract
is_contract

Flag indicating whether the address is a smart contract.

nickname

Optional human-readable label (e.g., contract name or token name).

streams
streams_from
streams_to
token: 'Token' | None

Optional back-reference to a Token if this address is one.

treasury_tx_from: Set['TreasuryTx']

Inverse relation for transactions sent from this address.

treasury_tx_to: Set['TreasuryTx']

Inverse relation for transactions sent to this address.

final class dao_treasury.db.Chain[source]

Bases: Entity

Pony ORM entity representing a blockchain network.

Stores human-readable network names and numeric chain IDs for reporting.

Examples

>>> Chain.get_dbid(1)  # Ethereum Mainnet
1

See also

get_or_insert()

__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
before_delete()
before_insert()
before_update()
delete()
find_updated_attributes()
flush()
static get_dbid(chainid=1)[source]

Get or create the record for chainid and return its database ID.

Parameters:

chainid (int) – Numeric chain identifier (default uses active RPC via CHAINID).

Return type:

int

Examples

>>> Chain.get_dbid(1)
1
static get_or_insert(chainid)[source]

Insert a new chain record if it does not exist.

Parameters:

chainid (int) – Numeric chain identifier.

Return type:

Chain

Examples

>>> chain = Chain.get_or_insert(1)
>>> chain.chain_name
'Mainnet'
get_pk()
load(*attrs)
set(**kwargs)
to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
addresses

Relationship to address records on this chain.

chain_dbid

Auto-incremented primary key for the chains table.

chain_name

Name of the blockchain network, e.g., ‘Mainnet’, ‘Polygon’.

chainid

Numeric chain ID matching the connected RPC via CHAINID.

tokens

Relationship to token records on this chain.

treasury_txs

Relationship to treasury transactions on this chain.

class dao_treasury.db.Stream[source]

Bases: Entity

__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
async amount_withdrawable(block)[source]
Parameters:

block (int)

Return type:

int

before_delete()
before_insert()
before_update()
static check_closed(stream_id)[source]
Parameters:

stream_id (HexStr)

Return type:

bool

delete()
find_updated_attributes()
flush()
get_pk()
load(*attrs)
pause()[source]
Return type:

None

print()[source]
Return type:

None

set(**kwargs)
stop_stream(block)[source]
Parameters:

block (int)

Return type:

None

to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
property amount_per_day: int
property amount_per_hour: int
property amount_per_minute: int
amount_per_second
contract
end_block
from_address
property is_alive: bool
reason
scale = 100000000000000000000
start_block
property start_date: date
status
property stream_contract: Contract
stream_id
streamed_funds
to_address
token
txgroup
class dao_treasury.db.StreamedFunds[source]

Bases: Entity

Each object represents one calendar day of tokens streamed for a particular stream.

__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
before_delete()
before_insert()
before_update()
classmethod create_entity(stream_id, date, price, seconds_active, is_last_day)[source]
Parameters:
Return type:

StreamedFunds

delete()
find_updated_attributes()
flush()
get_entity(date)[source]
Parameters:
Return type:

StreamedFunds

get_pk()
load(*attrs)
set(**kwargs)
to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
amount
date
is_last_day
price
seconds_active
stream
value_usd
final class dao_treasury.db.Token[source]

Bases: Entity

Pony ORM entity representing an ERC-20 token or native coin placeholder.

Stores symbol, name, and decimals for value scaling.

Examples

>>> Token.get_dbid("0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE")
1
>>> tok = Token.get_or_insert("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> tok.symbol
'DAI'

See also

scale_value()

__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
before_delete()
before_insert()
before_update()
delete()
find_updated_attributes()
flush()
static get_dbid(address)[source]

Get or insert a Token record and return its database ID.

Parameters:

address (HexAddress) – Token contract address or native coin placeholder.

Return type:

int

Examples

>>> Token.get_dbid("0x6B175474E89094C44Da98b954EedeAC495271d0F")
2
static get_or_insert(address)[source]

Insert or fetch a token record from the chain, resolving metadata on-chain.

Parameters:

address (HexAddress) – ERC-20 contract address or native coin placeholder.

Return type:

Token

Examples

>>> Token.get_or_insert("0x6B175474E89094C44Da98b954EedeAC495271d0F")
<Token ...>
get_pk()
load(*attrs)
scale_value(value)[source]

Convert an integer token amount into a Decimal accounting for decimals.

Parameters:

value (int) – Raw integer on-chain amount.

Return type:

Decimal

Examples

>>> t = Token.get_or_insert("0x...")
>>> t.scale_value(1500000000000000000)
Decimal('1.5')
set(**kwargs)
to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
address

Foreign key to the address record for this token contract.

chain

Foreign key linking to Chain.

property contract: Contract
decimals

Number of decimals used for value scaling.

name

Full human-readable name of the token.

property scale: int

Base for division according to decimals, e.g., 10**decimals.

Examples

>>> t = Token.get_or_insert("0x...")
>>> t.scale
1000000000000000000
streams
symbol

Short ticker symbol for the token.

token_id

Auto-incremented primary key for the tokens table.

treasury_tx: Set['TreasuryTx']

Inverse relation for treasury transactions involving this token.

class dao_treasury.db.TreasuryTx[source]

Bases: Entity

Pony ORM entity for on-chain treasury transactions.

Represents individual token or native transfers with pricing, grouping, and gas data.

Examples

>>> # After inserting, fetch sorted records
>>> with db_session:
...     txs = TreasuryTx.select(lambda tx: tx.txgroup == TxGroup.get_dbid("Revenue"))
...     for tx in txs:
...         print(tx.hash, tx.value_usd)
__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
before_delete()
before_insert()
before_update()
delete()
find_updated_attributes()
flush()
get_events(event_name)[source]
Parameters:

event_name (str)

Return type:

_EventItem

get_pk()
async static insert(entry)[source]

Asynchronously insert and sort a ledger entry.

Converts a LedgerEntry into a TreasuryTx record, then applies advanced sorting.

Parameters:

entry (Transaction | InternalTransfer | TokenTransfer) – A ledger entry representing a token or internal transfer.

Return type:

None

Examples

>>> import asyncio, eth_portfolio.structs as s
>>> asyncio.run(TreasuryTx.insert(s.TokenTransfer(...)))

See also

__insert()

load(*attrs)
set(**kwargs)
to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
amount

On-chain transfer amount as a Decimal with fixed precision.

block

Block number of the transaction.

chain

Foreign key to the network where the transaction occurred.

property events: EventDict

Decoded event logs for this transaction.

from_address

Foreign key to sender address record.

property from_nickname: str

Human-readable label for the sender address.

gas_price

Gas price paid, in native token units (native transfers only).

gas_used

Gas units consumed by this transaction (native transfers only).

hash

Hex string of the transaction hash.

log_index

Log index within the block (None for native transfers).

price

Token price at the time of transfer (if available).

property symbol: str

Ticker symbol for the transferred token.

timestamp

Block timestamp as Unix epoch seconds.

to_address

Foreign key to recipient address record.

property to_nickname: str | None

Human-readable label for the recipient address, if any.

token

Foreign key to the token record used in the transfer.

treasury_tx_id

Auto-incremented primary key for treasury transactions.

txgroup

Foreign key to the categorization group.

value_usd

USD value of the transfer, computed as amount * price.

class dao_treasury.db.TxGroup[source]

Bases: Entity

Pony ORM entity for hierarchical transaction groups.

Used to categorize treasury transactions into nested buckets.

Examples

>>> gid = TxGroup.get_dbid("Revenue")
>>> group = TxGroup.get_or_insert("Revenue", None)
>>> group.full_string
'Revenue'
__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
before_delete()
before_insert()
before_update()
delete()
find_updated_attributes()
flush()
static get_dbid(name, parent=None)[source]

Get or insert a transaction group and return its database ID.

Parameters:
Return type:

TxGroupDbid

Examples

>>> TxGroup.get_dbid("Expenses", None)
3
static get_fullname(dbid)[source]
Parameters:

dbid (TxGroupDbid)

Return type:

str

static get_or_insert(name, parent)[source]

Insert or fetch a transaction group.

Parameters:

  • name (str) – Category name.

  • parent (TxGroup | None) – Optional parent group.

Return type:

TxGroup

Examples

>>> TxGroup.get_or_insert("Expenses", None).name
'Expenses'
get_pk()
load(*attrs)
set(**kwargs)
to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
child_txgroups

Set of nested child groups.

property fullname: str

Return the colon-delimited path from root to this group.

Examples

>>> root = TxGroup.get_or_insert("Revenue", None)
>>> child = TxGroup.get_or_insert("Interest", root)
>>> child.full_string
'Revenue:Interest'
name

Name of the grouping category, e.g., ‘Revenue’, ‘Expenses’.

parent_txgroup

Optional reference to a parent group for nesting.

streams
property top_txgroup: TxGroup

Get the top-level ancestor in this group’s hierarchy.

treasury_tx

Inverse relation for treasury transactions assigned to this group.

txgroup_id

Auto-incremented primary key for transaction groups.

dao_treasury.db.create_general_ledger_view()[source]

Create or replace the SQL view general_ledger aggregating all treasury transactions.

Joins chains, tokens, addresses, and txgroups into a single chronological ledger.

Examples

>>> create_general_ledger_view()
Return type:

None

dao_treasury.db.create_monthly_pnl_view()[source]

Create or replace the SQL view monthly_pnl summarizing monthly profit and loss.

Aggregates categorized transactions by month and top-level category.

Examples

>>> create_monthly_pnl_view()
Return type:

None

dao_treasury.db.create_stream_ledger_view()[source]

Create or replace the SQL view stream_ledger for streamed funds reporting.

This view joins streamed funds, streams, tokens, addresses, and txgroups into a unified ledger of stream transactions.

Examples

>>> create_stream_ledger_view()
Return type:

None

dao_treasury.db.create_txgroup_hierarchy_view()[source]

Create or replace the SQL view txgroup_hierarchy for recursive txgroup hierarchy.

This view exposes txgroup_id, top_category, and parent_txgroup for all txgroups, matching the recursive CTE logic used in dashboards.

Return type:

None

dao_treasury.db.create_unsorted_txs_view()[source]

Create or replace the SQL view unsorted_txs for pending categorization.

Filters general_ledger for transactions still in ‘Categorization Pending’.

Examples

>>> create_unsorted_txs_view()
Return type:

None

dao_treasury.db.create_vesting_ledger_view()[source]

Create or replace the SQL view vesting_ledger for vesting escrow reporting.

This view joins vested funds, vesting escrows, tokens, chains, addresses, and txgroups to produce a vesting ledger.

Examples

>>> create_vesting_ledger_view()
Return type:

None

dao_treasury.db.get_transaction(txhash)[source]

Fetch and cache a transaction receipt from the connected chain.

Wraps brownie.network.chain.Chain.get_transaction().

Parameters:

txhash (str) – Hex string of the transaction hash.

Return type:

TransactionReceipt

Examples

>>> get_transaction("0xabcde...")
<Transaction '0xabcde...'>
dao_treasury.db.SQLITE_DIR = PosixPath('/home/runner/.dao-treasury')

Path to the directory in the user’s home where the DAO treasury SQLite database is stored.

dao_treasury.main module

dao_treasury.treasury module

Treasury orchestration and analytics interface.

This module defines the Treasury class, which aggregates DAO wallets, sets up sorting rules, and manages transaction ingestion and streaming analytics. It coordinates the end-to-end flow from wallet configuration to database population and dashboard analytics.

Key Responsibilities:

  • Aggregate and manage DAO-controlled wallets.

  • Ingest and process on-chain transactions.

  • Apply sorting/categorization rules.

  • Integrate with streaming protocols (e.g., LlamaPay).

  • Populate the database for analytics and dashboards.

This is the main entry point for orchestrating DAO treasury analytics.

class dao_treasury.treasury.Treasury[source]

Bases: ASyncGenericBase

__init__(wallets, sort_rules=None, start_block=0, label="your org's treasury", asynchronous=False)[source]

Initialize the Treasury singleton for managing DAO funds.

This class aggregates multiple treasury wallets, sets up sorting rules, and constructs an ExportablePortfolio for fetching balance and transaction history.

Parameters:

  • wallets (Iterable[TreasuryWallet | str]) – Iterable of wallet addresses or TreasuryWallet instances representing DAO-controlled wallets.

  • sort_rules (Path | None) – Directory path containing YAML rule files for sorting transactions. See dao_treasury.sorting._rules.Rules.

  • start_block (int) – Block number from which to start loading portfolio history.

  • label (str) – Descriptive label for the portfolio, used in exported data.

  • asynchronous (bool) – Whether methods default to asynchronous mode.

Raises:

  • RuntimeError – If a second Treasury instance is initialized.

  • TypeError – If any item in wallets is not a str or TreasuryWallet.

Return type:

None

Examples

# Create a synchronous Treasury
treasury = Treasury(
    wallets=["0xAbc123...", TreasuryWallet("0xDef456...", start_block=1000)],
    sort_rules=Path("/path/to/rules"),
    start_block=500,
    label="DAO Treasury",
    asynchronous=False
)

# Create an asynchronous Treasury
treasury_async = Treasury(
    wallets=["0xAbc123..."],
    asynchronous=True
)
asynchronous: Final

A boolean flag indicating whether the API for this Treasury object is sync or async by default

describe[source]
Parameters:

block (int)

Return type:

PortfolioBalances

populate_db[source]

Populate the database with treasury transactions and streams in parallel.

Parameters:
Return type:

None

portfolio: Final

An eth_portfolio.Portfolio object used for exporting tx and balance history

property txs: ASyncIterator[Union][Transaction | InternalTransfer | TokenTransfer]
wallets: Final[List[TreasuryWallet]]

The collection of wallets owned or controlled by the on-chain org

dao_treasury.types module

class dao_treasury.types.TxGroupDbid

NewType representing the primary key of a transaction group in the database.

This is the integer identifier returned by get_dbid().

Examples

Get the database ID for the “Expenses” group:
>>> dbid: TxGroupDbid = TxGroupDbid(3)

alias of int

__init__(name, tp)
__new__(**kwargs)
dao_treasury.types.TxGroupName

Alias for the human‐readable name of a transaction group.

Names are passed to get_dbid() to retrieve or create groups in the database.

Examples

Creating or retrieving the “Other Income” group:
>>> name: TxGroupName = "Other Income"
>>> dbid = TxGroup.get_dbid(name)
dao_treasury.types.Networks

Type alias for specifying one or more blockchain networks.

This can be a single Network enum member or any iterable of such members. It is used to restrict operations to specific networks when configuring or querying treasury data.

Examples

Specify a single network:
>>> net: Networks = Network.Mainnet
Specify multiple networks:
>>> nets: Networks = [Network.Mainnet, Network.Optimism]

See also

Network

alias of Network | Iterable[Network]

dao_treasury.types.SortFunction

Type for a function or coroutine that determines if a transaction matches a rule.

A sorting function takes a single TreasuryTx instance and returns a boolean or an awaitable resolving to a boolean to indicate whether the rule applies.

Examples

Synchronous matcher:
>>> def is_large(tx: TreasuryTx) -> bool:
...     return tx.amount > 1000
Asynchronous matcher:
>>> async def has_tag(tx: TreasuryTx) -> bool:
...     return await some_external_check(tx.hash)

See also

match

alias of Callable[[TreasuryTx], bool] | Callable[[TreasuryTx], Awaitable[bool]]

dao_treasury.types.SortRule

Union of all built‐in sort rule classes.

Each rule assigns transactions to a specific category by matching on transaction attributes or by invoking a custom function.

Examples

Define a list of available sort rules:
>>> from dao_treasury.sorting.rule import SORT_RULES
>>> rules: list[SortRule] = SORT_RULES

See also

RevenueSortRule ExpenseSortRule IgnoreSortRule CostOfRevenueSortRule OtherIncomeSortRule OtherExpenseSortRule

alias of RevenueSortRule | CostOfRevenueSortRule | ExpenseSortRule | OtherIncomeSortRule | OtherExpenseSortRule | IgnoreSortRule

dao_treasury.types.TopLevelCategory

Literal type defining the allowed top‐level categories for treasury transactions.

Each transaction must be assigned to one of these categories in order to generate financial reports.

Examples

Grouping a transaction as revenue:
>>> cat: TopLevelCategory = "Revenue"
Ignoring a transaction:
>>> cat: TopLevelCategory = "Ignore"

alias of Literal[‘Revenue’, ‘Cost of Revenue’, ‘Expenses’, ‘Other Income’, ‘Other Expenses’, ‘Ignore’]

Module contents

DAO Treasury package initializer.

Exposes the main public API for the library, including the Treasury class, wallet management, sorting rules, and database models. Sets up address nicknames and enables SQL debugging if configured.

Key Responsibilities:

  • Import and expose core classes and functions.

  • Initialize address nicknames in the database.

  • Configure SQL debugging for development.

This is the main import point for users and integrations.

class dao_treasury.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.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.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.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.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.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.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
class dao_treasury.Treasury[source]

Bases: ASyncGenericBase

__init__(wallets, sort_rules=None, start_block=0, label="your org's treasury", asynchronous=False)[source]

Initialize the Treasury singleton for managing DAO funds.

This class aggregates multiple treasury wallets, sets up sorting rules, and constructs an ExportablePortfolio for fetching balance and transaction history.

Parameters:

  • wallets (Iterable[TreasuryWallet | str]) – Iterable of wallet addresses or TreasuryWallet instances representing DAO-controlled wallets.

  • sort_rules (Path | None) – Directory path containing YAML rule files for sorting transactions. See dao_treasury.sorting._rules.Rules.

  • start_block (int) – Block number from which to start loading portfolio history.

  • label (str) – Descriptive label for the portfolio, used in exported data.

  • asynchronous (bool) – Whether methods default to asynchronous mode.

Raises:

  • RuntimeError – If a second Treasury instance is initialized.

  • TypeError – If any item in wallets is not a str or TreasuryWallet.

Return type:

None

Examples

# Create a synchronous Treasury
treasury = Treasury(
    wallets=["0xAbc123...", TreasuryWallet("0xDef456...", start_block=1000)],
    sort_rules=Path("/path/to/rules"),
    start_block=500,
    label="DAO Treasury",
    asynchronous=False
)

# Create an asynchronous Treasury
treasury_async = Treasury(
    wallets=["0xAbc123..."],
    asynchronous=True
)
asynchronous: Final

A boolean flag indicating whether the API for this Treasury object is sync or async by default

describe[source]
Parameters:

block (int)

Return type:

PortfolioBalances

populate_db[source]

Populate the database with treasury transactions and streams in parallel.

Parameters:
Return type:

None

portfolio: Final

An eth_portfolio.Portfolio object used for exporting tx and balance history

sort_rules: Final
property txs: ASyncIterator[Union][Transaction | InternalTransfer | TokenTransfer]
wallets: Final[List[TreasuryWallet]]

The collection of wallets owned or controlled by the on-chain org

class dao_treasury.TreasuryTx[source]

Bases: Entity

Pony ORM entity for on-chain treasury transactions.

Represents individual token or native transfers with pricing, grouping, and gas data.

Examples

>>> # After inserting, fetch sorted records
>>> with db_session:
...     txs = TreasuryTx.select(lambda tx: tx.txgroup == TxGroup.get_dbid("Revenue"))
...     for tx in txs:
...         print(tx.hash, tx.value_usd)
__init__(*args, **kwargs)
after_delete()
after_insert()
after_update()
before_delete()
before_insert()
before_update()
delete()
find_updated_attributes()
flush()
get_events(event_name)[source]
Parameters:

event_name (str)

Return type:

_EventItem

get_pk()
async static insert(entry)[source]

Asynchronously insert and sort a ledger entry.

Converts a LedgerEntry into a TreasuryTx record, then applies advanced sorting.

Parameters:

entry (Transaction | InternalTransfer | TokenTransfer) – A ledger entry representing a token or internal transfer.

Return type:

None

Examples

>>> import asyncio, eth_portfolio.structs as s
>>> asyncio.run(TreasuryTx.insert(s.TokenTransfer(...)))

See also

__insert()

load(*attrs)
set(**kwargs)
to_dict(only=None, exclude=None, with_collections=False, with_lazy=False, related_objects=False)
to_json(include=(), exclude=(), converter=None, with_schema=True, schema_hash=None)
amount

On-chain transfer amount as a Decimal with fixed precision.

block

Block number of the transaction.

chain

Foreign key to the network where the transaction occurred.

property events: EventDict

Decoded event logs for this transaction.

from_address

Foreign key to sender address record.

property from_nickname: str

Human-readable label for the sender address.

gas_price

Gas price paid, in native token units (native transfers only).

gas_used

Gas units consumed by this transaction (native transfers only).

hash

Hex string of the transaction hash.

log_index

Log index within the block (None for native transfers).

price

Token price at the time of transfer (if available).

property symbol: str

Ticker symbol for the transferred token.

timestamp

Block timestamp as Unix epoch seconds.

to_address

Foreign key to recipient address record.

property to_nickname: str | None

Human-readable label for the recipient address, if any.

token

Foreign key to the token record used in the transfer.

treasury_tx_id

Auto-incremented primary key for treasury transactions.

txgroup

Foreign key to the categorization group.

value_usd

USD value of the transfer, computed as amount * price.

final class dao_treasury.TreasuryWallet[source]

Bases: object

A dataclass used to supplement a treasury wallet address with some extra context if needed for your use case

__init__(address, start_block=None, end_block=None, start_timestamp=None, end_timestamp=None, networks=None)
Parameters:

  • address (EthAddress)

  • start_block (int | None)

  • end_block (int | None)

  • start_timestamp (int | None)

  • end_timestamp (int | None)

  • networks (List[int] | None)

Return type:

None

static check_membership(address, block=None)[source]
Parameters:
Return type:

bool

address: EthAddress

The wallet address you need to include with supplemental information.

end_block: int | None = None

The last block at which this wallet was considered owned by the DAO, if it wasn’t always included in the treasury. If end_block is provided, you cannot provide an end_timestamp.

end_timestamp: int | None = None

The last timestamp at which this wallet was considered owned by the DAO, if it wasn’t always included in the treasury. If end_timestamp is provided, you cannot provide an end_block.

networks: List[int] | None = None

The networks where the DAO owns this wallet. If not provided, the wallet will be active on all networks.

start_block: int | None = None

The first block at which this wallet was considered owned by the DAO, if it wasn’t always included in the treasury. If start_block is provided, you cannot provide a start_timestamp.

start_timestamp: int | None = None

The first timestamp at which this wallet was considered owned by the DAO, if it wasn’t always included in the treasury. If start_timestamp is provided, you cannot provide a start_block.

dao_treasury.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.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.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.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.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.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