add Kalshi prediction market

app/services/prediction_markets.py

@@ -34,6 +34,12 @@ class PredictionMarketsScraper:
             'weight': 1.8,
             'enabled': True
         },
+        'kalshi': {
+            'name': 'Kalshi',
+            'base_url': 'https://api.elections.kalshi.com/trade-api/v2',
+            'weight': 1.7,
+            'enabled': True
+        },
         'metaculus': {
             'name': 'Metaculus',
             'base_url': 'https://www.metaculus.com/api',
@@ -71,12 +77,15 @@ class PredictionMarketsScraper:
         seen_titles = set()
 
         # Parallel fetching
-        with ThreadPoolExecutor(max_workers=3) as executor:
+        with ThreadPoolExecutor(max_workers=4) as executor:
             futures = []
 
             if self.SOURCES['polymarket']['enabled']:
                 futures.append((executor.submit(self._fetch_polymarket), 'polymarket'))
 
+            if self.SOURCES['kalshi']['enabled']:
+                futures.append((executor.submit(self._fetch_kalshi), 'kalshi'))
+
             if self.SOURCES['metaculus']['enabled']:
                 futures.append((executor.submit(self._fetch_metaculus), 'metaculus'))
 
@@ -310,6 +319,96 @@ class PredictionMarketsScraper:
             logger.error(f"Error fetching Metaculus: {e}")
             return []
 
+    def _fetch_kalshi(self) -> List[Dict]:
+        """Fetch predictions from Kalshi public API (financial events only)"""
+        try:
+            base_url = self.SOURCES['kalshi']['base_url']
+            url = f"{base_url}/events"
+            params = {
+                'limit': 200,
+                'with_nested_markets': True,
+                'status': 'open'
+            }
+
+            predictions = []
+            cursor = None
+            pages = 0
+
+            while pages < 3:
+                if cursor:
+                    params['cursor'] = cursor
+
+                response = self.session.get(url, params=params, timeout=15)
+                response.raise_for_status()
+                data = response.json()
+
+                events = data.get('events', [])
+                for event in events:
+                    if not self._is_kalshi_financial_event(event):
+                        continue
+
+                    event_title = event.get('title', '')
+                    category = self._categorize_prediction(event_title)
+                    markets = event.get('markets', []) or []
+
+                    for market in markets:
+                        try:
+                            if market.get('market_type') and market.get('market_type') != 'binary':
+                                continue
+
+                            title = market.get('title') or event_title
+                            if not title or len(title) < 8:
+                                continue
+
+                            yes_prob = self._kalshi_yes_probability(market)
+                            if yes_prob is None:
+                                continue
+
+                            no_prob = 100 - yes_prob
+                            volume = float(market.get('volume', 0) or 0)
+                            impact = self._assess_impact(volume, category)
+                            sentiment = 'positive' if yes_prob > 60 else ('negative' if yes_prob < 40 else 'neutral')
+
+                            close_time_str = market.get('close_time') or market.get('expiration_time')
+                            end_date = self._parse_iso_datetime(close_time_str)
+
+                            market_ticker = market.get('ticker', '')
+
+                            predictions.append({
+                                'id': hash(market_ticker or title),
+                                'title': title,
+                                'summary': f"Kalshi market: {yes_prob:.1f}% YES, {no_prob:.1f}% NO",
+                                'source': 'Kalshi',
+                                'category': category,
+                                'timestamp': datetime.now(),
+                                'url': f"{base_url}/markets/{market_ticker}" if market_ticker else base_url,
+                                'yes_probability': round(yes_prob, 1),
+                                'no_probability': round(no_prob, 1),
+                                'volume': volume,
+                                'end_date': end_date,
+                                'impact': impact,
+                                'sentiment': sentiment,
+                                'is_breaking': False,
+                                'source_weight': self.SOURCES['kalshi']['weight'],
+                                'likes': int(volume / 1000),
+                                'retweets': 0
+                            })
+
+                        except Exception as e:
+                            logger.debug(f"Error parsing Kalshi market: {e}")
+                            continue
+
+                cursor = data.get('cursor')
+                pages += 1
+                if not cursor:
+                    break
+
+            return predictions
+
+        except Exception as e:
+            logger.error(f"Error fetching Kalshi: {e}")
+            return []
+
     def _fetch_cme_fedwatch(self) -> List[Dict]:
         """
         Fetch Fed rate probabilities from CME FedWatch Tool
@@ -387,6 +486,74 @@ class PredictionMarketsScraper:
         scores = {'macro': macro_score, 'markets': market_score, 'geopolitical': geo_score}
         return max(scores, key=scores.get) if max(scores.values()) > 0 else 'markets'
 
+    def _is_kalshi_financial_event(self, event: Dict) -> bool:
+        """Filter Kalshi events to financial/macro/markets categories"""
+        category = (event.get('category') or '').lower()
+        title = (event.get('title') or '').lower()
+        series_ticker = (event.get('series_ticker') or '').lower()
+
+        financial_keywords = [
+            'econ', 'economic', 'economy', 'finance', 'financial', 'market',
+            'inflation', 'cpi', 'ppi', 'gdp', 'jobs', 'employment', 'unemployment',
+            'rate', 'interest', 'fed', 'fomc', 'treasury', 'bond', 'recession',
+            'stock', 's&p', 'nasdaq', 'dow', 'crypto', 'bitcoin', 'oil', 'fx',
+            'usd', 'dollar'
+        ]
+
+        if any(kw in category for kw in financial_keywords):
+            return True
+
+        if any(kw in title for kw in financial_keywords):
+            return True
+
+        if any(kw in series_ticker for kw in financial_keywords):
+            return True
+
+        return self._categorize_prediction(event.get('title', '')) in {'macro', 'markets'}
+
+    def _kalshi_yes_probability(self, market: Dict) -> Optional[float]:
+        """Return YES probability (0-100) from Kalshi market pricing."""
+        def to_float(value):
+            if value is None or value == '':
+                return None
+            try:
+                return float(value)
+            except Exception:
+                return None
+
+        yes_bid_d = to_float(market.get('yes_bid_dollars'))
+        yes_ask_d = to_float(market.get('yes_ask_dollars'))
+        last_d = to_float(market.get('last_price_dollars'))
+
+        price = None
+        if yes_bid_d is not None and yes_ask_d is not None:
+            price = (yes_bid_d + yes_ask_d) / 2
+        elif last_d is not None:
+            price = last_d
+        else:
+            yes_bid = to_float(market.get('yes_bid'))
+            yes_ask = to_float(market.get('yes_ask'))
+            last = to_float(market.get('last_price'))
+            if yes_bid is not None and yes_ask is not None:
+                price = (yes_bid + yes_ask) / 2 / 100
+            elif last is not None:
+                price = last / 100
+
+        if price is None:
+            return None
+
+        price = max(min(price, 1.0), 0.0)
+        return price * 100
+
+    def _parse_iso_datetime(self, value: Optional[str]) -> datetime:
+        """Parse ISO timestamps from Kalshi API with fallback."""
+        if not value:
+            return datetime.now() + timedelta(days=30)
+        try:
+            return datetime.fromisoformat(value.replace('Z', '+00:00'))
+        except Exception:
+            return datetime.now() + timedelta(days=30)
+
     def _assess_impact(self, volume: float, category: str) -> str:
         """Assess market impact based on volume and category"""
         # Macro predictions are inherently high impact