pyth_tools.py•8.95 kB
"""Pyth Network API client functions."""
from typing import Optional
import httpx
BASE_URL = "https://hermes.pyth.network"
def get_price_feeds(
query: Optional[str] = None,
asset_type: Optional[str] = None,
) -> dict:
"""Get the set of price feeds from the Pyth network.
This endpoint fetches all price feeds from the Pyth network. It can be filtered
by asset type and query string.
Args:
query (optional): Filter results to price feeds whose symbol contains this string.
Query string is case insensitive.
asset_type (optional): Filter by asset type. Possible values are: crypto, equity,
fx, metal, rates. Filter string is case insensitive.
Returns:
List of price feed metadata with IDs and attributes.
"""
try:
params = {}
if query:
params["query"] = query
if asset_type:
params["asset_type"] = asset_type
response = httpx.get(f"{BASE_URL}/v2/price_feeds", params=params, timeout=30.0)
response.raise_for_status()
return {"price_feeds": response.json()}
except httpx.HTTPError as e:
return {"error": f"HTTP error occurred: {str(e)}"}
except Exception as e:
return {"error": f"Error occurred: {str(e)}"}
def get_latest_price_updates(
ids: list[str],
encoding: Optional[str] = "hex",
parsed: Optional[bool] = True,
ignore_invalid_price_ids: Optional[bool] = False,
include_binary: Optional[bool] = False,
) -> dict:
"""Get the latest price updates by price feed id.
Given a collection of price feed ids, retrieve the latest Pyth price for each price feed.
Args:
ids: List of price feed IDs to get updates for (required).
encoding (optional): Encoding type for binary data. Options: hex, base64. Default: hex.
parsed (optional): If true, include the parsed price update. Default: true.
ignore_invalid_price_ids (optional): If true, invalid price IDs are ignored. Default: false.
include_binary (optional): If true, include binary proof data. Default: false (only parsed data).
Returns:
Price updates with binary and/or parsed data.
"""
try:
params = [("ids[]", price_id) for price_id in ids]
if encoding:
params.append(("encoding", encoding))
if parsed is not None:
params.append(("parsed", str(parsed).lower()))
if ignore_invalid_price_ids:
params.append(("ignore_invalid_price_ids", str(ignore_invalid_price_ids).lower()))
response = httpx.get(f"{BASE_URL}/v2/updates/price/latest", params=params, timeout=30.0)
response.raise_for_status()
result = response.json()
# If user doesn't want binary data, return only parsed
if not include_binary and "parsed" in result:
return {"parsed": result["parsed"]}
return result
except httpx.HTTPError as e:
return {"error": f"HTTP error occurred: {str(e)}"}
except Exception as e:
return {"error": f"Error occurred: {str(e)}"}
def get_price_updates_at_time(
publish_time: int,
ids: list[str],
encoding: Optional[str] = "hex",
parsed: Optional[bool] = True,
ignore_invalid_price_ids: Optional[bool] = False,
include_binary: Optional[bool] = False,
) -> dict:
"""Get price updates at or after a specific timestamp.
Given a collection of price feed ids and a timestamp, retrieve the first price update
whose publish_time is >= the provided value.
Args:
publish_time: Unix timestamp in seconds (required).
ids: List of price feed IDs to get updates for (required).
encoding (optional): Encoding type for binary data. Options: hex, base64. Default: hex.
parsed (optional): If true, include the parsed price update. Default: true.
ignore_invalid_price_ids (optional): If true, invalid price IDs are ignored. Default: false.
include_binary (optional): If true, include binary proof data. Default: false (only parsed data).
Returns:
Price updates at or after the specified timestamp.
"""
try:
params = [("ids[]", price_id) for price_id in ids]
if encoding:
params.append(("encoding", encoding))
if parsed is not None:
params.append(("parsed", str(parsed).lower()))
if ignore_invalid_price_ids:
params.append(("ignore_invalid_price_ids", str(ignore_invalid_price_ids).lower()))
response = httpx.get(
f"{BASE_URL}/v2/updates/price/{publish_time}",
params=params,
timeout=30.0
)
response.raise_for_status()
result = response.json()
# If user doesn't want binary data, return only parsed
if not include_binary and "parsed" in result:
return {"parsed": result["parsed"]}
return result
except httpx.HTTPError as e:
return {"error": f"HTTP error occurred: {str(e)}"}
except Exception as e:
return {"error": f"Error occurred: {str(e)}"}
def get_publisher_stake_caps(
encoding: Optional[str] = "hex",
parsed: Optional[bool] = True,
include_binary: Optional[bool] = False,
) -> dict:
"""Get the most recent publisher stake caps update data.
This endpoint returns the latest publisher stake caps information.
Args:
encoding (optional): Encoding type for binary data. Options: hex, base64. Default: hex.
parsed (optional): If true, include the parsed update data. Default: true.
include_binary (optional): If true, include binary proof data. Default: false (only parsed data).
Returns:
Latest publisher stake caps update data.
"""
try:
params = {}
if encoding:
params["encoding"] = encoding
if parsed is not None:
params["parsed"] = str(parsed).lower()
response = httpx.get(
f"{BASE_URL}/v2/updates/publisher_stake_caps/latest",
params=params,
timeout=30.0
)
response.raise_for_status()
result = response.json()
# If user doesn't want binary data, return only parsed
if not include_binary and "parsed" in result:
return {"parsed": result["parsed"]}
return result
except httpx.HTTPError as e:
return {"error": f"HTTP error occurred: {str(e)}"}
except Exception as e:
return {"error": f"Error occurred: {str(e)}"}
def get_twap_latest(
window_seconds: int,
ids: list[str],
encoding: Optional[str] = "hex",
parsed: Optional[bool] = True,
ignore_invalid_price_ids: Optional[bool] = False,
include_binary: Optional[bool] = False,
) -> dict:
"""Get the latest TWAP (Time Weighted Average Price) with a custom time window.
Given a collection of price feed ids, retrieve the latest Pyth TWAP price for each
price feed over the specified time window.
Args:
window_seconds: Time window in seconds (1-600, required). For example, 300 = 5 minute TWAP.
ids: List of price feed IDs to get TWAP for (required).
encoding (optional): Encoding type for binary data. Options: hex, base64. Default: hex.
parsed (optional): If true, include calculated TWAP in parsed field. Default: true.
ignore_invalid_price_ids (optional): If true, invalid price IDs are ignored. Default: false.
include_binary (optional): If true, include binary proof data. Default: false (only parsed data).
Returns:
Time-weighted average prices for the specified window.
"""
try:
if window_seconds < 1 or window_seconds > 600:
return {"error": "window_seconds must be between 1 and 600 seconds"}
params = [("ids[]", price_id) for price_id in ids]
if encoding:
params.append(("encoding", encoding))
if parsed is not None:
params.append(("parsed", str(parsed).lower()))
if ignore_invalid_price_ids:
params.append(("ignore_invalid_price_ids", str(ignore_invalid_price_ids).lower()))
response = httpx.get(
f"{BASE_URL}/v2/updates/twap/{window_seconds}/latest",
params=params,
timeout=30.0
)
response.raise_for_status()
result = response.json()
# If user doesn't want binary data, return only parsed
if not include_binary and "parsed" in result:
return {"parsed": result["parsed"]}
return result
except httpx.HTTPError as e:
return {"error": f"HTTP error occurred: {str(e)}"}
except Exception as e:
return {"error": f"Error occurred: {str(e)}"}