Codex/fase 7

.env.example

@@ -12,5 +12,6 @@ POSTGRES_DB=chargegrid
 POSTGRES_USER=chargegrid
 POSTGRES_PASSWORD=chargegrid
 DATABASE_URL=postgresql+psycopg://chargegrid:chargegrid@localhost:5432/chargegrid
+DEMAND_MODEL_PATH=../data/models/demand_forecast.joblib
 
 VITE_API_URL=http://localhost:8000/api/v1

README.md

@@ -105,6 +105,7 @@ python3 scripts/sprint3_demo.py
 - [Fase 5 — Billing e histórico de invoices (concluída)](docs/PHASE_5.md)
 - [Fase 6 — Contratos da API de analytics](docs/PHASE_6_ANALYTICS.md)
 - [Fase 6 — Integração e limites](docs/PHASE_6.md)
+- [Fase 7 — ML, treinamento e inferência](docs/PHASE_7.md)
 
 ## Estado atual
 
@@ -157,7 +158,11 @@ atualização das respostas dos dashboards. Os gráficos do gestor somam leitura
 simultâneas para mostrar a demanda total de cada tick. Veja os resultados e
 limites em [docs/PHASE_6.md](docs/PHASE_6.md).
 
-A Fase 6 ainda não está concluída: o KPI obrigatório de risco de pico depende
-de uma previsão futura válida, e o fluxo automático de previsão/classificação
-da Fase 7 ainda não existe. Sem esses dados, a tela mostra um estado
-informativo. O simulador também depende de ticks manuais pela API; a Sprint 3 fornece seed e roteiro reproduzíveis para a demonstração completa.
+A Fase 7 disponibiliza inferência administrativa explícita em
+`POST /api/v1/predictions/demand/run`. Ela carrega o artefato configurado em
+`DEMAND_MODEL_PATH`, monta features apenas com leituras já disponíveis,
+persiste a previsão de 60 minutos, classifica o risco com os thresholds de
+`SystemConfiguration` e cria `PEAK_RISK` uma vez por episódio `HIGH`. A
+previsão é estritamente consultiva e não altera a alocação energética. Consulte
+[docs/PHASE_7.md](docs/PHASE_7.md) para preparo do modelo, pré-condições e
+respostas de erro. O simulador continua dependendo de ticks manuais pela API.

backend/app/api/routes/predictions.py

@@ -14,16 +14,21 @@
     commit_or_conflict,
     get_or_404,
 )
+from app.core.config import get_settings
+from app.ml.training import ModelArtifactError
 from app.models.infrastructure import ChargingStation
 from app.models.prediction import DemandPrediction, SystemConfiguration
+from app.schemas.common import ErrorResponse
 from app.schemas.prediction import (
     DemandPredictionCreate,
     DemandPredictionResponse,
+    DemandPredictionRun,
     SystemConfigurationCreate,
     SystemConfigurationResponse,
     SystemConfigurationUpdate,
     SystemConfigurationValues,
 )
+from app.services.demand_predictions import PredictionDataError, run_demand_prediction
 
 predictions_router = APIRouter(prefix="/predictions", tags=["predictions"])
 configuration_router = APIRouter(prefix="/system-configuration", tags=["configuration"])
