GMXMarketCache

Documentation for eth_defi.gmx.cache.GMXMarketCache Python class.

class GMXMarketCache

Bases: eth_defi.sqlite_cache.PersistentKeyValueStore

Persistent cache for GMX market data with TTL support.

Extends PersistentKeyValueStore to add: - JSON encoding/decoding for market data - TTL (time-to-live) support with expiry checking - Chain-specific cache keys - Loading mode separation (rpc/graphql/rest_api)

Cache entries are stored with metadata: {

“data”: <market_data>, “timestamp”: <unix_timestamp>, “ttl”: <seconds>, “loading_mode”: “rest_api|graphql|rpc”

}

Example:

cache = GMXMarketCache.get_cache("arbitrum")

# Try to get cached markets
markets = cache.get_markets("rest_api")
if markets is None:
    # Cache miss or expired - load from API
    markets = load_markets_from_api()
    cache.set_markets(markets, "rest_api", ttl=3600)

Initialise GMX market cache.

Parameters

  • filename – Path to SQLite database file

  • autocommit – Whether to commit after each write

Attributes summary

conn

One connection per thread

Methods summary

__init__(filename[, autocommit])

Initialise GMX market cache.

clear()

clear_expired()

Remove all expired cache entries.

close()

commit()

copy()

decode_value(value)

Decode JSON string to Python dict.

encode_value(value)

Encode Python dict as JSON string.

fromkeys([value])

Create a new dictionary with keys from iterable and values set to value.

get(key[, default])

Return the value for key if key is in the dictionary, else default.

get_apy([period, check_expiry])

Get cached APY data.

get_cache(chain[, cache_dir, disabled])

Get or create cache instance for a chain.

get_file_size()

get_markets(loading_mode[, check_expiry])

Get cached markets data.

items()

iteritems()

iterkeys()

itervalues()

keys()

pop(k[,d])

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

purge()

Delete all keys and save.

set_apy(data[, period, ttl])

Store APY data in cache.

set_markets(data, loading_mode[, ttl])

Store markets data in cache.

setdefault(key[, default])

Insert key with a value of default if key is not in the dictionary.

update([E, ]**F)

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values()

__init__(filename, autocommit=True)

Initialise GMX market cache.

Parameters

  • filename (pathlib.Path) – Path to SQLite database file

  • autocommit (bool) – Whether to commit after each write

classmethod get_cache(chain, cache_dir=None, disabled=False)

Get or create cache instance for a chain.

Parameters

  • chain (str) – Chain name (arbitrum, avalanche, etc.)

  • cache_dir (Optional[pathlib.Path]) – Custom cache directory (optional, uses default if None)

  • disabled (bool) – If True, returns None (cache disabled)

Returns

GMXMarketCache instance or None if disabled

Return type

Optional[eth_defi.gmx.cache.GMXMarketCache]

encode_value(value)

Encode Python dict as JSON string.

Parameters

value (dict) – Dictionary to encode

Returns

JSON string

Return type

str

decode_value(value)

Decode JSON string to Python dict.

Parameters

value (str) – JSON string to decode

Returns

Dictionary

Return type

dict

get_markets(loading_mode, check_expiry=True)

Get cached markets data.

Parameters

  • loading_mode (str) – Loading mode (‘rest_api’, ‘graphql’, or ‘rpc’)

  • check_expiry (bool) – Whether to check TTL and return None if expired

Returns

Market data dict or None if not found/expired

Return type

Optional[dict]

set_markets(data, loading_mode, ttl=None)

Store markets data in cache.

Parameters

  • data (dict) – Markets dictionary to cache

  • loading_mode (str) – Loading mode (‘rest_api’, ‘graphql’, or ‘rpc’)

  • ttl (Optional[int]) – Time-to-live in seconds (uses default if None)

Return type

None

get_apy(period='30d', check_expiry=True)

Get cached APY data.

Parameters

  • period (str) – APY period (1d, 7d, 30d, etc.)

  • check_expiry (bool) – Whether to check TTL

Returns

APY data dict or None if not found/expired

Return type

Optional[dict]

set_apy(data, period='30d', ttl=None)

Store APY data in cache.

Parameters

  • data (dict) – APY dictionary to cache

  • period (str) – APY period

  • ttl (Optional[int]) – Time-to-live in seconds (uses default if None)

Return type

None

clear_expired()

Remove all expired cache entries.

Returns

Number of entries removed

Return type

int

__new__(**kwargs)
clear() → None.  Remove all items from D.
property conn: sqlite3.Connection

One connection per thread

copy() → a shallow copy of D
fromkeys(value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None)

Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items
keys() → a set-like object providing a view on D's keys
pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

purge()

Delete all keys and save.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values() → an object providing a view on D's values