479M Tuples Crashed a Pod — Python itertools Permutations

import itertools
import sys


# ── 1. Memory footprint: eager vs lazy ───────────────────────────────────────
# The eager list allocates every integer object reference upfront.
# The lazy iterator is a fixed-size C object regardless of how many values it produces.

eager_numbers = list(range(1_000_000))
lazy_numbers = itertools.islice(itertools.count(0), 1_000_000)

print(f"Eager list size  : {sys.getsizeof(eager_numbers):>12,} bytes")
print(f"Lazy iterator    : {sys.getsizeof(lazy_numbers):>12,} bytes")
print(f"Memory ratio     : {sys.getsizeof(eager_numbers) // sys.getsizeof(lazy_numbers):>12,}x more memory for eager")


# ── 2. Single-consumption: the most common itertools production bug ──────────
# There is no error, no warning — just empty results on the second pass.

it = itertools.chain([1, 2, 3], [4, 5, 6])
first_pass = list(it)
second_pass = list(it)  # exhausted — always empty

print(f"\nFirst  pass : {first_pass}")
print(f"Second pass : {second_pass}  <-- empty, not a bug — expected single-consumption behaviour")


# ── 3. The correct pattern: recreate from source for each pass ───────────────
# When materialising is not an option, wrap in a factory and call fresh.

def make_pipeline():
    """Returns a fresh iterator every time. No state shared between calls."""
    return itertools.chain([1, 2, 3], [4, 5, 6])

for pass_number in range(1, 4):
    result = list(make_pipeline())
    print(f"Pass {pass_number}: {result}")  # Always [1, 2, 3, 4, 5, 6]


# ── 4. chain object size: constant regardless of input ───────────────────────

chain_small = itertools.chain(range(10), range(10))
chain_large = itertools.chain(range(10_000_000), range(10_000_000))

print(f"\nchain over 20 items       : {sys.getsizeof(chain_small)} bytes")
print(f"chain over 20,000,000 items: {sys.getsizeof(chain_large)} bytes")
print("Both the same — chain holds references, not copies.")

import itertools


# ── 1. count + islice: bounded ID generator ──────────────────────────────────
# count() is infinite. islice() imposes the finite boundary.
# Never use count() in a for loop without a termination condition.

def generate_ids(prefix: str, count: int):
    for i in itertools.islice(itertools.count(1), count):
        yield f"{prefix}-{i:07d}"

ids = list(generate_ids("JOB", 5))
print(f"Generated IDs: {ids}")


# ── 2. cycle + islice: round-robin load balancer ─────────────────────────────

servers = ['api-a', 'api-b', 'api-c']
requests = [f"req-{i}" for i in range(9)]

for request, server in zip(requests, itertools.cycle(servers)):
    print(f"{request} -> {server}")


# ── 3. chain.from_iterable: lazy log file merger ─────────────────────────────
# WRONG pattern: chain(*handles) — all files opened at once at call time
# handles = [open(f) for f in file_list]  # all open simultaneously
# merged = itertools.chain(*handles)       # file descriptor exhaustion risk

# CORRECT pattern: chain.from_iterable with a generator that opens lazily
# Each file is opened only when the previous one is exhausted.

def open_files_lazily(file_paths):
    """Opens and yields lines from each file, one file at a time."""
    for path in file_paths:
        try:
            with open(path, 'r', encoding='utf-8') as f:
                yield from f
        except FileNotFoundError:
            pass  # skip missing log files gracefully

# Usage: processes 10,000 log files with at most 1 open file descriptor at a time
# for line in open_files_lazily(all_log_paths):
#     process(line)


# ── 4. takewhile + dropwhile: conditional stream windowing ───────────────────

data = [0, 0, 0, 1, 2, 3, 0, 4, 5]

take = list(itertools.takewhile(lambda x: x == 0, data))  # leading zeros
skip = list(itertools.dropwhile(lambda x: x == 0, data))  # skip leading zeros

print(f"\nOriginal       : {data}")
print(f"Leading zeros  : {take}")
print(f"After zeros    : {skip}")


# ── 5. zip_longest: safe merging of unequal-length streams ───────────────────
# Standard zip silently truncates to the shorter sequence.
# zip_longest makes the length mismatch explicit and fills gaps.

stream_a = [1, 2, 3, 4, 5]
stream_b = ['a', 'b', 'c']  # shorter — zip would silently drop items 4 and 5

print("\nWith zip (silent truncation):")
for pair in zip(stream_a, stream_b):
    print(f"  {pair}")

