Advanced Functional Programming in Python

This notebook provides a practical introduction to advanced functional programming concepts in Python, with a focus on patterns and techniques that are useful in real-world, production scenarios. You will explore how to write cleaner, more modular, and powerful Python code by leveraging functional programming tools and thinking.

Advanced Pandas Functional Patterns

In this section, we will leverage real transaction data to demonstrate how pandas enables advanced, functional-style data analysis. You will see how to apply group-by transformations, aggregate computations, mapping, filtering, and custom functions to efficiently derive insights from tabular data. The goal is to showcase patterns that echo core functional programming concepts—such as immutability of inputs, first-class functions, chaining of operations, and high-level data abstractions—using pandas’ powerful APIs. We’ll cover practical techniques for condensing complex business analysis steps into concise, modular, and reusable workflows suitable for production data pipelines.

Data

import pandas as pd

transactions_df = pd.DataFrame([
    {"txn_id": "T1001", "customer": "Alice", "country": "US", "amount": 120.0, "fee": 3.6, "processing_seconds": 2.1, "status": "success", "channel": "card"},
    {"txn_id": "T1002", "customer": "Bob", "country": "US", "amount": 80.5, "fee": 2.4, "processing_seconds": 3.7, "status": "failed", "channel": "wallet"},
    {"txn_id": "T1003", "customer": "Carol", "country": "CA", "amount": 240.0, "fee": 7.2, "processing_seconds": 1.8, "status": "success", "channel": "bank"},
    {"txn_id": "T1004", "customer": "Dan", "country": "US", "amount": 15.0, "fee": 0.4, "processing_seconds": 4.4, "status": "success", "channel": "card"},
    {"txn_id": "T1005", "customer": "Eve", "country": "UK", "amount": 500.0, "fee": 15.0, "processing_seconds": 2.9, "status": "success", "channel": "card"},
    {"txn_id": "T1006", "customer": "Frank", "country": "UK", "amount": 65.0, "fee": 1.9, "processing_seconds": 5.2, "status": "failed", "channel": "wallet"},
])

print("Transaction data:")
print(transactions_df)

GroupBy Operations (Like tapply)

GroupBy operations in pandas allow you to split your data into groups based on values in one or more columns, and then perform aggregate computations or transformations on each group, independently. This is similar to the ‘tapply’ function in R, which applies a function to subsets of a vector, split by a factor. In pandas, the groupby() method enables this pattern in a flexible and expressive way.

Typical GroupBy workflow involves three steps, often referred to as “split-apply-combine”: 1. Split: Divide the data into groups based on one or more keys (e.g., ‘country’). 2. Apply: Apply a function to each group independently, such as computing mean, sum, count, or a custom aggregation. 3. Combine: Merge the results back into a new DataFrame or Series.

This approach enables complex and modular data analysis, where you can easily compare metrics across different segments of your data.

# Group by country and calculate statistics
grouped_stats = transactions_df.groupby('country').agg({
    'amount': ['mean', 'std', 'min', 'max'],
    'fee': ['mean', 'median', 'count']
}).round(2)

print("Grouped statistics:")
print(grouped_stats)

# More complex grouping with custom functions
def amount_range(series):
    return series.max() - series.min()

country_summary = transactions_df.groupby('country').agg(
    avg_amount=('amount', 'mean'),
    amount_range=('amount', amount_range),
    avg_processing_seconds=('processing_seconds', 'mean'),
    num_transactions=('txn_id', 'count')
).round(2)

print("\nCustom groupby summary:")
print(country_summary)

Conditional Operations (Like keep/discard)

Conditional operations allow you to selectively filter or exclude rows from your pandas DataFrame based on specified criteria, similar to “keep” or “discard” functions in functional programming.

• Keep: Use boolean indexing to retain (keep) rows that meet a condition. For example, filtering transactions with amounts above a certain threshold.
• Discard: Similarly, you can discard rows by selecting those that do not match a condition.

These operations provide fine-grained control over your data and enable more focused analyses by working only with relevant subsets.

# Filter rows (equivalent to keep)
high_value_txns = transactions_df[transactions_df['amount'] > 100]
print("High value transactions (amount > 100):")
print(high_value_txns)# Functional Programming in Python — Advanced

# Advanced notebook focused on production-like functional patterns.

