""" PolyMarket API Integration Handles interactions with Polymarket's Gamma API and CLOB API """ import requests from typing import Dict, Optional, List import re class PolymarketAPI: """Class to handle Polymarket API operations""" GAMMA_API_BASE = "https://gamma-api.polymarket.com" CLOB_API_BASE = "https://clob.polymarket.com" def __init__(self): pass def extract_slug_from_url(self, url: str) -> Optional[str]: """ Extract the slug from a Polymarket URL Args: url: Polymarket URL (e.g., https://polymarket.com/event/will-the-supreme-court-rule-in-favor-of-trumps-tariffs) Returns: The slug string or None if not found """ # Handle different URL patterns # Pattern 1: https://polymarket.com/event/slug-here # Pattern 2: https://polymarket.com/market/slug-here patterns = [ r'polymarket\.com/event/([a-z0-9\-]+)', r'polymarket\.com/market/([a-z0-9\-]+)', r'polymarket\.com/([a-z0-9\-]+)', ] for pattern in patterns: match = re.search(pattern, url.lower()) if match: return match.group(1) return None def get_market_info(self, url: str) -> Optional[Dict]: """ Get market information from Polymarket using the URL Args: url: Polymarket URL Returns: Dictionary containing market information including: - id: market id - question: market question - conditionId: condition id - clobTokenIds: array of token ids """ slug = self.extract_slug_from_url(url) if not slug: raise ValueError(f"Could not extract slug from URL: {url}") # Call Gamma API with the slug endpoint = f"{self.GAMMA_API_BASE}/markets" params = {"slug": slug} try: response = requests.get(endpoint, params=params, timeout=10) response.raise_for_status() data = response.json() # The API returns a singleton array if isinstance(data, list) and len(data) > 0: market_info = data[0] # Validate required fields required_fields = ['id', 'question', 'conditionId', 'clobTokenIds'] for field in required_fields: if field not in market_info: raise ValueError(f"Missing required field: {field}") return market_info else: raise ValueError(f"No market found for slug: {slug}") except requests.exceptions.RequestException as e: raise Exception(f"Error calling Gamma API: {str(e)}") def get_first_token_id(self, market_info: Dict) -> str: """ Extract the first token ID from market information Args: market_info: Dictionary containing market information Returns: The first clobTokenId as a string """ clob_token_ids = market_info.get('clobTokenIds', []) if isinstance(clob_token_ids, str): # Sometimes it might be a string representation of an array import json try: clob_token_ids = json.loads(clob_token_ids) except: pass if not clob_token_ids or len(clob_token_ids) == 0: raise ValueError("No clobTokenIds found in market info") return str(clob_token_ids[0]) def get_price_history( self, token_id: str, start_unix: int, end_unix: int, fidelity: int = 1 ) -> List[Dict]: """ Get price history from CLOB API If the requested time period is longer than 5 days, this method automatically breaks it into chunks of up to 5 days and combines the results. Args: token_id: The clobTokenId to fetch prices for start_unix: Start time as Unix timestamp end_unix: End time as Unix timestamp fidelity: Time granularity in minutes (default: 1 minute) Returns: List of price points, each containing: - t: Unix timestamp - p: Price (0-1 probability) Example: [ {"t": 1763121606, "p": 0.26}, {"t": 1763121664, "p": 0.26}, ... ] """ # Maximum interval is 5 days (in seconds) MAX_INTERVAL_SECONDS = 5 * 24 * 60 * 60 # 432000 seconds # Calculate the time difference time_diff = end_unix - start_unix # If the interval is within the limit, make a single API call if time_diff <= MAX_INTERVAL_SECONDS: return self._fetch_price_history_chunk(token_id, start_unix, end_unix, fidelity) # Otherwise, break it into chunks all_history = [] current_start = start_unix while current_start < end_unix: # Calculate chunk end (current_start + 5 days, or end_unix if smaller) chunk_end = min(current_start + MAX_INTERVAL_SECONDS, end_unix) # Fetch this chunk chunk_history = self._fetch_price_history_chunk( token_id, current_start, chunk_end, fidelity ) # Add to combined results all_history.extend(chunk_history) # Move to next chunk (start from the end of this chunk) current_start = chunk_end # Sort by timestamp and remove duplicates # Use a set to track seen timestamps to avoid duplicates seen_timestamps = set() unique_history = [] for point in sorted(all_history, key=lambda x: x['t']): if point['t'] not in seen_timestamps: seen_timestamps.add(point['t']) unique_history.append(point) return unique_history def _fetch_price_history_chunk( self, token_id: str, start_unix: int, end_unix: int, fidelity: int ) -> List[Dict]: """ Internal method to fetch a single chunk of price history from CLOB API Args: token_id: The clobTokenId to fetch prices for start_unix: Start time as Unix timestamp end_unix: End time as Unix timestamp fidelity: Time granularity in minutes Returns: List of price points """ endpoint = f"{self.CLOB_API_BASE}/prices-history" params = { "market": token_id, "startTs": start_unix, "endTs": end_unix, "fidelity": fidelity } try: response = requests.get(endpoint, params=params, timeout=30) response.raise_for_status() data = response.json() # Extract history array if 'history' not in data: raise ValueError("No 'history' field in API response") history = data['history'] if not isinstance(history, list): raise ValueError("'history' field is not a list") # Validate data points for point in history: if 't' not in point or 'p' not in point: raise ValueError("Invalid data point format in history") return history except requests.exceptions.RequestException as e: raise Exception(f"Error calling CLOB API: {str(e)}") def reconstruct_timeseries( self, history: List[Dict], start_unix: int, end_unix: int, granularity_minutes: int = 1 ) -> List[Dict]: """ Reconstruct a complete time series from sparse price history data The API may return sparse data points. This method fills in gaps by forward-filling prices (last known price carries forward). Args: history: List of price points from get_price_history() start_unix: Start time as Unix timestamp end_unix: End time as Unix timestamp granularity_minutes: Desired time granularity in minutes Returns: List of complete time series data points: [ {"timestamp": 1763121600, "price": 0.26, "datetime": "..."}, {"timestamp": 1763121660, "price": 0.26, "datetime": "..."}, ... ] """ if not history: return [] # Sort history by timestamp history = sorted(history, key=lambda x: x['t']) # Initialize result timeseries = [] # Convert granularity to seconds granularity_seconds = granularity_minutes * 60 # Current price (forward fill) current_price = None history_idx = 0 # Iterate through each time point current_time = start_unix while current_time <= end_unix: # Update current price if we have new data points while history_idx < len(history) and history[history_idx]['t'] <= current_time: current_price = history[history_idx]['p'] history_idx += 1 # Add data point if we have a price if current_price is not None: timeseries.append({ "timestamp": current_time, "price": current_price }) # Move to next time point current_time += granularity_seconds return timeseries # Example usage if __name__ == "__main__": api = PolymarketAPI() # Test with example URL test_url = "https://polymarket.com/event/will-the-supreme-court-rule-in-favor-of-trumps-tariffs" try: print(f"Testing with URL: {test_url}") market_info = api.get_market_info(test_url) print(f"\nMarket Info:") print(f" ID: {market_info['id']}") print(f" Question: {market_info['question']}") print(f" Condition ID: {market_info['conditionId']}") token_id = api.get_first_token_id(market_info) print(f"\nFirst Token ID: {token_id}") # Test price history (example timestamps) # This is just for demonstration - use real timestamps when testing print("\nTo test price history, use:") print(f" history = api.get_price_history('{token_id}', start_unix, end_unix, fidelity=1)") print(f" timeseries = api.reconstruct_timeseries(history, start_unix, end_unix, 1)") except Exception as e: print(f"Error: {str(e)}")