폴리마켓 날씨 트레이딩 봇 with Claude

# config.py CITIES = { "NYC": {"station": "KLGA", "name": "LaGuardia"}, "Chicago": {"station": "KMDW", "name": "Midway"}, "Miami": {"station": "KMIA", "name": "Miami Intl"}, "LA": {"station": "KLAX", "name": "Los Angeles Intl"}, "Dallas": {"station": "KDAL", "name": "Love Field"}, } EDGE_THRESHOLD = 0.08 # 8% minimum edge KELLY_FRACTION = 0.15 # 15% fractional Kelly MAX_POSITION_SIZE = 50.0 # USD cap per trade POLL_INTERVAL_HOURS = 1 API_BASE = "https://clob.polymarket.com" OPEN_METEO_BASE = "https://api.open-meteo.com/v1/forecast" ``` ```python # weather_bot.py import json import logging import math import time from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple import requests import pandas as pd import numpy as np from scipy import stats import config logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler("weather_bot.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class PolymarketWeatherBot: def __init__(self): self.config = config self.session = requests.Session() self.positions = {} # market_id -> position info self.pnl = 0.0 self.trade_log = [] def get_active_temperature_markets(self) -> List[Dict]: """Fetch active temperature markets from Polymarket CLOB API.""" url = f"{self.config.API_BASE}/markets" params = { "active": "true", "closed": "false", "limit": 100, "tag": "weather" } try: resp = self.session.get(url, params=params, timeout=30) resp.raise_for_status() markets = resp.json() # Filter for temperature-related markets temp_markets = [ m for m in markets if "temperature" in m.get("question", "").lower() or "high" in m.get("question", "").lower() or "low" in m.get("question", "").lower() ] logger.info(f"Found {len(temp_markets)} active temperature markets") return temp_markets except Exception as e: logger.error(f"Failed to fetch markets: {e}") return [] def parse_market_details(self, market: Dict) -> Tuple[Optional[str], Optional[str]]: """Extract city and date from market question.""" question = market.get("question", "") # Example: "What will the high temperature be in NYC on 2024-03-15?" for city_key, city_info in self.config.CITIES.items(): if city_key.lower() in question.lower() or city_info["name"].lower() in question.lower(): # Try to extract date for word in question.split(): if len(word) == 10 and word[4] == "-" and word[7] == "-": try: datetime.strptime(word, "%Y-%m-%d") return city_key, word except ValueError: continue # If no date found, assume tomorrow return city_key, (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d") return None, None def get_gfs_ensemble_forecast(self, city_key: str, date_str: str) -> Optional[np.ndarray]: """Fetch 31-member GFS ensemble forecast for a city.""" city_info = self.config.CITIES[city_key] station = city_info["station"] # Open-Meteo ensemble API params = { "latitude": self._get_city_coords(city_key)[0], "longitude": self._get_city_coords(city_key)[1], "hourly": "temperature_2m", "models": "gfs_seamless", "ensemble": "true", "past_days": "0", "forecast_days": "7", "timezone": "UTC" } try: resp = self.session.get(self.config.OPEN_METEO_BASE, params=params, timeout=30) resp.raise_for_status() data = resp.json() # Extract ensemble members for the target date target_date = datetime.strptime(date_str, "%Y-%m-%d") hourly_data = data.get("hourly", {}) times = hourly_data.get("time", []) # Find indices for target date (daily max) daily_maxes = [] for member_idx in range(31): # 31 ensemble members member_key = f"temperature_2m_{member_idx}" if member_idx > 0 else "temperature_2m" if member_key not in hourly_data: continue temps = hourly_data[member_key] day_temps = [] for i, t in enumerate(times): dt = datetime.fromisoformat(t) if dt.date() == target_date.date(): day_temps.append(temps[i]) if day_temps: daily_maxes.append(max(day_temps)) if daily_maxes: return np.array(daily_maxes) return None except Exception as e: logger.error(f"Failed to fetch forecast for {city_key}: {e}") return None def _get_city_coords(self, city_key: str) -> Tuple[float, float]: """Get coordinates for a city (approximate, for forecast).""" coords = { "NYC": (40.7128, -74.0060), "Chicago": (41.8781, -87.6298), "Miami": (25.7617, -80.1918), "LA": (34.0522, -118.2437), "Dallas": (32.7767, -96.7970) } return coords.get(city_key, (0, 0)) def calculate_bucket_probabilities(self, ensemble_forecast: np.ndarray, market: Dict) -> Dict[str, float]: """Calculate probability for each price bucket based on ensemble.""" # Parse market outcomes/buckets outcomes = market.get("outcomes", []) if not outcomes: return {} # Convert outcomes to numeric buckets buckets = [] for outcome in outcomes: try: # Handle formats like ">=80", "<70", "75-80", "75" if ">=" in outcome: lower = int(outcome.replace(">=", "")) buckets.append((lower, float('inf'))) elif "<" in outcome: upper = int(outcome.replace("<", "")) buckets.append((-float('inf'), upper)) elif "-" in outcome: parts = outcome.split("-") buckets.append((int(parts[0]), int(parts[1]))) else: val = int(outcome) buckets.append((val, val)) except (ValueError, IndexError): continue if not buckets: return {} # Fit normal distribution to ensemble mu = np.mean(ensemble_forecast) sigma = np.std(ensemble_forecast) # Calculate probabilities for each bucket probabilities = {} for i, (lower, upper) in enumerate(buckets): if upper == float('inf'): prob = 1 - stats.norm.cdf(lower, mu, sigma) elif lower == -float('inf'): prob = stats.norm.cdf(upper, mu, sigma) else: prob = stats.norm.cdf(upper, mu, sigma) - stats.norm.cdf(lower, mu, sigma) probabilities[outcomes[i]] = max(0.0, min(1.0, prob)) return probabilities def calculate_edge(self, calculated_prob: float, market_price: float) -> float: """Calculate edge as difference between calculated probability and market price.""" return calculated_prob - (market_price / 100.0) def kelly_criterion(self, prob: float, price: float, fraction: float = 0.15) -> float: """Calculate Kelly criterion position size.""" # Convert price to decimal odds odds = 1.0 / (price / 100.0) # Kelly formula: f* = (bp - q) / b # where b = odds - 1, p = probability, q = 1 - p b = odds - 1 p = prob q = 1 - p if b <= 0: return 0 kelly = (b * p - q) / b return max(0, kelly * fraction) def place_order(self, market_id: str, outcome: str, size: float, price: int) -> bool: """Place a buy order on Polymarket.""" # This would integrate with Polymarket's CLOB API # For demo purposes, we just log it logger.info(f"PLACE ORDER: market={market_id}, outcome={outcome}, size=${size:.2f}, price={price}¢") # Simulate order placement order = { "market_id": market_id, "outcome": outcome, "size": size, "price": price, "timestamp": datetime.now().isoformat() } # Store position self.positions[market_id] = { "outcome": outcome, "size": size, "entry_price": price, "entry_time": datetime.now() } return True def check_and_close_positions(self, markets: List[Dict]) -> None: """Check if any positions have resolved and update P&L.""" for market in markets: market_id = market.get("id") if market_id in self.positions: if market.get("closed"): position = self.positions.pop(market_id) # Calculate P&L based on resolution winning_outcome = market.get("winning_outcome") if winning_outcome == position["outcome"]: profit = position["size"] * (100 - position["entry_price"]) / 100 self.pnl += profit logger.info(f"POSITION WON: market={market_id}, profit=${profit:.2f}") else: loss = position["size"] self.pnl -= loss logger.info(f"POSITION LOST: market={market_id}, loss=${loss:.2f}") def log_trade(self, market_id: str, forecast: np.ndarray, market_price: float, calc_prob: float, edge: float) -> None: """Log trade details.""" trade_entry = { "timestamp": datetime.now().isoformat(), "market_id": market_id, "forecast_mean": float(np.mean(forecast)), "forecast_std": float(np.std(forecast)), "market_price": market_price, "calculated_probability": calc_prob, "edge_percentage": edge * 100 } self.trade_log.append(trade_entry) logger.info(f"TRADE LOG: {json.dumps(trade_entry)}") def run(self) -> None: """Main bot loop.""" logger.info("Starting Polymarket Weather Bot") while True: try: # 1. Fetch active markets markets = self.get_active_temperature_markets() # 9. Check for resolved positions self.check_and_close_positions(markets) # 2-6. Process each market for market in markets: city, date_str = self.parse_market_details(market) if not city or not date_str: continue # 3. Get ensemble forecast forecast = self.get_gfs_ensemble_forecast(city, date_str) if forecast is None: continue # 4. Calculate probabilities probabilities = self.calculate_bucket_probabilities(forecast, market) if not probabilities: continue # 5-6. Compare and trade for outcome, calc_prob in probabilities.items(): # Get market price for this outcome outcome_data = market.get("outcomePrices", []) if not outcome_data: continue try: market_price = float(outcome_data[market["outcomes"].index(outcome)]) * 100 except (ValueError, IndexError): continue # 5. Calculate edge edge = self.calculate_edge(calc_prob, market_price) # 6. Trade if edge > threshold if edge > self.config.EDGE_THRESHOLD: # 6. Kelly sizing kelly_size = self.kelly_criterion( calc_prob, market_price, self.config.KELLY_FRACTION ) # 7. Cap position size position_size = min(kelly_size * 100, self.config.MAX_POSITION_SIZE) if position_size > 0: # 8. Log trade self.log_trade( market["id"], forecast, market_price, calc_prob, edge ) # Place order self.place_order( market["id"], outcome, position_size, int(market_price) ) # Sleep until next poll logger.info(f"Sleeping for {self.config.POLL_INTERVAL_HOURS} hour(s)") time.sleep(self.config.POLL_INTERVAL_HOURS * 3600) except KeyboardInterrupt: logger.info("Bot stopped by user") break except Exception as e: logger.error(f"Unexpected error: {e}") time.sleep(60) if __name__ == "__main__": bot = PolymarketWeatherBot() bot.run()