print("\nWith zip_longest (explicit fill):")
for pair in itertools.zip_longest(stream_a, stream_b, fillvalue='MISSING'):
    print(f"  {pair}")

import itertools
import sys

PYTHON_VERSION = sys.version_info


# ── pairwise (Python 3.10+) ───────────────────────────────────────────────────
# Returns successive overlapping pairs from an iterable.
# Essential for time-series deltas, state transitions, consecutive comparisons.

if PYTHON_VERSION >= (3, 10):
    # Canonical — clear, lazy, correct
    prices = [10.0, 10.5, 9.8, 11.2, 11.0, 12.3]
    deltas = [(b - a) for a, b in itertools.pairwise(prices)]
    print(f"Prices : {prices}")
    print(f"Deltas : {[round(d, 2) for d in deltas]}")

    # State transition detection in a log stream
    states = ['idle', 'idle', 'processing', 'processing', 'idle', 'error']
    transitions = [
        (prev, curr)
        for prev, curr in itertools.pairwise(states)
        if prev != curr
    ]
    print(f"\nState transitions: {transitions}")
else:
    print("pairwise requires Python 3.10+ — using manual equivalent")
    def pairwise_compat(iterable):
        """Pre-3.10 equivalent. Materialises one copy."""
        a, b = itertools.tee(iterable)
        next(b, None)
        return zip(a, b)


# ── batched (Python 3.12+) ────────────────────────────────────────────────────
# Chunks an iterable into tuples of at most n items.
# Final chunk may be shorter if the iterable does not divide evenly.

if PYTHON_VERSION >= (3, 12):
    records = list(range(1, 11))  # [1, 2, 3, ..., 10]
    print(f"\nRecords: {records}")
    print("Batched into groups of 3:")
    for batch in itertools.batched(records, 3):
        print(f"  {batch}")

    # Real-world use: chunked API writes
    def insert_in_chunks(items, chunk_size: int, insert_fn):
        """Processes items in chunks without materialising the full list."""
        for batch in itertools.batched(items, chunk_size):
            insert_fn(batch)

    def simulated_db_insert(batch):
        print(f"  Inserting batch of {len(batch)}: {batch}")

    print("\nChunked DB inserts:")
    insert_in_chunks(range(1, 8), 3, simulated_db_insert)
else:
    print("batched requires Python 3.12+ — using manual equivalent")
    def batched_compat(iterable, n):
        """Pre-3.12 equivalent using islice."""
        it = iter(iterable)
        while True:
            batch = tuple(itertools.islice(it, n))
            if not batch:
                return
            yield batch


# ── Combining pairwise + batched: sliding window analysis ────────────────────
# Real use case: detect anomalies in sensor readings by comparing consecutive
# values in batches, without loading the entire stream.

if PYTHON_VERSION >= (3, 12):
    sensor_readings = [100, 102, 98, 250, 101, 99, 103, 400, 97]
    threshold = 50

    print("\nAnomaly detection (consecutive delta > threshold):")
    for a, b in itertools.pairwise(sensor_readings):
        delta = abs(b - a)
        if delta > threshold:
            print(f"  Anomaly: {a} -> {b} (delta={delta})")

import itertools
import math
import sys


# ── 1. Size check before materialising: always do this first ─────────────────

n, r = 12, 12

perm_count = math.perm(n, r)   # 479,001,600
comb_count = math.comb(n, r)   # 1 (choosing all 12 from 12 is one combination)
comb_10 = math.comb(n, 10)     # 66 (more interesting: choosing 10 from 12)

print(f"permutations({n}, {r}) = {perm_count:>15,} tuples")
print(f"combinations({n}, {r}) = {comb_count:>15,} tuples")
print(f"combinations({n}, 10) = {comb_10:>15,} tuples")

# Memory estimate on 64-bit CPython 3.11
# Each tuple of 12 elements: 56 byte header + 12 pointers * 8 bytes = 152 bytes
# Small integers (0-11) are interned so no per-tuple allocation for the ints themselves
bytes_per_tuple = 56 + (r * 8)
print(f"\nBytes per tuple of {r} elements: {bytes_per_tuple}")
print(f"Memory for list(permutations({n},{r})): ~{perm_count * bytes_per_tuple / 1e9:.1f} GB")
print(f"Memory for list(combinations({n},10)): ~{comb_10 * (56 + 10*8) / 1e6:.1f} MB  <- safe")


