REE MCP Server

Instructions

Analyze volatility for a month:
>>> await analyze_demand_volatility("2025-10-01", "2025-10-31")

Implementation Reference

• src/ree_mcp/interface/mcp_server.py:1077-1112 (handler)

  MCP tool handler decorated with @mcp.tool(). Instantiates DemandAnalysisService and calls its analyze_demand_volatility method. Handles exceptions and formats response as JSON string.

  @mcp.tool()
  async def analyze_demand_volatility(start_date: str, end_date: str) -> str:
      """Analyze demand volatility patterns over a period.
  
      Calculates daily demand swings, load factors, and volatility levels
      to identify high-variability days and overall stability patterns.
  
      Args:
          start_date: Start date in YYYY-MM-DD format
          end_date: End date in YYYY-MM-DD format
  
      Returns:
          JSON string with volatility analysis and stability assessment.
  
      Examples:
          Analyze volatility for a week:
          >>> await analyze_demand_volatility("2025-10-01", "2025-10-07")
  
          Analyze volatility for a month:
          >>> await analyze_demand_volatility("2025-10-01", "2025-10-31")
      """
      try:
          async with ToolExecutor() as executor:
              use_case = executor.create_get_indicator_data_use_case()
              data_fetcher = DataFetcher(use_case)
              service = DemandAnalysisService(data_fetcher)
  
              result = await service.analyze_demand_volatility(start_date, end_date)
  
          return ResponseFormatter.success(result)
  
      except DomainException as e:
          return ResponseFormatter.domain_exception(e)
      except Exception as e:
          return ResponseFormatter.unexpected_error(e, context="Error analyzing demand volatility")

• src/ree_mcp/interface/tool_services.py:589-696 (helper)

  Main business logic for analyzing demand volatility within DemandAnalysisService class. Fetches daily max and min demand data, computes daily swings, percentages, classifies volatility levels, and provides overall analysis including stability assessment.

  async def analyze_demand_volatility(self, start_date: str, end_date: str) -> dict[str, Any]:
      """Analyze demand volatility patterns.
  
      Args:
          start_date: Start date in YYYY-MM-DD format
          end_date: End date in YYYY-MM-DD format
  
      Returns:
          Volatility analysis with swings and stability metrics
      """
      # Fetch max and min daily demand
      indicators = {
          "max_daily": IndicatorIDs.MAX_DAILY_DEMAND,
          "min_daily": IndicatorIDs.MIN_DAILY_DEMAND,
      }
  
      raw_data = await self.data_fetcher.fetch_multiple_indicators(
          indicators, start_date, end_date, "day"
      )
  
      result: dict[str, Any] = {
          "period": {"start": start_date, "end": end_date},
          "daily_volatility": [],
          "analysis": {},
      }
  
      max_values = raw_data.get("max_daily", {}).get("values", [])
      min_values = raw_data.get("min_daily", {}).get("values", [])
  
      daily_swings = []
      load_factors = []
  
      # Calculate daily volatility
      for i, max_point in enumerate(max_values):
          if i >= len(min_values):
              break
  
          date = max_point["datetime"][:10]
          max_mw = max_point["value"]
          min_mw = min_values[i]["value"]
  
          swing_mw = max_mw - min_mw
          swing_pct = (swing_mw / max_mw) * 100 if max_mw > 0 else 0
          load_factor = (min_mw / max_mw) * 100 if max_mw > 0 else 0
  
          volatility_data = {
              "date": date,
              "max_demand_mw": round(max_mw, 2),
              "min_demand_mw": round(min_mw, 2),
              "daily_swing_mw": round(swing_mw, 2),
              "swing_percentage": round(swing_pct, 2),
              "load_factor_pct": round(load_factor, 2),
          }
  
          # Volatility classification
          if swing_pct < 20:
              volatility_data["volatility_level"] = "low"
          elif swing_pct < 40:
              volatility_data["volatility_level"] = "moderate"
          elif swing_pct < 60:
              volatility_data["volatility_level"] = "high"
          else:
              volatility_data["volatility_level"] = "very_high"
  
          result["daily_volatility"].append(volatility_data)
          daily_swings.append(swing_mw)
          load_factors.append(load_factor)
  
      # Overall analysis
      if daily_swings:
          avg_swing = sum(daily_swings) / len(daily_swings)
          max_swing = max(daily_swings)
          min_swing = min(daily_swings)
          avg_load_factor = sum(load_factors) / len(load_factors)
  
          # Count days by volatility level
          volatility_counts = {"low": 0, "moderate": 0, "high": 0, "very_high": 0}
          for day_data in result["daily_volatility"]:
              level = day_data["volatility_level"]
              volatility_counts[level] += 1
  
          result["analysis"] = {
              "average_daily_swing_mw": round(avg_swing, 2),
              "max_daily_swing_mw": round(max_swing, 2),
              "min_daily_swing_mw": round(min_swing, 2),
              "average_load_factor_pct": round(avg_load_factor, 2),
              "volatility_distribution": volatility_counts,
              "stability_assessment": (
                  "excellent"
                  if avg_load_factor >= 80
                  else "good"
                  if avg_load_factor >= 70
                  else "moderate"
                  if avg_load_factor >= 60
                  else "concerning"
              ),
              "interpretation": {
                  "load_factor": "Higher load factor (closer to 100%) indicates more stable demand",
                  "volatility_levels": {
                      "low": "<20% swing",
                      "moderate": "20-40% swing",
                      "high": "40-60% swing",
                      "very_high": ">60% swing",
                  },
              },
          }
  
      return result