Performance & Profiling

import time

def slow_function():
    total = 0
    for i in range(1_000_000):
        total += i
    return total

# Basic timing
start = time.perf_counter()
result = slow_function()
end = time.perf_counter()

print(f"Elapsed time: {end - start:.4f} seconds")

# Context manager for cleaner timing
from contextlib import contextmanager

@contextmanager
def timer(name="Operation"):
    start = time.perf_counter()
    yield
    end = time.perf_counter()
    print(f"{name} took {end - start:.4f}s")

# Usage
with timer("Slow function"):
    result = slow_function()

import timeit

# Compare different approaches
# Approach 1: List concatenation
time1 = timeit.timeit(
    stmt="result = []\nfor i in range(1000): result = result + [i]",
    number=1000
)

# Approach 2: List append
time2 = timeit.timeit(
    stmt="result = []\nfor i in range(1000): result.append(i)",
    number=1000
)

# Approach 3: List comprehension
time3 = timeit.timeit(
    stmt="result = [i for i in range(1000)]",
    number=1000
)

print(f"Concatenation: {time1:.4f}s")  # Slowest
print(f"Append:        {time2:.4f}s")  # Medium
print(f"Comprehension: {time3:.4f}s")  # Fastest

# Using repeat for more accurate results
times = timeit.repeat(
    stmt="sum(range(1000))",
    repeat=5,
    number=10000
)
print(f"Best of 5: {min(times):.6f}s")
print(f"Average:   {sum(times)/len(times):.6f}s")

# Save as profiler.py, then run: python profiler.py
import cProfile
import pstats

def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

def calculate_all():
    results = []
    for i in range(30):
        results.append(fibonacci(i))
    return results

# Method 1: Quick profiling
cProfile.run('calculate_all()')

# Method 2: Save and analyze results
profiler = cProfile.Profile()
profiler.enable()

calculate_all()

profiler.disable()

# Print stats
stats = pstats.Stats(profiler)
stats.strip_dirs()
stats.sort_stats('cumulative')
stats.print_stats(10)  # Top 10 functions

# Install: pip install line_profiler
# Run with: kernprof -l -v profile_demo.py

def complex_calculation(value):
    return sum(i * value for i in range(500))

# kernprof injects @profile into builtins, so no import is needed
@profile
def process_data(items):
    result = []
    for item in items:
        # Expensive operation
        processed = complex_calculation(item)
        result.append(processed)
    return result

process_data(range(1000))

import time
from functools import lru_cache

# Without caching - exponential time complexity
def fibonacci_slow(n):
    if n <= 1:
        return n
    return fibonacci_slow(n-1) + fibonacci_slow(n-2)

# With caching - linear time complexity
@lru_cache(maxsize=None)
def fibonacci_fast(n):
    if n <= 1:
        return n
    return fibonacci_fast(n-1) + fibonacci_fast(n-2)

start = time.perf_counter()
fibonacci_slow(35)
without_cache = time.perf_counter() - start

start = time.perf_counter()
fibonacci_fast(35)
with_cache = time.perf_counter() - start

print(f"Without cache: {without_cache:.3f}s")
print(f"With cache:    {with_cache:.6f}s")
print(f"Speedup:       {without_cache / with_cache:,.0f}x")

# Check cache stats
print(fibonacci_fast.cache_info())

# Clear the cache if the underlying data changes
fibonacci_fast.cache_clear()
print(fibonacci_fast.cache_info())

import sys

def process_line(line):
    return line.strip().upper()

# Bad: Load everything into memory
def process_large_file_slow(filename):
    with open(filename) as f:
        lines = f.readlines()  # Loads entire file!

    results = []
    for line in lines:
        results.append(process_line(line))
    return results

# Good: Process one line at a time
def process_large_file_fast(filename):
    def line_generator():
        with open(filename) as f:
            for line in f:  # Lazy iteration
                yield process_line(line)
    return line_generator()

# A small stand-in for the huge file
with open('huge_file.txt', 'w') as f:
    f.writelines(f"record {i}\n" for i in range(1, 4))

# Usage
for result in process_large_file_fast('huge_file.txt'):
    print(result)  # Memory stays constant!

# Generator expressions vs list comprehensions
# Bad: Creates entire list in memory
squares_list = [x**2 for x in range(1_000_000)]

# Good: Creates values on demand
squares_gen = (x**2 for x in range(1_000_000))

print(f"List:      {sys.getsizeof(squares_list):>9,} bytes")
print(f"Generator: {sys.getsizeof(squares_gen):>9,} bytes")

# Use case: Sum of squares
total = sum(x**2 for x in range(1_000_000))  # Memory efficient!
print(f"Sum of squares: {total:,}")

import timeit

words = ["alpha", "beta", "gamma", "delta", "epsilon"] * 200

# Slow: explicit loop
def squares_loop():
    result = []
    for x in range(1000):
        if x % 2 == 0:
            result.append(x ** 2)
    return result

# Fast: list comprehension
def squares_comprehension():
    return [x ** 2 for x in range(1000) if x % 2 == 0]

# Slow: explicit loop
def lengths_loop():
    word_lengths = {}
    for word in words:
        word_lengths[word] = len(word)
    return word_lengths

# Fast: dict comprehension
def lengths_comprehension():
    return {word: len(word) for word in words}