# ── 2. Safe lazy processing: filter directly on the iterator ─────────────────
# Rule: filter before you materialise, never after.

def is_ascending(t):
    """True if the tuple values are strictly increasing."""
    return all(a < b for a, b in itertools.pairwise(t))  # pairwise from 3.10+

# Process lazily: tuples that don't pass the filter are never stored
ascending_count = 0
for perm in itertools.permutations(range(6)):  # 720 permutations — safe to iterate fully
    if is_ascending(perm):
        ascending_count += 1

print(f"\nAscending permutations of range(6): {ascending_count}")  # should be 1: (0,1,2,3,4,5)


# ── 3. islice: bounded sampling from a huge combinatoric space ───────────────

print("\nFirst 5 permutations of range(12) (sample only — not materialised fully):")
for p in itertools.islice(itertools.permutations(range(12)), 5):
    print(f"  {p}")


# ── 4. product: more predictable growth than permutations ────────────────────
# product(n, repeat=r) = n^r — exponential but predictable and often much smaller

config_values = {
    'workers': [2, 4, 8],
    'timeout': [30, 60],
    'retries': [1, 3, 5],
}
keys = list(config_values.keys())
value_lists = list(config_values.values())

grid_size = math.prod(len(v) for v in value_lists)
print(f"\nConfig grid size: {grid_size} combinations")
print("Configs:")
for combo in itertools.product(*value_lists):
    config = dict(zip(keys, combo))
    print(f"  {config}")


# ── 5. Pre-flight assertion: catch it before it crashes ──────────────────────

MAX_SAFE_MATERIALISE = 1_000_000

def safe_permutations(items, r=None):
    """
    Yields permutations lazily after confirming the count is within budget.
    Raises ValueError before allocating anything if the count exceeds the limit.
    """
    n = len(items)
    r = r if r is not None else n
    count = math.perm(n, r)
    if count > MAX_SAFE_MATERIALISE:
        raise ValueError(
            f"permutations({n}, {r}) would produce {count:,} tuples — "
            f"exceeds safe limit of {MAX_SAFE_MATERIALISE:,}. "
            f"Use filter() and islice() on the lazy iterator instead."
        )
    return itertools.permutations(items, r)

try:
    result = safe_permutations(list(range(12)))
except ValueError as e:
    print(f"\nPre-flight check caught: {e}")

import itertools
import timeit
import sys


# ── 1. compress vs list comprehension filter ─────────────────────────────────
# compress applies a boolean mask in C — no Python call per element.
# List comprehension executes Python bytecode for every element.

records = [f"row-{i}" for i in range(500_000)]
mask = [i % 3 == 0 for i in range(500_000)]

def list_comp_filter():
    return [r for r, m in zip(records, mask) if m]

def itertools_compress_filter():
    return list(itertools.compress(records, mask))

time_comp = timeit.timeit(list_comp_filter, number=10)
time_compress = timeit.timeit(itertools_compress_filter, number=10)

print("── compress vs list comprehension (500k records) ──")
print(f"list comprehension : {time_comp:.4f}s (10 runs)")
print(f"itertools.compress : {time_compress:.4f}s (10 runs)")
print(f"Speedup            : {time_comp / time_compress:.1f}x")


# ── 2. chain vs list concatenation ───────────────────────────────────────────
# + operator allocates a new list — O(n) memory, fast CPU.
# chain holds references — O(1) memory, similar CPU, no intermediate allocation.

list_a = list(range(500_000))
list_b = list(range(500_000))

def eager_plus():
    result = list_a + list_b
    return len(result)  # force evaluation

def lazy_chain():
    result = itertools.chain(list_a, list_b)
    return sum(1 for _ in result)  # force evaluation without materialising

time_plus = timeit.timeit(eager_plus, number=20)
time_chain = timeit.timeit(lazy_chain, number=20)

print("\n── chain vs list concatenation (1M elements total) ──")
print(f"list_a + list_b    : {time_plus:.4f}s (20 runs)")
print(f"itertools.chain    : {time_chain:.4f}s (20 runs)")
print(f"Speed difference   : {abs(time_plus - time_chain):.4f}s")
print(f"Memory difference  : list allocates ~{sys.getsizeof(list_a + list_b):,} bytes extra")
print(f"                     chain allocates ~{sys.getsizeof(itertools.chain(list_a, list_b))} bytes")


# ── 3. islice on a generator vs materialise-then-slice ───────────────────────
# islice wins when the source is lazy (generator) — avoids materialisation entirely.
# On a pre-built list, direct slicing is faster — no per-element overhead.

