Python para Data Science: mas alla de la sintaxis

# Dos formas de comparar objetos
a = [1, 2, 3]
b = [1, 2, 3]
c = a

print(a == b)   # True  - mismo VALOR
print(a is b)   # False - distinta IDENTIDAD (distintos objetos)
print(a is c)   # True  - misma IDENTIDAD (mismo objeto)
print(id(a), id(b), id(c))  # Direcciones de memoria

from typing import Callable, Iterable

# Una funcion es un objeto. Se puede asignar, pasar, retornar.
def aplicar_a_lista(func: Callable, datos: Iterable):
    return [func(x) for x in datos]

def cuadrado(x):
    return x ** 2

# La funcion cuadrado SE PASA como argumento
resultado = aplicar_a_lista(cuadrado, [1, 2, 3, 4])
print(resultado)  # [1, 4, 9, 16]

# Tambien se puede hacer al vuelo con lambda
resultado = aplicar_a_lista(lambda x: x ** 3, [1, 2, 3])
print(resultado)  # [1, 8, 27]

def crear_multiplicador(factor):
    """Retorna una funcion que multiplica por `factor`."""
    def multiplicar(x):
        return x * factor  # 'factor' viene del ambito exterior
    return multiplicar

duplicar = crear_multiplicador(2)
triplicar = crear_multiplicador(3)

print(duplicar(5))   # 10
print(triplicar(5))  # 15
print(duplicar(10))  # 20

# El closure mantiene su propio 'factor' aunque crear_multiplicador ya termino
print(duplicar.__closure__[0].cell_contents)  # 2

import time
import functools
from typing import Callable

def timing(func: Callable) -> Callable:
    """Mide el tiempo de ejecucion de una funcion."""
    @functools.wraps(func)  # preserva __name__, __doc__, etc.
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"[{func.__name__}] ejecutado en {elapsed:.4f}s")
        return result
    return wrapper

def logger(func: Callable) -> Callable:
    """Loguea argumentos de entrada y retorno."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"[{func.__name__}] llamado con args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"[{func.__name__}] retorno {result}")
        return result
    return wrapper

@timing
@logger
def entrenar_modelo(epochs, lr):
    # Simular entrenamiento
    time.sleep(0.1)
    return f"modelo entrenado en {epochs} epochs, lr={lr}"

entrenar_modelo(epochs=10, lr=0.001)
# [entrenar_modelo] llamado con args=(), kwargs={'epochs': 10, 'lr': 0.001}
# [entrenar_modelo] retorno modelo entrenado en 10 epochs, lr=0.001
# [entrenar_modelo] ejecutado en 0.1008s

import time
import random

random.seed(42)
N = 10_000_000
datos = [random.randint(1, 100) for _ in range(N)]

# MANERA 1: loop + append (lenta, verbosa)
start = time.perf_counter()
result = []
for x in datos:
    if x % 2 == 0:
        result.append(x ** 2)
print(f"Loop:        {time.perf_counter() - start:.3f}s")

# MANERA 2: list comprehension (rapida, declarativa)
start = time.perf_counter()
result = [x ** 2 for x in datos if x % 2 == 0]
print(f"List comp:   {time.perf_counter() - start:.3f}s")

# MANERA 3: generator (memoria minima, mismo rendimiento)
start = time.perf_counter()
result = sum(x ** 2 for x in datos if x % 2 == 0)
print(f"Generator:   {time.perf_counter() - start:.3f}s")

# Tipicamente la list comp es 30-50% mas rapida que el loop

from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np

class LogTransformer(BaseEstimator, TransformerMixin):
    """Aplica np.log1p a columnas numericas. Encaja en un Pipeline."""

    def __init__(self, columns=None):
        self.columns = columns

    def fit(self, X, y=None):
        # No aprende nada, solo retorna self
        return self

    def transform(self, X):
        X = X.copy() if hasattr(X, 'copy') else np.array(X, copy=True)
        cols = self.columns if self.columns is not None else (X.columns if hasattr(X, 'columns') else range(X.shape[1]))
        for c in cols:
            X[c] = np.log1p(X[c])
        return X

# Esto ahora se puede usar en un Pipeline de sklearn
# from sklearn.pipeline import Pipeline
# pipe = Pipeline([
#     ('log', LogTransformer(columns=['precio', 'ingreso'])),
#     ('scaler', StandardScaler()),
# ])

import csv
from typing import Iterator, Dict

def leer_csv_lazy(path: str) -> Iterator[Dict[str, str]]:
    """Lee un CSV linea por linea. Memoria: O(1)."""
    with open(path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:  # 'yield' implicito (es un generator function)
            yield row

# Procesar 5 GB de transacciones sin cargar todo a memoria
def calcular_monto_total(path: str) -> float:
    total = 0.0
    for row in leer_csv_lazy(path):
        if 'monto' in row:
            total += float(row['monto'])
    return total

# Memoria constante, no importa el tamaño del CSV
print(calcular_monto_total('transacciones.csv'))

# Los generators SON iterables pero NO indexables
gen = (x for x in range(10))
# gen[0]  # TypeError!
list(gen)  # OK, los consume

import numpy as np
from scipy import stats

# Distribucion normal: media=0, std=1
X = stats.norm(loc=0, scale=1)

# PDF (densidad de probabilidad)
print(f"PDF en x=0: {X.pdf(0):.4f}")  # 0.3989
print(f"PDF en x=1: {X.pdf(1):.4f}")  # 0.2419

# CDF (probabilidad acumulada)
print(f"P(X <= 0): {X.cdf(0):.4f}")  # 0.5
print(f"P(X <= 1.96): {X.cdf(1.96):.4f}")  # 0.975 (el famoso 95%)

# Cuantil inverso (ppf = percent point function)
print(f"95% quantile: {X.ppf(0.95):.4f}")  # 1.6449

# Muestreo
samples = X.rvs(size=10_000, random_state=42)
print(f"Media de la muestra: {samples.mean():.3f}")  # ~0
print(f"Std de la muestra: {samples.std():.3f}")    # ~1

# Test de hipotesis: t-test de una muestra
t_stat, p_value = stats.ttest_1samp(samples, 0)
print(f"t-stat: {t_stat:.3f}, p-value: {p_value:.3f}")  # p > 0.05: no rechazamos H0