Skip to main content
The composable module provides functionality for creating and managing programmatic orders that execute based on specific conditions.

ConditionalOrder

Abstract base class for all programmatic orders.

Constructor

from cowdao_cowpy.composable import ConditionalOrder

class MyOrder(ConditionalOrder[DataType, StructType]):
    def __init__(
        self,
        handler: HexStr,
        data: DataType,
        salt: HexStr | None = None,
        has_off_chain_input: bool = False,
        chain: Chain = Chain.MAINNET,
    ):
        super().__init__(handler, data, salt, has_off_chain_input, chain)
ParameterTypeDefaultDescription
handlerHexStrRequiredHandler contract address
dataDRequiredOrder parameters in friendly format
saltHexStr | NoneNone32-byte hex salt (random if not provided)
has_off_chain_inputboolFalseWhether order requires off-chain input
chainChainChain.MAINNETBlockchain network

Properties

PropertyTypeDescription
idHexStrUnique identifier (keccak256(serialize()))
ctxstrContext key for cabinet lookups
leafConditionalOrderParamsLeaf data for Merkle trees
create_calldataHexStrCalldata for on-chain creation
remove_calldatastrCalldata for removing the order
off_chain_inputHexStrOff-chain input data

Abstract Methods

MethodReturnsDescription
is_valid()IsValidResultValidate order parameters
poll_validate(params)Optional[PollResultError]Validate during polling
transform_data_to_struct(data)SConvert data to struct format
transform_struct_to_data(struct)DConvert struct to data format
to_string(token_formatter)strHuman-readable representation
serialize()HexStrABI-encoded serialization
encode_static_input()HexStrEncode static input

Methods

poll

async def poll(self, params: PollParams) -> PollResult
Poll the programmatic order to check if it’s tradeable.

is_authorized

async def is_authorized(self, params: OwnerParams) -> bool
Check if the owner has authorized this programmatic order.

cabinet

async def cabinet(self, params: OwnerParams) -> str
Get the cabinet value for this order and owner.

Multiplexer

Manages multiple programmatic orders using Merkle trees.

Constructor

from cowdao_cowpy.composable import Multiplexer, ProofLocation

multiplexer = Multiplexer(
    orders={"id1": order1, "id2": order2},
    root="0xabc...",
    location=ProofLocation.PRIVATE
)
ParameterTypeDefaultDescription
ordersOptional[Dict[str, ConditionalOrder]]NoneOrder ID to order mapping
rootOptional[str]NoneMerkle tree root
locationProofLocationProofLocation.PRIVATEProof storage location

Properties

PropertyTypeDescription
rootstrMerkle tree root hash
order_idsList[str]List of all order IDs

Methods

MethodDescription
add(order)Add a new order
remove(id)Remove an order by ID
update(id, updater)Update an order with a function
get_by_id(id)Get order by ID
get_by_index(i)Get order by index
get_proofs(filter)Get Merkle proofs for orders
prepare_proof_struct(...)Prepare proof data for on-chain submission
to_json()Serialize to JSON
from_json(s)Deserialize from JSON (classmethod)
register_order_type(type, cls)Register a new order type (classmethod)

TWAP Order Type

TwapData

@dataclass
class TwapData:
    sell_token: str
    buy_token: str
    receiver: str
    sell_amount: int
    buy_amount: int
    start_type: StartType
    number_of_parts: int
    time_between_parts: int
    duration_type: DurationType
    app_data: str
    start_time_epoch: int = 0
    duration_of_part: int = 0

Twap

from cowdao_cowpy.composable import Twap, TwapData, StartType, DurationType

twap = Twap.from_data(TwapData(
    sell_token="0x...", buy_token="0x...", receiver="0x...",
    sell_amount=1000000, buy_amount=900000,
    start_type=StartType.AT_MINING_TIME,
    number_of_parts=10, time_between_parts=3600,
    duration_type=DurationType.AUTO, app_data="0x..."
))

validation = twap.is_valid()
if validation.is_valid:
    print(f"Order ID: {twap.id}")
    print(f"Create calldata: {twap.create_calldata}")
TWAP handler address: 0x6cF1e9cA41f7611dEf408122793c358a3d11E5a5

Types

ProofLocation

class ProofLocation(Enum):
    PRIVATE = 0
    EMITTED = 1
    SWARM = 2
    WAKU = 3
    RESERVED = 4
    IPFS = 5

PollResultCode

class PollResultCode(Enum):
    SUCCESS = "SUCCESS"
    UNEXPECTED_ERROR = "UNEXPECTED_ERROR"
    TRY_NEXT_BLOCK = "TRY_NEXT_BLOCK"
    TRY_ON_BLOCK = "TRY_ON_BLOCK"
    TRY_AT_EPOCH = "TRY_AT_EPOCH"
    DONT_TRY_AGAIN = "DONT_TRY_AGAIN"

IsValidResult

@dataclass
class IsValidResult:
    is_valid: bool
    reason: Optional[str] = None

ConditionalOrderParams

@dataclass
class ConditionalOrderParams:
    handler: HexStr
    salt: HexStr
    static_input: HexStr

Utility Functions

FunctionDescription
encode_params(params)Encode programmatic order parameters to ABI format
decode_params(encoded)Decode ABI-encoded parameters
hash_order(domain, order)Compute EIP-712 hash of an order
hash_order_cancellation(domain, order_uid)Compute hash for cancelling a single order
hash_order_cancellations(domain, order_uids)Compute hash for cancelling multiple orders
encode_order(order)Encode an order to bytes for hashing
Last modified on March 12, 2026