get_weather_signals

Instructions

Implementation Reference

• src/weather_edge_mcp/mcp_server.py:36-44 (handler)

  The MCP tool handler for get_weather_signals. Decorated with @mcp.tool(), it takes a city name, looks up the city config, runs compute_signals asynchronously, and formats the result via format_weather_signals.

  @mcp.tool()
  def get_weather_signals(city: str) -> str:
      """Get calibrated edge signals for one city's Kalshi weather markets.
  
      Args:
          city: One of nyc, chicago, denver, miami, la.
      """
      cfg = get_city(city)
      return format_weather_signals(_run(compute_signals(cfg.key)))

• src/weather_edge_mcp/mcp_server.py:21-29 (registration)

  FastMCP server registration with instructions mentioning get_weather_signals in the usage description.

  mcp = FastMCP(
      name="weather-edge",
      instructions=(
          "Weather Edge MCP Server for calibrated Kalshi weather-market intelligence. "
          "Use list_cities for supported markets, get_weather_signals for one city, "
          "get_all_signals for a full scan, get_forecast for raw forecast context, and "
          "get_station_observation for live settlement-station readings."
      ),
  )

• src/weather_edge_mcp/core.py:239-254 (helper)

  format_weather_signals - formats the computed signal data dict into a human-readable string for the get_weather_signals tool response.

  def format_weather_signals(data: dict[str, Any]) -> str:
      if data.get("error"):
          return f"Error: {data['error']}"
      lines = [f"# Weather Edge — {data['city_label']}", f"Station: {data['station']}", ""]
      forecast = data["forecast"]
      lines.append(f"Forecast: {forecast['high_f']}°F — {forecast['forecast']}")
      lines.append("")
      for signal in data.get("signals", [])[:10]:
          if not signal.get("verdict"):
              continue
          lines.append(
              f"- [{signal['verdict']}] {signal['bucket']} | NWS {signal['nws_prob']}% vs market {signal['market_price']}% | edge {signal['edge']:+.1f} pts | EV {signal['net_ev_cents']:+.1f}c"
          )
      if len(lines) <= 4:
          lines.append("No positive-EV signals found.")
      return "\n".join(lines)

• src/weather_edge_mcp/core.py:161-236 (helper)

  compute_signals - the core async function that fetches NWS forecast and Kalshi markets, computes probabilities and edges, and returns a structured signal dict ready for formatting.

  async def compute_signals(city_key: str) -> dict[str, Any]:
      cached = get_cached(f"signals_{city_key}")
      if cached:
          return cached
  
      cfg = CITIES[city_key]
      forecast = await fetch_nws_forecast(city_key)
      if not forecast:
          return {"city": city_key, "error": "NWS unavailable", "signals": []}
  
      markets = await fetch_kalshi_markets(city_key)
      signals: list[dict[str, Any]] = []
  
      for market in markets:
          subtitle = market.get("subtitle", market.get("yes_sub_title", ""))
          yes_bid = float(market.get("yes_bid_dollars", 0) or 0)
          yes_ask = float(market.get("yes_ask_dollars", 0) or 0)
          volume = float(market.get("volume_fp", market.get("volume", 0)) or 0)
  
          is_over = "or above" in subtitle.lower() or "greater" in market.get("strike_type", "")
          is_under = "or below" in subtitle.lower()
          low_f = high_f = None
          nums = re.findall(r"(\d+)", subtitle)
          if is_over and nums:
              low_f = int(nums[0])
          elif is_under and nums:
              high_f = int(nums[0])
          elif len(nums) >= 2:
              low_f, high_f = int(nums[0]), int(nums[1])
  
          nws_prob = compute_probability(
              forecast["high_f"],
              low_f,
              high_f,
              is_over,
              is_under,
              sigma=cfg.sigma,
              forecast_bias=cfg.forecast_bias,
          )
          mid_price = (yes_bid + yes_ask) / 2 if yes_ask > 0 else yes_bid
          if mid_price <= 0:
              continue
  
          edge = nws_prob - mid_price
          fee = 0.07 * mid_price * (1 - mid_price)
          net_ev = nws_prob * (1 - yes_ask) - (1 - nws_prob) * yes_ask - fee if yes_ask > 0 else 0
          verdict = "STRONG" if net_ev > 0.05 else ("GOOD" if net_ev > 0.02 else ("MARGINAL" if net_ev > 0 else ""))
  
          signals.append(
              {
                  "ticker": market.get("ticker", ""),
                  "bucket": subtitle,
                  "date": forecast["date"],
                  "nws_high": forecast["high_f"],
                  "nws_prob": round(nws_prob * 100, 1),
                  "market_price": round(mid_price * 100, 1),
                  "edge": round(edge * 100, 1),
                  "net_ev_cents": round(net_ev * 100, 1),
                  "volume": int(volume),
                  "yes_bid": yes_bid,
                  "yes_ask": yes_ask,
                  "verdict": verdict,
              }
          )
  
      signals.sort(key=lambda signal: signal["net_ev_cents"], reverse=True)
      result = {
          "city": city_key,
          "city_label": cfg.label,
          "station": cfg.station,
          "forecast": forecast,
          "signals": signals,
          "generated_at": datetime.now(timezone.utc).isoformat(),
      }
      set_cached(f"signals_{city_key}", result)
      return result

• src/weather_edge_mcp/core.py:48-52 (helper)

  get_city - looks up a CityConfig by city key, used by the handler to resolve the city string to a configuration.

  def get_city(city: str) -> CityConfig:
      key = city.lower().strip()
      if key not in CITIES:
          raise ValueError(f"Unknown city '{city}'. Valid: {', '.join(CITIES.keys())}")
      return CITIES[key]