ramseshk
c93af97059
- Open-Meteo WeatherNext API client for HK forecasts - HKO public data client (current conditions, 9-day forecast, typhoon warnings) - HK-specific weather extraction and calibration - Polymarket market scanning, price discovery, and market creation proposals - Trading strategy engine: edge detection, Kelly criterion sizing, probability calibration - End-to-end pipeline with dry-run mode and scheduled runner - Interactive dashboard with live HK weather + forecasts + trading signals Dependencies: Python 3.10+, openmeteo-requests, pandas No API keys needed for dry-run mode. Polymarket trading requires private key in .env.
207 lines
7.7 KiB
Python
207 lines
7.7 KiB
Python
"""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)
|