Added kalshi bet predictor example

8-application-demos/6-kalshi-bet-predictor/.gitignore

@@ -0,0 +1,4 @@
+.DS_Store
+.env
+.venv/
+.vscode/

8-application-demos/6-kalshi-bet-predictor/README.md

@@ -0,0 +1,55 @@
+# Kalshi Bet Predictor
+This repository contains a set of Python scripts designed to find equivalent binary markets across Kalshi and Polymarket, use an LLM via OpenAI and web search via Exa to generate an independent prediction, and then calculate the trading "edge" on both platforms.
+
+Core Components
+---------------
+
+The project is structured around three main scripts:
+
+**`find_equiv_markets.py`**: A utility script to automatically search Kalshi and Polymarket APIs, use a Sentence Transformer and FAISS vector index to find markets with similar titles (i.e., equivalent questions), and save potential matches to a CSV file for manual review.
+
+**`analyst.py`** and **`main.py`**: These scripts form the core prediction engine.
+
+*   `analyst.py` handles API interactions with OpenAI (for prediction and question generation) and Exa (for web information retrieval).
+
+*   `main.py` fetches the current prices from both Kalshi and Polymarket, runs the prediction via `BetAnalyst`, and calculates the trading edge against the model's prediction. This logic is intended to be hosted on Cerebrium
+
+*   **`compare.py`**: This script reads the market pairs from the CSV, asynchronously calls the hosted prediction endpoint for each pair, and compiles statistics on the trading edge and which platform offers a better opportunity more frequently.
+
+
+
+Prerequisites
+-------------
+
+You will need API keys for the following services:
+
+*   **OpenAI**: For the large language model (`BetAnalyst` class).
+
+*   **Exa**: For semantic search/information retrieval (`BetAnalyst` class).
+
+*   **Cerebrium** (or similar hosting platform): To deploy the `main.py` and `analyst.py` logic as a prediction endpoint.
+
+
+Create a `.env` file in your project root to store your keys:
+
+``` OPENAI_API_KEY="your_openai_key" EXA_API_KEY="your_exa_key" ```
+
+Setup and Installation
+----------------------
+
+### Dependencies
+
+Install the required Python packages:
+
+```bash
+pip install -r requirements.txt
+```
+
+
+Workflow
+--------
+
+1. Host the prediction service by deploying `main.py` and `analyst.py` on Cerebrium to expose a `predict` endpoint that runs the `BetAnalyst` logic.
+2. Run `find_equiv_markets.py` to identify equivalent Kalshi and Polymarket markets and export the candidate pairs to a CSV file.
+3. Execute `compare.py`, which loads the CSV pairs, calls the hosted prediction endpoint for each pair, and aggregates the edge statistics to highlight the most favorable markets.
+

8-application-demos/6-kalshi-bet-predictor/analyst.py