# Slow: explicit loop
def unique_loop():
    unique_lengths = set()
    for word in words:
        unique_lengths.add(len(word))
    return unique_lengths

# Fast: set comprehension
def unique_comprehension():
    return {len(word) for word in words}

for loop_fn, comp_fn in [(squares_loop, squares_comprehension),
                         (lengths_loop, lengths_comprehension),
                         (unique_loop, unique_comprehension)]:
    assert loop_fn() == comp_fn()  # Same result, different syntax
    loop_time = min(timeit.repeat(loop_fn, repeat=5, number=2000)) / 2000
    comp_time = min(timeit.repeat(comp_fn, repeat=5, number=2000)) / 2000
    name = loop_fn.__name__.removesuffix("_loop")
    print(f"{name:8s} loop {loop_time * 1e6:5.1f} us   "
          f"comprehension {comp_time * 1e6:5.1f} us   "
          f"{loop_time / comp_time:.2f}x")

# Save as memory_demo.py, then run: python memory_demo.py
import os
import tracemalloc

# Start tracking
tracemalloc.start()

# Your code here
data = [i ** 2 for i in range(1_000_000)]

# Get current memory usage
current, peak = tracemalloc.get_traced_memory()
print(f"Current memory: {current / 1024 / 1024:.2f} MB")
print(f"Peak memory: {peak / 1024 / 1024:.2f} MB")

# Get top memory allocations
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("Top 3 allocation sites:")
for stat in top_stats[:3]:
    frame = stat.traceback[0]
    print(f"  {os.path.basename(frame.filename)}:{frame.lineno}  {stat.size:,} bytes")

tracemalloc.stop()

# Install: pip install memory_profiler
# Run with: python -m memory_profiler mem_demo.py

from memory_profiler import profile

@profile
def memory_heavy_function():
    # Creates large list
    a = [1] * (10 ** 6)

    # Creates another large list
    b = [2] * (2 * 10 ** 7)

    # Delete first list
    del a

    return sum(b)

memory_heavy_function()

# Problem: count 100,000 activity records and report the 10 most active users
import random
import time
from collections import Counter, defaultdict

random.seed(42)
logs = [{"user_id": f"u{random.randint(1, 5000)}"} for _ in range(100_000)]

# SLOW VERSION: hand-rolled counting and a hand-rolled sort
def find_top_users_slow(logs):
    user_activity = {}

    # Count activities per user
    for log in logs:
        user_id = log['user_id']
        if user_id in user_activity:
            user_activity[user_id] += 1
        else:
            user_activity[user_id] = 1

    # Sort to find top 10
    pairs = []
    for user_id, count in user_activity.items():
        pairs.append((user_id, count))

    pairs.sort(key=lambda pair: pair[1], reverse=True)
    return pairs[:10]

# VERSION 1: defaultdict drops the branch, sorted() replaces the manual pass
def find_top_users_faster(logs):
    user_activity = defaultdict(int)

    for log in logs:
        user_activity[log['user_id']] += 1

    return sorted(user_activity.items(),
                  key=lambda pair: pair[1],
                  reverse=True)[:10]

# VERSION 2: Counter counts in C, most_common uses a heap for small k
def find_top_users_fastest(logs):
    user_activity = Counter(log['user_id'] for log in logs)
    return user_activity.most_common(10)

def benchmark(fn, repeat=10):
    timings = []
    for _ in range(repeat):
        start = time.perf_counter()
        fn(logs)
        timings.append(time.perf_counter() - start)
    return min(timings)

# All three must agree before any of them can be called an optimization
assert find_top_users_slow(logs) == find_top_users_faster(logs) == find_top_users_fastest(logs)

timings = [(fn.__name__, benchmark(fn))
           for fn in (find_top_users_slow, find_top_users_faster, find_top_users_fastest)]
baseline = timings[0][1]
for name, elapsed in timings:
    print(f"{name:24s} {elapsed * 1000:6.1f} ms   {baseline / elapsed:.2f}x")

print(f"Top 3: {find_top_users_fastest(logs)[:3]}")

# VERSION 3: stream the file instead of loading it, when the logs live on disk
import json
import random
import tracemalloc
from collections import Counter

random.seed(42)
with open('activity.jsonl', 'w') as f:
    for _ in range(100_000):
        f.write(json.dumps({"user_id": f"u{random.randint(1, 5000)}"}) + "\n")

# Loads every record before counting any of them
def top_users_in_memory(log_file):
    with open(log_file) as f:
        logs = [json.loads(line) for line in f]
    return Counter(log['user_id'] for log in logs).most_common(10)

# Counts a batch at a time, so only batch_size records are ever live
def top_users_streaming(log_file, batch_size=10_000):
    user_activity = Counter()
    batch = []

    with open(log_file) as f:
        for line in f:
            batch.append(json.loads(line))

            if len(batch) >= batch_size:
                user_activity.update(log['user_id'] for log in batch)
                batch = []

        # Process remaining
        if batch:
            user_activity.update(log['user_id'] for log in batch)

    return user_activity.most_common(10)

for fn in (top_users_in_memory, top_users_streaming):
    tracemalloc.start()
    result = fn('activity.jsonl')
    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    print(f"{fn.__name__:22s} peak {peak / 1024 / 1024:6.2f} MiB   top user {result[0]}")