"""Signal generator for HK weather prediction market trading. Combines model forecasts, probability calibration, and Kelly sizing to generate trading signals for Polymarket execution. """ from datetime import datetime, timedelta from typing import Optional, Dict, List from weather.hk_extractor import HKExtractor from strategy.calibrator import ProbabilityCalibrator from strategy.kelly import KellyCriterion, KellyResult from markets.polymarket_client import PolymarketClient from markets.trader import TradeSignal from config import MIN_EDGE_BPS class SignalGenerator: """Generate trading signals from weather forecasts and market prices.""" def __init__( self, bankroll_usdc: float = 1000.0, min_edge_bps: float = MIN_EDGE_BPS, ): self.weather = HKExtractor() self.polymarket = PolymarketClient() self.calibrator = ProbabilityCalibrator() self.kelly = KellyCriterion(bankroll_usdc=bankroll_usdc) self.min_edge_bps = min_edge_bps self.signals: List[TradeSignal] = [] def generate_signals(self) -> List[TradeSignal]: """Generate all trading signals for available markets.""" self.signals = [] markets = self.polymarket.find_relevant_weather_markets() if not markets: print("No relevant markets found on Polymarket") self._generate_standalone_signals() return self.signals for market in markets: signal = self._analyze_market(market) if signal: self.signals.append(signal) self.signals.sort(key=lambda s: abs(s.edge_bps), reverse=True) return self.signals def _analyze_market(self, market: Dict) -> Optional[TradeSignal]: """Analyze a single market and generate a signal.""" condition_id = market["condition_id"] question = market["question"].lower() if not market.get("active") or market.get("closed"): return None if market.get("liquidity", 0) < 50: return None # Too illiquid # Get market-implied probability market_prob = self.polymarket.get_market_implied_probability(condition_id, 0) if market_prob is None: return None # Determine what we're predicting model_prob, variable = self._get_model_probability(question) if model_prob is None: return None # Calibrate our probability cal_prob = self.calibrator.calibrate(variable, model_prob) # Calculate edge edge_bps = (cal_prob - market_prob) * 100 # Convert to basis points if abs(edge_bps) < self.min_edge_bps: return TradeSignal( market_id=market["id"], condition_id=condition_id, question=question, outcome_index=0, outcome_label=market["outcomes"][0] if market.get("outcomes") else "Yes", model_probability=cal_prob, market_probability=market_prob, edge_bps=edge_bps, recommended_size_usdc=0, max_size_usdc=0, signal_type="pass", ) # Determine side side = "buy_yes" if edge_bps > 0 else "buy_no" side_prob = cal_prob if side == "buy_yes" else 100 - cal_prob # Kelly sizing kelly_result = self.kelly.size_bet( our_probability=cal_prob, market_probability=market_prob, side=side, ) return TradeSignal( market_id=market["id"], condition_id=condition_id, question=question, outcome_index=0, outcome_label=market["outcomes"][0] if market.get("outcomes") else "Yes", model_probability=cal_prob, market_probability=market_prob, edge_bps=edge_bps, recommended_size_usdc=kelly_result.size_usdc, max_size_usdc=kelly_result.size_usdc, signal_type=side, ) def _get_model_probability(self, question: str) -> tuple: """Get our model's probability for a given market question.""" question = question.lower() if "rain" in question or "precipitation" in question: prob = self.weather.should_bet_rain_tomorrow() return (prob, "rain") if prob is not None else (None, "") if "temperature" in question and ("above" in question or "exceed" in question): if "30" in question or "thirty" in question: prob = self.weather.should_bet_temp_above(30.0) elif "35" in question or "thirty five" in question: prob = self.weather.should_bet_temp_above(35.0) else: prob = self.weather.should_bet_temp_above(33.0) return (prob, "temperature") if prob is not None else (None, "") if "typhoon" in question or "t8" in question or "tropical cyclone" in question: forecast = self.weather.get_hk_forecast() typhoon = forecast.get("typhoon_info", {}) prob = 30.0 if typhoon else 5.0 return (prob, "typhoon") if "weather" in question or "storm" in question: forecast = self.weather.get_combined_tomorrow_forecast() tomorrow = forecast.get("tomorrow", {}) if tomorrow: rain_prob = tomorrow.get("precipitation_probability_calibrated", 50) return (rain_prob, "rain") return (None, "") def _generate_standalone_signals(self): """Generate signals even when no Polymarket markets exist. Useful for tracking model predictions and for creating new markets. """ tomorrow = datetime.now() + timedelta(days=1) forecast = self.weather.get_hk_forecast() consensus = forecast.get("consensus", {}) tmrw = consensus.get("tomorrow", {}) if tmrw: self.signals.append(TradeSignal( market_id="standalone", condition_id="standalone", question=f"Will it rain in Hong Kong on {tomorrow:%Y-%m-%d}?", outcome_index=0, outcome_label="Yes", model_probability=tmrw.get("precipitation_probability_calibrated", 50), market_probability=50.0, edge_bps=0, recommended_size_usdc=0, max_size_usdc=0, signal_type="pass", )) print(f"\nGenerated {len(self.signals)} standalone signals") for s in self.signals: print(f" {s.question} -> P={s.model_probability:.1f}%") def get_signal_summary(self) -> str: """Get a human-readable summary of current signals.""" if not self.signals: return "No signals generated." lines = [] active = [s for s in self.signals if s.signal_type != "pass"] passed = [s for s in self.signals if s.signal_type == "pass"] lines.append(f"\n=== Signal Summary ({datetime.now():%Y-%m-%d %H:%M}) ===") lines.append(f"Active signals: {len(active)}") lines.append(f"Passed (no edge): {len(passed)}") lines.append("") if active: lines.append("TRADE SIGNALS:") for s in active: lines.append(f" [{s.signal_type.upper()}] {s.question}") lines.append(f" Model: {s.model_probability:.1f}% | Market: {s.market_probability:.1f}%") lines.append(f" Edge: {s.edge_bps:.0f}bps | Size: ${s.recommended_size_usdc:.2f}") if passed: lines.append("PASSED (edge < threshold):") for s in passed[:5]: # Limit to 5 lines.append(f" {s.question} (edge: {s.edge_bps:.0f}bps)") return "\n".join(lines)