@@ -0,0 +1,127 @@
+from typing import Tuple
+from dotenv import load_dotenv
+import os
+import json
+from exa_py import Exa
+from openai import OpenAI
+from pydantic import BaseModel
+
+class BetAnalyst:
+    def __init__(self, model_name: str = "gpt-5-nano"):
+        """Initializes the API clients and loads necessary API keys from environment variables"""
+        load_dotenv()
+
+        exa_api_key = os.environ.get("EXA_API_KEY")
+        openai_api_key = os.environ.get("OPENAI_API_KEY")
+
+        if not exa_api_key:
+            raise EnvironmentError("Missing EXA_API_KEY in environment variables")
+        if not openai_api_key: 
+            raise EnvironmentError("Missing OPENAI_API_KEY in environment variables")
+
+        self.exa = Exa(exa_api_key)
+        self.client = OpenAI(api_key=openai_api_key)
+        self.model_name = model_name
+
+        print(f"Using model: {model_name}")
+
+    def _generate_response(self, prompt: str, text_format = None):
+        """Sends a prompt to the OpenAI API and optionally parses the output into a structured format"""
+        try:
+            response = self.client.responses.create(
+                model=self.model_name,
+                input=prompt,
+            )
+
+            output_text = response.output_text.strip()
+            print(f"Generated raw response: {output_text}")
+
+            if text_format is not None:
+                # If a Pydantic model (text_format) is provided, re-parse the raw output into that structure.
+                parsed = self.client.responses.parse(
+                    model=self.model_name,
+                    input=[
+                        {
+                            "role": "user",
+                            "content": output_text
+                        },
+                    ],
+                    text_format=text_format,
+                )
+                print(f"Parsed structured response: {parsed.output_parsed}")
+                return parsed.output_parsed
+
+            return output_text
+
+        except Exception as e:
+            raise RuntimeError(f"Error during API call: {e}") from e
+
+    def get_relevant_questions(self, question: str) -> list[str]:
+        """Generates a list of related search queries based on an initial user question"""
+        prompt = (
+            "Based on the following question, generate a list of 5 relevant questions "
+            "that one could search online to gather more information. "
+            "These questions should yield information that would be helpful to answering "
+            "the following question in an objective manner.\n\n"
+            "Your response SHOULD ONLY BE the following lines, in this exact format:\n"
+            "1. <question 1>\n"
+            "2. <question 2>\n"
+            "3. <question 3>\n"
+            "4. <question 4>\n"
+            "5. <question 5>\n"
+            "Do not add ANY preamble, conclusion, or extra text.\n\n"
+            f"Question: \"{question}\""
+        )
+
+        raw_response = self._generate_response(prompt)
+
+        relevant_questions = []
+        for line in raw_response.split('\n'):
+            line = line.strip()
+            if line and line[0].isdigit():
+                # Parse lines like "1. What is..." into "What is..."
+                clean_question = line.split('.', 1)[-1].strip()
+                relevant_questions.append(clean_question)
+
+        print(f"Generated relevant questions: {relevant_questions}")
+
+        return relevant_questions
+
+
+    def get_web_info(self, questions):
+        """Uses the Exa API to find answers for a list of questions."""
+        results = [self.exa.answer(q, text=True) for q in questions]
+        answers = [r.answer for r in results]
+        return answers
+
+    def get_binary_answer_with_percentage(self, information: str, question: str) -> Tuple[str, str, str]:
+        """Analyzes provided information to return a Yes/No probability and explanation for a given question"""
+        prompt = (
+            "Analyze the provided information below to answer the given binary question. "
+            "Based on the information, determine the probability that the answer is 'Yes' or 'No'.\n\n"
+            "--- Information ---\n"
+            f"{information}\n\n"
+            "--- Question ---\n"
+            f"{question}\n\n"
+            "IMPORTANT INSTRUCTIONS:\n"
+            "1. Your response MUST ONLY be a single line in THIS EXACT FORMAT:\n"
+            "   Yes: <YES PERCENTAGE>%, No: <NO PERCENTAGE>%, Explanation: <EXPLANATION>\n"
+            "2. Percentages must sum to 100%.\n"
+            "3. Do NOT include any preamble, summary, or additional text.\n"
+            "4. Provide a brief but clear explanation supporting your probabilities.\n\n"
+        )
+
+        # Define the expected Pydantic structure for the _generate_response 'text_format' parameter
+        class Response(BaseModel):
+            yes_percentage: str
+            no_percentage: str
+            explanation: str
+
+        response = self._generate_response(prompt, Response)
+        print(f"HELLO {response}")
+
+        try:
+            return response.yes_percentage, response.no_percentage, response.explanation
+        except json.JSONDecodeError:
+            raise RuntimeError(f"Failed to parse output as JSON: {response}")
+

8-application-demos/6-kalshi-bet-predictor/cerebrium.toml

@@ -0,0 +1,23 @@
+[cerebrium.deployment]
+name = "kalshi-bet-predictor"
+python_version = "3.11"
+docker_base_image_url = "debian:bookworm-slim"
+disable_auth = true
+include = ['./*', 'main.py', 'cerebrium.toml']
+exclude = ['.*']
+
+[cerebrium.dependencies.paths]
+pip = "cerebrium_requirements.txt"
+
+[cerebrium.hardware]
+cpu = 4
+memory = 16
+compute = "CPU"
+
+[cerebrium.scaling]
+min_replicas = 0
+max_replicas = 100
+cooldown = 30
+replica_concurrency = 1
+scaling_metric = "concurrency_utilization"
+

8-application-demos/6-kalshi-bet-predictor/cerebrium_requirements.txt

@@ -0,0 +1,22 @@
+annotated-types==0.7.0
+anyio==4.11.0
+certifi==2025.10.5
+charset-normalizer==3.4.4
+distro==1.9.0
+dotenv==0.9.9
+exa-py==1.16.1
+h11==0.16.0
+httpcore==1.0.9
+httpx==0.28.1
+idna==3.11
+jiter==0.11.1
+openai==2.6.0
+pydantic==2.12.3
+pydantic_core==2.41.4
+python-dotenv==1.1.1
+requests==2.32.5
+sniffio==1.3.1
+tqdm==4.67.1
+typing-inspection==0.4.2
+typing_extensions==4.15.0
+urllib3==2.5.0

8-application-demos/6-kalshi-bet-predictor/compare.py

