Python Advanced Data Structures & Collections

1. Overview & CPython Memory Architecture#

Writing high-performance Python code requires selecting the right algorithmic data structure and understanding how CPython manages memory beneath the abstraction layer.

mermaid
Ask AICopy
graph TD
 DS["Python Data Structure Landscape"]
 DS --> Seq["Sequential Data<br/>(list, tuple, deque)"]
 DS --> HashD["Hash-Based Mappings & Sets<br/>(dict, set, defaultdict, Counter)"]
 DS --> HeapD["Priority Queues & Trees<br/>(heapq, bisect, Tries)"]
 DS --> CustomD["Specialized & Extension Classes<br/>(UserDict, UserList, ChainMap)"]

1.1 CPython Compact Dictionary Internals#

Since Python 3.6+, CPython uses a compact dictionary representation that reduces memory consumption by ~25% and guarantees insertion-order preservation.

Architecture & Data Flow
Copy
Traditional Sparse Hash Table (Pre-3.6):
[hash, key_ptr, val_ptr] -> Heavy empty rows (lots of NULL memory wasted)

Modern Compact Hash Table (Python 3.6+):
Indices Table (Sparse array of small bytes): [0, None, 2, None, 1]
Entries Table (Dense contiguous array):
 Index 0: [hash, key_ptr, val_ptr]
 Index 1: [hash, key_ptr, val_ptr]
 Index 2: [hash, key_ptr, val_ptr]

---

2. Double-Ended Queues: collections.deque#

Standard Python list.pop(0) or list.insert(0, val) has an $O(n)$ time complexity because all remaining elements must shift in contiguous memory. collections.deque is implemented as a doubly-linked list of fixed-size blocks (64 elements per block), providing $O(1)$ operations at both ends.

mermaid
Ask AICopy
graph LR
 subgraph Deque ["collections.deque (Doubly-Linked Block List)"]
 Head["Block 0 (64 elements)"] <--> Body["Block 1 (64 elements)"] <--> Tail["Block 2 (64 elements)"]
 end
 LeftPush["appendleft() O(1)"] --> Head
 RightPush["append() O(1)"] --> Tail
 LeftPop["popleft() O(1)"] --> Head
 RightPop["pop() O(1)"] --> Tail

2.1 Ring Buffer / Moving Average Implementation#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from collections import deque
from typing import Iterable, List


class MovingAverageFilter:
 """Fixed-memory streaming window for sensor/telemetry analysis."""

 def __init__(self, window_size: int) -> None:
 self.window = deque(maxlen=window_size)
 self._running_sum = 0.0

 def add(self, sample: float) -> float:
 if len(self.window) == self.window.maxlen:
 # Subtract the item that is automatically discarded from the head
 self._running_sum -= self.window[0]

 self.window.append(sample)
 self._running_sum += sample
 return self._running_sum / len(self.window)


filter_stream = MovingAverageFilter(window_size=3)
for val in [10.0, 20.0, 30.0, 40.0, 50.0]:
 avg = filter_stream.add(val)
 print(f"Added: {val:<5} | Window: {list(filter_stream.window)} | Rolling Avg: {avg:.2f}")

---

3. The collections Toolkit#

3.1 collections.defaultdict#

Avoids missing-key checks by providing a zero-argument callable factory:

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from collections import defaultdict

# 1. Inverted Index for Text Search
corpus = {
 "doc1": "machine learning python algorithms",
 "doc2": "deep learning neural networks python",
 "doc3": "data analytics visualization python"
}

inverted_index = defaultdict(set)
for doc_id, text in corpus.items():
 for word in text.split():
 inverted_index[word].add(doc_id)

print(dict(inverted_index))
# {'machine': {'doc1'}, 'learning': {'doc1', 'doc2'}, 'python': {'doc1', 'doc2', 'doc3'}, ...}

---

3.2 collections.Counter (Multiset Operations)#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from collections import Counter

# 1. Anagram and Frequency Verification
def is_anagram(str1: str, str2: str) -> bool:
 return Counter(str1.lower()) == Counter(str2.lower())

print(is_anagram("Astronomer", "Moon starer")) # True

# 2. Token Bag-of-Words & Top-K Sampling
text_tokens = "the quick brown fox jumps over the lazy dog and the dog barked".split()
counter = Counter(text_tokens)

print(counter.most_common(2)) # [('the', 3), ('dog', 2)]

# 3. Multiset Subtraction
required_ingredients = Counter(flour=3, sugar=2, eggs=4)
pantry = Counter(flour=5, sugar=1, eggs=4)

