hcheng826 · November 10, 2025 03:10
diff --git a/compute_twap.py b/compute_twap.py
 #!/usr/bin/env python3
 """
 Script to compute TWAP (Time Weighted Average Price) from Pyth Network price data.
 Queries prices for Nov 7 0:00 UTC and 29 days before it, then calculates simple average.
 """

 import urllib.request
 import urllib.parse
 import urllib.error
 import json
 from datetime import datetime, timezone, timedelta
 from typing import Optional, List, Tuple

 PRICE_ID = ""
 BASE_URL = "https://benchmarks.pyth.network/v1/updates/price"


 def get_timestamp_for_date(date: datetime) -> int:
    """Convert a datetime to Unix timestamp."""
    return int(date.timestamp())


 def get_daily_timestamps(start_date: datetime, num_days: int) -> List[int]:
    """
    Generate timestamps for num_days days starting from start_date at 0:00 UTC.
    Returns list of Unix timestamps.
    """
    timestamps = []
    for i in range(num_days):
        # Go backwards: start_date, start_date - 1 day, start_date - 2 days, etc.
        target_date = start_date - timedelta(days=i)
        timestamp = get_timestamp_for_date(target_date)
        timestamps.append(timestamp)
    return timestamps


 def query_pyth_price(timestamp: int, price_id: str) -> Optional[float]:
    """
    Query Pyth Network API for price at a specific timestamp.
    Returns the price as a float, or None if query fails.
    """
    url = f"{BASE_URL}/{timestamp}"
    params = {
        "ids": price_id,
        "encoding": "hex",
        "parsed": "true"
    }

    # Build URL with query parameters
    query_string = urllib.parse.urlencode(params)
    full_url = f"{url}?{query_string}"

    try:
        # Create request with headers
        req = urllib.request.Request(full_url, headers={"accept": "application/json"})

        # Make request with timeout
        with urllib.request.urlopen(req, timeout=10) as response:
            if response.status != 200:
                print(f"Error: HTTP {response.status} for timestamp {timestamp}")
                return None

            data = json.loads(response.read().decode('utf-8'))

        # Extract price from parsed data
        if "parsed" in data and len(data["parsed"]) > 0:
            price_info = data["parsed"][0].get("price", {})
            price_str = price_info.get("price")
            expo = price_info.get("expo", 0)

            if price_str:
                # Convert price: price * 10^expo
                price = float(price_str) * (10 ** expo)
                return price

        print(f"Warning: No price data found for timestamp {timestamp}")
        return None

    except urllib.error.URLError as e:
        print(f"Error querying timestamp {timestamp}: {e}")
        return None
    except (KeyError, ValueError, TypeError, json.JSONDecodeError) as e:
        print(f"Error parsing response for timestamp {timestamp}: {e}")
        return None


 def compute_twap(price_id: str, start_date: datetime, num_days: int = 30) -> Tuple[float, List[float]]:
    """
    Compute TWAP (simple average) for the given price ID.

    Args:
        price_id: Pyth Network price ID
        start_date: Starting date (datetime object, should be at 0:00 UTC)
        num_days: Number of days to query (default 30)

    Returns:
        Tuple of (average_price, list_of_prices)
    """
    timestamps = get_daily_timestamps(start_date, num_days)
    prices = []

    print(f"Querying {num_days} days of price data starting from {start_date.strftime('%Y-%m-%d %H:%M:%S UTC')}...")
    print(f"Price ID: {price_id}\n")

    for i, timestamp in enumerate(timestamps):
        date = datetime.fromtimestamp(timestamp, tz=timezone.utc)
        print(f"Day {i+1}/{num_days}: Querying {date.strftime('%Y-%m-%d %H:%M:%S UTC')} (timestamp: {timestamp})...", end=" ")

        price = query_pyth_price(timestamp, price_id)

        if price is not None:
            prices.append(price)
            print(f"Price: {price}")
        else:
            print("Failed to get price")

    if not prices:
        raise ValueError("No prices were successfully retrieved")

    average_price = sum(prices) / len(prices)

    print(f"\n{'='*60}")
    print(f"Successfully retrieved {len(prices)}/{num_days} prices")
    print(f"TWAP (Simple Average): {average_price}")
    print(f"{'='*60}")

    return average_price, prices


 def main():
    """Main function to compute TWAP for Nov 7, 2025 0:00 UTC and 29 days before."""
    # Nov 7, 2025 0:00 UTC
    start_date = datetime(2025, 11, 7, 0, 0, 0, tzinfo=timezone.utc)

    # Compute TWAP
    try:
        twap, prices = compute_twap(PRICE_ID, start_date, num_days=30)

        print(f"\nDetailed Results:")
        print(f"Number of data points: {len(prices)}")
        print(f"TWAP: {twap}")
        print(f"Min price: {min(prices)}")
        print(f"Max price: {max(prices)}")

    except Exception as e:
        print(f"Error: {e}")
        return 1

    return 0


 if __name__ == "__main__":
    exit(main())
	#!/usr/bin/env python3
	"""
	Script to compute TWAP (Time Weighted Average Price) from Pyth Network price data.
	Queries prices for Nov 7 0:00 UTC and 29 days before it, then calculates simple average.
	"""

	import urllib.request
	import urllib.parse
	import urllib.error
	import json
	from datetime import datetime, timezone, timedelta
	from typing import Optional, List, Tuple

	PRICE_ID = ""
	BASE_URL = "https://benchmarks.pyth.network/v1/updates/price"


	def get_timestamp_for_date(date: datetime) -> int:
	"""Convert a datetime to Unix timestamp."""
	return int(date.timestamp())


	def get_daily_timestamps(start_date: datetime, num_days: int) -> List[int]:
	"""
	Generate timestamps for num_days days starting from start_date at 0:00 UTC.
	Returns list of Unix timestamps.
	"""
	timestamps = []
	for i in range(num_days):
	# Go backwards: start_date, start_date - 1 day, start_date - 2 days, etc.
	target_date = start_date - timedelta(days=i)
	timestamp = get_timestamp_for_date(target_date)
	timestamps.append(timestamp)
	return timestamps


	def query_pyth_price(timestamp: int, price_id: str) -> Optional[float]:
	"""
	Query Pyth Network API for price at a specific timestamp.
	Returns the price as a float, or None if query fails.
	"""
	url = f"{BASE_URL}/{timestamp}"
	params = {
	"ids": price_id,
	"encoding": "hex",
	"parsed": "true"
	}

	# Build URL with query parameters
	query_string = urllib.parse.urlencode(params)
	full_url = f"{url}?{query_string}"

	try:
	# Create request with headers
	req = urllib.request.Request(full_url, headers={"accept": "application/json"})

	# Make request with timeout
	with urllib.request.urlopen(req, timeout=10) as response:
	if response.status != 200:
	print(f"Error: HTTP {response.status} for timestamp {timestamp}")
	return None

	data = json.loads(response.read().decode('utf-8'))

	# Extract price from parsed data
	if "parsed" in data and len(data["parsed"]) > 0:
	price_info = data["parsed"][0].get("price", {})
	price_str = price_info.get("price")
	expo = price_info.get("expo", 0)

	if price_str:
	# Convert price: price * 10^expo
	price = float(price_str) * (10 ** expo)
	return price

	print(f"Warning: No price data found for timestamp {timestamp}")
	return None

	except urllib.error.URLError as e:
	print(f"Error querying timestamp {timestamp}: {e}")
	return None
	except (KeyError, ValueError, TypeError, json.JSONDecodeError) as e:
	print(f"Error parsing response for timestamp {timestamp}: {e}")
	return None


	def compute_twap(price_id: str, start_date: datetime, num_days: int = 30) -> Tuple[float, List[float]]:
	"""
	Compute TWAP (simple average) for the given price ID.

	Args:
	price_id: Pyth Network price ID
	start_date: Starting date (datetime object, should be at 0:00 UTC)
	num_days: Number of days to query (default 30)

	Returns:
	Tuple of (average_price, list_of_prices)
	"""
	timestamps = get_daily_timestamps(start_date, num_days)
	prices = []

	print(f"Querying {num_days} days of price data starting from {start_date.strftime('%Y-%m-%d %H:%M:%S UTC')}...")
	print(f"Price ID: {price_id}\n")

	for i, timestamp in enumerate(timestamps):
	date = datetime.fromtimestamp(timestamp, tz=timezone.utc)
	print(f"Day {i+1}/{num_days}: Querying {date.strftime('%Y-%m-%d %H:%M:%S UTC')} (timestamp: {timestamp})...", end=" ")

	price = query_pyth_price(timestamp, price_id)

	if price is not None:
	prices.append(price)
	print(f"Price: {price}")
	else:
	print("Failed to get price")

	if not prices:
	raise ValueError("No prices were successfully retrieved")

	average_price = sum(prices) / len(prices)

	print(f"\n{'='*60}")
	print(f"Successfully retrieved {len(prices)}/{num_days} prices")
	print(f"TWAP (Simple Average): {average_price}")
	print(f"{'='*60}")

	return average_price, prices


	def main():
	"""Main function to compute TWAP for Nov 7, 2025 0:00 UTC and 29 days before."""
	# Nov 7, 2025 0:00 UTC
	start_date = datetime(2025, 11, 7, 0, 0, 0, tzinfo=timezone.utc)

	# Compute TWAP
	try:
	twap, prices = compute_twap(PRICE_ID, start_date, num_days=30)

	print(f"\nDetailed Results:")
	print(f"Number of data points: {len(prices)}")
	print(f"TWAP: {twap}")
	print(f"Min price: {min(prices)}")
	print(f"Max price: {max(prices)}")

	except Exception as e:
	print(f"Error: {e}")
	return 1

	return 0


	if __name__ == "__main__":
	exit(main())
No results found