def make_generator(n):
    return (x * x for x in range(n))

def materialise_then_slice(n):
    # Materialise the full generator, then slice
    return list(make_generator(n))[100:200]

def islice_on_generator(n):
    # Take only elements 100..199 from the generator — never materialises the rest
    return list(itertools.islice(make_generator(n), 100, 200))

time_mat = timeit.timeit(lambda: materialise_then_slice(500_000), number=50)
time_isl = timeit.timeit(lambda: islice_on_generator(500_000), number=50)

print("\n── islice vs materialise-then-slice on a generator (500k elements) ──")
print(f"Materialise + slice: {time_mat:.4f}s (50 runs)")
print(f"islice on generator: {time_isl:.4f}s (50 runs)")
print(f"Speedup            : {time_mat / time_isl:.1f}x")
print("Note: islice wins because it never touches elements 200..499,999")


# ── 4. starmap vs map+lambda ─────────────────────────────────────────────────
# starmap avoids the lambda call overhead for functions that take multiple args.

coords = [(x, y) for x in range(1000) for y in range(10)]

def with_lambda():
    return list(map(lambda args: args[0] ** 2 + args[1] ** 2, coords))

def with_starmap():
    import math
    return list(itertools.starmap(lambda x, y: x ** 2 + y ** 2, coords))

time_lambda = timeit.timeit(with_lambda, number=50)
time_star = timeit.timeit(with_starmap, number=50)

print("\n── starmap vs map+lambda (10k two-arg operations) ──")
print(f"map + lambda       : {time_lambda:.4f}s (50 runs)")
print(f"itertools.starmap  : {time_star:.4f}s (50 runs)")
print(f"Speedup            : {time_lambda / time_star:.1f}x")

import itertools
import operator


# ── 1. groupby: sort first or get duplicate groups ───────────────────────────

data = [
    {'department': 'engineering', 'name': 'Alice'},
    {'department': 'sales',       'name': 'Bob'},
    {'department': 'engineering', 'name': 'Charlie'},  # same dept, not consecutive
    {'department': 'sales',       'name': 'Diana'},
]

key_func = lambda x: x['department']

print("── groupby WITHOUT sorting (wrong) ──")
for dept, group in itertools.groupby(data, key_func):
    members = [p['name'] for p in group]
    print(f"  {dept}: {members}")
# engineering and sales each appear twice — duplicate groups, wrong aggregation

print("\n── groupby WITH sorting (correct) ──")
sorted_data = sorted(data, key=key_func)
for dept, group in itertools.groupby(sorted_data, key_func):
    members = [p['name'] for p in group]
    print(f"  {dept}: {members}")
# Each department appears once with all its members


# ── 2. tee: buffer holds the delta between consumers, not everything ──────────
# The buffer grows when one consumer advances ahead of the other.
# Items consumed by BOTH are released — tee does not hold the entire stream.

source = iter(range(10))
t1, t2 = itertools.tee(source, 2)

# Advance t1 by 5 items — tee buffers items 0..4 for t2
for _ in range(5):
    next(t1)

# t2 hasn't consumed anything yet — the 5 buffered items are waiting
print("\n── tee: consuming the buffered items ──")
print(f"t2 sees: {list(t2)}")  # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
print(f"t1 sees remaining: {list(t1)}")  # [5, 6, 7, 8, 9]

# Safe tee pattern: lockstep consumers via zip
source2 = iter(range(8))
t1, t2 = itertools.tee(source2)
consecutive_pairs = list(zip(t1, t2))
# Note: this is equivalent to pairwise in Python 3.10+
print(f"\nLockstep tee (consecutive pairs): {consecutive_pairs}")


# ── 3. accumulate: arbitrary binary functions, factorially growing integers ──

values = [3, 1, 4, 1, 5, 9, 2, 6]

# Running sum (default)
running_sum = list(itertools.accumulate(values))
print(f"\nRunning sum : {running_sum}")

# Running maximum
running_max = list(itertools.accumulate(values, max))
print(f"Running max : {running_max}")

# Running minimum
running_min = list(itertools.accumulate(values, min))
print(f"Running min : {running_min}")

# Running product — integers grow factorially in magnitude, not overflowing
# (Python has arbitrary-precision integers) but getting expensive to compute
running_product = list(itertools.accumulate(values, operator.mul))
print(f"Running prod: {running_product}")
print("Note: no overflow — Python ints are arbitrary precision.")
print("But: each multiplication involves larger numbers than the last.")
print("For large sequences with large values this becomes a CPU cost, not a correctness issue.")

