1# Python Performance Optimization — Advanced Reference
2 
3Advanced optimization techniques including NumPy vectorization, caching, memory management, parallelization, async I/O, database optimization, and benchmarking tools.
4 
5## Advanced Optimization
6 
7### Pattern 11: NumPy for Numerical Operations
8 
9```python
10import timeit
11import numpy as np
12 
13def python_sum(n):
14    """Sum using pure Python."""
15    return sum(range(n))
16 
17def numpy_sum(n):
18    """Sum using NumPy."""
19    return np.arange(n).sum()
20 
21n = 1000000
22 
23python_time = timeit.timeit(lambda: python_sum(n), number=100)
24numpy_time = timeit.timeit(lambda: numpy_sum(n), number=100)
25 
26print(f"Python: {python_time:.4f}s")
27print(f"NumPy: {numpy_time:.4f}s")
28print(f"Speedup: {python_time/numpy_time:.2f}x")
29 
30# Vectorized operations
31def python_multiply():
32    """Element-wise multiplication in Python."""
33    a = list(range(100000))
34    b = list(range(100000))
35    return [x * y for x, y in zip(a, b)]
36 
37def numpy_multiply():
38    """Vectorized multiplication in NumPy."""
39    a = np.arange(100000)
40    b = np.arange(100000)
41    return a * b
42 
43py_time = timeit.timeit(python_multiply, number=100)
44np_time = timeit.timeit(numpy_multiply, number=100)
45 
46print(f"\nPython multiply: {py_time:.4f}s")
47print(f"NumPy multiply: {np_time:.4f}s")
48print(f"Speedup: {py_time/np_time:.2f}x")
49```
50 
51### Pattern 12: Caching with functools.lru_cache
52 
53```python
54from functools import lru_cache
55import timeit
56 
57def fibonacci_slow(n):
58    """Recursive fibonacci without caching."""
59    if n < 2:
60        return n
61    return fibonacci_slow(n-1) + fibonacci_slow(n-2)
62 
63@lru_cache(maxsize=None)
64def fibonacci_fast(n):
65    """Recursive fibonacci with caching."""
66    if n < 2:
67        return n
68    return fibonacci_fast(n-1) + fibonacci_fast(n-2)
69 
70# Massive speedup for recursive algorithms
71n = 30
72 
73slow_time = timeit.timeit(lambda: fibonacci_slow(n), number=1)
74fast_time = timeit.timeit(lambda: fibonacci_fast(n), number=1000)
75 
76print(f"Without cache (1 run): {slow_time:.4f}s")
77print(f"With cache (1000 runs): {fast_time:.4f}s")
78 
79# Cache info
80print(f"Cache info: {fibonacci_fast.cache_info()}")
81```
82 
83### Pattern 13: Using __slots__ for Memory
84 
85```python
86import sys
87 
88class RegularClass:
89    """Regular class with __dict__."""
90    def __init__(self, x, y, z):
91        self.x = x
92        self.y = y
93        self.z = z
94 
95class SlottedClass:
96    """Class with __slots__ for memory efficiency."""
97    __slots__ = ['x', 'y', 'z']
98 
99    def __init__(self, x, y, z):
100        self.x = x
101        self.y = y
102        self.z = z
103 
104# Memory comparison
105regular = RegularClass(1, 2, 3)
106slotted = SlottedClass(1, 2, 3)
107 
108print(f"Regular class size: {sys.getsizeof(regular)} bytes")
109print(f"Slotted class size: {sys.getsizeof(slotted)} bytes")
110 
111# Significant savings with many instances
112regular_objects = [RegularClass(i, i+1, i+2) for i in range(10000)]
113slotted_objects = [SlottedClass(i, i+1, i+2) for i in range(10000)]
114 
115print(f"\nMemory for 10000 regular objects: ~{sys.getsizeof(regular) * 10000} bytes")
116print(f"Memory for 10000 slotted objects: ~{sys.getsizeof(slotted) * 10000} bytes")
117```
118 
119### Pattern 14: Multiprocessing for CPU-Bound Tasks
120 
121```python
122import multiprocessing as mp
123import time
124 
125def cpu_intensive_task(n):
126    """CPU-intensive calculation."""
127    return sum(i**2 for i in range(n))
128 
129def sequential_processing():
130    """Process tasks sequentially."""
131    start = time.time()
132    results = [cpu_intensive_task(1000000) for _ in range(4)]
133    elapsed = time.time() - start
134    return elapsed, results
135 
136def parallel_processing():
137    """Process tasks in parallel."""
138    start = time.time()
139    with mp.Pool(processes=4) as pool:
140        results = pool.map(cpu_intensive_task, [1000000] * 4)
141    elapsed = time.time() - start
142    return elapsed, results
143 
144if __name__ == "__main__":
145    seq_time, seq_results = sequential_processing()
146    par_time, par_results = parallel_processing()
147 
148    print(f"Sequential: {seq_time:.2f}s")
149    print(f"Parallel: {par_time:.2f}s")
150    print(f"Speedup: {seq_time/par_time:.2f}x")
151```
152 
153### Pattern 15: Async I/O for I/O-Bound Tasks
154 
155```python
156import asyncio
157import aiohttp
158import time
159import requests
160 
161urls = [
162    "https://httpbin.org/delay/1",
163    "https://httpbin.org/delay/1",
164    "https://httpbin.org/delay/1",
165    "https://httpbin.org/delay/1",
166]
167 
168def synchronous_requests():
169    """Synchronous HTTP requests."""
170    start = time.time()
171    results = []
172    for url in urls:
173        response = requests.get(url)
174        results.append(response.status_code)
175    elapsed = time.time() - start
176    return elapsed, results
177 
178async def async_fetch(session, url):
179    """Async HTTP request."""
180    async with session.get(url) as response:
181        return response.status
182 
183async def asynchronous_requests():
184    """Asynchronous HTTP requests."""
185    start = time.time()
186    async with aiohttp.ClientSession() as session:
187        tasks = [async_fetch(session, url) for url in urls]
188        results = await asyncio.gather(*tasks)
189    elapsed = time.time() - start
190    return elapsed, results
191 
192# Async is much faster for I/O-bound work
193sync_time, sync_results = synchronous_requests()
194async_time, async_results = asyncio.run(asynchronous_requests())
195 
196print(f"Synchronous: {sync_time:.2f}s")
197print(f"Asynchronous: {async_time:.2f}s")
198print(f"Speedup: {sync_time/async_time:.2f}x")
199```
200 
201## Database Optimization
202 
203### Pattern 16: Batch Database Operations
204 
205```python
206import sqlite3
207import time
208 
209def create_db():
210    """Create test database."""
211    conn = sqlite3.connect(":memory:")
212    conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
213    return conn
214 
215def slow_inserts(conn, count):
216    """Insert records one at a time."""
217    start = time.time()
218    cursor = conn.cursor()
219    for i in range(count):
220        cursor.execute("INSERT INTO users (name) VALUES (?)", (f"User {i}",))
221        conn.commit()  # Commit each insert
222    elapsed = time.time() - start
223    return elapsed
224 
225def fast_inserts(conn, count):
226    """Batch insert with single commit."""
227    start = time.time()
228    cursor = conn.cursor()
229    data = [(f"User {i}",) for i in range(count)]
230    cursor.executemany("INSERT INTO users (name) VALUES (?)", data)
231    conn.commit()  # Single commit
232    elapsed = time.time() - start
233    return elapsed
234 
235# Benchmark
236conn1 = create_db()
237slow_time = slow_inserts(conn1, 1000)
238 
239conn2 = create_db()
240fast_time = fast_inserts(conn2, 1000)
241 
242print(f"Individual inserts: {slow_time:.4f}s")
243print(f"Batch insert: {fast_time:.4f}s")
244print(f"Speedup: {slow_time/fast_time:.2f}x")
245```
246 
247### Pattern 17: Query Optimization
248 
249```python
250# Use indexes for frequently queried columns
251"""
252-- Slow: No index
253SELECT * FROM users WHERE email = '[email protected]';
254 
255-- Fast: With index
256CREATE INDEX idx_users_email ON users(email);
257SELECT * FROM users WHERE email = '[email protected]';
258"""
259 
260# Use query planning
261import sqlite3
262 
263conn = sqlite3.connect("example.db")
264cursor = conn.cursor()
265 
266# Analyze query performance
267cursor.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ("[email protected]",))
268print(cursor.fetchall())
269 
270# Use SELECT only needed columns
271# Slow: SELECT *
272# Fast: SELECT id, name
273```
274 
275## Memory Optimization
276 
277### Pattern 18: Detecting Memory Leaks
278 
279```python
280import tracemalloc
281import gc
282 
283def memory_leak_example():
284    """Example that leaks memory."""
285    leaked_objects = []
286 
287    for i in range(100000):
288        # Objects added but never removed
289        leaked_objects.append([i] * 100)
290 
291    # In real code, this would be an unintended reference
292 
293def track_memory_usage():
294    """Track memory allocations."""
295    tracemalloc.start()
296 
297    # Take snapshot before
298    snapshot1 = tracemalloc.take_snapshot()
299 
300    # Run code
301    memory_leak_example()
302 
303    # Take snapshot after
304    snapshot2 = tracemalloc.take_snapshot()
305 
306    # Compare
307    top_stats = snapshot2.compare_to(snapshot1, 'lineno')
308 
309    print("Top 10 memory allocations:")
310    for stat in top_stats[:10]:
311        print(stat)
312 
313    tracemalloc.stop()
314 
315# Monitor memory
316track_memory_usage()
317 
318# Force garbage collection
319gc.collect()
320```
321 
322### Pattern 19: Iterators vs Lists
323 
324```python
325import sys
326 
327def process_file_list(filename):
328    """Load entire file into memory."""
329    with open(filename) as f:
330        lines = f.readlines()  # Loads all lines
331        return sum(1 for line in lines if line.strip())
332 
333def process_file_iterator(filename):
334    """Process file line by line."""
335    with open(filename) as f:
336        return sum(1 for line in f if line.strip())
337 
338# Iterator uses constant memory
339# List loads entire file into memory
340```
341 
342### Pattern 20: Weakref for Caches
343 
344```python
345import weakref
346 
347class CachedResource:
348    """Resource that can be garbage collected."""
349    def __init__(self, data):
350        self.data = data
351 
352# Regular cache prevents garbage collection
353regular_cache = {}
354 
355def get_resource_regular(key):
356    """Get resource from regular cache."""
357    if key not in regular_cache:
358        regular_cache[key] = CachedResource(f"Data for {key}")
359    return regular_cache[key]
360 
361# Weak reference cache allows garbage collection
362weak_cache = weakref.WeakValueDictionary()
363 
364def get_resource_weak(key):
365    """Get resource from weak cache."""
366    resource = weak_cache.get(key)
367    if resource is None:
368        resource = CachedResource(f"Data for {key}")
369        weak_cache[key] = resource
370    return resource
371 
372# When no strong references exist, objects can be GC'd
373```
374 
375## Benchmarking Tools
376 
377### Custom Benchmark Decorator
378 
379```python
380import time
381from functools import wraps
382 
383def benchmark(func):
384    """Decorator to benchmark function execution."""
385    @wraps(func)
386    def wrapper(*args, **kwargs):
387        start = time.perf_counter()
388        result = func(*args, **kwargs)
389        elapsed = time.perf_counter() - start
390        print(f"{func.__name__} took {elapsed:.6f} seconds")
391        return result
392    return wrapper
393 
394@benchmark
395def slow_function():
396    """Function to benchmark."""
397    time.sleep(0.5)
398    return sum(range(1000000))
399 
400result = slow_function()
401```
402 
403### Performance Testing with pytest-benchmark
404 
405```python
406# Install: pip install pytest-benchmark
407 
408def test_list_comprehension(benchmark):
409    """Benchmark list comprehension."""
410    result = benchmark(lambda: [i**2 for i in range(10000)])
411    assert len(result) == 10000
412 
413def test_map_function(benchmark):
414    """Benchmark map function."""
415    result = benchmark(lambda: list(map(lambda x: x**2, range(10000))))
416    assert len(result) == 10000
417 
418# Run with: pytest test_performance.py --benchmark-compare
419```
420