# Filter with multiple conditions
fast_successful_txns = transactions_df[
    (transactions_df['processing_seconds'] <= 3.0) &
    (transactions_df['status'] == 'success')
]
print("\nFast successful transactions:")
print(fast_successful_txns)

# Exclude rows (equivalent to discard)
non_failed_txns = transactions_df[transactions_df['status'] != 'failed']
print("\nNon-failed transactions:")
print(non_failed_txns)

The Power of Chaining: Functional Pipelines

Python does not have a native dataframe pipe operator like some other languages, but pandas method chaining gives a clean functional pipeline style.

# Complex data transformation pipeline
pipeline_result = (
    transactions_df
    .assign(
        amount_after_fee=lambda x: x['amount'] - x['fee'],
        is_fast=lambda x: x['processing_seconds'] <= 3.0
    )
    .query('amount >= 50')
    .groupby('country')
    .agg({
        'amount_after_fee': ['mean', 'std'],
        'processing_seconds': 'mean',
        'txn_id': 'count'
    })
    .round(2)
    .reset_index()
)

print("Pipeline result:")
print(pipeline_result)

Decorators (retry) for unstable external calls

A decorator is a function that wraps another function to add behavior without changing the original function body. This is ideal for cross-cutting concerns such as retries, logging, timing, authentication, and caching.

In real systems, external calls can fail temporarily due to network jitter, timeouts, or rate limits. A retry decorator centralizes retry logic so business functions stay clean.

How this retry decorator works

1. retry(max_attempts=3) configures policy.
2. decorator(func) receives the target function.
3. wrapper(*args, **kwargs) executes the function with retry attempts.
4. On success, result is returned immediately.
5. On failure, it retries until max_attempts is reached.
6. Final failure is re-raised so callers can handle it upstream.

@wraps(func) preserves metadata (__name__, docstring), which helps debugging and tooling.

Production notes

• Retry only transient failures (timeouts, temporary 5xx errors), not validation/business errors.
• Use idempotent operations (or request IDs) to avoid duplicate side effects.
• Add backoff + jitter in production to reduce retry storms.
• Log attempt counts and final errors for observability.

This pattern makes reliability concerns reusable and keeps domain code focused on business logic.

from functools import wraps
import random