@@ -0,0 +1,131 @@
+import csv
+import json
+from typing import Dict, List, Tuple
+import asyncio
+import aiohttp
+
+def load_markets(csv_path: str) -> List[Tuple[str, str]]:
+    # Loads market pairs (Kalshi ticker, Polymarket slug) from a CSV file.
+    markets = []
+    with open(csv_path, 'r') as f:
+        reader = csv.reader(f)
+        next(reader) # skip header
+        for row in reader:
+            if len(row) >= 2:
+                markets.append((row[0], row[1]))
+    return markets
+
+async def get_market_data(session: aiohttp.ClientSession, kalshi_ticker: str, 
+                         poly_slug: str, endpoint_url: str) -> Dict:
+    # Asynchronously fetches and processes edge data for a single market pair from a specified API endpoint.
+    payload = json.dumps({
+        'kalshi_ticker': kalshi_ticker,
+        'poly_slug': poly_slug
+    })
+
+    headers = {
+        'Authorization': '<YOUR AUTHORIZATION>',
+        'Content-Type': 'application/json'
+    }
+
+    try:
+        async with session.post(endpoint_url, headers=headers, data=payload) as response:
+            response.raise_for_status()
+            data = await response.json()
+            print(data)
+            data = data['result']
+
+            kalshi_data = data['kalshi']
+            poly_data = data['polymarket']
+
+            return {
+                'kalshi_ticker': kalshi_ticker,
+                'poly_slug': poly_slug,
+                'kalshi_edge_value': kalshi_data['edge'],
+                'poly_edge_value': poly_data['edge'],
+                'kalshi_is_buy_yes': kalshi_data['buy_yes'],
+                'kalshi_is_buy_no': kalshi_data['buy_no'],
+                'poly_is_buy_yes': poly_data['buy_yes'],
+                'poly_is_buy_no': poly_data['buy_no'],
+            }
+    except Exception as e:
+        print(f"Error fetching data for {kalshi_ticker}/{poly_slug}: {e}")
+        return None
+
+async def analyze_markets_async(csv_path: str, endpoint_url: str) -> List[Dict]:
+    # Orchestrates the asynchronous fetching of data for all markets listed in the CSV.
+    markets = load_markets(csv_path)
+
+    print(f"Fetching data for {len(markets)} markets all at once...")
+
+    async with aiohttp.ClientSession() as session:
+        tasks = [get_market_data(session, kalshi_ticker, poly_slug, endpoint_url) 
+                for kalshi_ticker, poly_slug in markets]
+
+        results = await asyncio.gather(*tasks)
+
+    # Filter out None results from failed requests
+    return [r for r in results if r is not None]
+
+def compute_statistics(results: List[Dict]) -> None:
+    # Calculates and prints summary statistics comparing Kalshi and Polymarket edge data.
+    print("\n" + "="*80)
+    print("STATISTICS")
+    print("="*80)
+
+    if not results:
+        print("No results to analyze")
+        return
+
+    total_markets = len(results)
+
+    kalshi_edges_values = [r['kalshi_edge_value'] for r in results]
+    kalshi_edge_sum = sum(kalshi_edges_values)
+
+    poly_edges_values = [r['poly_edge_value'] for r in results]
+    poly_edge_sum = sum(poly_edges_values)
+
+    kalshi_better_count = sum(1 for r in results if r['kalshi_edge_value'] > r['poly_edge_value'])
+    poly_better_count = sum(1 for r in results if r['poly_edge_value'] > r['kalshi_edge_value'])
+    equal_count = total_markets - kalshi_better_count - poly_better_count
+
+    edge_differences = [abs(r['kalshi_edge_value'] - r['poly_edge_value']) for r in results]
+    avg_edge_difference = sum(edge_differences) / total_markets
+    max_edge_difference = max(edge_differences)
+
+    print(f"\nTotal markets analyzed: {total_markets}")
+    print("\n" + "-"*80)
+    print("COMPARISON")
+    print("-"*80)
+    print(f"Markets with greater Kalshi edge:      {kalshi_better_count} ({kalshi_better_count/total_markets*100:.1f}%)")
+    print(f"Markets with greater Polymarket edge:  {poly_better_count} ({poly_better_count/total_markets*100:.1f}%)")
+    print(f"Markets with equal edge:               {equal_count} ({equal_count/total_markets*100:.1f}%)")
+    print(f"\nAverage edge difference: {avg_edge_difference:.4f} cents")
+    print(f"Max edge difference:     {max_edge_difference:.4f} cents")
+
+    print("\n" + "="*80)
+    if kalshi_edge_sum > poly_edge_sum:
+        advantage = kalshi_edge_sum - poly_edge_sum
+        print(f"OVERALL: Kalshi has greater total edge (+{advantage:.4f}) cents")
+        print(f"OVERALL: Kalshi has an average edge of (+{advantage/total_markets:.4f}) cents per market")
+    elif poly_edge_sum > kalshi_edge_sum:
+        advantage = poly_edge_sum - kalshi_edge_sum
+        print(f"OVERALL: Polymarket has greater total edge (+{advantage:.4f}) cents")
+        print(f"OVERALL: Polymarket has an average edge of (+{advantage/total_markets:.4f}) cents per market")
+    else:
+        print("OVERALL: Both platforms have equal total edge")
+    print("="*80)
+
+def main():
+    CSV_PATH = "<PATH_TO_YOUR_CSV_FILE>"
+    ENDPOINT_URL = "<YOUR_CEREBRIUM_PREDICT_URL>"
+
+    print("Starting async market analysis...")
+    results = asyncio.run(analyze_markets_async(CSV_PATH, ENDPOINT_URL))
+
+    print(f"\nSuccessfully fetched {len(results)} markets")
+
+    compute_statistics(results)
+
+if __name__ == "__main__":    
+    main()