Python itertools Module: Lazy Iterators, Combinatorics and Performance Secrets

import itertools
import sys

# --- Eager approach: builds the entire list in memory ---
eager_numbers = list(range(1_000_000))          # Allocates ~8 MB on 64-bit Python
eager_size = sys.getsizeof(eager_numbers)       # Size of the list object itself

# --- Lazy approach: just a tiny C iterator object ---
lazy_numbers = itertools.islice(itertools.count(0), 1_000_000)  # Describes the sequence
lazy_size = sys.getsizeof(lazy_numbers)         # Size of the iterator object

print(f"Eager list size  : {eager_size:>10,} bytes")
print(f"Lazy iterator    : {lazy_size:>10,} bytes")
print(f"Memory ratio     : {eager_size // lazy_size:>10,}x more memory for the eager list")

# --- Proving single-consumption behaviour ---
letters = iter(['a', 'b', 'c'])                 # A plain iterator (same rule applies)
chained = itertools.chain(letters, ['d', 'e'])  # Wrap it in chain

first_pass  = list(chained)   # Consumes the chain object completely
second_pass = list(chained)   # chained is now exhausted — returns []

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

# --- Proof that the C iterator holds no data copy ---
big_range = range(50_000_000)                   # range object, not a list
chain_obj = itertools.chain(big_range, big_range)
print(f"\nchain object type : {type(chain_obj)}")
print(f"chain object size : {sys.getsizeof(chain_obj)} bytes (constant, not proportional to input)")

import itertools
from datetime import datetime, timedelta

# ── 1. count() + islice(): generate timestamped event IDs lazily ──────────────
start_time = datetime(2024, 1, 1, 9, 0, 0)

def timestamped_events(start: datetime, interval_minutes: int):
    """Infinite generator of (event_id, timestamp) tuples."""
    for event_id in itertools.count(start=1001):           # Starts at 1001, never stops
        yield event_id, start
        start += timedelta(minutes=interval_minutes)

# Pull only the first 5 events — islice imposes a safe boundary on the infinite stream
first_five_events = list(itertools.islice(timestamped_events(start_time, 15), 5))
print("First 5 events:")
for eid, ts in first_five_events:
    print(f"  Event {eid} at {ts.strftime('%H:%M')}")

# ── 2. cycle(): round-robin assignment across servers ─────────────────────────
servers          = ['server-A', 'server-B', 'server-C']
server_carousel  = itertools.cycle(servers)               # Loops forever: A, B, C, A, B, C ...
incoming_requests = [f"req-{i:03d}" for i in range(8)]   # 8 incoming requests

print("\nRound-robin routing:")
for request in incoming_requests:
    assigned_server = next(server_carousel)               # Pulls one from the cycle
    print(f"  {request} → {assigned_server}")

# ── 3. chain.from_iterable() vs chain(*args): lazy vs eager unpacking ─────────
def load_log_chunk(day: int):
    """Simulates reading a log file for a given day (a generator, not a list)."""
    print(f"  [Loading chunk for day {day}]")
    yield from [f"day{day}-line{i}" for i in range(3)]

log_days = range(1, 4)   # Days 1, 2, 3

# chain.from_iterable: each chunk is loaded ONLY when the previous is exhausted
print("\nLazy chaining with from_iterable:")
lazy_pipeline = itertools.chain.from_iterable(
    load_log_chunk(day) for day in log_days   # Generator expression — not evaluated yet
)
for log_line in itertools.islice(lazy_pipeline, 4):       # Only read first 4 lines
    print(f"  → {log_line}")
# Notice: day 3 never even loads because islice stops at 4 items

# ── 4. dropwhile + takewhile: extract a specific window without loading all ───
sensor_readings = [0.1, 0.2, 0.5, 3.8, 4.1, 4.7, 4.2, 0.8, 0.3]

# Skip readings below threshold, then take readings until they drop back below
threshold = 1.0
abnormal_window = list(
    itertools.takewhile(
        lambda reading: reading >= threshold,                         # Stop when reading drops
        itertools.dropwhile(lambda r: r < threshold, sensor_readings) # Skip until first spike
    )
)
print(f"\nAbnormal sensor window: {abnormal_window}")

import itertools
import math

# ── 1. product(): test all combinations of config flags ───────────────────────
debug_modes    = [True, False]
log_levels     = ['INFO', 'WARNING', 'ERROR']
cache_sizes_mb = [128, 256]

print("All configuration combinations (product):")
config_space = itertools.product(debug_modes, log_levels, cache_sizes_mb)
for debug, level, cache in config_space:
    print(f"  debug={str(debug):<5}  log={level:<7}  cache={cache}MB")

expected_count = len(debug_modes) * len(log_levels) * len(cache_sizes_mb)
print(f"Total: {expected_count} configs  (math: {math.prod([2, 3, 2])})")

# ── 2. permutations vs combinations: the classic confusion ───────────────────
task_names = ['deploy', 'migrate', 'notify']

# permutations: order matters (deploy→migrate is different from migrate→deploy)
ordered_plans = list(itertools.permutations(task_names, r=2))
print(f"\nPermutations (ordered pairs, r=2): {len(ordered_plans)} results")
for plan in ordered_plans:
    print(f"  {plan[0]} then {plan[1]}")

