Common types#

Asset#

class pragma_sdk.common.types.asset.Asset[source]#

Bases: object

__init__(data_type: DataTypes, asset_id: str | int, expiration_timestamp: int | None = None)[source]#

asset_id: int#

data_type: DataTypes#

expiration_timestamp: int | None#

serialize() → Dict[str, int | Tuple[int, int | None]][source]#: Serialize method used to interact with Cairo contracts.

to_dict() → Dict[str, int | str | None][source]#

PragmaClient#

class pragma_sdk.common.types.client.PragmaClient[source]#

Bases: ABC

abstract async publish_entries(entries: List[Entry]) → Tuple[Dict | None, Dict | None] | List[InvokeResult][source]#

Publish entries to some destination.

Parameters:

entries – List of Entry objects
execution_config – ExecutionConfig object. Only used for on-chain publishing.

Returns:

Tuple of responses for spot and future entries

abstract publish_entries_sync(entries: List[Entry]) → Tuple[Dict | None, Dict | None] | List[InvokeResult]#: Synchronous version of the method.

Currency#

class pragma_sdk.common.types.currency.Currency[source]#

Bases: object

__init__(currency_id: str, decimals: int, is_abstract_currency: bool, starknet_address: int | str | None = None, ethereum_address: int | str | None = None)[source]#

decimals: int#

ethereum_address: int#

classmethod from_asset_config(config: AssetConfig) → Self[source]#

id: str#

is_abstract_currency: bool#

serialize() → Tuple[str, int, bool, int, int][source]#

starknet_address: int#

to_dict() → Dict[str, int | str | bool][source]#

Entries#

class pragma_sdk.common.types.entry.BaseEntry[source]#

Bases: object

BaseEntry is a dataclass that represents the common fields between SpotEntry and FutureEntry.

__init__(*args: Any, **kwargs: Any) → None#

publisher: int#

source: int#

timestamp: int#

class pragma_sdk.common.types.entry.Entry[source]#

Bases: ABC

Abstract class that represents an Entry. All entries must implement this class.

static flatten_entries(entries: List[Entry]) → List[int][source]#: This flattens entriees to tuples. Useful when you need the raw felt array

static from_oracle_response(pair: Pair, oracle_response: OracleResponse, publisher_name: str, source_name: str) → Entry | None[source]#

abstract get_asset_type() → DataTypes[source]#

abstract get_expiry() → str | None[source]#

abstract get_pair_id() → str[source]#

abstract get_source() → str[source]#

abstract get_timestamp() → int[source]#

abstract offchain_serialize() → Dict[str, object][source]#

static offchain_serialize_entries(entries: List[Entry]) → List[Dict[str, int]][source]#

abstract serialize() → Dict[str, object][source]#

static serialize_entries(entries: List[Entry]) → List[Dict[str, int]][source]#

abstract to_tuple() → Tuple[Any][source]#

class pragma_sdk.common.types.entry.FutureEntry[source]#

Bases: Entry

Represents a Future Entry.

Also used to represent a Perp Entry - the only difference is that a perpetual future has no expiry timestamp.

__init__(pair_id: str | int, price: int, timestamp: int, source: str | int, publisher: str | int, expiry_timestamp: int | None = None, volume: float | int | None = None)[source]#

base: BaseEntry#

expiry_timestamp: int#

static from_dict(entry_dict: Any) → FutureEntry[source]#

static from_oracle_response(pair: Pair, oracle_response: OracleResponse, publisher_name: str, source_name: str) → FutureEntry | None[source]#: Builds the object from a PragmaAsset and an OracleResponse. Method primarly used by our price pusher package when we’re retrieving lastest oracle prices for comparisons with the latest prices of various APIs (binance etc).

get_asset_type() → DataTypes[source]#

get_expiry() → str | None[source]#

get_pair_id() → str[source]#

get_source() → str[source]#

get_timestamp() → int[source]#

offchain_serialize() → Dict[str, object][source]#

pair_id: int#

price: int#

serialize() → Dict[str, object][source]#

to_tuple() → Tuple[int, int, int, int, int, int, int][source]#

volume: int#

class pragma_sdk.common.types.entry.GenericEntry[source]#

