rubi.contracts.contract_types package

Submodules

rubi.contracts.contract_types.events module

class rubi.contracts.contract_types.events.BaseEvent(block_number: int, **args)

Bases: ABC

Base class for events to define the structure of an Event from a Rubicon contract.

static builder(name: str, **kwargs) → EmitOfferEvent | EmitTakeEvent | EmitCancelEvent | EmitSwap

client_filter(wallet: eth_typing.ChecksumAddress) → bool

Filter function for client-side filtering. By default, no filtering is done but this can optionally be overwritten by each subclass.

Parameters:: wallet (ChecksumAddress) – The wallet address.
Returns:: True if the event passes the filter, False otherwise.
Return type:: bool

abstract static create_event_filter(contract: web3.contract.Contract, argument_filters: Dict[str, Any] | None = None) → web3._utils.filters.LogFilter

Abstract method to create an event filter for the given contract with optional argument filters. Must be overridden in each event subclass.

Parameters:

contract (Contract) – The contract instance.
argument_filters (Optional[Dict[str, Any]]) – Optional filters. Only events that match these filters will be returned when the filter is queried.

Returns:

The created event filter.

Return type:

LogFilter

abstract static default_filters(bid_identifier: str, ask_identifier: str) → dict

Get the default filters for an event. These are used if no argument filters are provided. By default, these filters make sure we only receive events that relate to us. E.g on markets we care about.

Parameters:

bid_identifier (str) – The identifier for bid events.
ask_identifier (str) – The identifier for ask events.

Returns:

The default filters.

Return type:

dict

classmethod default_handler(pair_name: str, event_type: Type[BaseEvent], event_data: web3.types.EventData) → None

A default event handler. If no handler is provided then this one is used.

Parameters:

pair_name (str) – The name of the pair.
event_type (Type["BaseEvent"]) – The type of the event.
event_data (EventData) – The data of the event.

abstract static get_event_contract(market: _RubiconMarket, router: _RubiconRouter) → _RubiconMarket | _RubiconRouter

Abstract method to determine the contract an event corresponds to. Must be overridden in subclasses.

Parameters:

market (_RubiconMarket) – The RubiconMarket instance.
router (_RubiconRouter) – The RubiconRouter instance.

Returns:

The contract corresponding to the event.

Return type:

Union[_RubiconMarket, _RubiconRouter]

class rubi.contracts.contract_types.events.BaseMarketEvent(id: bytes, pair: bytes, **args)

Bases: BaseEvent, ABC

This class is a base class for all MarketEvents

static get_event_contract(market: _RubiconMarket, router: _RubiconRouter) → _RubiconMarket | _RubiconRouter: implementation of BaseEvent get_event_contract

class rubi.contracts.contract_types.events.EmitCancelEvent(maker: eth_typing.ChecksumAddress, pay_gem: eth_typing.ChecksumAddress, buy_gem: eth_typing.ChecksumAddress, pay_amt: int, buy_amt: int, **args)

Bases: BaseMarketEvent

Event emitted whenever an offer is cancelled on the RubiconMarket

static create_event_filter(contract: web3.contract.Contract, argument_filters: Dict[str, Any] | None = None) → web3._utils.filters.LogFilter: implementation of BaseEvent create_event_filter

static default_filters(bid_identifier: str, ask_identifier: str) → dict: implementation of BaseEvent default_filters

class rubi.contracts.contract_types.events.EmitDeleteEvent(maker: eth_typing.ChecksumAddress, **args)

Bases: BaseMarketEvent

Event emitted whenever an offer is fully taken by a market order on the RubiconMarket resulting in the offer being closed

static create_event_filter(contract: web3.contract.Contract, argument_filters: Dict[str, Any] | None = None) → web3._utils.filters.LogFilter: implementation of BaseEvent create_event_filter

static default_filters(bid_identifier: str, ask_identifier: str) → dict: implementation of BaseEvent default_filters

class rubi.contracts.contract_types.events.EmitFeeEvent(taker: eth_typing.ChecksumAddress, feeTo: eth_typing.ChecksumAddress, asset: eth_typing.ChecksumAddress, feeAmt: int, **args)

Bases: BaseMarketEvent

Event emitted whenever an offer is taken on the RubiconMarket that results in a fee being paid to the maker.

static create_event_filter(contract: web3.contract.Contract, argument_filters: Dict[str, Any] | None = None) → web3._utils.filters.LogFilter: implementation of BaseEvent create_event_filter

static default_filters(bid_identifier: str, ask_identifier: str) → dict: implementation of BaseEvent default_filters

class rubi.contracts.contract_types.events.EmitOfferEvent(maker: eth_typing.ChecksumAddress, pay_gem: eth_typing.ChecksumAddress, buy_gem: eth_typing.ChecksumAddress, pay_amt: int, buy_amt: int, **args)

Bases: BaseMarketEvent

Event emitted whenever a new offer is made on the RubiconMarket

static create_event_filter(contract: web3.contract.Contract, argument_filters: Dict[str, Any] | None = None) → web3._utils.filters.LogFilter: implementation of BaseEvent create_event_filter

static default_filters(bid_identifier: str, ask_identifier: str) → dict: implementation of BaseEvent default_filters

class rubi.contracts.contract_types.events.EmitSwap(recipient: eth_typing.ChecksumAddress, inputERC20: eth_typing.ChecksumAddress, targetERC20: eth_typing.ChecksumAddress, pair: bytes, inputAmount: int, realizedFill: int, hurdleBuyAmtMin: int, **args)

Bases: BaseEvent

Event emitted whenever swap is executed on the RubiconRouter

static create_event_filter(contract: web3.contract.Contract, argument_filters: Dict[str, Any] | None = None) → web3._utils.filters.LogFilter: implementation of BaseEvent create_event_filter

static default_filters(bid_identifier: str, ask_identifier: str) → dict: implementation of BaseEvent default_filters

static get_event_contract(market: _RubiconMarket, router: _RubiconRouter) → _RubiconMarket | _RubiconRouter: implementation of BaseEvent get_event_contract

class rubi.contracts.contract_types.events.EmitTakeEvent(maker: eth_typing.ChecksumAddress, taker: eth_typing.ChecksumAddress, pay_gem: eth_typing.ChecksumAddress, buy_gem: eth_typing.ChecksumAddress, take_amt: int, give_amt: int, **args)

Bases: BaseMarketEvent

Event emitted whenever an offer is taken by a market order on the RubiconMarket

client_filter(wallet: eth_typing.ChecksumAddress) → bool: overwriting of BaseEvent client_filter to only filter when our wallet is either the maker or taker

static create_event_filter(contract: web3.contract.Contract, argument_filters: Dict[str, Any] | None = None) → web3._utils.filters.LogFilter: implementation of BaseEvent create_event_filter

static default_filters(bid_identifier: eth_typing.HexStr, ask_identifier: eth_typing.HexStr) → dict: implementation of BaseEvent default_filters