Python heapq: Colas de prioridad y operaciones de heap simplificadas

Name

Soren Atelier

Updated on 10/2/2026

Necesitas procesar elementos por prioridad -- tareas en un planificador de trabajos, eventos en una simulación o el camino más corto en un grafo. Una lista ordenada parece la opción obvia, pero insertar en una lista ordenada es O(n), lo que se convierte en un cuello de botella cuando procesas millones de elementos. Reordenar toda la lista cada vez que agregas un elemento es aún peor con O(n log n).

El costo se acumula rápidamente. Un planificador de trabajos que reordena su cola en cada inserción desperdicia la mayor parte de su tiempo en tareas administrativas. Un algoritmo de búsqueda de caminos que escanea una lista completa buscando la distancia mínima en cada paso convierte una operación de tiempo lineal en una cuadrática.

El módulo heapq de Python proporciona una cola de prioridad basada en heap que ofrece inserción O(log n) y extracción O(log n) del elemento más pequeño. También incluye funciones optimizadas para encontrar los N elementos más grandes o más pequeños de cualquier iterable sin ordenar toda la colección. Esta guía cubre cada función en heapq con ejemplos prácticos.

¿Qué es un Heap?

Un heap es un árbol binario donde cada nodo padre es menor o igual que sus hijos (min-heap). El heapq de Python implementa un min-heap usando una lista regular, donde el elemento más pequeño siempre está en el índice 0.

Las operaciones clave y sus complejidades temporales:

Operaciones básicas del Heap

Crear y usar un Heap

import heapq
 
# Start with an empty heap (just a regular list)
heap = []
 
# Push items
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
heapq.heappush(heap, 2)
 
print(heap)        # [1, 2, 3, 5] (heap order, not sorted)
print(heap[0])     # 1 (smallest element, always at index 0)
 
# Pop smallest item
smallest = heapq.heappop(heap)
print(smallest)    # 1
print(heap)        # [2, 5, 3]

Convertir una lista en un Heap

import heapq
 
data = [5, 3, 8, 1, 2, 7, 4, 6]
heapq.heapify(data)  # In-place, O(n)
print(data)  # [1, 2, 4, 3, 5, 7, 8, 6] (heap order)
 
# Pop elements in sorted order
sorted_data = []
while data:
    sorted_data.append(heapq.heappop(data))
print(sorted_data)  # [1, 2, 3, 4, 5, 6, 7, 8]

heappushpop y heapreplace

Estas operaciones combinadas son más eficientes que llamadas separadas de push/pop:

import heapq
 
heap = [1, 3, 5, 7]
heapq.heapify(heap)
 
# heappushpop: push first, then pop smallest
result = heapq.heappushpop(heap, 2)
print(result)  # 1 (pushed 2, popped 1)
print(heap)    # [2, 3, 5, 7]
 
# heapreplace: pop smallest first, then push
result = heapq.heapreplace(heap, 10)
print(result)  # 2 (popped 2, pushed 10)
print(heap)    # [3, 7, 5, 10]

nlargest y nsmallest

Estas funciones encuentran los N elementos más grandes o más pequeños sin ordenar toda la colección.

import heapq
 
scores = [85, 92, 78, 95, 88, 72, 91, 83, 97, 69]
 
# Top 3 scores
top_3 = heapq.nlargest(3, scores)
print(top_3)  # [97, 95, 92]
 
# Bottom 3 scores
bottom_3 = heapq.nsmallest(3, scores)
print(bottom_3)  # [69, 72, 78]

Con una función clave

Ambas aceptan un parámetro key para comparación personalizada:

import heapq
 
students = [
    {'name': 'Alice', 'gpa': 3.9},
    {'name': 'Bob', 'gpa': 3.5},
    {'name': 'Charlie', 'gpa': 3.8},
    {'name': 'Diana', 'gpa': 3.95},
    {'name': 'Eve', 'gpa': 3.2},
]
 
top_students = heapq.nlargest(3, students, key=lambda s: s['gpa'])
for s in top_students:
    print(f"{s['name']}: {s['gpa']}")
# Diana: 3.95
# Alice: 3.9
# Charlie: 3.8

Cuándo usar nlargest/nsmallest vs sorted()

Implementar un Max-Heap

El heapq de Python solo proporciona un min-heap. Para crear un max-heap, niega los valores:

import heapq
 
# Max-heap using negation
max_heap = []
for val in [5, 1, 3, 8, 2]:
    heapq.heappush(max_heap, -val)
 
# Pop largest items
while max_heap:
    print(-heapq.heappop(max_heap), end=' ')
# 8 5 3 2 1

Cola de prioridad con objetos personalizados

Para objetos que no son directamente comparables, usa tuplas (priority, counter, item):

import heapq
from dataclasses import dataclass
 
@dataclass
class Task:
    name: str
    description: str
 
# Use (priority, tie_breaker, task) tuples
task_queue = []
counter = 0
 
