Filtración intuitiva de marcos de datos temporales

Deshágase de su código ineficaz para filtrar datos de series temporales

Cada vez que trabajo con datos de series temporales, termino escribiendo código complejo y no reutilizable para filtrarlos. Ya sea que esté haciendo técnicas de filtrado simples, como eliminar fines de semana, o más complejas, como eliminar ventanas de tiempo específicas, siempre recurro a escribir una función rápida y sucia que funcione para lo específico que estoy filtrando en este momento, pero nunca más. .

Finalmente decidí romper ese horrible ciclo escribiendo un procesador que me permita filtrar series temporales sin importar cuán compleja sea la condición usando entradas muy simples y concisas.

Sólo un ejemplo de cómo funciona en la práctica:

• Entre semana, quiero eliminar < 6 a. m. y ≥ 8 p. m., y los fines de semana quiero eliminar < 8 a. m. y ≥ 10 p. m.

df = pl.DataFrame(
   {"date": [
      # -- may 24th is a Friday, weekday
      '2024-05-24 00:00:00',  # < 6 am, should remove
      '2024-05-24 06:00:00',  # not < 6 am, should keep
      '2024-05-24 06:30:00',  # not < 6 am, should keep
      '2024-05-24 20:00:00',  # >= 8 pm, should remove

      # -- may 25th is a Saturday, weekend
      '2024-05-25 00:00:00',  # < 8 am, should remove
      '2024-05-25 06:00:00',  # < 8 am, should remove
      '2024-05-25 06:30:00',  # < 8 am, should remove
      '2024-05-25 20:00:00',  # not >= 10 pm, should keep

      ]
   }
).with_columns(pl.col("date").str.strptime(pl.Datetime, "%Y-%m-%d %H:%M:%S"))

• Sin procesador: expresivo, pero prolijo y no reutilizable

df.filter(   
    pl.Expr.not_(
        (
            (pl.col("date").dt.weekday() < 6)
            .and_(
                (pl.col("date").dt.hour() < 6)
                .or_(pl.col("date").dt.hour() >= 20)
            )
        )
        .or_(
            (pl.col("date").dt.weekday() >= 6)
            .and_(
                (pl.col("date").dt.hour() < 8)
                .or_(pl.col("date").dt.hour() >= 22)
            )
        )
    )
)

• Con procesador: igual de expresivo, conciso y reutilizable

processor = FilterDataBasedOnTime(
    "date", time_patterns=[
        "<6wd<6h",
        "<6wd>=20h",
        ">=6wd<8h",
        ">=6wd>=22h",
    ]
)
processor.transform(df)

En este artículo explicaré cómo se me ocurrió esta solución, comenzando con el formato de cadena que elegí para definir las condiciones del filtro, seguido por el diseño del procesador en sí. Hacia el final del artículo, describiré cómo se puede utilizar este canal junto con otros canales para permitir el procesamiento complejo de series de tiempo con solo unas pocas líneas de código.

Si solo está interesado en el código, salte hasta el final del artículo para obtener un enlace al repositorio.

¿Condiciones de tiempo expresivas, concisas y flexibles?

Esta fue, con diferencia, la parte más difícil de esta tarea. Filtrar series temporales basadas en el tiempo es conceptualmente fácil, pero es mucho más difícil hacerlo con código. Mi idea inicial fue utilizar un patrón de cuerdas que me resulte más intuitivo:

# -- remove values between 6 am (inclusive) and 2 pm (exclusive)
pattern = '>=06:00,<14:00'

Pero con esto inmediatamente nos topamos con un problema: perdemos flexibilidad. Esto se debe a que 06:00 es ambiguo, ya que podría significar min:sec o hr:min . Por lo que casi siempre tendríamos que definir el formato de fecha a priori.

Esto nos impide permitir técnicas de filtrado complejas, como filtrar una hora específica EN días específicos (por ejemplo, eliminar solo valores en [6am, 2pm) on a Saturday).

Extending my pattern to something resembling cron would not help either:

# cronlike pattern
pattern = ‘>=X-X-X 06:00:X, <X-X-X 20:00:X’

The above can help with selecting specific months or years, but doesn’t allow flexibility of things like weekdays. Further, it is not very expressive with all the X’s and it’s really verbose.

I knew that I needed a pattern that allows chaining of individual time series components or units. Effectively something that is just like an if-statement:

• IF day == Saturday
• AND time ≥ 06:00
• AND time < 14:00

So then I thought, why not use a pattern where you can add any conditions to a time-components, with the implicit assumption that they are all AND conditions?

# -- remove values in [6am, 2pm) on Saturday
pattern = 'day==6,time>=06:00,time<14:00'