missing = required_ingredients - pantry
print(f"Missing ingredients: {missing}") # Counter({'sugar': 1})

---

3.3 collections.ChainMap (Contextual Scope Hierarchy)#

ChainMap links multiple dictionaries into an ordered lookup sequence without copying data. Updates mutate the first mapping:

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from collections import ChainMap

cli_args = {"log_level": "DEBUG"}
env_vars = {"log_level": "INFO", "database_host": "10.0.0.1"}
defaults = {"log_level": "WARNING", "database_host": "localhost", "timeout": 30}

# Scoped context lookup
config = ChainMap(cli_args, env_vars, defaults)

print(config["log_level"]) # "DEBUG" (from CLI)
print(config["database_host"]) # "10.0.0.1" (from ENV)
print(config["timeout"]) # 30 (from defaults)

# Mutating updates the primary (first) layer only
config["timeout"] = 60
print(cli_args["timeout"]) # 60
print(defaults["timeout"]) # 30 (untouched)

---

4. Subclassing UserDict vs. Built-in dict#

The Built-in Subclassing Trap: In CPython, C-level methods of built-in dict (like dict.update() or __getitem__ inside C code) bypass overridden Python methods. Always subclass collections.UserDict or collections.UserList instead!

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from collections import UserDict

# INCORRECT: INCORRECT (Bypasses __setitem__ in C-calls like update())
class BrokenCaseInsensitiveDict(dict):
 def __setitem__(self, key, value):
 super().__setitem__(key.lower(), value)

d_bad = BrokenCaseInsensitiveDict()
d_bad.update({"UserName": "Alice"})
print(d_bad.keys()) # dict_keys(['UserName']) — OVERRIDE WAS BYPASSED!

# CORRECT: CORRECT: collections.UserDict routes all mutations through Python methods
class CaseInsensitiveDict(UserDict):
 def __setitem__(self, key: str, value):
 super().__setitem__(key.lower(), value)

 def __getitem__(self, key: str):
 return super().__getitem__(key.lower())

 def __contains__(self, key: object) -> bool:
 if isinstance(key, str):
 return super().__contains__(key.lower())
 return False

d_good = CaseInsensitiveDict()
d_good.update({"UserName": "Alice"})
print(d_good.keys()) # dict_keys(['username'])
print(d_good["USERNAME"]) # "Alice"

---

5. Priority Queues & Heap Operations with heapq#

heapq implements a binary Min-Heap on native Python lists.

• Child of element $k$: at indices $2k + 1$ and $2k + 2$.
• Parent of element $k$: at index $\lfloor(k - 1) / 2\rfloor$.

mermaid
Ask AICopy
graph TD
 N1["1 (Root - Min)"] --> N5["5"]
 N1 --> N3["3"]
 N5 --> N12["12"]
 N5 --> N8["8"]
 N3 --> N10["10"]
 N3 --> N4["4"]

5.1 Real-Time Streaming Median Calculator#

Using two heaps (Max-Heap for lower half, Min-Heap for upper half):

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import heapq


class MedianFinder:
 """Calculates running median of a streaming dataset in O(log n) insert and O(1) lookup."""

 def __init__(self):
 # max_heap stores lower half (negated values since heapq is min-heap)
 self.max_heap = []
 # min_heap stores upper half
 self.min_heap = []

 def add_num(self, num: float) -> None:
 # Step 1: Add to max_heap (invert sign)
 heapq.heappush(self.max_heap, -num)

 # Balance: largest in max_heap must be <= smallest in min_heap
 if self.max_heap and self.min_heap and (-self.max_heap[0] > self.min_heap[0]):
 val = -heapq.heappop(self.max_heap)
 heapq.heappush(self.min_heap, val)

 # Size balance: max_heap can have at most 1 more element than min_heap
 if len(self.max_heap) > len(self.min_heap) + 1:
 val = -heapq.heappop(self.max_heap)
 heapq.heappush(self.min_heap, val)
 elif len(self.min_heap) > len(self.max_heap):
 val = heapq.heappop(self.min_heap)
 heapq.heappush(self.max_heap, -val)

 def find_median(self) -> float:
 if len(self.max_heap) > len(self.min_heap):
 return float(-self.max_heap[0])
 return (-self.max_heap[0] + self.min_heap[0]) / 2.0


mf = MedianFinder()
for n in [5, 15, 1, 3]:
 mf.add_num(n)
 print(f"Added {n:<2} | Running Median: {mf.find_median()}")
# Added 5 | Running Median: 5.0
# Added 15 | Running Median: 10.0
# Added 1 | Running Median: 5.0
# Added 3 | Running Median: 4.0

