get_all_signals

Instructions

Implementation Reference

• src/weather_edge_mcp/mcp_server.py:47-53 (handler)

  The MCP tool handler for 'get_all_signals'. Iterates over all cities, runs compute_signals for each, then formats the results via format_all_signals.

  @mcp.tool()
  def get_all_signals() -> str:
      """Run a full scan across all supported cities and rank top weather-market signals."""
      items = []
      for key, cfg in CITIES.items():
          items.append((cfg.label, _run(compute_signals(key))))
      return format_all_signals(items)

• src/weather_edge_mcp/mcp_server.py:47-48 (registration)

  The @mcp.tool() decorator registers this function as an MCP tool named 'get_all_signals'.

  @mcp.tool()
  def get_all_signals() -> str:

• src/weather_edge_mcp/core.py:257-271 (helper)

  The format_all_signals helper function formats the full scan results into a readable string, sorting signals by net EV across all cities.

  def format_all_signals(items: list[tuple[str, dict[str, Any]]]) -> str:
      summaries: list[tuple[float, str, dict[str, Any]]] = []
      for city_label, data in items:
          for signal in data.get("signals", [])[:5]:
              if signal.get("verdict"):
                  summaries.append((signal["net_ev_cents"], city_label, signal))
      summaries.sort(key=lambda row: row[0], reverse=True)
      lines = ["# Weather Edge — Full Scan", ""]
      for _, city_label, signal in summaries[:15]:
          lines.append(
              f"- {city_label}: [{signal['verdict']}] {signal['bucket']} | market {signal['market_price']}% | edge {signal['edge']:+.1f} pts | EV {signal['net_ev_cents']:+.1f}c"
          )
      if len(lines) == 2:
          lines.append("No positive-EV signals found across supported cities.")
      return "\n".join(lines)

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

  The compute_signals helper computes all signals for a single city, fetching NWS forecast and Kalshi markets, then calculating probabilities and edges. Called by get_all_signals for each city.

  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:19-31 (schema)

  The CityConfig dataclass defines the input configuration schema used by get_all_signals to iterate over cities.

  @dataclass(frozen=True)
  class CityConfig:
      key: str
      label: str
      station: str
      metar_station: str
      nws_office: str
      nws_grid_x: int
      nws_grid_y: int
      kalshi_series: str
      sigma: float
      forecast_bias: float