diff --git a/README.md b/README.md index cc81aa1..e7f0d95 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,27 @@ +# Cross-Platform Market Matcher + +This module allows you to compare and log markets, events, and profiles between Kalshi and Polymarket using their public APIs. + +### How to Run + +**One-time scan:** +```bash +python cross_platform_runner.py --once --category Politics +``` + +**Continuous monitoring (every 60 seconds):** +```bash +python cross_platform_runner.py --watch 60 --category Politics +``` + +You can change the category (e.g., "Economy", "Crypto", "Sports") as needed. + +**What it does:** +- Fetches all Kalshi series for the given category +- For each series, searches Polymarket for related markets, events, and profiles using the series title +- Logs all checked markets, events, and profiles, including reasons for non-matches + +This is useful for research, debugging, and monitoring cross-platform market overlap. # Arbitrage Bot Automated arbitrage detection bot for prediction markets (Polymarket & Kalshi) made by Jasper Buffet, Mark Burry and Duco Munger. diff --git a/config.yaml b/config.yaml index ce5ee7d..64cc1fc 100644 --- a/config.yaml +++ b/config.yaml @@ -1,17 +1,17 @@ apis: polymarket: - base_url: "https://gamma-api.polymarket.com" + base_url: "https://matic-poly.market" timeout: 30 kalshi: - base_url: "https://api.elections.kalshi.com" + base_url: "https://trading-api.kalshi.com" api_key_id: "${KALSHI_API_KEY_ID}" private_key: "${KALSHI_PRIVATE_KEY}" timeout: 30 matching: min_confidence: 0.65 - use_semantic: true + use_semantic: false semantic_model: "all-MiniLM-L6-v2" arbitrage: diff --git a/cross_platform_runner.py b/cross_platform_runner.py new file mode 100644 index 0000000..5b8b1c7 --- /dev/null +++ b/cross_platform_runner.py @@ -0,0 +1,137 @@ +import argparse +import asyncio +import time +import aiohttp +from src.matching.cross_platform_matcher import cross_platform_match + +KALSHI_URL = "https://api.elections.kalshi.com" +POLYMARKET_URL = "https://gamma-api.polymarket.com" + +async def run_once(category): + print(f"Fetching Kalshi series for category '{category}'...") + from src.matching.cross_platform_matcher import fetch_kalshi_series, search_polymarket + kalshi_series = await fetch_kalshi_series(KALSHI_URL) + checked = 0 + matched = 0 + cross_reference = [] + for series in kalshi_series: + if category and series.get("category", "").lower() != category.lower(): + continue + title = series.get("title", "") + if not title: + print(f"Skipping Kalshi series (no title): {series}") + continue + print(f"Checking Kalshi series: {series.get('ticker','')} | {title} | {series.get('category','')}") + async def full_search(base_url, query): + url = f"{base_url}/public-search?q={query}" + async with aiohttp.ClientSession() as session: + async with session.get(url) as resp: + resp.raise_for_status() + return await resp.json() + + try: + search_results = await full_search(POLYMARKET_URL, title) + except aiohttp.ClientResponseError as e: + print(f" Polymarket search failed for title '{title}': {e.status} {e.message}") + checked += 1 + print(f" No Polymarket markets, events, or profiles for title '{title}' (API error)") + continue + except Exception as e: + print(f" Polymarket search failed for title '{title}': {e}") + checked += 1 + print(f" No Polymarket markets, events, or profiles for title '{title}' (API error)") + continue + checked += 1 + found_any = False + for key in ["markets", "events", "profiles"]: + items = search_results.get(key, []) + if items: + found_any = True + print(f" Found {len(items)} Polymarket {key} for title '{title}':") + for item in items: + if key == "markets": + print(f" Market: {item.get('id','')} | {item.get('question','')}" ) + elif key == "events": + event_id = item.get('id','') + event_title = item.get('name','') + ticker = None + event_data = None + market_list = [] + if event_id: + event_url = f"https://gamma-api.polymarket.com/events/{event_id}" + async with aiohttp.ClientSession() as session: + try: + async with session.get(event_url) as event_resp: + event_resp.raise_for_status() + event_data = await event_resp.json() + ticker = event_data.get('ticker') or event_data.get('name') + market_list = event_data.get('markets', []) + except Exception as e: + print(f" Error fetching event details for {event_id}: {e}") + print(f" Event: {event_id} | {ticker if ticker else event_title}") + open_markets = [m for m in market_list if m.get('active', True) and not m.get('closed', False)] + print(f" Open markets for event {event_id}: {len(open_markets)}") + for market in open_markets: + market_id = market.get('id', '') + question = market.get('question', '') + volume = market.get('volume', market.get('volumeNum', 0)) + outcomes = market.get('outcomes', []) + outcome_prices = market.get('outcomePrices', []) + last_volume = market.get('volume24hr', 0) + # Parse outcomes and prices if they are JSON strings + import json as _json + if isinstance(outcomes, str): + try: + outcomes = _json.loads(outcomes) + except: + pass + if isinstance(outcome_prices, str): + try: + outcome_prices = _json.loads(outcome_prices) + except: + pass + print(f" Market: {market_id} | {question}") + print(f" Volume: {volume}") + print(f" Last 24hr Volume: {last_volume}") + print(f" Outcomes: {outcomes}") + print(f" Prices: {outcome_prices}") + # Save standardized info for cross referencing + cross_reference.append({ + "kalshi_series": series, + "polymarket_event": event_data if event_data else item, + "polymarket_markets": open_markets + }) + elif key == "profiles": + print(f" Profile: {item.get('id','')} | {item.get('name','')}") + if found_any: + matched += 1 + else: + print(f" No Polymarket markets, events, or profiles for title '{title}'") + # Output standardized cross-reference data + import json + print("\nStandardized cross-reference data:") + print(json.dumps(cross_reference, indent=2, default=str)) + print(f"Checked {checked} Kalshi series, {matched} had Polymarket matches.") + +async def run_watch(category, interval): + while True: + await run_once(category) + print(f"Waiting {interval} seconds before next check...") + await asyncio.sleep(interval) + +def main(): + parser = argparse.ArgumentParser(description="Cross-platform market matcher") + parser.add_argument("--category", type=str, default="Politics", help="Market category to check") + parser.add_argument("--once", action="store_true", help="Run one-time market scan") + parser.add_argument("--watch", type=int, metavar="SECONDS", help="Continuously check every N seconds") + args = parser.parse_args() + + if args.once: + asyncio.run(run_once(args.category)) + elif args.watch: + asyncio.run(run_watch(args.category, args.watch)) + else: + print("Specify --once for one-time scan or --watch N for continuous mode.") + +if __name__ == "__main__": + main() diff --git a/src/api/kalshi.py b/src/api/kalshi.py index fc7ea74..d80e3ca 100644 --- a/src/api/kalshi.py +++ b/src/api/kalshi.py @@ -1,13 +1,24 @@ -from typing import List, Optional +from typing import List, Optional, Dict from datetime import datetime -from .base import BaseAPIClient, Market -from ..utils.logger import logger +import aiohttp +import asyncio import time import jwt from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend +from .base import BaseAPIClient, Market +from ..utils.logger import logger +from ..core import cache class KalshiClient(BaseAPIClient): + # Map our standard categories to Kalshi series tags + CATEGORY_MAP = { + "politics": ["POLITICS", "ELECTORAL"], + "economy": ["ECONOMIC", "FED", "MACRO"], + "crypto": ["CRYPTO"], + "sports": ["SPORTS", "NFL", "NBA", "MLB", "NHL"], + } + def __init__(self, base_url: str, timeout: int = 30, api_key_id: Optional[str] = None, private_key_str: Optional[str] = None): @@ -17,12 +28,21 @@ def __init__(self, base_url: str, timeout: int = 30, self.token = None self.token_expiry = 0 + async def _make_async_request(self, session: aiohttp.ClientSession, + method: str, endpoint: str, **kwargs) -> Dict: + """Make an async HTTP request.""" + url = f"{self.base_url}{endpoint}" + + try: + async with session.request(method, url, **kwargs) as response: + response.raise_for_status() + return await response.json() + except Exception as e: + logger.error(f"Async request failed for {url}: {e}") + raise + def _ensure_authenticated(self) -> None: - """ - Attempt to authenticate with Kalshi API. - Note: Authentication is NOT required for read-only market data access. - It's only needed for trading operations (placing orders, etc). - """ + """Ensure we have a valid JWT token.""" if self.token and time.time() < self.token_expiry: return @@ -31,15 +51,13 @@ def _ensure_authenticated(self) -> None: return try: - # Generate JWT token signed with private key current_time = int(time.time()) payload = { 'iss': self.api_key_id, - 'exp': current_time + 1800, # 30 minutes expiry + 'exp': current_time + 1800, 'iat': current_time } - # Load the private key (replace literal \n with actual newlines) private_key_pem = self.private_key_str.replace('\\n', '\n') private_key = serialization.load_pem_private_key( private_key_pem.encode(), @@ -47,52 +65,102 @@ def _ensure_authenticated(self) -> None: backend=default_backend() ) - # Sign the JWT signed_jwt = jwt.encode(payload, private_key, algorithm='RS256') - - # Set JWT as authorization header for authenticated requests - # Note: Kalshi may not have a login endpoint, JWT is used directly - self.session.headers.update({"Authorization": f"Bearer {signed_jwt}"}) self.token = signed_jwt self.token_expiry = current_time + 1800 - logger.info("Kalshi API authentication configured (JWT Bearer token)") + + logger.info("Kalshi API authentication configured") + except Exception as e: - logger.debug(f"Kalshi authentication setup failed (not required for read-only): {e}") + logger.error(f"Kalshi authentication failed: {e}") + raise - def get_markets(self, limit: int = 200, status: str = "open") -> List[Market]: - cache_key = f"kalshi_markets_{limit}_{status}" - cached = self._get_cached(cache_key) + async def get_series(self) -> List[Dict]: + """Fetch all market series from Kalshi.""" + cache_key = "kalshi_series" + cached = cache.load_data(cache_key, max_age_hours=24) if cached: return cached + self._ensure_authenticated() + headers = {"Authorization": f"Bearer {self.token}"} if self.token else {} + + async with aiohttp.ClientSession(headers=headers) as session: + data = await self._make_async_request( + session, "GET", "/trading-api/v2/series" + ) + series = data.get("series", []) + cache.save_data(cache_key, series) + return series + + async def get_markets_by_category(self, category: str, + limit: int = 200) -> List[Market]: + """ + Fetch markets for a specific category. + + Args: + category: Category to fetch markets for (politics, economy, crypto, sports) + limit: Maximum number of markets to fetch + """ + cache_key = f"kalshi_markets_{category}_{limit}" + cached = cache.load_data(cache_key, max_age_hours=1) + if cached: + return [Market(**m) for m in cached] + try: self._ensure_authenticated() + headers = {"Authorization": f"Bearer {self.token}"} if self.token else {} - endpoint = "/trade-api/v2/markets" - params = { - "limit": limit, - "status": status - } + # Get relevant series tags for category + category_tags = self.CATEGORY_MAP.get(category.lower(), []) + if not category_tags: + logger.warning(f"Unknown category: {category}") + return [] - data = self._make_request("GET", endpoint, params=params) - market_data = data.get("markets", []) + # Get all series first + series = await self.get_series() + relevant_series = [ + s for s in series + if any(tag in s.get("tags", []) for tag in category_tags) + ] markets = [] - for item in market_data: - try: - market = self._parse_market(item) - if market: - markets.append(market) - except Exception as e: - logger.warning(f"Failed to parse Kalshi market: {e}") - continue - - self._set_cache(cache_key, markets) - logger.info(f"Fetched {len(markets)} markets from Kalshi") + async with aiohttp.ClientSession(headers=headers) as session: + for series_data in relevant_series: + series_id = series_data["id"] + + # Fetch markets in this series + endpoint = f"/trading-api/v2/markets" + params = { + "series_id": series_id, + "status": "active", + "limit": limit + } + + try: + data = await self._make_async_request( + session, "GET", endpoint, params=params + ) + + for item in data.get("markets", []): + market = self._parse_market(item) + if market: + markets.append(market) + + if len(markets) >= limit: + break + + except Exception as e: + logger.error(f"Failed to fetch markets for series {series_id}: {e}") + continue + + # Cache the results + cache.save_data(cache_key, [m.__dict__ for m in markets]) + logger.info(f"Fetched {len(markets)} {category} markets from Kalshi") return markets except Exception as e: - logger.error(f"Failed to fetch Kalshi markets: {e}") + logger.error(f"Failed to fetch {category} markets from Kalshi: {e}") return [] def _parse_market(self, data: dict) -> Optional[Market]: diff --git a/src/api/polymarket.py b/src/api/polymarket.py index 7d77326..f0e7504 100644 --- a/src/api/polymarket.py +++ b/src/api/polymarket.py @@ -1,44 +1,131 @@ -from typing import List, Optional +from typing import List, Optional, Dict from datetime import datetime +import aiohttp +import json from .base import BaseAPIClient, Market from ..utils.logger import logger +from ..core import cache class PolymarketClient(BaseAPIClient): + # Map our standard categories to Polymarket tags/keywords + CATEGORY_MAP = { + "politics": ["election", "politics", "president", "congress", "senate"], + "economy": ["gdp", "cpi", "inflation", "rate", "fed", "recession"], + "crypto": ["bitcoin", "ethereum", "crypto", "btc", "eth"], + "sports": ["nfl", "nba", "mlb", "nhl", "sports", "football", "basketball"], + } + def __init__(self, base_url: str, timeout: int = 30): super().__init__(base_url, timeout) + self.graphql_url = f"{base_url}/graphql" - def get_markets(self, limit: int = 100, active_only: bool = True) -> List[Market]: - cache_key = f"polymarket_markets_{limit}_{active_only}" - cached = self._get_cached(cache_key) - if cached: - return cached + async def _make_async_request(self, session: aiohttp.ClientSession, + method: str, endpoint: str, **kwargs) -> Dict: + """Make an async HTTP request.""" + url = f"{self.base_url}{endpoint}" + + try: + async with session.request(method, url, **kwargs) as response: + response.raise_for_status() + return await response.json() + except Exception as e: + logger.error(f"Async request failed for {url}: {e}") + raise + async def _make_graphql_request(self, session: aiohttp.ClientSession, + query: str, variables: Dict = None) -> Dict: + """Make an async GraphQL request.""" try: - endpoint = "/markets" - params = { - "limit": limit, - "closed": "false" if active_only else None + payload = { + "query": query, + "variables": variables or {} } - params = {k: v for k, v in params.items() if v is not None} + + async with session.post(self.graphql_url, json=payload) as response: + response.raise_for_status() + return await response.json() + except Exception as e: + logger.error(f"GraphQL request failed: {e}") + raise + + async def get_markets_by_category(self, category: str, + limit: int = 100) -> List[Market]: + """ + Fetch markets for a specific category using the GraphQL API. + + Args: + category: Category to fetch markets for (politics, economy, crypto, sports) + limit: Maximum number of markets to fetch + """ + cache_key = f"polymarket_markets_{category}_{limit}" + cached = cache.load_data(cache_key, max_age_hours=1) + if cached: + return [Market(**m) for m in cached] + + try: + keywords = self.CATEGORY_MAP.get(category.lower(), []) + if not keywords: + logger.warning(f"Unknown category: {category}") + return [] - data = self._make_request("GET", endpoint, params=params) + # GraphQL query for markets with category filtering + query = """ + query GetMarkets($keyword: String!) { + markets( + where: { + question_contains_nocase: $keyword, + closed: false + } + first: 100, + orderBy: liquidityNum, + orderDirection: desc + ) { + id + question + outcomePrices + liquidity + volumeNum + endDate + } + } + """ markets = [] - for item in data: - try: - market = self._parse_market(item) - if market: - markets.append(market) - except Exception as e: - logger.warning(f"Failed to parse Polymarket market: {e}") - continue - - self._set_cache(cache_key, markets) - logger.info(f"Fetched {len(markets)} markets from Polymarket") + async with aiohttp.ClientSession() as session: + # Query for each keyword in the category + for keyword in keywords: + try: + result = await self._make_graphql_request( + session, + query, + variables={"keyword": keyword} + ) + + if result.get("errors"): + logger.error(f"GraphQL error: {result['errors']}") + continue + + market_data = result.get("data", {}).get("markets", []) + for item in market_data: + market = self._parse_market(item) + if market: + markets.append(market) + + except Exception as e: + logger.error(f"Failed to fetch markets for keyword '{keyword}': {e}") + continue + + # Remove duplicates based on market_id + unique_markets = list({m.market_id: m for m in markets}.values()) + markets = sorted(unique_markets, key=lambda x: x.liquidity or 0, reverse=True)[:limit] + + # Cache the results + cache.save_data(cache_key, [m.__dict__ for m in markets]) + logger.info(f"Fetched {len(markets)} {category} markets from Polymarket") return markets except Exception as e: - logger.error(f"Failed to fetch Polymarket markets: {e}") + logger.error(f"Failed to fetch {category} markets from Polymarket: {e}") return [] def _parse_market(self, data: dict) -> Optional[Market]: diff --git a/src/core/__init__.py b/src/core/__init__.py new file mode 100644 index 0000000..9967b0d --- /dev/null +++ b/src/core/__init__.py @@ -0,0 +1,8 @@ +""" +Core module initialization. +""" + +from .caching import MarketCache, cache +from .knowledge import knowledge_base + +__all__ = ['MarketCache', 'cache', 'knowledge_base'] \ No newline at end of file diff --git a/src/core/caching.py b/src/core/caching.py new file mode 100644 index 0000000..ffa6091 --- /dev/null +++ b/src/core/caching.py @@ -0,0 +1,115 @@ +""" +Persistent caching system for market data and related information. +Handles saving and loading of market data to/from JSON files. +""" + +import os +import json +from typing import Dict, List, Any +from datetime import datetime +from pathlib import Path +from ..utils.logger import logger + +class MarketCache: + def __init__(self, cache_dir: str = "data/cache"): + """ + Initialize the market cache with a specified cache directory. + + Args: + cache_dir: Directory to store cache files + """ + self.cache_dir = Path(cache_dir) + self.cache_dir.mkdir(parents=True, exist_ok=True) + + def _get_cache_path(self, filename: str) -> Path: + """Get the full path for a cache file.""" + return self.cache_dir / filename + + def save_data(self, filename: str, data: Any) -> None: + """ + Save data to a JSON cache file. + + Args: + filename: Name of the cache file + data: Data to cache (must be JSON serializable) + """ + try: + cache_path = self._get_cache_path(filename) + + # Ensure the cache directory exists + cache_path.parent.mkdir(parents=True, exist_ok=True) + + # Add metadata to cached data + cache_data = { + "timestamp": datetime.utcnow().isoformat(), + "data": data + } + + with cache_path.open('w') as f: + json.dump(cache_data, f, indent=2) + + logger.debug(f"Saved cache file: {filename}") + + except Exception as e: + logger.error(f"Failed to save cache file {filename}: {e}") + raise + + def load_data(self, filename: str, max_age_hours: float = 24.0) -> Any: + """ + Load data from a JSON cache file if it exists and is not too old. + + Args: + filename: Name of the cache file + max_age_hours: Maximum age of cache in hours (default: 24) + + Returns: + Cached data if valid, None otherwise + """ + try: + cache_path = self._get_cache_path(filename) + + if not cache_path.exists(): + return None + + with cache_path.open('r') as f: + cache_data = json.load(f) + + # Check cache age + timestamp = datetime.fromisoformat(cache_data["timestamp"]) + age = (datetime.utcnow() - timestamp).total_seconds() / 3600 + + if age > max_age_hours: + logger.debug(f"Cache file {filename} is too old ({age:.1f} hours)") + return None + + return cache_data["data"] + + except Exception as e: + logger.error(f"Failed to load cache file {filename}: {e}") + return None + + def clear_cache(self, filename: str = None) -> None: + """ + Clear specific cache file or all cache files. + + Args: + filename: Specific file to clear, or None to clear all + """ + try: + if filename: + cache_path = self._get_cache_path(filename) + if cache_path.exists(): + cache_path.unlink() + logger.debug(f"Cleared cache file: {filename}") + else: + # Clear all .json files in cache directory + for cache_file in self.cache_dir.glob("*.json"): + cache_file.unlink() + logger.debug("Cleared all cache files") + + except Exception as e: + logger.error(f"Failed to clear cache: {e}") + raise + +# Initialize global cache instance +cache = MarketCache() \ No newline at end of file diff --git a/src/core/knowledge.py b/src/core/knowledge.py new file mode 100644 index 0000000..36cdde6 --- /dev/null +++ b/src/core/knowledge.py @@ -0,0 +1,152 @@ +""" +Semantic embeddings storage and management for market questions. +Uses sentence-transformers for generating embeddings. +""" + +import numpy as np +from typing import List, Dict, Any +from pathlib import Path +import json +from sentence_transformers import SentenceTransformer +from ..api.base import Market +from .caching import cache +from ..utils.logger import logger + +class KnowledgeBase: + def __init__(self, model_name: str = 'all-MiniLM-L6-v2'): + """ + Initialize the knowledge base with a specified sentence transformer model. + + Args: + model_name: Name of the sentence-transformers model to use + """ + self.model_name = model_name + self._model = None # Lazy load model + + def _ensure_model_loaded(self): + """Ensure the transformer model is loaded.""" + if self._model is None: + try: + self._model = SentenceTransformer(self.model_name) + logger.info(f"Loaded semantic embedding model: {self.model_name}") + except Exception as e: + logger.error(f"Failed to load semantic model: {e}") + raise + + def generate_embedding(self, text: str) -> np.ndarray: + """ + Generate embedding for a single piece of text. + + Args: + text: Text to generate embedding for + + Returns: + Numpy array containing the embedding + """ + self._ensure_model_loaded() + return self._model.encode([text])[0] + + def generate_embeddings(self, texts: List[str]) -> np.ndarray: + """ + Generate embeddings for multiple texts. + + Args: + texts: List of texts to generate embeddings for + + Returns: + Numpy array containing embeddings + """ + self._ensure_model_loaded() + return self._model.encode(texts) + + def update_embeddings(self, markets: List[Market]) -> None: + """ + Update embeddings cache with new market data. + + Args: + markets: List of Market objects to update embeddings for + """ + try: + # Load existing embeddings if any + existing = cache.load_data("embeddings_cache.json") or {} + + # Convert existing embeddings from lists back to numpy arrays + existing_embeddings = { + k: np.array(v) for k, v in existing.items() + if isinstance(v, list) + } + + # Get questions that need new embeddings + new_questions = [ + market.question for market in markets + if market.market_id not in existing_embeddings + ] + + if new_questions: + logger.info(f"Generating embeddings for {len(new_questions)} new markets") + new_embeddings = self.generate_embeddings(new_questions) + + # Add new embeddings + for question, embedding in zip(new_questions, new_embeddings): + existing_embeddings[question] = embedding + + # Save updated embeddings (convert numpy arrays to lists) + cache_data = { + k: v.tolist() for k, v in existing_embeddings.items() + } + cache.save_data("embeddings_cache.json", cache_data) + + except Exception as e: + logger.error(f"Failed to update embeddings: {e}") + raise + + def find_similar_markets(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]: + """ + Find markets with questions similar to the query. + + Args: + query: Query text to find similar markets for + top_k: Number of similar markets to return + + Returns: + List of dicts containing market info and similarity scores + """ + try: + # Load cached embeddings + cached = cache.load_data("embeddings_cache.json") + if not cached: + return [] + + # Convert cached embeddings back to numpy arrays + embeddings_dict = { + k: np.array(v) for k, v in cached.items() + if isinstance(v, list) + } + + if not embeddings_dict: + return [] + + # Generate query embedding + query_embedding = self.generate_embedding(query) + + # Calculate similarities + similarities = [] + for question, embedding in embeddings_dict.items(): + similarity = np.dot(query_embedding, embedding) / ( + np.linalg.norm(query_embedding) * np.linalg.norm(embedding) + ) + similarities.append({ + "question": question, + "similarity": float(similarity) + }) + + # Sort by similarity and return top-k + similarities.sort(key=lambda x: x["similarity"], reverse=True) + return similarities[:top_k] + + except Exception as e: + logger.error(f"Failed to find similar markets: {e}") + return [] + +# Initialize global knowledge base instance +knowledge_base = KnowledgeBase() \ No newline at end of file diff --git a/src/main.py b/src/main.py index 500251e..03441f1 100644 --- a/src/main.py +++ b/src/main.py @@ -1,45 +1,198 @@ +import asyncio import time import argparse -from .arbitrage import detector +from typing import List, Dict +from datetime import datetime +from pathlib import Path +from .api import KalshiClient, PolymarketClient, Market from .utils import config, logger, notifier +from .core import cache, knowledge_base +from .arbitrage import detector -def run_once(): - logger.info("Starting arbitrage detection run...") +CATEGORIES = ["politics", "economy", "crypto", "sports"] +DATA_DIR = Path("data") - opportunities = detector.detect_opportunities() +async def fetch_markets_for_category( + kalshi: KalshiClient, + polymarket: PolymarketClient, + category: str, + limit: int = 100 +) -> List[Market]: + """ + Fetch markets from both platforms for a specific category. + + Args: + kalshi: Kalshi API client + polymarket: Polymarket API client + category: Category to fetch markets for + limit: Maximum markets per platform + """ + try: + # Fetch from both platforms concurrently + kalshi_markets, poly_markets = await asyncio.gather( + kalshi.get_markets_by_category(category, limit), + polymarket.get_markets_by_category(category, limit) + ) + + return kalshi_markets + poly_markets + + except Exception as e: + logger.error(f"Failed to fetch markets for category {category}: {e}") + return [] + +async def update_all_markets() -> Dict[str, List[Market]]: + """ + Fetch and cache markets from all platforms, organized by category. + Updates semantic embeddings for all markets. + """ + try: + # Initialize API clients + kalshi = KalshiClient( + base_url=config.get("apis", "kalshi", "base_url"), + timeout=config.get("apis", "kalshi", "timeout"), + api_key_id=config.get("apis", "kalshi", "api_key_id"), + private_key_str=config.get("apis", "kalshi", "private_key") + ) + + polymarket = PolymarketClient( + base_url=config.get("apis", "polymarket", "base_url"), + timeout=config.get("apis", "polymarket", "timeout") + ) + + # Ensure data directory exists + DATA_DIR.mkdir(exist_ok=True) + + markets_by_category = {} + all_markets = [] + + # Fetch markets for each category concurrently + for category in CATEGORIES: + markets = await fetch_markets_for_category( + kalshi, polymarket, category + ) + markets_by_category[category] = markets + all_markets.extend(markets) + + # Cache category markets + cache.save_data( + f"{category}_markets.json", + [m.__dict__ for m in markets] + ) + + logger.info( + f"Cached {len(markets)} {category} markets " + f"({len([m for m in markets if m.platform == 'kalshi'])} Kalshi, " + f"{len([m for m in markets if m.platform == 'polymarket'])} Polymarket)" + ) + + # Cache combined markets + cache.save_data( + "combined_markets.json", + [m.__dict__ for m in all_markets] + ) + + # Update semantic embeddings + knowledge_base.update_embeddings(all_markets) + logger.info(f"Updated semantic embeddings for {len(all_markets)} markets") + + return markets_by_category + + except Exception as e: + logger.error(f"Failed to update markets: {e}") + return {} +def get_cached_markets(category: str = None) -> List[Market]: + """ + Get cached markets, optionally filtered by category. + + Args: + category: Optional category to filter by + """ + try: + if category: + cached = cache.load_data(f"{category}_markets.json") + if cached: + return [Market(**m) for m in cached] + else: + cached = cache.load_data("combined_markets.json") + if cached: + return [Market(**m) for m in cached] + + return [] + + except Exception as e: + logger.error(f"Failed to load cached markets: {e}") + return [] + +async def run_once(): + """Run one iteration of market updates and arbitrage detection.""" + logger.info("Starting market data update and arbitrage detection...") + + start_time = datetime.now() + markets = await update_all_markets() + update_duration = (datetime.now() - start_time).total_seconds() + + total_markets = sum(len(m) for m in markets.values()) + logger.info(f"Market update complete - Fetched {total_markets} markets in {update_duration:.1f}s") + + opportunities = detector.detect_opportunities() + if opportunities: logger.info(f"\nFound {len(opportunities)} arbitrage opportunities:\n") for opp in opportunities: formatted_opp = detector.format_opportunity(opp) print(formatted_opp) - # Send Discord notification for each opportunity notifier.send_arbitrage_opportunity(formatted_opp) else: logger.info("No arbitrage opportunities found") - + return len(opportunities) -def run_continuous(): +async def run_continuous(): + """Run continuous market updates and arbitrage detection.""" check_interval = config.get("arbitrage", "check_interval") logger.info(f"Starting continuous monitoring (checking every {check_interval}s)...") logger.info("Press Ctrl+C to stop\n") - + try: while True: - count = run_once() - + await run_once() logger.info(f"\nWaiting {check_interval} seconds until next check...\n") - time.sleep(check_interval) - + await asyncio.sleep(check_interval) + except KeyboardInterrupt: logger.info("\nStopping arbitrage bot...") def main(): - parser = argparse.ArgumentParser(description='Arbitrage Detection Bot for Polymarket & Kalshi') - parser.add_argument('--once', action='store_true', help='Run detection once and exit') - parser.add_argument('--continuous', action='store_true', help='Run continuous monitoring') - parser.add_argument('--interval', type=int, help='Check interval in seconds (overrides config)') + """Main entry point for the arbitrage bot.""" + parser = argparse.ArgumentParser( + description='Arbitrage Detection Bot for Polymarket & Kalshi' + ) + parser.add_argument( + '--once', + action='store_true', + help='Run detection once and exit' + ) + parser.add_argument( + '--continuous', + action='store_true', + help='Run continuous monitoring' + ) + parser.add_argument( + '--interval', + type=int, + help='Check interval in seconds (overrides config)' + ) + parser.add_argument( + '--update-only', + action='store_true', + help='Only update market data without checking for arbitrage' + ) + parser.add_argument( + '--category', + choices=CATEGORIES, + help='Only process specific market category' + ) args = parser.parse_args() @@ -49,12 +202,19 @@ def main(): logger.info("Arbitrage Detection Bot") logger.info(f"Min profit threshold: {config.get('arbitrage', 'min_profit_pct')}%") logger.info(f"Min liquidity: ${config.get('arbitrage', 'min_liquidity')}") - logger.info(f"Match confidence: {config.get('matching', 'min_confidence')}\n") - - if args.continuous: - run_continuous() - else: - run_once() + logger.info(f"Match confidence: {config.get('matching', 'min_confidence')}") + if args.category: + logger.info(f"Processing category: {args.category}") + logger.info("") + try: + if args.continuous: + asyncio.run(run_continuous()) + else: + asyncio.run(run_once()) + + except KeyboardInterrupt: + logger.info("\nStopping arbitrage bot...") + if __name__ == "__main__": main() diff --git a/src/matching/cross_platform_matcher.py b/src/matching/cross_platform_matcher.py new file mode 100644 index 0000000..1188417 --- /dev/null +++ b/src/matching/cross_platform_matcher.py @@ -0,0 +1,53 @@ +import aiohttp +from typing import List, Dict, Any + +async def fetch_kalshi_series(base_url: str) -> List[Dict[str, Any]]: + url = f"{base_url}/trade-api/v2/series" + async with aiohttp.ClientSession() as session: + async with session.get(url) as resp: + resp.raise_for_status() + data = await resp.json() + return data.get("series", []) + +async def search_polymarket(base_url: str, query: str, events_tag: str = None) -> List[Dict[str, Any]]: + # Build query params + params = [f"q={query}", "keep_closed_markets=0"] + if events_tag: + params.append(f"events_tag={events_tag}") + url = f"{base_url}/public-search?{'&'.join(params)}" + async with aiohttp.ClientSession() as session: + async with session.get(url) as resp: + resp.raise_for_status() + data = await resp.json() + return data.get("markets", []) + +async def cross_platform_match(kalshi_base_url: str, polymarket_base_url: str, category: str = None) -> List[Dict[str, Any]]: + kalshi_series = await fetch_kalshi_series(kalshi_base_url) + results = [] + for series in kalshi_series: + if category and series.get("category", "").lower() != category.lower(): + continue + # Use ticker, title, or description as keywords + keywords = series.get("ticker", "") + if not keywords: + keywords = series.get("title", "") + if not keywords: + continue + # Use Kalshi category as events_tag for Polymarket search + events_tag = series.get("category", None) + polymarket_markets = await search_polymarket(polymarket_base_url, keywords, events_tag=events_tag) + for market in polymarket_markets: + results.append({ + "kalshi_series": series, + "polymarket_market": market + }) + return results + +# Example usage: +# import asyncio +# matches = asyncio.run(cross_platform_match( +# kalshi_base_url="https://api.elections.kalshi.com", +# polymarket_base_url="https://gamma-api.polymarket.com", +# category="Politics" +# )) +# print(matches) diff --git a/test_bot.py b/test_bot.py index fe17557..4a1f2b5 100755 --- a/test_bot.py +++ b/test_bot.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python3 +#!/usr/bin/env python3 """ Manual test script for the arbitrage bot. @@ -8,6 +8,7 @@ """ import sys +import asyncio import argparse from src.api import PolymarketClient, KalshiClient, Market from src.matching.semantic_matcher import matcher @@ -58,7 +59,7 @@ def create_mock_markets(): return polymarket_markets, kalshi_markets -def test_real_apis(): +async def test_real_apis(): """Test with real API data""" logger.info("=" * 60) logger.info("TESTING WITH REAL APIs") @@ -80,16 +81,24 @@ def test_real_apis(): private_key_str=kalshi_private_key ) - # Fetch real markets - logger.info("Fetching Polymarket markets...") - polymarket_markets = polymarket_client.get_markets() - logger.info(f"✓ Got {len(polymarket_markets)} Polymarket markets") + all_polymarket_markets = [] + all_kalshi_markets = [] - logger.info("Fetching Kalshi markets...") - kalshi_markets = kalshi_client.get_markets() - logger.info(f"✓ Got {len(kalshi_markets)} Kalshi markets") + for category in ["politics", "economy", "crypto", "sports"]: + try: + logger.info(f"Fetching {category} markets from Polymarket...") + poly_markets = await polymarket_client.get_markets_by_category(category, limit=100) + all_polymarket_markets.extend(poly_markets) + logger.info(f"✓ Got {len(poly_markets)} Polymarket {category} markets") - return polymarket_markets, kalshi_markets + logger.info(f"Fetching {category} markets from Kalshi...") + kalshi_markets = await kalshi_client.get_markets_by_category(category, limit=200) + all_kalshi_markets.extend(kalshi_markets) + logger.info(f"✓ Got {len(kalshi_markets)} Kalshi {category} markets") + except Exception as e: + logger.error(f"Failed to fetch {category} markets: {e}") + + return all_polymarket_markets, all_kalshi_markets def test_mock_data(): """Test with mock data that has a guaranteed arbitrage opportunity""" @@ -202,7 +211,7 @@ def run_arbitrage_test(polymarket_markets, kalshi_markets): logger.info(f" → {strategy['explanation']}") logger.info("") -def main(): +async def main(): parser = argparse.ArgumentParser(description="Test the arbitrage bot") parser.add_argument("--mock", action="store_true", help="Use mock data instead of real APIs") args = parser.parse_args() @@ -211,7 +220,7 @@ def main(): if args.mock: polymarket_markets, kalshi_markets = test_mock_data() else: - polymarket_markets, kalshi_markets = test_real_apis() + polymarket_markets, kalshi_markets = await test_real_apis() run_arbitrage_test(polymarket_markets, kalshi_markets) @@ -224,4 +233,4 @@ def main(): sys.exit(1) if __name__ == "__main__": - main() + asyncio.run(main())