def add_task(priority, task):
    global counter
    heapq.heappush(task_queue, (priority, counter, task))
    counter += 1
 
def get_next_task():
    if task_queue:
        priority, _, task = heapq.heappop(task_queue)
        return task
    return None
 
add_task(3, Task("Low", "Clean up logs"))
add_task(1, Task("Critical", "Fix production bug"))
add_task(2, Task("Medium", "Update docs"))
add_task(1, Task("Critical", "Patch security hole"))
 
while task_queue:
    task = get_next_task()
    print(f"{task.name}: {task.description}")
 
# Critical: Fix production bug
# Critical: Patch security hole
# Medium: Update docs
# Low: Clean up logs

El contador sirve como desempate para que los elementos con igual prioridad se devuelvan en orden de inserción (FIFO).

Casos de uso prácticos

Fusionar iterables ordenados

import heapq
 
list1 = [1, 4, 7, 10]
list2 = [2, 5, 8, 11]
list3 = [3, 6, 9, 12]
 
merged = list(heapq.merge(list1, list2, list3))
print(merged)  # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]

heapq.merge() es eficiente en memoria -- no carga todas las listas en memoria a la vez.

Mediana en ejecución

import heapq
 
class RunningMedian:
    """Track the median of a stream of numbers."""
    def __init__(self):
        self.low = []   # max-heap (negated values)
        self.high = []  # min-heap
 
    def add(self, num):
        heapq.heappush(self.low, -num)
        heapq.heappush(self.high, -heapq.heappop(self.low))
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))
 
    def median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2
 
rm = RunningMedian()
for num in [2, 1, 5, 7, 2, 0, 5]:
    rm.add(num)
    print(f"Added {num}, median = {rm.median()}")

Camino más corto de Dijkstra

import heapq
 
def dijkstra(graph, start):
    """Find shortest paths from start to all nodes."""
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    pq = [(0, start)]
 
    while pq:
        dist, node = heapq.heappop(pq)
        if dist > distances[node]:
            continue
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            if new_dist < distances[neighbor]:
                distances[neighbor] = new_dist
                heapq.heappush(pq, (new_dist, neighbor))
 
    return distances
 
graph = {
    'A': [('B', 1), ('C', 4)],
    'B': [('C', 2), ('D', 5)],
    'C': [('D', 1)],
    'D': [],
}
 
print(dijkstra(graph, 'A'))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}

Analizar el rendimiento del Heap en Jupyter

Al comparar algoritmos basados en heap contra alternativas o perfilar el rendimiento de colas de prioridad, la experimentación interactiva acelera el proceso. RunCell (opens in a new tab) es un agente de IA para Jupyter que puede ayudarte a hacer benchmarks de operaciones de heap, visualizar características de rendimiento y optimizar tus decisiones de estructuras de datos de forma interactiva.

FAQ

¿Qué es heapq en Python?

heapq es un módulo de la biblioteca estándar que implementa un min-heap usando una lista regular de Python. Proporciona funciones para insertar y extraer elementos en tiempo O(log n) mientras mantiene la propiedad del heap (el elemento más pequeño siempre en el índice 0). Es la forma estándar de implementar colas de prioridad en Python.

¿Cómo creo un max-heap en Python?

El heapq de Python solo soporta min-heaps. Para simular un max-heap, niega los valores al insertar y niega de nuevo al extraer: heapq.heappush(heap, -value) para insertar, -heapq.heappop(heap) para obtener el más grande. Para objetos complejos, niega la prioridad en una tupla: (-priority, counter, item).

¿Cuándo debería usar heapq vs sorted()?

Usa heapq.nlargest(n, data) o heapq.nsmallest(n, data) cuando solo necesites los top N elementos y N sea pequeño relativo al tamaño total de los datos. Usa sorted() cuando necesites todos los elementos en orden o cuando N esté cerca del tamaño total. Usa min()/max() cuando N sea 1.

¿Es heapq seguro para hilos?

No. Las llamadas individuales de heappush y heappop no son atómicas. Si múltiples hilos acceden al mismo heap, necesitas sincronización externa. Para colas de prioridad seguras para hilos, usa queue.PriorityQueue, que envuelve heapq con bloqueo adecuado.

¿Cuál es la complejidad temporal de las operaciones de heapq?

heappush y heappop son ambos O(log n). heapify convierte una lista en un heap en O(n). nlargest(k, data) y nsmallest(k, data) se ejecutan en O(n log k) donde n es el tamaño de los datos y k es el número de elementos solicitados.

Conclusión

El módulo heapq de Python es la herramienta estándar para colas de prioridad y selección top-N. Su enfoque basado en heap proporciona operaciones de push y pop en O(log n), haciéndolo dramáticamente más rápido que mantener una lista ordenada. Usa heappush/heappop para patrones de colas de prioridad, nlargest/nsmallest para selección top-N, merge para combinar secuencias ordenadas y heapify cuando tengas una lista existente para convertir.