def retry(max_attempts=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as exc:
                    if attempt == max_attempts:
                        raise
                    print(f"Attempt {attempt} failed: {exc}")
        return wrapper
    return decorator

@retry(max_attempts=3)
def fetch_rate():
    if random.random() < 0.6:
        raise RuntimeError("temporary API issue")
    return 1.1

fetch_rate()

Composition-style ETL

ETL means Extract, Transform, Load: - Extract: read raw transaction records from a source. - Transform: clean, filter, and reshape records into analysis-ready form. - Load: write results to a target table/file or pass to downstream analytics.

In this notebook, composition is used to build ETL as a chain of small functions: - filter_success keeps only successful transactions. - filter_large keeps only high-value transactions. - select_country_amount reshapes each record to (country, amount). - list materializes the final lazy iterator.

Why composition helps

• Modularity: each step does one thing well.
• Reusability: you can swap steps in/out without rewriting everything.
• Testability: each function can be unit-tested independently.
• Readability: pipeline order documents business logic clearly.

Data flow in this pipeline

transactions -> filter_success -> filter_large -> select_country_amount -> records

After this, reduce(...) aggregates records into country_totals, which is a common analytics load target (for reporting or dashboards).

def pipe(*funcs):
    def run(data):
        for f in funcs:
            data = f(data)
        return data
    return run

# Example transaction data to avoid NameError
transactions = [
    {"status": "success", "amount": 150, "country": "US"},
    {"status": "failed", "amount": 90, "country": "CA"},
    {"status": "success", "amount": 200, "country": "CA"},
    {"status": "success", "amount": 50, "country": "US"},
    {"status": "success", "amount": 100, "country": "MX"},
    {"status": "failed", "amount": 130, "country": "US"},
]

filter_success = lambda rows: filter(lambda r: r["status"] == "success", rows)
filter_large = lambda rows: filter(lambda r: r["amount"] >= 100, rows)
select_country_amount = lambda rows: map(lambda r: (r["country"], r["amount"]), rows)

pipeline = pipe(filter_success, filter_large, select_country_amount, list)
records = pipeline(transactions)
records

from functools import reduce

country_totals = reduce(
    lambda acc, item: {**acc, item[0]: acc.get(item[0], 0) + item[1]},
    records,
    {},
)
country_totals

Lazy stream processing + safe parsing

This section combines two important production ideas:

1. Lazy stream processing
   • Data is processed one item at a time, only when needed.
   • This reduces memory usage and works well for large logs, event streams, and file-based pipelines.
   • Instead of creating many intermediate lists, generator-style functions (yield) pass values through stages on demand.
2. Safe parsing with Ok / Err style results
   • Parsing often fails in real systems (bad formats, missing fields, invalid numbers).
   • Returning Ok(value) or Err(error) makes failures explicit and composable.
   • This avoids deeply nested try/except blocks and keeps transformation pipelines predictable.

Why this pattern is useful

• Scalable: handles high-volume streams without loading everything into memory.
• Readable: each step has one clear responsibility (filter, transform, validate).
• Robust: invalid inputs are represented as data (Err) instead of crashing the full pipeline.

Mental model

Think of the pipeline as a factory belt: - lazy_filter(...) decides which records stay on the belt. - Mapping/transformation converts valid records into new shapes. - Parsing/validation wraps outcomes as Ok (good) or Err (bad) so downstream code can handle both safely.

This is a core functional approach for real-time analytics, ETL pipelines, and API input validation.

events = [
    {"user_id": 1, "event": "purchase", "latency_ms": 180},
    {"user_id": 2, "event": "click", "latency_ms": 250},
    {"user_id": 3, "event": "purchase", "latency_ms": 320},
]

def lazy_filter(pred, iterable):
    for x in iterable:
        if pred(x):
            yield x

high_latency_purchase = lazy_filter(
    lambda e: e["event"] == "purchase" and e["latency_ms"] > 200,
    events,
)
list(high_latency_purchase)

class Ok:
    def __init__(self, value):
        self.value = value
    def bind(self, fn):
        return fn(self.value)

class Err:
    def __init__(self, error):
        self.error = error
    def bind(self, fn):
        return self

def parse_float(text):
    try:
        return Ok(float(text))
    except ValueError:
        return Err(f"Invalid number: {text}")

parse_float("42.0").bind(lambda x: Ok(x * 2)).value

Real-World Example: Processing Multiple Files

Let’s recreate the cancer mortality analysis example from the R tutorial using Python.

import pandas as pd
import numpy as np
import os
from pathlib import Path
import re
import matplotlib.pyplot as plt
import seaborn as sns

# For this example, let's create synthetic data similar to the R example
np.random.seed(42)
years = list(range(1998, 2011))
data_rows = []

for year in years:
    n_counties = 1000  # Simulate ~1000 counties per year
    for _ in range(n_counties):
        row = {
            'Year': year,
            'SMOKING': np.random.normal(20, 5),  # % smoking prevalence
            'POVERTY': np.random.normal(15, 4),   # % poverty
            'PM25': np.random.normal(12, 3),      # PM2.5 concentration
            'NO2': np.random.normal(25, 6),       # NO2 concentration
            'SO2': np.random.normal(8, 2),        # SO2 concentration
            'RATE': np.random.normal(60, 15)      # Cancer mortality rate
        }
        data_rows.append(row)

# Create DataFrame
data = pd.DataFrame(data_rows)
print(f"Data shape: {data.shape}")
print(data.head())

data = pd.read_csv('https://github.com/zia207/Python_for_Beginners/raw/refs/heads/main/Data/data_1998_2010_long_lbc.csv')
print(f"Data shape: {data.shape}")
print(data.head())

def write_yearly_files(data, output_dir="output"):
    """Write separate CSV files for each year (equivalent to purrr::walk)"""
    Path(output_dir).mkdir(exist_ok=True)

    # Get unique years
    unique_years = data['Year'].unique()

    for year in unique_years:
        year_data = data[data['Year'] == year]
        filename = f"{output_dir}/data_{year}.csv"
        year_data.to_csv(filename, index=False)
        print(f"Written file: {filename}")

# Write files
write_yearly_files(data)

# Verify files were created
files = sorted(Path("output").glob("data_*.csv"))
print(f"\nCreated {len(files)} files:")
for f in files:
    print(f"  {f.name}")

Reading Multiple Files into a Dictionary

def load_yearly_data(directory="output"):
    """Load all yearly CSV files into a dictionary (equivalent to purrr::map)"""
    files = sorted(Path(directory).glob("data_*.csv"))
    
    # Extract years from filenames using regex
    year_pattern = r"data_(\d{4})\.csv"
    data_dict = {}
    
    for file_path in files:
        match = re.search(year_pattern, file_path.name)
        if match:
            year = int(match.group(1))
            df = pd.read_csv(file_path)
            data_dict[year] = df
    
    return data_dict

# Load data
data_list = load_yearly_data()
print(f"Loaded {len(data_list)} years of data")
print(f"Years available: {sorted(data_list.keys())}")

Descriptive Statistics (Like compute_descriptive_stats)

def compute_descriptive_stats(df, year):
    """
    Compute descriptive statistics for numeric columns
    Similar to the R function using dplyr and tidyr
    """
    # Select only numeric columns
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    
    # Calculate statistics
    stats = {}
    for col in numeric_cols:
        col_data = df[col].dropna()  # Remove NaN values
        
        mean_val = np.mean(col_data)
        std_val = np.std(col_data, ddof=1)  # Sample standard deviation
        n = len(col_data)
        
        # Calculate 95% CI (t-distribution approximation)
        if n > 1:
            t_critical = 1.96  # Approximation for large samples
            margin_error = t_critical * (std_val / np.sqrt(n))
            ci_lower = mean_val - margin_error
            ci_upper = mean_val + margin_error
        else:
            ci_lower = ci_upper = np.nan
        
        stats[col] = {
            'Min': np.min(col_data),
            'Max': np.max(col_data),
            'Mean': mean_val,
            'CI_lower': ci_lower,
            'CI_upper': ci_upper
        }
    
    # Convert to DataFrame
    stats_df = pd.DataFrame(stats).T.reset_index()
    stats_df.columns = ['Variable', 'Min', 'Max', 'Mean', 'CI_lower', 'CI_upper']
    stats_df['Year'] = year
    
    return stats_df

# Apply to all years
desc_stats_list = []
for year, df in data_list.items():
    stats_df = compute_descriptive_stats(df, year)
    desc_stats_list.append(stats_df)

# Combine all results
desc_stats_table = pd.concat(desc_stats_list, ignore_index=True)
desc_stats_table = desc_stats_table.round(3)

print("Descriptive Statistics Table:")
print(desc_stats_table.head(10))

Correlation Analysis (Like compute_correlations)

def compute_correlations(df, year):
    """
    Compute correlation between RATE and other variables
    """
    predictors = ['SMOKING', 'POVERTY', 'PM25', 'NO2', 'SO2']
    correlations = []

    for predictor in predictors:
        # Drop NA values for both variables
        valid_data = df[[predictor, 'RATE']].dropna()

        if len(valid_data) > 1:
            # Pearson correlation coefficient
            r = np.corrcoef(valid_data[predictor], valid_data['RATE'])[0, 1]

            # Simple p-value approximation (not exact, for illustration)
            n = len(valid_data)
            if abs(r) < 1:
                t_stat = r * np.sqrt((n - 2) / (1 - r**2))
                # Approximate p-value using normal distribution for large n
                from scipy.stats import norm
                p_value = 2 * (1 - norm.cdf(abs(t_stat)))
            else:
                p_value = 0

            correlations.append({
                'Variable': predictor,
                'r': r,
                'p_value': p_value
            })

    result_df = pd.DataFrame(correlations)
    result_df['Year'] = year
    return result_df

# Apply to all years
correlation_list = []
for year, df in data_list.items():
    corr_df = compute_correlations(df, year)
    correlation_list.append(corr_df)

cor_table = pd.concat(correlation_list, ignore_index=True)
cor_table = cor_table.round(3)

print("Correlation Table (RATE vs Predictors):")
print(cor_table.head(15))

Correlation Matrix Plots

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path

# --- Correlation function ---
def create_correlation_plot(df, year, ax=None):
    """
    Create a correlation matrix heatmap on the provided axis.
    If ax is None, creates a new subplot.
    """
    # Select only numeric columns for correlation
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    corr_matrix = df[numeric_cols].corr()
    
    # Plot on the provided axis
    sns.heatmap(corr_matrix, annot=True, cmap='RdBu_r', center=0,
                square=True, linewidths=.5, cbar_kws={"shrink": .8}, ax=ax)
    ax.set_title(f'Correlation Matrix - {year}', fontsize=10, pad=10)
    return ax

# --- Load data (assuming you have this from earlier) ---
# Example: load_yearly_data() or similar — replace with your actual loader
# data_list = load_yearly_data()  # <- This should be defined elsewhere

# For demonstration, let's simulate some data:
np.random.seed(42)
years = list(range(1998, 2011))
data_list = {}
for year in years:
    n = 100
    data_list[year] = pd.DataFrame({
        'SMOKING': np.random.normal(20, 5, n),
        'POVERTY': np.random.normal(15, 4, n),
        'PM25': np.random.normal(12, 3, n),
        'NO2': np.random.normal(25, 6, n),
        'SO2': np.random.normal(8, 2, n),
        'RATE': np.random.normal(60, 15, n)
    })

# --- Create grid of all plots ---
n_plots = len(data_list)
n_col = 3
n_row = (n_plots + n_col - 1) // n_col  # Ceiling division

fig, axes = plt.subplots(n_row, n_col, figsize=(15, 5 * n_row))
fig.suptitle('Correlation Matrices by Year', fontsize=14, fontweight='bold', y=0.98)

# Flatten axes array for easy indexing
axes_flat = axes.flatten() if n_row > 1 else axes if n_col > 1 else [axes]

# Generate each correlation plot
for i, (year, df) in enumerate(data_list.items()):
    ax = axes_flat[i]
    create_correlation_plot(df, year, ax=ax)

# Hide unused subplots
for i in range(len(data_list), len(axes_flat)):
    axes_flat[i].set_visible(False)

# Adjust layout to prevent overlap
plt.tight_layout(rect=[0, 0, 1, 0.96])  # Leave space for suptitle

# Display the combined plot
plt.show()

# Optional: Save the figure
# plt.savefig("correlation_matrices_panel.png", dpi=300, bbox_inches='tight')

Regression Analysis (Multiple Linear Regression)

from sklearn.linear_model import LinearRegression
from sklearn.metrics import r2_score

def fit_linear_regression(df, year):
    """
    Fit multiple linear regression: RATE ~ SMOKING + POVERTY + PM25 + NO2 + SO2
    """
    # Select predictors and target
    predictors = ['SMOKING', 'POVERTY', 'PM25', 'NO2', 'SO2']
    target = 'RATE'
    
    # Prepare data (remove rows with missing values)
    X = df[predictors].dropna()
    y = df.loc[X.index, target]
    
    # Fit model
    model = LinearRegression()
    model.fit(X, y)
    
    # Get coefficients and R-squared
    coefficients = dict(zip(predictors, model.coef_))
    intercept = model.intercept_
    r_squared = r2_score(y, model.predict(X))
    
    # Create result DataFrame
    result = {
        'Year': year,
        'Intercept': intercept,
        'R_squared': r_squared
    }
    result.update(coefficients)
    
    return result

# Fit models for all years
regression_results = []
for year, df in data_list.items():
    result = fit_linear_regression(df, year)
    regression_results.append(result)

# Convert to DataFrame
results_table = pd.DataFrame(regression_results)
results_table = results_table.round(3)

print("Multiple Linear Regression Results:")
print(results_table)

Scatter plots

# fig.width: 12
# fig.height: 16
import pandas as pd
import numpy as np
import re
import matplotlib.pyplot as plt
from pathlib import Path

# --- Step 1: Load Data ---
def load_yearly_data(directory="output"):
    """Load all yearly CSV files into a dictionary with year as key."""
    files = sorted(Path(directory).glob("data_*.csv"))

    year_pattern = r"data_(\d{4})\.csv"
    data_dict = {}

    for file_path in files:
        match = re.search(year_pattern, file_path.name)
        if match:
            year = int(match.group(1))
            df = pd.read_csv(file_path)
            data_dict[year] = df

    return data_dict

# Load your data (adjust path if needed)
data_list = load_yearly_data()

# Sort years for consistent ordering
sorted_years = sorted(data_list.keys())

# --- Step 2: Define Plot Function ---
def create_scatter_plot(ax, df, year):
    """Create a scatter plot of SMOKING vs RATE on the given axis."""
    ax.scatter(df['SMOKING'], df['RATE'],
               color='#1f77b4', alpha=0.6, s=30, edgecolors='none')

    ax.set_title(f'SMOKING vs RATE - {year}', fontsize=10, pad=8)
    ax.set_xlabel('SMOKING', fontsize=8)
    ax.set_ylabel('RATE', fontsize=8)
    ax.tick_params(axis='both', labelsize=7)
    ax.grid(True, alpha=0.2)

    # Remove top and right spines for minimal theme
    ax.spines['top'].set_visible(False)
    ax.spines['right'].set_visible(False)

# --- Step 3: Create Grid of Plots ---
n_plots = len(sorted_years)
n_col = 3
n_row = (n_plots + n_col - 1) // n_col

fig, axes = plt.subplots(n_row, n_col, figsize=(15, 5 * n_row))
fig.suptitle('Scatter Plots of SMOKING vs RATE by Year', fontsize=14, fontweight='bold', y=0.98)

# Flatten axes array for easy indexing
axes_flat = axes.flatten() if n_row > 1 else axes if n_col > 1 else [axes]

# Generate each plot
for i, year in enumerate(sorted_years):
    df = data_list[year]
    df_subset = df[['SMOKING', 'RATE']].dropna()
    create_scatter_plot(axes_flat[i], df_subset, year)

# Hide unused subplots
for i in range(len(sorted_years), len(axes_flat)):
    axes_flat[i].set_visible(False)

# Adjust layout to prevent overlap
plt.tight_layout(rect=[0, 0, 1, 0.96])

# Display the combined plot
plt.show()

Best Practices for Functional Programming in Python

1. Keep Functions Pure

# BAD: Side effects inside function
def bad_function(x):
    print(f"Processing: {x}")  # Side effect - printing to console
    return x * 2

# GOOD: Pure function
def good_function(x):
    return x * 2

# Handle side effects outside
values = [1, 2, 3, 4, 5]
results = list(map(good_function, values))
print("Results:", results)  # Print outside the function

2. Embrace Immutability

# BAD: Modifying original data
def bad_transform(df):
    df['new_col'] = df['age'] * 2  # Modifies original DataFrame
    return df

# GOOD: Create new DataFrame
def good_transform(df):
    new_df = df.copy()  # Explicit copy
    new_df['new_col'] = new_df['age'] * 2
    return new_df

# Original data unchanged
original_df = pd.DataFrame({'age': [20, 30, 40]})
transformed_df = good_transform(original_df)
print("Original:", original_df)
print("Transformed:", transformed_df)

3. Use Composition for Complex Logic

import pandas as pd
import numpy as np

# Define small, reusable functions
def clean_names(name_series):
    return name_series.str.strip().str.title()

def convert_to_numeric(series):
    return pd.to_numeric(series, errors='coerce')

def normalize_column(series):
    cleaned = series.dropna()
    if len(cleaned) == 0:
        return pd.Series([np.nan] * len(series), index=series.index)
    mean_val = cleaned.mean()
    std_val = cleaned.std()
    if std_val == 0:
        return pd.Series([0] * len(series), index=series.index)
    return (series - mean_val) / std_val

# Compose functions for complex transformations
def process_data(df):
    return (df
        .assign(name=clean_names(df['name']))
        .assign(age=convert_to_numeric(df['age']))
        .assign(normalized_age=lambda x: normalize_column(x['age']))
    )

sample_df = pd.DataFrame({
    'name': [' alice ', 'BOB', ' charlie '],
    'age': ['25', '30abc', '35']
})

processed = process_data(sample_df)
print("Processed data:")
print(processed)

Summary and Conclusion

This tutorial demonstrated how to implement functional programming principles in Python for data science tasks.

Key Takeaways:

1. Use vectorized operations whenever possible (fastest and most Pythonic)
2. Embrace immutability by creating new objects rather than modifying existing ones
3. Keep functions pure - avoid side effects like printing, file I/O, or global state changes
4. Chain operations using method chaining for readable, maintainable code
5. Use appropriate tools: map(), filter(), reduce() for basic operations; pandas apply(), groupby() for data frames

Benefits of Functional Programming in Python:

• Cleaner, more readable code
• Easier testing (pure functions are deterministic)
• Better parallelization potential
• Fewer bugs due to reduced side effects
• More modular code that’s easier to reuse

When to Avoid Strict FP:

• Performance-critical code where mutation is faster
• Interfacing with external systems requiring side effects
• Very simple scripts where imperative style is clearer

The functional approach shown here scales beautifully to large datasets and complex analyses, making your code more robust, testable, and maintainable.