Bases: Entry

Represents a Generic Entry.

Currently used this way: instead of publishing all the future options for all availables instruments from Deribit, we place them in all a Merkle tree & we only publish the merkle root through this Generic entry. So the key will be DERIBIT_OPTIONS_DATA and the value the merkle root containing all the price feeds.

__init__(key: str | int, value: int, timestamp: int, source: str | int, publisher: str | int)[source]#

base: BaseEntry#

static from_dict(entry_dict: Any) → GenericEntry[source]#

static from_oracle_response(pair: Pair, oracle_response: OracleResponse, publisher_name: str, source_name: str) → GenericEntry | None[source]#: Builds the object from a PragmaAsset and an OracleResponse. Method primarly used by our price pusher package when we’re retrieving lastest oracle prices for comparisons with the latest prices of various APIs (binance etc).

get_asset_type() → DataTypes[source]#

get_expiry() → str | None[source]#

get_pair_id() → str[source]#

get_source() → str[source]#

get_timestamp() → int[source]#

key: int#

offchain_serialize() → Dict[str, object][source]#

serialize() → Dict[str, object][source]#

to_tuple() → Tuple[int, int, int, int, int][source]#

value: int#

class pragma_sdk.common.types.entry.SpotEntry[source]#

Bases: Entry

Represents a Spot Entry.

__init__(pair_id: str | int, price: int, timestamp: int, source: str | int, publisher: str | int, volume: float | int | None = None) → None[source]#

base: BaseEntry#

static from_dict(entry_dict: Any) → SpotEntry[source]#

static from_oracle_response(pair: Pair, oracle_response: OracleResponse, publisher_name: str, source_name: str) → SpotEntry | None[source]#: Builds a SpotEntry object from a Pair and an OracleResponse. Method primarly used by our price pusher package when we’re retrieving lastest oracle prices for comparisons with the latest prices of various APIs (binance etc).

get_asset_type() → DataTypes[source]#

get_expiry() → str | None[source]#

get_pair_id() → str[source]#

get_source() → str[source]#

get_timestamp() → int[source]#

offchain_serialize() → Dict[str, object][source]#

pair_id: int#

price: int#

serialize() → Dict[str, object][source]#

set_publisher(publisher: int) → SpotEntry[source]#

to_tuple() → Tuple[int, int, int, int, int, int][source]#

volume: int#

Pair#

class pragma_sdk.common.types.pair.Pair[source]#

Bases: object

Pair class to represent a trading pair.

Parameters:

id – Corresponds to the felt representation of the pair e.g str_to_felt(“ETH/USD”)
base_currency – Base currency
quote_currency – Quote currency

__init__(base_currency: Currency, quote_currency: Currency)[source]#

base_currency: Currency#

decimals() → int[source]#: Returns the decimals of the pair. Corresponds to the minimum of both currencies’ decimals.

classmethod from_asset_configs(base_asset: AssetConfig, quote_asset: AssetConfig) → Pair[source]#

Return a Pair from two AssetConfigs. Return None if the base and quote assets are the same.

Parameters:

base_asset – Base asset
quote_asset – Quote asset

Returns:

Pair

static from_tickers(base_ticker: str, quote_ticker: str) → Pair[source]#

Return a Pair from two tickers. Return None if the base and quote tickers are the same.

Parameters:

base_ticker – Base ticker
quote_ticker – Quote ticker

Returns:

Pair

id: int#

quote_currency: Currency#

serialize() → Tuple[int, str, str][source]#

to_dict() → Dict[str, int | str][source]#

to_tuple() → Tuple[str, str][source]#

Misc#

class pragma_sdk.common.types.types.AggregationMode[source]#

Bases: StrEnum

AVERAGE = 'Mean'#

ERROR = 'Error'#

MEDIAN = 'Median'#

serialize() → Dict[str, None][source]#

class pragma_sdk.common.types.types.DataTypes[source]#

Bases: StrEnum

FUTURE = 'Future'#

GENERIC = 'Generic'#

SPOT = 'Spot'#

class pragma_sdk.common.types.types.Environment[source]#

Bases: StrEnum

DEV = 'dev'#

PROD = 'prod'#