Now we have a pattern that is expressive, however it can still be ambiguous, since time implicitly assumes a date fomat. So I decided to go further:

# -- remove values in [6am, 2pm) on Saturday
pattern = 'day==6,hour>=6,hour<14'

Now to make it less verbose, I borrowed the Polars duration string format (this is the equivalent of “frequency” if you are more familiar with Pandas), and viola:

# -- remove values in [6am, 2pm) on Saturday
pattern = '==6wd,>=6h,<14h'

What About Time Conditions that Need the OR Operator?

Let’s consider a different condition: to filter anything LESS than 6 am (inclusive) and > 2 pm (exclusive). A pattern like below would fail:

# -- remove values in (-inf, 6am]y (2pm, inf)
patrón = '<=6h,>14h'

Ya que lo leeríamos como: ≤ 6 am Y > 2 pm

¡No existe ningún valor que satisfaga estas dos condiciones!

Pero la solución a esto es simple: aplicar condiciones AND dentro de un patrón y aplicar condiciones OR en diferentes patrones. Entonces:

# -- remove values in (-inf, 6am], and (2pm, inf)
patterns = ['<=6h', '>14h']

Se leería como: ≤ 6 am O > 2 pm

¿Por qué no permitir declaraciones OR dentro de un patrón?

Consideré agregar soporte para una declaración OR dentro del patrón, por ejemplo, usando | o, alternativamente, dejar que , denote la diferencia entre una condición “izquierda” y “derecha”. Sin embargo, descubrí que esto agregaría complejidad innecesaria al análisis del patrón, sin hacer que el código sea más expresivo.

Lo prefiero simple: dentro de un patrón aplicamos Y, entre patrones aplicamos O.

Casos extremos

Hay un caso extremo que vale la pena discutir aquí. El patrón tipo “declaración if” no siempre funciona.

Consideremos filtrar marcas de tiempo> 06:00. Si simplemente definimos:

# -- pattern to remove values > 06:00
pattern = '>6h'

Entonces interpretamos esto como:

• Eliminar todos los valores donde hora>6
• ¿O eliminar todos los valores donde time>06:00?

Esto último tiene más sentido, pero el patrón actual no nos permite expresarlo. Entonces, para indicar explícitamente qué incluir marcas de tiempo mayores que la sexta hora del día, debemos agregar lo que yo llamo el operador en cascada:

# -- pattern to remove values > 06:00
pattern = '>6h*'

Que se leería como:

• hora > 6
• O (hora == 6 Y cualquier (minuto, segundo, milisegundo, etc… > 0)

¡Lo cual sería una condición precisa para capturar la hora> 06:00!

El código

Aquí destaco partes de diseño importantes para crear un procesador para filtrar datos de series de tiempo.

Lógica de análisis

Dado que el patrón es bastante simple, analizarlo es realmente fácil. Todo lo que tenemos que hacer es recorrer cada patrón y realizar un seguimiento de los caracteres del operador. Lo que queda entonces es una lista de operadores y una lista de duraciones a las que se aplican.

# -- code for parsing a time pattern, e.g. "==7d<7h"
pattern = pattern.replace(" ", "")
operator = ""
operators = []
duration_string = ""
duration_strings = []
for char in pattern:
    if char in {">", "<", "=", "!"}:
        operator += char
        if duration_string:
            duration_strings.append(duration_string)
            duration_string = ""
    else:
        duration_string += char
        if operator:
            operators.append(operator)
            operator = ""
duration_strings.append(duration_string)

Ahora, para cada operador y cadena de duración, podemos extraer metadatos que nos ayudarán a crear las reglas booleanas reales más adelante.

# -- code for extracting metadata from a parsed pattern

# -- mapping to convert each operator to the Polars method
OPERATOR_TO_POLARS_METHOD_MAPPING = {
    "==": "eq",
    "!=": "ne",
    "<=": "le",
    "<": "lt",
    ">": "gt",
    ">=": "ge",
}

operator_method = (
    OPERATOR_TO_POLARS_METHOD_MAPPING[operator]
)

# -- identify cascade operations
if duration_string.endswith("*"):
    duration_string = duration_string[:-1]
    how = "cascade"
else:
    how = "simple"

# -- extract a polars duration, e.g. 7d7h into it's components: [(7, "d"), (7, "h")]
polars_duration = PolarsDuration(duration=duration_string)
decomposed_duration = polars_duration.decomposed_duration

# -- ensure that cascade operator only applied to durations that accept it
if how == "cascade" and any(
    unit not in POLARS_DURATIONS_TO_IMMEDIATE_CHILD_MAPPING
    for _, unit in decomposed_duration
):
    raise ValueError(
        (
            "You requested a cascade condition on an invalid "
            "duration. Durations supporting cascade: "
            f"{list(POLARS_DURATIONS_TO_IMMEDIATE_CHILD_MAPPING.keys())}"
        )
    )

rule_metadata = {
    "operator": operator_method,
    "decomposed_duration": decomposed_duration,
    "how": how,
}

Ahora tenemos, para cada patrón, diccionarios sobre cómo definir las reglas para cada uno de sus componentes. Entonces, si optamos por un ejemplo complicado:

pattern = '==1m>6d6h' # remove if month = Jan, and day > 6 and hour > 6

# parsed pattern

[
   [
      {
         "operator": "eq",
         "decomposed_duration": [(1, "m")],
         "how": "simple"
      },
      { 
         "operator": "gt",
         "decomposed_duration": [(6, "d"), (6, "h")],
         "how": "simple"
      }
   ]
]

Tenga en cuenta que un único patrón se puede dividir en varios dictados de metadatos porque puede estar compuesto de varias duraciones y operaciones.

Crear reglas a partir de metadatos

Después de haber creado metadatos para cada patrón, ¡ahora viene la parte divertida de crear reglas Polars!

Recuerde que dentro de cada patrón aplicamos una condición Y, pero entre patrones aplicamos una condición O. Entonces, en el caso más simple, necesitamos un contenedor que pueda tomar una lista de todos los metadatos para un patrón específico y luego aplicarle la condición and. Podemos almacenar esta expresión en una lista junto con las expresiones de todos los demás patrones, antes de aplicar la condición OR.

# -- dictionary to contain each unit along with the polars method to extract it's value
UNIT_TO_POLARS_METHOD_MAPPING = {
        "d": "day",
        "h": "hour",
        "m": "minute",
        "s": "second",
        "ms": "millisecond",
        "us": "microsecond",
        "ns": "nanosecond",
        "wd": "weekday",
}

patterns = ["==6d<6h6s"]
patterns_metadata = get_rule_metadata_from_patterns(patterns)

# -- create an expression for the rule pattern
pattern_metadata = patterns_metadata[0]  # list of length two

# -- let's consider the condition for ==6d
condition = pattern_metadata[0]

decomposed_duration = condition["decomposed_duration"]  # [(6, 'd')]
operator = condition["operator"]  # eq
conditions = [
    getattr(  # apply the operator method, e.g. pl.col("date").dt.hour().eq(value)
        getattr(  # get the value of the unit, e.g. pl.col("date").dt.hour()
            pl.col(time_column).dt,  
            UNIT_TO_POLARS_METHOD_MAPPING[unit], 
        )(),
        operator,
    )(value) for value, unit in decomposed_duration  # for each unit separately
]

# -- finally, we aggregate the separate conditions using an AND condition
final_expression = conditions.pop()
for expression in conditions:
    final_expression = getattr(final_expression, 'and_')(expression)

Esto parece complejo… pero podemos convertir partes de él en funciones y el código final se ve bastante limpio y legible:

rules = []  # list to store expressions for each time pattern
for rule_metadata in patterns_metadata:
    rule_expressions = []
    for condition in rule_metadata:
        how = condition["how"]
        decomposed_duration = condition["decomposed_duration"]
        operator = condition["operator"]
        if how == "simple":
            expression = generate_polars_condition(  # function to do the final combination of expressions
                [
                    self._generate_simple_condition(
                        unit, value, operator
                    )  # this is the complex "getattr" code
                    for value, unit in decomposed_duration
                ],
                "and_",
            )
        rule_expressions.append(expression)

    rule_expression = generate_polars_condition(
        rule_expressions, "and_"
    )
    rules.append(rule_expression)

overall_rule_expression = generate_polars_condition(
    rules, "or_"
).not_()  # we must negate because we're filtering!

Creando reglas para el operador en cascada

En el código anterior, tenía una condición if solo para las condiciones “simples”… ¿cómo hacemos las condiciones en cascada?

Recuerde de nuestra discusión anterior que un patrón de “>6h*” significa:

hora > 6 O (hora == 6 Y cualquiera (min, s, ms, etc… > 0)

Entonces, lo que necesitamos es saber para cada unidad cuáles son las unidades más pequeñas posteriores.

Por ejemplo, si tuviera “>6d*”, debería saber incluir “hora” en mi cualquier condición, así:

día > 6 O (día == 6 Y cualquiera (h, min, s, ms, etc… > 0)

Esto se logra fácilmente usando un diccionario que asigna cada unidad a su “siguiente” unidad más pequeña. Ej.: día → hora, hora → segundo, etc…

POLARS_DURATIONS_TO_IMMEDIATE_CHILD_MAPPING = {
    "y": {"next": "mo", "start": 1},
    "mo": {"next": "d", "start": 1},
    "d": {"next": "h", "start": 0},
    "wd": {"next": "h", "start": 0},
    "h": {"next": "m", "start": 0},
    "m": {"next": "s", "start": 0},
    "s": {"next": "ms", "start": 0},
    "ms": {"next": "us", "start": 0},
    "us": {"next": "ns", "start": 0},
}

El valor inicial es necesario porque cualquier condición no siempre es > 0. Porque si quiero filtrar cualquier valor > febrero, entonces 2023–02–02 debería ser parte de él, pero no 2023–02–01.

Con este diccionario en mente, podemos crear fácilmente cualquier condición:

# -- pattern example: >6h* cascade
simple_condition = self._generate_simple_condition(
    unit, value, operator
)  # generate the simple condition, e.g. hour>6
all_conditions = [simple_condition]
if operator == "gt":  # cascade only affects > operator
    equality_condition = self._generate_simple_condition(
        unit, value, "eq"
    )  # generate hour==6
    child_unit_conditions = []
    child_unit_metadata = (
        POLARS_DURATIONS_TO_IMMEDIATE_CHILD_MAPPING.get(unit, None)
    )  # get the next smallest unit, e.g. minute
    while child_unit_metadata is not None:
        start_value = child_unit_metadata["start"]
        child_unit = child_unit_metadata["next"]
        child_unit_condition = self._generate_simple_condition(
            child_unit, start_value, "gt"
        )  # generate minute > 0
        child_unit_conditions.append(child_unit_condition)
        child_unit_metadata = (
            POLARS_DURATIONS_TO_IMMEDIATE_CHILD_MAPPING.get(
                child_unit, None
            )
        )  # now go on to seconds, and so on...

    cascase_condition = generate_polars_condition(
        [
            equality_condition,  # and condition for the hour unit
            generate_polars_condition(child_unit_conditions, "or_"),  # any condition for all the child units
        ],
        "and_",
    )

    all_conditions.append(cascase_condition)

# -- final condition is hour>6 AND the cascade condition
overall_condition = generate_polars_condition(all_conditions, "or_")

La fotografía más grande

Un procesador como este no sólo es útil para análisis ad hoc. Puede ser un componente central de sus procesos de procesamiento de datos. Un caso de uso realmente útil para mí es usarlo junto con el remuestreo. Un paso de filtrado sencillo me permitiría calcular fácilmente métricas en series temporales con interrupciones o tiempos de inactividad regulares.

Además, con unas pocas modificaciones sencillas puedo ampliar este procesador para permitir un etiquetado sencillo de mis series temporales. Esto me permite agregar regresores a bits que sé que se comportan de manera diferente; por ejemplo, si estoy modelando una serie de tiempo que salta en horas específicas, puedo agregar un regresor escalonado solo a esas partes.

Observaciones finales

En este artículo, describí un procesador que permite un filtrado de series temporales fácil, flexible y conciso en conjuntos de datos Polars. La lógica discutida se puede extender a su biblioteca de procesamiento de marcos de datos favorita, como Pandas, con algunos cambios menores.

El procesador no sólo es útil para el análisis de series temporales ad hoc, sino que también puede ser la columna vertebral del procesamiento de datos si se encadena con otras operaciones como el remuestreo o si se utiliza para crear características adicionales para el modelado.

Concluiré con algunas extensiones que tengo en mente para mejorar aún más el código:

• Estoy pensando en crear un atajo para definir “fin de semana”, por ejemplo, “==nosotros”. De esta manera no necesitaría definir explícitamente “>=6wd”, lo cual puede ser menos claro.
• Con un diseño adecuado, creo que es posible permitir la adición de identificadores de tiempo personalizados. Por ejemplo, “==víspera” para indicar la noche, cuya hora puede ser definida por el usuario.
• Definitivamente voy a agregar soporte para simplemente etiquetar los datos, en lugar de filtrarlos.
• Y voy a agregar soporte para poder definir los límites como “mantener”, por ejemplo, en lugar de definir [“<6h”, “>=20hr”] puedo hacer [“>=6h<20hr”]

Dónde encontrar el código

Este proyecto está en su infancia, por lo que los elementos pueden moverse. A partir del 23.05.2024, puede encontrar FilterDataBasedOnTime en mix_n_match/main.py.

GitHub – namiyousef/mix-n-match: repositorio para procesar marcos de datos

Todo el código, datos e imágenes por autor a menos que se especifique lo contrario.

Filtración intuitiva de marcos de datos temporales fue publicado originalmente en Hacia la ciencia de datos en Medium, donde las personas continúan la conversación resaltando y respondiendo a esta historia.