# With initial value (Python 3.8+)
running_with_initial = list(itertools.accumulate(values, initial=100))
print(f"\nWith initial=100: {running_with_initial}")

// io.thecodeforge — python tutorial

import itertools
import tracemalloc

def process_with_list(data_stream):
    # Bad: builds full list in memory
    all_data = list(data_stream)
    evens = [x for x in all_data if x % 2 == 0]
    squares = [x*x for x in evens]
    return sum(squares)

def process_with_itertools(data_stream):
    # Good: lazy pipeline, constant memory
    evens = itertools.filterfalse(lambda x: x % 2 != 0, data_stream)
    squares = map(lambda x: x * x, evens)
    return sum(squares)

# Simulate 10M row stream
tracemalloc.start()
_ = process_with_list(iter(range(10_000_000)))
current, peak = tracemalloc.get_traced_memory()
print(f"List version: {peak / 1024 / 1024:.0f} MB peak")
tracemalloc.stop()

tracemalloc.start()
_ = process_with_itertools(iter(range(10_000_000)))
current, peak = tracemalloc.get_traced_memory()
print(f"Itertools version: {peak / 1024 / 1024:.0f} MB peak")

// io.thecodeforge — python tutorial

import itertools

def grouper(iterable, chunk_size, fill_value=None):
    "Collect data into fixed-length chunks or blocks"
    # grouper('ABCDEFG', 3, 'x') --> ABC DEF Gxx
    args = [iter(iterable)] * chunk_size
    return itertools.zip_longest(*args, fillvalue=fill_value)

# Real scenario: batch API calls
api_ids = [101, 202, 303, 404, 505, 606, 707, 808]
batch_size = 3

for batch_num, batch in enumerate(grouper(api_ids, batch_size, fill_value=None), 1):
    clean_ids = [i for i in batch if i is not None]
    print(f"Batch {batch_num}: {clean_ids}")

// io.thecodeforge — python tutorial

import itertools
import random

suits = ['S', 'H', 'D', 'C']
ranks = list(range(1, 14))  # 1=Ace, 11=Jack, etc.

def fresh_deck():
    return list(itertools.product(ranks, suits))

def deal_hands(seed=42, hand_size=5, num_hands=4):
    deck = fresh_deck()
    rng = random.Random(seed)
    rng.shuffle(deck)
    it = iter(deck)
    return [list(itertools.islice(it, hand_size)) for _ in range(num_hands)]

hands = deal_hands()
for i, hand in enumerate(hands, 1):
    print(f"Hand {i}: {hand}")

// io.thecodeforge — python tutorial

import itertools

swimmers = [
    ('Alice', {'free': 54.2, 'back': 59.1, 'breast': 63.7, 'fly': 58.0}),
    ('Bob',   {'free': 53.8, 'back': 60.2, 'breast': 65.0, 'fly': 57.4}),
    ('Carol', {'free': 55.0, 'back': 58.5, 'breast': 62.1, 'fly': 59.2}),
    ('Dave',  {'free': 52.9, 'back': 61.0, 'breast': 64.3, 'fly': 56.8}),
]

strokes = ['back', 'breast', 'fly', 'free']
best_time = float('inf')
best_team = None

for combo in itertools.combinations(swimmers, 4):
    for perm in itertools.permutations(combo):
        total = sum(swimmer[1][style] for swimmer, style in zip(perm, strokes))
        if total < best_time:
            best_time = total
            best_team = list(zip(perm, strokes))

print(f"Best relay team: {best_team}, total: {best_time}")

// io.thecodeforge — python tutorial

from itertools import accumulate, islice

def fib_stream():
    seed = (0, 1)
    rule = lambda state: (state[1], state[0] + state[1])
    yield from (pair[0] for pair in accumulate(repeat(seed), lambda x, _: rule(x)))

first_10 = list(islice(fib_stream(), 10))
print(first_10)

// io.thecodeforge — python tutorial

from itertools import islice, accumulate, tee

def rolling_max_drawdown(prices):
    it1, it2 = tee(prices, 2)
    highs = accumulate(it1, max)
    for price, high in zip(it2, highs):
        yield high - price  # drawdown at each tick

sample = [4500, 4510, 4490, 4480, 4520]
drawdowns = list(rolling_max_drawdown(iter(sample)))
print(drawdowns)