@@ -67,6 +72,44 @@ async def create_demand_prediction(
     return prediction
 
 
+@predictions_router.post(
+    "/demand/run",
+    response_model=DemandPredictionResponse,
+    status_code=status.HTTP_201_CREATED,
+    responses=UNAUTHORIZED_RESPONSE | FORBIDDEN_RESPONSE | NOT_FOUND_RESPONSE | {
+        422: {"model": ErrorResponse, "description": "Inference features are insufficient"},
+        503: {"model": ErrorResponse, "description": "Demand model is unavailable"},
+    },
+)
+async def execute_demand_prediction(
+    payload: DemandPredictionRun, db: DbSession, _: AdminUser
+) -> DemandPrediction:
+    station = db.scalar(
+        select(ChargingStation)
+        .where(ChargingStation.id == payload.station_id)
+        .with_for_update()
+    )
+    if station is None:
+        raise HTTPException(status_code=404, detail="Resource not found")
+    configuration = db.scalar(select(SystemConfiguration).limit(1))
+    if configuration is None:
+        raise HTTPException(status_code=404, detail="Configuration not found")
+    try:
+        prediction = run_demand_prediction(
+            db,
+            station=station,
+            configuration=configuration,
+            artifact_path=get_settings().demand_model_path,
+        )
+    except PredictionDataError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    except ModelArtifactError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    commit_or_conflict(db)
+    db.refresh(prediction)
+    return prediction
+
+
 @configuration_router.get(
     "",
     response_model=SystemConfigurationResponse,

backend/app/core/config.py

@@ -1,5 +1,6 @@
 from datetime import UTC, datetime
 from functools import lru_cache
+from pathlib import Path
 from typing import Literal
 
 from pydantic import Field, field_validator, model_validator
@@ -22,6 +23,7 @@ class Settings(BaseSettings):
     app_cors_origins: list[str] = Field(default_factory=lambda: ["http://localhost:5173"])
     api_v1_prefix: str = "/api/v1"
     grid_emission_factor_kg_per_kwh: float = Field(default=0.0, ge=0)
+    demand_model_path: Path = Path("../data/models/demand_forecast.joblib")
     demo_simulation_start_utc: datetime | None = None
     database_url: str = "postgresql+psycopg://chargegrid:chargegrid@localhost:5432/chargegrid"
     jwt_secret_key: str = Field(default=DEFAULT_JWT_SECRET, min_length=32)

backend/app/ml/baseline.py

@@ -0,0 +1,94 @@
+"""Chronological evaluation of the hour/day-of-week demand baseline."""
+
+import math
+from collections import defaultdict
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime
+
+from app.ml.dataset import DemandDatasetRow
+
+
+@dataclass(frozen=True, slots=True)
+class RegressionMetrics:
+    mae: float
+    rmse: float
+    r2: float
+
+
+@dataclass(frozen=True, slots=True)
+class BaselineEvaluation:
+    strategy: str
+    train_rows: int
+    test_rows: int
+    train_start: datetime
+    train_end: datetime
+    test_start: datetime
+    test_end: datetime
+    metrics: RegressionMetrics
+
+
+def chronological_split(
+    rows: Sequence[DemandDatasetRow], test_fraction: float = 0.2
+) -> tuple[Sequence[DemandDatasetRow], Sequence[DemandDatasetRow]]:
+    if not 0 < test_fraction < 1:
+        raise ValueError("test_fraction must be between zero and one")
+    if len(rows) < 2:
+        raise ValueError("at least two rows are required")
+    if any(
+        current.timestamp <= previous.timestamp
+        for previous, current in zip(rows, rows[1:], strict=False)
+    ):
+        raise ValueError("rows must be strictly ordered by timestamp")
+    split_index = int(len(rows) * (1 - test_fraction))
+    if split_index == 0 or split_index == len(rows):
+        raise ValueError("split must leave rows in train and test")
+    return rows[:split_index], rows[split_index:]
+
+
+def regression_metrics(actual: Sequence[float], predicted: Sequence[float]) -> RegressionMetrics:
+    if len(actual) != len(predicted) or not actual:
+        raise ValueError("actual and predicted must have the same nonzero length")
+    errors = [
+        actual_value - predicted_value
+        for actual_value, predicted_value in zip(actual, predicted, strict=True)
+    ]
+    mae = sum(abs(error) for error in errors) / len(errors)
+    mse = sum(error**2 for error in errors) / len(errors)
+    actual_mean = sum(actual) / len(actual)
+    total_variance = sum((value - actual_mean) ** 2 for value in actual)
+    residual_variance = sum(error**2 for error in errors)
+    r2 = 1 - residual_variance / total_variance if total_variance else float(residual_variance == 0)
+    return RegressionMetrics(mae=mae, rmse=math.sqrt(mse), r2=r2)
+
+
+def evaluate_hour_weekday_baseline(
+    rows: Sequence[DemandDatasetRow], test_fraction: float = 0.2
+) -> BaselineEvaluation:
+    train, test = chronological_split(rows, test_fraction)
+    totals: defaultdict[tuple[int, int], float] = defaultdict(float)
+    counts: defaultdict[tuple[int, int], int] = defaultdict(int)
+    for row in train:
+        bucket = (row.day_of_week, row.hour)
+        totals[bucket] += row.demand_kw_next_60_minutes
+        counts[bucket] += 1
+    global_average = sum(row.demand_kw_next_60_minutes for row in train) / len(train)
+    predictions = [
+        totals[(row.day_of_week, row.hour)] / counts[(row.day_of_week, row.hour)]
+        if counts[(row.day_of_week, row.hour)]
+        else global_average
+        for row in test
+    ]
+    metrics = regression_metrics(
+        [row.demand_kw_next_60_minutes for row in test], predictions
+    )
+    return BaselineEvaluation(
+        strategy="historical_mean_by_hour_and_day_of_week",
+        train_rows=len(train),
+        test_rows=len(test),
+        train_start=train[0].timestamp,
+        train_end=train[-1].timestamp,
+        test_start=test[0].timestamp,
+        test_end=test[-1].timestamp,
+        metrics=metrics,
+    )

backend/app/ml/dataset.py

@@ -0,0 +1,190 @@
+"""Deterministic temporal dataset generation for demand forecasting.
+
+The generator deliberately produces station-level observations without calling
+the production power allocator.  It is training-data scaffolding, not a second
+implementation of the energy-management business rules.
+"""
+
+import csv
+import math
+import random
+from collections import defaultdict
+from collections.abc import Sequence
+from dataclasses import asdict, dataclass, fields
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+INTERVAL_MINUTES = 5
+FORECAST_HORIZON_MINUTES = 60
+HORIZON_STEPS = FORECAST_HORIZON_MINUTES // INTERVAL_MINUTES
+
+
+@dataclass(frozen=True, slots=True)
+class DatasetConfig:
+    days: int = 90
+    seed: int = 42
+    start: datetime = datetime(2026, 1, 1, tzinfo=UTC)
+    grid_limit_kw: float = 180.0
+    solar_peak_kw: float = 90.0
+
+
+@dataclass(frozen=True, slots=True)
+class DemandObservation:
+    timestamp: datetime
+    active_sessions: int
+    requested_power_kw: float
+    allocated_power_kw: float
+    solar_available_kw: float
+    solar_used_kw: float
+    grid_power_kw: float
+    total_demand_kw: float
+
+
+@dataclass(frozen=True, slots=True)
+class DemandDatasetRow:
+    timestamp: datetime
+    hour: int
+    day_of_week: int
+    is_weekend: bool
+    active_sessions: int
+    requested_power_kw: float
+    allocated_power_kw: float
+    solar_available_kw: float
+    solar_used_kw: float
+    grid_power_kw: float
+    total_demand_kw: float
+    current_demand_kw: float
+    historical_avg_demand_kw: float
+    demand_kw_next_60_minutes: float
+
+
+DATASET_COLUMNS = tuple(field.name for field in fields(DemandDatasetRow))
+FEATURE_COLUMNS = tuple(
+    column for column in DATASET_COLUMNS if column != "demand_kw_next_60_minutes"
+)
+
+
+def _validate_config(config: DatasetConfig) -> None:
+    if config.days <= 0:
+        raise ValueError("days must be positive")
+    if config.start.tzinfo is None or config.start.utcoffset() is None:
+        raise ValueError("start must be timezone-aware")
+    if config.grid_limit_kw <= 0 or config.solar_peak_kw < 0:
+        raise ValueError("power limits must be valid")
+
+
+def generate_observations(config: DatasetConfig) -> list[DemandObservation]:
+    """Generate 90 days plus the look-ahead needed to label the final row."""
+
+    _validate_config(config)
+    rng = random.Random(config.seed)
+    row_count = config.days * 24 * 60 // INTERVAL_MINUTES + HORIZON_STEPS
+    observations: list[DemandObservation] = []
+
+    for step in range(row_count):
+        timestamp = config.start + timedelta(minutes=step * INTERVAL_MINUTES)
+        fractional_hour = timestamp.hour + timestamp.minute / 60
+        weekday_factor = 0.78 if timestamp.weekday() >= 5 else 1.0
+        morning = math.exp(-((fractional_hour - 8.0) / 2.8) ** 2)
+        evening = math.exp(-((fractional_hour - 18.0) / 3.4) ** 2)
+        activity_level = weekday_factor * (1.6 + 8.2 * morning + 11.5 * evening)
+        active_sessions = max(0, round(activity_level + rng.gauss(0, 1.15)))
+
+        average_request_kw = max(3.6, 9.2 + rng.gauss(0, 1.1))
+        requested_power_kw = active_sessions * average_request_kw
+
+        daylight = 0.0
+        if 6 < fractional_hour < 18:
+            daylight = math.sin(math.pi * (fractional_hour - 6) / 12)
+        seasonal_factor = 0.82 + 0.18 * math.sin(2 * math.pi * timestamp.timetuple().tm_yday / 365)
+        cloud_factor = min(1.0, max(0.35, 0.84 + rng.gauss(0, 0.07)))
+        solar_available_kw = config.solar_peak_kw * daylight * seasonal_factor * cloud_factor
+
+        # Synthetic capacity is intentionally local to the data generator.  It
+        # preserves physical consistency without owning production allocation.
+        allocated_power_kw = min(requested_power_kw, config.grid_limit_kw + solar_available_kw)
+        solar_used_kw = min(allocated_power_kw, solar_available_kw)
+        grid_power_kw = allocated_power_kw - solar_used_kw
+        observations.append(
+            DemandObservation(
+                timestamp=timestamp,
+                active_sessions=active_sessions,
+                requested_power_kw=round(requested_power_kw, 6),
+                allocated_power_kw=round(allocated_power_kw, 6),
+                solar_available_kw=round(solar_available_kw, 6),
+                solar_used_kw=round(solar_used_kw, 6),
+                grid_power_kw=round(grid_power_kw, 6),
+                total_demand_kw=round(allocated_power_kw, 6),
+            )
+        )
+    return observations
+
+
+def build_supervised_dataset(
+    observations: Sequence[DemandObservation],
+    *,
+    output_rows: int | None = None,
+) -> list[DemandDatasetRow]:
+    """Build causal features and a label exactly 60 minutes in the future.
+
+    Historical averages are calculated from earlier observations only.  The
+    current observation is added to the accumulator after its feature value is
+    read, which makes the no-look-ahead property explicit.
+    """
+
+    if len(observations) <= HORIZON_STEPS:
+        raise ValueError("observations must cover more than the forecast horizon")
+    for previous, current in zip(observations, observations[1:], strict=False):
+        if current.timestamp - previous.timestamp != timedelta(minutes=INTERVAL_MINUTES):
+            raise ValueError("observations must be unique, ordered 5-minute intervals")
+
+    available_rows = len(observations) - HORIZON_STEPS
+    row_count = available_rows if output_rows is None else output_rows
+    if not 0 < row_count <= available_rows:
+        raise ValueError("output_rows exceeds observations with known targets")
+
+    totals: defaultdict[tuple[int, int], float] = defaultdict(float)
+    counts: defaultdict[tuple[int, int], int] = defaultdict(int)
+    rows: list[DemandDatasetRow] = []
+    for index, observation in enumerate(observations[:row_count]):
+        bucket = (observation.timestamp.weekday(), observation.timestamp.hour)
+        historical_average = totals[bucket] / counts[bucket] if counts[bucket] else 0.0
+        target = observations[index + HORIZON_STEPS].total_demand_kw
+        rows.append(
+            DemandDatasetRow(
+                timestamp=observation.timestamp,
+                hour=observation.timestamp.hour,
+                day_of_week=observation.timestamp.weekday(),
+                is_weekend=observation.timestamp.weekday() >= 5,
+                active_sessions=observation.active_sessions,
+                requested_power_kw=observation.requested_power_kw,
+                allocated_power_kw=observation.allocated_power_kw,
+                solar_available_kw=observation.solar_available_kw,
+                solar_used_kw=observation.solar_used_kw,
+                grid_power_kw=observation.grid_power_kw,
+                total_demand_kw=observation.total_demand_kw,
+                current_demand_kw=observation.total_demand_kw,
+                historical_avg_demand_kw=round(historical_average, 6),
+                demand_kw_next_60_minutes=target,
+            )
+        )
+        totals[bucket] += observation.total_demand_kw
+        counts[bucket] += 1
+    return rows
+
+
+def generate_dataset(config: DatasetConfig | None = None) -> list[DemandDatasetRow]:
+    config = config or DatasetConfig()
+    expected_rows = config.days * 24 * 60 // INTERVAL_MINUTES
+    return build_supervised_dataset(generate_observations(config), output_rows=expected_rows)
+
+
+def write_dataset_csv(rows: Sequence[DemandDatasetRow], output_path: Path) -> None:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open("w", encoding="utf-8", newline="") as output_file:
+        writer = csv.DictWriter(output_file, fieldnames=DATASET_COLUMNS)
+        writer.writeheader()
+        for row in rows:
+            values = asdict(row)
+            values["timestamp"] = row.timestamp.isoformat()
+            writer.writerow(values)