Pre-processamento de Dados

Pré-processamento de dados e engenharia de recursos¶

Registros de viagem brutos são registros de transações (eventos individuais). Para construir um modelo de previsão de séries temporais, devemos limpar, agregar e enriquecer esses eventos em um conjunto de dados estruturado e tabular que represente a demanda horária.

1. O pipeline da transformação do pré-processamento¶

Nossa etapa de pré-processamento limpa os registros de parquet crus e os agrega:

1. Filtragem & Limpeza:
   • As durações da viagem devem ser estritamente positivas (tempo de entrega > tempo de captação).
   • Os locais de retirada devem corresponder a IDs de zona TLC válidos ($1 \le \text{PULocationID} \le 263$).
   • As distâncias de viagem devem ser não negativas.
2. Time-Series Agrupamento horário:
   • As viagens são agrupadas por PULocationID (ID da zona) e pickup_hour (por exemplo, 2023-01-01 14:00:00). Isso produz uma série de contagens regulares de coleta para cada zona.
3. Criação de Alvo Supervisionado:
   • Para cada zona, as contagens de coleta são ordenadas cronologicamente. A coluna de destino next_hour_pickup_count é criada deslocando as contações de coleta por hora para 1 hora: $$\text{Target}(z, t) = \text{PickupCount}(z, t+1)$$

2. Engenharia Avançada de Características & Codificações¶

Para ajudar modelos lineares e baseados em árvores a capturar padrões temporais, usamos duas técnicas avançadas de engenharia de recursos:

Lags Históricos¶

Para previsão de séries temporais, o melhor preditor da demanda de amanhã é a demanda de ontem, e o melhor preditivo da demanda da próxima hora é a procura da última hora. Nós alavancamos lags históricos da contagem de alvos:

• pickup_count_lag_1: Demanda na hora anterior (captura o momentum de curto prazo).
• pickup_count_lag_2: Demanda há duas horas (captura tendências de curto prazo).
• pickup_count_lag_24: Demand há 24 horas (captures diurnal sazonality).
• pickup_count_lag_168: Demanda 168 horas atrás (captura a sazonalidade semanal a mesma hora do mesmo dia da semana passada).

Cíclica Temporal Codificação¶

Características temporais padrão (como hour de $0$ para $23$) são entradas lineares. Um modelo linear se esforça para entender que a hora 23 e a hora 0 são consecutivas. Para preservar essa proximidade periódica, transformamos as características usando Codificações seno e cosseno:

$$\text{hour\_sin} = \sin\left(\frac{2\pi \cdot \text{hour}}{24}\right)$$ $$\text{hour\_cos} = \cos\left(\frac{2\pi \cdot \text{hour}}{24}\right)$$ $$\text{day\_of\_week\_sin} = \sin\left(\frac{2\pi \cdot \text{day\_of\_week}}{7}\right)$$ $$\text{day\_of\_week\_cos} = \cos\left(\frac{2\pi \cdot \text{day\_of\_week}}{7}\right)$$

Isso mapeia variáveis de hora / dia em um círculo unitário, garantindo que os limites periódicos sejam representados sem problemas.

3. Point-in-Time Joins & Leakage Prevention¶

Em sistemas de séries temporais de produção, vazamento de dados é um modo de falha comum. O vazamento ocorre quando recursos do futuro (por exemplo, demanda de atraso calculada usando informações após a hora de destino) são acidentalmente incluídos no quadro de treinamento.

Para evitar isso, nosso Feature Store (Feast) executa Point-in-Time (As-Of) Junta-se:

• For each record in our entity dataframe (representing a specific zone $z$ and event timestamp $t$), Feast joins features (such as historical pickup lags) using only the values that were available em ou antes timestamp $t$.
• Isso garante que nosso conjunto de dados de treinamento corresponda ao estado de informação exato que existiria na produção durante a inferência em tempo real.

4. Referência do código¶

Abaixo está o código que constrói os conjuntos de dados de treinamento supervisionados:

from pathlib import Path

import pandas as pd


class NextHourDemandDatasetBuilder:
    """Build zone-hour demand rows with the next hour as target.

    Example:
        NextHourDemandDatasetBuilder().build(interim_path, processed_path)
    """

    def build(self, trips_path: Path, output_path: Path) -> Path:
        output_path.parent.mkdir(parents=True, exist_ok=True)
        trips = pd.read_parquet(trips_path)
        dataset = self.build_from_trips(trips)
        dataset.to_parquet(output_path, index=False)
        return output_path

    def build_from_trips(self, trips: pd.DataFrame) -> pd.DataFrame:
        hourly_demand = self._hourly_pickups(trips)
        # Join on the actual next hour: a next-observed-row shift mislabels
        # across gaps, and an absent (location, hour+1) row is ambiguous
        # (idle hour vs uncollected data), so only observed labels survive.
        next_hour = hourly_demand.loc[
            :, ["pickup_location_id", "pickup_hour", "pickup_count"]
        ].copy()
        next_hour["pickup_hour"] -= pd.Timedelta(hours=1)
        labeled = hourly_demand.merge(
            next_hour.rename(
                columns={"pickup_count": "next_hour_pickup_count"}
            ),
            on=["pickup_location_id", "pickup_hour"],
            how="left",
        )
        complete_rows = labeled.dropna(
            subset=["next_hour_pickup_count"]
        ).copy()
        complete_rows["next_hour_pickup_count"] = complete_rows[
            "next_hour_pickup_count"
        ].astype(int)
        return complete_rows.reset_index(drop=True)

    def _hourly_pickups(self, trips: pd.DataFrame) -> pd.DataFrame:
        pickup_hour = pd.to_datetime(trips["tpep_pickup_datetime"]).dt.floor(
            "h"
        )
        grouped = trips.assign(pickup_hour=pickup_hour).groupby(
            ["PULocationID", "pickup_hour"], as_index=False
        )
        demand = grouped.size().rename(
            columns={
                "PULocationID": "pickup_location_id",
                "size": "pickup_count",
            }
        )
        return self._add_calendar_columns(
            demand.sort_values(["pickup_location_id", "pickup_hour"])
        )

    def _add_calendar_columns(self, demand: pd.DataFrame) -> pd.DataFrame:
        demand["hour"] = demand["pickup_hour"].dt.hour
        demand["day_of_week"] = demand["pickup_hour"].dt.dayofweek
        demand["is_weekend"] = demand["day_of_week"].isin([5, 6])
        demand["month"] = demand["pickup_hour"].dt.month
        return demand