Composable Orders

from cowdao_cowpy.composable import ConditionalOrder
from cowdao_cowpy.common.chains import Chain
from eth_typing import HexStr

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)

def is_valid(self) -> IsValidResult:
    """Validate the conditional order's parameters."""
    pass

async def poll_validate(self, params: PollParams) -> Optional[PollResultError]:
    """Perform order-specific validation during polling."""
    pass

async def handle_poll_failed_already_present(
    self, order_uid: str, order: Order, params: PollParams
) -> Optional[PollResultError]:
    """Handle the case when the polled order is already in the orderbook."""
    pass

def transform_data_to_struct(self, params: D) -> S:
    """Convert friendly data format to contract struct format."""
    pass

def transform_struct_to_data(self, params: S) -> D:
    """Convert contract struct format to friendly data format."""
    pass

def to_string(self, token_formatter: Optional[Callable] = None) -> str:
    """Create a human-readable representation of the order."""
    pass

def serialize(self) -> HexStr:
    """Serialize the order to its ABI-encoded form."""
    pass

def encode_static_input(self) -> HexStr:
    """Encode the staticInput for the conditional order."""
    pass

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

from cowdao_cowpy.composable.types import PollParams

result = await order.poll(PollParams(
    owner="0x...",
    chain=Chain.MAINNET,
    provider=web3,
    order_book_api=api
))

if result.result == PollResultCode.SUCCESS:
    print(f"Order ready: {result.order}")
    print(f"Signature: {result.signature}")

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

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

def compute_order_uid(self, chain: Chain, owner: str, order: Order) -> str:
    """Compute the unique identifier for an order."""

def encode_static_input_helper(
    self, order_data_types: Iterable[TypeStr], static_input: S
) -> HexStr:
    """Helper function for generically serializing static input."""

@staticmethod
def leaf_to_id(leaf: ConditionalOrderParams) -> str:
    """Calculate the ID from a conditional order leaf."""

@staticmethod
def deserialize_helper(
    encoded_data: HexStr,
    handler: str,
    order_data_types: List[str],
    callback: Callable[[Any, HexStr], T],
) -> T:
    """Helper function for generically deserializing a conditional order."""

from cowdao_cowpy.composable import Multiplexer, ProofLocation

multiplexer = Multiplexer(
    orders={"order_id_1": order1, "order_id_2": order2},
    root="0xabc...",
    location=ProofLocation.PRIVATE
)

def add(self, order: ConditionalOrder):
    """Add a new order to the multiplexer."""

from cowdao_cowpy.composable import Twap, TwapData

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..."
))

multiplexer.add(twap)

def remove(self, id: str):
    """Remove an order by ID."""

def update(self, id: str, updater: Callable):
    """Update an order using an updater function."""

def get_by_id(self, id: str) -> ConditionalOrder:
    """Get an order by its ID."""

def get_by_index(self, i: int) -> ConditionalOrder:
    """Get an order by its index in the order list."""

def get_proofs(self, filter: Optional[Callable] = None) -> List[ProofWithParams]:
    """Get merkle proofs for orders, optionally filtered."""

async def prepare_proof_struct(
    self,
    location: ProofLocation = None,
    filter: Callable = None,
    uploader: Callable = None,
) -> tuple[int, HexBytes]:
    """Prepare proof data for on-chain submission."""

@staticmethod
async def poll(
    owner: HexStr,
    p: ProofWithParams,
    chain: Chain,
    off_chain_input_fn: Optional[Callable] = None,
) -> Tuple[Order, HexStr]:
    """Poll a specific order from the multiplexer."""

def to_json(self) -> str:
    """Serialize multiplexer to JSON."""

@classmethod
def from_json(cls, s: str) -> "Multiplexer":
    """Deserialize multiplexer from JSON."""

@classmethod
def register_order_type(cls, order_type: str, conditional_order_class: type):
    """Register a new order type with the multiplexer."""

Multiplexer.register_order_type("twap", Twap)

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

twap = Twap(
    handler=TWAP_ADDRESS,
    data=TwapData(...),
    salt=None
)

@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

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

# Create a TWAP that sells 1000 tokens in 10 parts over 10 hours
twap_data = TwapData(
    sell_token="0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2",  # WETH
    buy_token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",   # USDC
    receiver="0x...",
    sell_amount=1000000000000000000000,  # 1000 WETH
    buy_amount=3000000000,  # 3000 USDC minimum
    start_type=StartType.AT_MINING_TIME,
    number_of_parts=10,
    time_between_parts=3600,  # 1 hour
    duration_type=DurationType.AUTO,
    app_data="0x..."
)

twap = Twap.from_data(twap_data)

# Validate the order
validation = twap.is_valid()
if validation.is_valid:
    print(f"Order ID: {twap.id}")
    print(f"Create calldata: {twap.create_calldata}")

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

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"

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

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

def encode_params(params: ConditionalOrderParams) -> HexStr:
    """Encode conditional order parameters to ABI format."""

def decode_params(encoded: HexStr) -> ConditionalOrderParams:
    """Decode ABI-encoded parameters to ConditionalOrderParams."""

def hash_order(domain: Dict[str, Any], order: Dict[str, Any]) -> str:
    """Compute the EIP-712 hash of an order."""

def hash_order_cancellation(domain: Dict[str, Any], order_uid: str) -> str:
    """Compute the EIP-712 hash for cancelling a single order."""

def hash_order_cancellations(domain: Dict[str, Any], order_uids: List[str]) -> str:
    """Compute the EIP-712 hash for cancelling multiple orders."""

def encode_order(order: Dict[str, Any]) -> bytes:
    """Encode an order to bytes for hashing."""