File size: 8,068 Bytes
6993919 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 | """
Pyth Network Price Feeds Provider
=================================
This provider integrates with Pyth Network's Hermes API to fetch real-time price feeds
for cryptocurrencies and other financial assets.
Pyth provides high-quality, real-time price feeds from 120+ first-party providers
including leading exchanges, banks, and trading venues.
API Documentation:
- https://docs.pyth.network/price-feeds
- https://pyth.dourolabs.app/docs/?urls.primaryName=Hermes+API
"""
import asyncio
import logging
from typing import Any
import httpx
logger = logging.getLogger(__name__)
# Pyth Hermes API endpoint
PYTH_HERMES_BASE_URL = "https://hermes.pyth.network"
class PythPriceFeedProvider:
"""Pyth Network price feed provider"""
def __init__(self):
self.client = httpx.AsyncClient(timeout=30.0)
async def get_price_feeds_list(self) -> dict[str, Any]:
"""
Get the list of all available price feeds from Pyth Network.
Returns:
Dict containing price feeds metadata
"""
try:
response = await self.client.get(f"{PYTH_HERMES_BASE_URL}/v2/price_feeds")
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error fetching price feeds list: {e}")
return {}
async def get_latest_price_updates(self, price_feed_ids: list) -> dict[str, Any]:
"""
Get the latest price updates for specified price feed IDs.
Args:
price_feed_ids: List of price feed IDs to fetch
Returns:
Dict containing price updates
"""
if not price_feed_ids:
return {}
try:
# Build query parameters
params = []
for feed_id in price_feed_ids:
# Remove 0x prefix if present
clean_id = feed_id.replace("0x", "") if feed_id.startswith("0x") else feed_id
params.append(f"ids[]={clean_id}")
query_string = "&".join(params)
url = f"{PYTH_HERMES_BASE_URL}/v2/updates/price/latest?{query_string}"
logger.info(f"Fetching price updates from: {url}")
response = await self.client.get(url)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error fetching price updates: {e}")
return {}
async def get_single_price_feed(self, price_feed_id: str) -> dict[str, Any]:
"""
Get a single price feed by ID.
Args:
price_feed_id: Price feed ID
Returns:
Dict containing price feed data
"""
try:
# Remove 0x prefix if present
clean_id = (
price_feed_id.replace("0x", "") if price_feed_id.startswith("0x") else price_feed_id
)
url = f"{PYTH_HERMES_BASE_URL}/v2/updates/price/latest?ids[]={clean_id}"
response = await self.client.get(url)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error fetching single price feed: {e}")
return {}
async def parse_price_data(self, price_feed_id: str) -> dict[str, Any] | None:
"""
Parse price data for a specific feed into a standardized format.
Args:
price_feed_id: Price feed ID
Returns:
Dict with parsed price data or None if error
"""
try:
data = await self.get_single_price_feed(price_feed_id)
if not data or "parsed" not in data or not data["parsed"]:
logger.warning(f"No data returned for price feed {price_feed_id}")
return None
feed_data = data["parsed"][0]
# Extract price information
price_info = feed_data.get("price", {})
price = price_info.get("price")
conf = price_info.get("conf")
expo = price_info.get("expo")
publish_time = price_info.get("publish_time")
# Convert price to decimal format
if price and expo:
# Price is stored as integer with exponent
price_decimal = int(price) * (10**expo)
else:
price_decimal = None
return {
"id": feed_data.get("id"),
"price": price_decimal,
"price_raw": price,
"confidence": conf,
"exponent": expo,
"publish_time": publish_time,
"timestamp": publish_time,
}
except Exception as e:
logger.error(f"Error parsing price data for {price_feed_id}: {e}")
return None
async def close(self):
"""Close the HTTP client"""
await self.client.aclose()
# Provider functions for DataBus integration
async def _pyth_price_feed_list(**kwargs) -> dict[str, Any]:
"""Get list of all Pyth price feeds"""
provider = PythPriceFeedProvider()
try:
result = await provider.get_price_feeds_list()
# Return in the expected DataBus format
return {"source": "pyth", "data": result, "count": len(result) if result else 0}
finally:
await provider.close()
async def _pyth_latest_price_updates(price_feed_ids: list, **kwargs) -> dict[str, Any]:
"""Get latest price updates for specified feeds"""
provider = PythPriceFeedProvider()
try:
result = await provider.get_latest_price_updates(price_feed_ids)
return result
finally:
await provider.close()
async def _pyth_single_price_feed(price_feed_id: str, **kwargs) -> dict[str, Any]:
"""Get a single price feed by ID"""
provider = PythPriceFeedProvider()
try:
result = await provider.get_single_price_feed(price_feed_id)
return result
finally:
await provider.close()
async def _pyth_parsed_price(price_feed_id: str, **kwargs) -> dict[str, Any]:
"""Get parsed price data for a single feed"""
provider = PythPriceFeedProvider()
try:
result = await provider.parse_price_data(price_feed_id)
# Return in the expected DataBus format
return {"source": "pyth", "data": result} if result else None
finally:
await provider.close()
# Example usage functions
async def get_bitcoin_price_feed() -> dict[str, Any]:
"""
Get Bitcoin/USD price feed.
This is just an example - you would need to find the actual Pyth ID for BTC/USD
"""
# This is a placeholder - actual BTC/USD feed ID would need to be looked up
btc_usd_feed_id = "0xe62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b41"
return await _pyth_parsed_price(btc_usd_feed_id)
async def get_ethereum_price_feed() -> dict[str, Any]:
"""
Get Ethereum/USD price feed.
This is just an example - you would need to find the actual Pyth ID for ETH/USD
"""
# This is a placeholder - actual ETH/USD feed ID would need to be looked up
eth_usd_feed_id = "0xff61491a931112ddf1bd8147cd1b641375f79f5825126d665480874634fd0ace"
return await _pyth_parsed_price(eth_usd_feed_id)
if __name__ == "__main__":
# Test the provider
async def test_provider():
provider = PythPriceFeedProvider()
try:
# Test getting price feeds list
logger.info("Getting price feeds list...")
feeds = await provider.get_price_feeds_list()
logger.info(f"Found {len(feeds)} price feeds")
if feeds:
# Test getting a single feed (using the first one as example)
first_feed_id = feeds[0]["id"]
logger.info(f"\nGetting price feed for ID: {first_feed_id}")
price_data = await provider.parse_price_data(first_feed_id)
logger.info(f"Price data: {price_data}")
finally:
await provider.close()
# Run the test
asyncio.run(test_provider())
|