# combinations: order doesn't matter (a team of 2, not a sequence)
unordered_pairs = list(itertools.combinations(task_names, r=2))
print(f"\nCombinations (unordered pairs, r=2): {len(unordered_pairs)} results")
for pair in unordered_pairs:
    print(f"  {pair}")

# ── 3. Always check size before materialising ─────────────────────────────────
items_count = 12
r_choose    = 4

perm_count  = math.perm(items_count, r_choose)  # n! / (n-r)!
comb_count  = math.comb(items_count, r_choose)  # n! / (r! * (n-r)!)

print(f"\nWith n={items_count}, r={r_choose}:")
print(f"  permutations → {perm_count:,} tuples  (~{perm_count * 48 / 1e6:.1f} MB if materialised)")
print(f"  combinations → {comb_count:,} tuples  (~{comb_count * 48 / 1e6:.1f} MB if materialised)")

# ── 4. combinations_with_replacement: sampling WITH replacement ───────────────
dice_faces     = [1, 2, 3, 4, 5, 6]
two_dice_rolls = list(itertools.combinations_with_replacement(dice_faces, r=2))
print(f"\nUnique unordered two-dice outcomes: {len(two_dice_rolls)}")
print(f"  First five: {two_dice_rolls[:5]}")
print(f"  Last  five: {two_dice_rolls[-5:]}")

# ── 5. Lazy combinatorics with filter: only materialise valid items ───────────
password_chars = 'abc'
four_char_passwords = itertools.product(password_chars, repeat=4)

# Only keep passwords that start AND end with 'a' — filter lazily before list()
valid_passwords = [
    ''.join(pwd)
    for pwd in four_char_passwords
    if pwd[0] == 'a' and pwd[-1] == 'a'    # Constraint applied on-the-fly
]
print(f"\n4-char passwords from 'abc' starting and ending with 'a':")
print(f"  {valid_passwords}")
print(f"  {len(valid_passwords)} valid out of {len(password_chars)**4} total")

import itertools
import operator
from collections import namedtuple

Transaction = namedtuple('Transaction', ['date', 'category', 'amount'])

transactions = [
    Transaction('2024-01-05', 'food',    12.50),
    Transaction('2024-01-06', 'food',    8.75),
    Transaction('2024-01-06', 'travel',  45.00),   # travel after food — but consecutive?
    Transaction('2024-01-07', 'food',    6.00),    # food again — NOT consecutive with earlier food
    Transaction('2024-01-08', 'travel',  120.00),
    Transaction('2024-01-08', 'travel',  30.00),
]

# ── 1. groupby GOTCHA: data must be sorted by key first ──────────────────────
print("=== groupby WITHOUT sorting (wrong) ===")
for category, group_iter in itertools.groupby(transactions, key=lambda t: t.category):
    items = list(group_iter)
    total = sum(t.amount for t in items)
    print(f"  {category:<8}: {len(items)} items, total={total:.2f}")  # 'food' appears TWICE!

print("\n=== groupby WITH sorting (correct) ===")
sorted_transactions = sorted(transactions, key=lambda t: t.category)  # Sort first!
for category, group_iter in itertools.groupby(sorted_transactions, key=lambda t: t.category):
    items = list(group_iter)                   # Materialise the group — group_iter is also lazy
    total = sum(t.amount for t.amount in [t.amount for t in items])
    total = sum(t.amount for t in items)       # Clean version
    print(f"  {category:<8}: {len(items)} items, total={total:.2f}")

# ── 2. accumulate: running totals and rolling maximums ───────────────────────
daily_sales = [320, 415, 290, 560, 480, 390, 610]

running_total   = list(itertools.accumulate(daily_sales))                        # Default: add
running_max     = list(itertools.accumulate(daily_sales, func=max))              # Rolling max
running_product = list(itertools.accumulate(daily_sales, func=operator.mul))     # Running product

print("\nDaily sales  :", daily_sales)
print("Running total:", running_total)
print("Running max  :", running_max)

# accumulate with initial value (Python 3.8+): prepend a starting accumulator
cumulative_with_carry = list(
    itertools.accumulate(daily_sales, initial=1000)  # Start at 1000 carry-over
)
print("With 1000 carry:", cumulative_with_carry)      # First value is 1000 itself

# ── 3. tee: splitting a stream for two consumers advancing in lockstep ────────
def sensor_stream():
    """Simulates a live sensor stream — we can only iterate it once."""
    for reading in [1.2, 0.8, 2.4, 1.9, 0.5, 3.1]:
        yield reading

raw_stream = sensor_stream()
stream_for_avg, stream_for_max = itertools.tee(raw_stream, 2)  # Split into 2 independent copies

# Both consumers advance together via zip — tee buffer stays small
print("\nStep-by-step: running avg vs running max")
avg_accumulator = itertools.accumulate(stream_for_avg,
                                       func=lambda running, new: (running + new) / 2)
max_accumulator = itertools.accumulate(stream_for_max, func=max)

for step, (avg_val, max_val) in enumerate(zip(avg_accumulator, max_accumulator), start=1):
    print(f"  Step {step}: running_avg={avg_val:.3f}  running_max={max_val:.1f}")