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, Currency, Currency][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'#