compare_forecast_actual
Analyze electricity demand forecast accuracy by comparing predicted vs actual values and calculating error metrics like MAE and RMSE for specific dates.
Instructions
Compare forecasted vs actual electricity demand.
Calculates forecast accuracy metrics (error, MAE, RMSE) for demand predictions.
Args: date: Date in YYYY-MM-DD format
Returns: JSON string with forecast comparison and accuracy metrics.
Examples: Compare forecast accuracy for Oct 8: >>> await compare_forecast_actual("2025-10-08")
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| date | Yes |
Implementation Reference
- The primary handler function implementing the 'compare_forecast_actual' MCP tool. It retrieves hourly demand forecast (IndicatorIDs.DEMAND_FORECAST) and actual demand (IndicatorIDs.REAL_DEMAND_PENINSULAR) data for the specified date, performs pairwise comparisons to compute absolute/percentage errors, and aggregates accuracy metrics including MAE, RMSE, MAPE, and forecast bias direction. The function is decorated with @mcp.tool() for automatic registration and schema generation via FastMCP.@mcp.tool() async def compare_forecast_actual(date: str) -> str: """Compare forecasted vs actual electricity demand. Calculates forecast accuracy metrics (error, MAE, RMSE) for demand predictions. Args: date: Date in YYYY-MM-DD format Returns: JSON string with forecast comparison and accuracy metrics. Examples: Compare forecast accuracy for Oct 8: >>> await compare_forecast_actual("2025-10-08") """ try: start_date, end_date = DateTimeHelper.build_day_range(date) async with ToolExecutor() as executor: use_case = executor.create_get_indicator_data_use_case() # Get forecast data forecast_request = GetIndicatorDataRequest( indicator_id=IndicatorIDs.DEMAND_FORECAST.id, start_date=start_date, end_date=end_date, time_granularity="hour", ) forecast_response = await use_case.execute(forecast_request) forecast_data = forecast_response.model_dump() # Get actual data actual_request = GetIndicatorDataRequest( indicator_id=IndicatorIDs.REAL_DEMAND_PENINSULAR.id, start_date=start_date, end_date=end_date, time_granularity="hour", ) actual_response = await use_case.execute(actual_request) actual_data = actual_response.model_dump() # Compare values forecast_values = forecast_data.get("values", []) actual_values = actual_data.get("values", []) comparisons = [] errors = [] absolute_errors = [] squared_errors = [] for forecast, actual in zip(forecast_values, actual_values, strict=False): forecast_mw = forecast["value"] actual_mw = actual["value"] error_mw = forecast_mw - actual_mw error_pct = (error_mw / actual_mw * 100) if actual_mw > 0 else 0 comparisons.append( { "datetime": forecast["datetime"], "forecast_mw": forecast_mw, "actual_mw": actual_mw, "error_mw": round(error_mw, 2), "error_percentage": round(error_pct, 2), } ) errors.append(error_mw) absolute_errors.append(abs(error_mw)) squared_errors.append(error_mw**2) # Calculate accuracy metrics accuracy_metrics = {} if errors: mae = sum(absolute_errors) / len(absolute_errors) rmse = (sum(squared_errors) / len(squared_errors)) ** 0.5 mean_error = sum(errors) / len(errors) mape = sum( abs(e / a["value"]) * 100 for e, a in zip(errors, actual_values, strict=False) ) / len(errors) accuracy_metrics = { "mean_absolute_error_mw": round(mae, 2), "root_mean_squared_error_mw": round(rmse, 2), "mean_error_mw": round(mean_error, 2), "mean_absolute_percentage_error": round(mape, 2), "bias": ( "overforecast" if mean_error > 0 else "underforecast" if mean_error < 0 else "unbiased" ), } result = { "date": date, "comparisons": comparisons, "accuracy_metrics": accuracy_metrics, } return ResponseFormatter.success(result) except Exception as e: return ResponseFormatter.unexpected_error(e, context="Error comparing forecast")