---

6. Binary Search & Sorted Insertion with bisect#

The bisect module implements binary search algorithms over sorted sequences in $O(\log n)$ time.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import bisect
from dataclasses import dataclass


@dataclass
class TimestampedMetric:
 timestamp: int
 metric_value: float

 # Define custom comparison for bisect searching
 def __lt__(self, other):
 if isinstance(other, TimestampedMetric):
 return self.timestamp < other.timestamp
 return self.timestamp < other


class MetricTimeSeries:
 def __init__(self):
 self.series: list[TimestampedMetric] = []

 def insert(self, ts: int, value: float) -> None:
 item = TimestampedMetric(ts, value)
 # Keeps list sorted upon insert in O(n) due to array shift, but O(log n) search
 bisect.insort(self.series, item)

 def find_closest_at(self, target_ts: int) -> TimestampedMetric:
 # Binary search for closest timestamp
 idx = bisect.bisect_left(self.series, target_ts)
 if idx == 0:
 return self.series[0]
 if idx == len(self.series):
 return self.series[-1]

 before = self.series[idx - 1]
 after = self.series[idx]
 return after if (after.timestamp - target_ts) < (target_ts - before.timestamp) else before

---

7. Production LRU Cache From Scratch#

An LRU (Least Recently Used) Cache combines a Hash Map ($O(1)$ lookup) with a Doubly Linked List ($O(1)$ node relocation):

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
from typing import Optional, Any


class Node:
 __slots__ = ("key", "val", "prev", "next")

 def __init__(self, key: Any, val: Any):
 self.key = key
 self.val = val
 self.prev: Optional["Node"] = None
 self.next: Optional["Node"] = None


class LRUCache:
 """Production LRU Cache with strictly O(1) get and put operations."""

 def __init__(self, capacity: int):
 self.capacity = capacity
 self.cache: dict[Any, Node] = {}

 # Dummy head and tail nodes to eliminate edge-case boundary checks
 self.head = Node(0, 0)
 self.tail = Node(0, 0)
 self.head.next = self.tail
 self.tail.prev = self.head

 def _remove(self, node: Node) -> None:
 """Unlink node from doubly linked list."""
 prev_node = node.prev
 next_node = node.next
 prev_node.next = next_node
 next_node.prev = prev_node

 def _add_to_front(self, node: Node) -> None:
 """Insert node right after dummy head (most recently used)."""
 node.next = self.head.next
 node.prev = self.head
 self.head.next.prev = node
 self.head.next = node

 def get(self, key: Any) -> Any:
 if key in self.cache:
 node = self.cache[key]
 self._remove(node)
 self._add_to_front(node)
 return node.val
 return -1

 def put(self, key: Any, value: Any) -> None:
 if key in self.cache:
 self._remove(self.cache[key])

 new_node = Node(key, value)
 self._add_to_front(new_node)
 self.cache[key] = new_node

 if len(self.cache) > self.capacity:
 # Evict least recently used (node right before dummy tail)
 lru_node = self.tail.prev
 self._remove(lru_node)
 del self.cache[lru_node.key]

---

8. High-Performance Prefix Search: The Trie Data Structure#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
class TrieNode:
 __slots__ = ("children", "is_end_of_word")

 def __init__(self):
 self.children: dict[str, "TrieNode"] = {}
 self.is_end_of_word = False


class AutocompleteTrie:
 """Fast prefix search engine for autocomplete and NLP dictionary lookups."""

 def __init__(self):
 self.root = TrieNode()

 def insert(self, word: str) -> None:
 curr = self.root
 for char in word:
 if char not in curr.children:
 curr.children[char] = TrieNode()
 curr = curr.children[char]
 curr.is_end_of_word = True

 def starts_with_prefix(self, prefix: str) -> list[str]:
 curr = self.root
 for char in prefix:
 if char not in curr.children:
 return []
 curr = curr.children[char]

 # DFS traversal to collect all words under this prefix branch
 results = []
 def _dfs(node: TrieNode, path: list[str]):
 if node.is_end_of_word:
 results.append(prefix + "".join(path))
 for ch, child_node in node.children.items():
 path.append(ch)
 _dfs(child_node, path)
 path.pop()

 _dfs(curr, [])
 return results


trie = AutocompleteTrie()
for term in ["python", "pytorch", "pydantic", "pytest", "pandas", "algorithm"]:
 trie.insert(term)

print(trie.starts_with_prefix("py")) # ['python', 'pytorch', 'pydantic', 'pytest']

---

9. Comprehensive Time & Space Complexity Matrix#