Python for AI & Data.

import math, numpy as np, pandas as pd

# ── Float precision in metric comparisons ────────────────
correct, total = 19, 20
accuracy = correct / total           # 0.95

a = 0.1 + 0.2
print(a == 0.3)                      # False (0.30000000000000004)
print(math.isclose(a, 0.3, rel_tol=1e-9))  # True ✓

# Safe threshold comparison for model evaluation
threshold = 0.95
if math.isclose(accuracy, threshold, rel_tol=1e-6) or accuracy > threshold:
    print("Model meets target")

# ── RMSE comparison ───────────────────────────────────────
rmse_v1, rmse_v2 = 0.4230000000001, 0.423
improved = not math.isclose(rmse_v1, rmse_v2, rel_tol=1e-6)
print("V2 improved:", improved)      # False — same within tolerance

# ── NumPy dtype: int64 cannot hold NaN ───────────────────
arr_int = np.array([1, 2, 3], dtype=np.int64)
# arr_int[0] = np.nan  → ValueError: cannot convert float NaN to int

arr_float = np.array([1, 2, 3], dtype=np.float64)
arr_float[0] = np.nan               # works — float64 holds NaN

# ── Pandas silently promotes int → float64 with NaN ──────
s = pd.Series([1, 2, 3])
print(s.dtype)    # int64
s[0] = None
print(s.dtype)    # float64 ← automatic promotion!
print(s)          # [NaN, 2.0, 3.0]

# Use nullable Int64 to keep integers with NaN
s2 = pd.array([1, 2, None], dtype=pd.Int64Dtype())
print(s2)         # [1, 2, <NA>]  ← no float promotion

import re

# ── Text normalisation pipeline ───────────────────────────
def clean_text(text: str) -> str:
    text = text.lower().strip()
    text = re.sub(r'[^a-z0-9 ]', '', text)   # keep alphanumeric + spaces
    return ' '.join(text.split())             # collapse whitespace

raw = "  Hello, World! This is a TEST.  "
print(clean_text(raw))    # "hello world this is a test"

# ── f-strings for experiment logging ─────────────────────
epoch, loss, acc = 5, 0.234, 0.912
log_line = f"Epoch {epoch:02d} | loss={loss:.4f} | acc={acc:.2%}"
print(log_line)           # "Epoch 05 | loss=0.2340 | acc=91.20%"

# ── LLM prompt building ───────────────────────────────────
def build_prompt(query: str, context: list[str], max_ctx: int = 3) -> str:
    ctx_block = "
".join(f"  - {c}" for c in context[:max_ctx])
    return (
        "You are a helpful assistant. Answer based on the context below.

"
        f"Context:
{ctx_block}

"
        f"Question: {query}
Answer:"
    )

prompt = build_prompt(
    "What is RAG?",
    ["RAG retrieves relevant documents", "LLMs use retrieved context to answer"]
)

# ── Efficient string building (join not +) ────────────────
tokens = ["the", "model", "predicted", "positive"]
# BAD — O(n²): creates a new string object on every iteration
result = ""
for t in tokens: result += t + " "

# GOOD — O(n): one allocation, one pass
result = " ".join(tokens)

# ── Regex for feature extraction ──────────────────────────
text = "Order #12345 confirmed. Contact [email protected]"
order_ids = re.findall(r'#(\d+)', text)     # ['12345']
emails    = re.findall(r'[\w.]+@[\w.]+', text)  # ['[email protected]']

from collections import defaultdict

# ── Label encoding with dict ─────────────────────────────
labels = ["cat", "dog", "cat", "bird", "dog", "cat"]

label2id = {label: idx for idx, label in enumerate(sorted(set(labels)))}
id2label = {v: k for k, v in label2id.items()}
# label2id = {'bird': 0, 'cat': 1, 'dog': 2}

encoded = [label2id[l] for l in labels]        # [1, 2, 1, 0, 2, 1]
decoded = [id2label[i] for i in encoded]

# Safe lookup for unknown labels at inference time
def encode_label(label: str, fallback: int = -1) -> int:
    return label2id.get(label, fallback)

encode_label("fish")    # -1 — unknown class, no KeyError

# ── Deduplication with set ────────────────────────────────
urls = ["http://a.com", "http://b.com", "http://a.com", "http://c.com"]
unique_urls = list(dict.fromkeys(urls))   # preserves insertion order (3.7+)

# ── O(1) vocabulary membership check ────────────────────
vocab = {"cat", "dog", "bird", "fish"}         # set, not list!
word = "fish"
print(word in vocab)    # O(1)   ← hashed lookup
# AVOID: word in ["cat", "dog", "bird", "fish"] → O(n) linear scan

# ── List comprehension for feature extraction ─────────────
records = [
    {"text": "great product", "stars": 5},
    {"text": "terrible",      "stars": 1},
    {"text": "it's ok",       "stars": 3},
]
texts   = [r["text"]                      for r in records]
ratings = [r["stars"]                     for r in records]
is_pos  = [r["stars"] >= 4                for r in records]  # binary label

# ── defaultdict for group-by ──────────────────────────────
buckets = defaultdict(list)
for r in records:
    key = "positive" if r["stars"] >= 4 else "negative"
    buckets[key].append(r["text"])
# {"positive": ["great product"], "negative": ["terrible", "it's ok"]}

import numpy as np, pandas as pd

# ── None vs NaN comparison ────────────────────────────────
print(None == None)         # True
print(np.nan == np.nan)     # False  ← IEEE 754: NaN != NaN always
print(np.nan is np.nan)     # True   ← same object in CPython

value = np.nan
print(value == np.nan)      # False — always False, even when it IS NaN!
print(pd.isna(value))       # True  ✓  use pd.isna()

# pd.isna handles both None and NaN uniformly
print(pd.isna(None))        # True
print(pd.isna(np.nan))      # True
print(pd.isna(0))           # False
print(pd.isna(""))          # False — empty string is NOT missing!

# ── Handling missing values in a DataFrame ────────────────
df = pd.DataFrame({
    "age":    [25.0, None, 30.0, np.nan],
    "income": [50000, 60000, None, 80000],
    "label":  [1, 0, 1, 0],
})

print(df.isna().sum())          # missing count per column

# Strategy 1: drop rows with any missing value
df_clean = df.dropna()

# Strategy 2: fill with statistics
df["age"]    = df["age"].fillna(df["age"].median())
df["income"] = df["income"].fillna(df["income"].mean())

# ── Validate before model.predict() ──────────────────────
def validate_features(X: pd.DataFrame) -> None:
    missing = X.isna().any()
    if missing.any():
        cols = missing[missing].index.tolist()
        raise ValueError(f"Missing values in columns: {cols}")
    if not np.isfinite(X.to_numpy(dtype=float)).all():
        raise ValueError("Non-finite values (inf / -inf) detected")

validate_features(df[["age", "income"]])   # must pass before predict()

import pandas as pd, numpy as np
from functools import partial, reduce
from typing import Callable

# ── Single-responsibility transforms ─────────────────────
def clip_outliers(series: pd.Series, lower=0.01, upper=0.99) -> pd.Series:
    lo, hi = series.quantile(lower), series.quantile(upper)
    return series.clip(lo, hi)

def log_transform(series: pd.Series) -> pd.Series:
    return np.log1p(series)         # log(1 + x) — safe for zeros

def standardise(series: pd.Series) -> pd.Series:
    return (series - series.mean()) / series.std()

# ── Pipeline composition ──────────────────────────────────
def apply_pipeline(series: pd.Series,
                   *transforms: Callable[[pd.Series], pd.Series]) -> pd.Series:
    return reduce(lambda s, fn: fn(s), transforms, series.copy())

income = pd.Series([0, 150000, 1_000_000, 50000, 75000])
income_clean = apply_pipeline(income, clip_outliers, log_transform, standardise)

# ── **kwargs for passing model config ─────────────────────
def train_model(X_train, y_train, model_class, **model_params):
    model = model_class(**model_params)
    model.fit(X_train, y_train)
    return model

from sklearn.ensemble import RandomForestClassifier
config = {"n_estimators": 200, "max_depth": 10, "random_state": 42}
model = train_model(X_train, y_train, RandomForestClassifier, **config)

# ── Partial application: pre-configure transforms ─────────
clip_top5pct = partial(clip_outliers, upper=0.95)
age_clean    = apply_pipeline(df["age"], clip_top5pct, standardise)

# ── Mutable default argument bug (common in data code) ────
def add_feature(df, new_cols=[]):   # BUG: shared across calls
    new_cols.append("new")          # silently grows each call

def add_feature_fixed(df, new_cols=None):   # CORRECT
    if new_cols is None: new_cols = []
    new_cols.append("new")
    return new_cols

# ── List comprehension: feature extraction ───────────────
records = [
    {"text": "great product", "stars": 5},
    {"text": "terrible",      "stars": 1},
    {"text": "it's ok",       "stars": 3},
]

texts    = [r["text"]           for r in records]
ratings  = [r["stars"]          for r in records]
is_pos   = [r["stars"] >= 4     for r in records]   # binary label
lengths  = [len(r["text"])      for r in records]   # text length feature

# Dict comprehension: build label encoding in one line
words  = ["cat", "dog", "cat", "bird"]
vocab  = {w: i for i, w in enumerate(sorted(set(words)))}
# {'bird': 0, 'cat': 1, 'dog': 2}

# ── Generator: memory-efficient large-file processing ─────
import csv

def batch_csv(filepath: str, batch_size: int = 1000):
    """Yield batches of rows — O(batch_size) memory, not O(file)."""
    with open(filepath, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        batch = []
        for row in reader:
            batch.append(row)
            if len(batch) == batch_size:
                yield batch
                batch.clear()
        if batch:
            yield batch

# Process a 10M-row file with constant memory
for chunk in batch_csv("train.csv", batch_size=500):
    feats = extract_features(chunk)
    model.partial_fit(feats)         # online/incremental learning

# ── Generator expression: lazy embedding pipeline ─────────
texts = load_all_texts()             # could be millions
# Compute embeddings one-by-one — never all in RAM
embedding_gen = (embed(t) for t in texts)

import numpy as np
# Materialise only when you need the full matrix
X = np.array(list(embedding_gen))   # fits in memory? materialise.

import functools, time, logging
from typing import Any

log = logging.getLogger(__name__)

# ── @timer: profile any function ─────────────────────────
def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        t0 = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - t0
        log.info(f"{func.__name__} completed in {elapsed:.3f}s")
        return result
    return wrapper

@timer
def preprocess_batch(texts: list[str]) -> list[str]:
    return [t.lower().strip() for t in texts]

# ── @retry: LLM API with exponential backoff ─────────────
def retry(max_attempts: int = 3, delay: float = 1.0,
          exceptions: tuple = (Exception,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exc = RuntimeError("No attempts made")
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    wait = delay * (2 ** attempt)
                    log.warning(f"{func.__name__} attempt {attempt+1} failed: {e}."
                                f" Retrying in {wait:.1f}s...")
                    if attempt < max_attempts - 1:
                        time.sleep(wait)
            raise last_exc
        return wrapper
    return decorator

import openai

@retry(max_attempts=5, delay=2.0,
       exceptions=(openai.RateLimitError, openai.APITimeoutError))
def call_llm(prompt: str) -> str:
    resp = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    return resp.choices[0].message.content

# ── @lru_cache: memoise embedding computations ────────────
from functools import lru_cache

@lru_cache(maxsize=10_000)    # str is hashable — works directly
def get_embedding(text: str) -> tuple[float, ...]:
    return tuple(embedding_model.encode(text))  # tuple is hashable

import numpy as np
embed_text = lambda t: np.array(get_embedding(t))  # back to array

print(get_embedding.cache_info())
# CacheInfo(hits=42, misses=100, maxsize=10000, currsize=100)

from pathlib import Path
import json, csv, logging
from datetime import datetime

# ── pathlib: readable cross-platform paths ────────────────
DATA_DIR  = Path("data") / "raw"          # works on Win & Linux
MODEL_DIR = Path("models")
RESULTS   = Path("results") / "metrics.json"

MODEL_DIR.mkdir(parents=True, exist_ok=True)  # create if missing

csvs = list(DATA_DIR.glob("*.csv"))           # find all CSVs

# Read and write without open()
text   = (DATA_DIR / "readme.txt").read_text(encoding="utf-8")
config = {"lr": 0.001, "epochs": 50}
(MODEL_DIR / "config.json").write_text(json.dumps(config, indent=2))

# ── json: persist model results ───────────────────────────
results = {
    "model":      "RandomForest",
    "accuracy":   0.924,
    "f1":         0.891,
    "trained_at": datetime.now().isoformat(),
    "params":     {"n_estimators": 200, "max_depth": 10},
}

with open(RESULTS, "w", encoding="utf-8") as f:
    json.dump(results, f, indent=2)

with open(RESULTS, encoding="utf-8") as f:
    loaded = json.load(f)
print(loaded["accuracy"])     # 0.924

# ── csv: chunked iteration for large files ────────────────
def iter_csv_chunks(path: Path, chunk_size: int = 5000):
    with open(path, encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        batch = []
        for row in reader:
            batch.append(row)
            if len(batch) >= chunk_size:
                yield batch
                batch.clear()
        if batch: yield batch

for chunk in iter_csv_chunks(DATA_DIR / "train.csv"):
    df = pd.DataFrame(chunk)   # process chunk, not full file

# ── logging: structured experiment tracking ───────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    handlers=[logging.FileHandler("run.log"), logging.StreamHandler()],
)
log = logging.getLogger("trainer")

log.info("Training started: model=rf, n_samples=50000")
log.warning("Class imbalance: positive_rate=%.3f", 0.05)

for epoch in range(10):
    loss = train_one_epoch()
    log.info(f"Epoch {epoch+1:02d} | loss={loss:.4f}")

from typing import NamedTuple

# ── NamedTuple: typed readable prediction records ─────────
class Prediction(NamedTuple):
    label:       str
    confidence:  float
    latency_ms:  float

pred = Prediction("dog", 0.91, 12.5)
print(pred.label)       # "dog"
print(pred.confidence)  # 0.91
print(pred._asdict())   # {'label': 'dog', 'confidence': 0.91, 'latency_ms': 12.5}

# ── Rank predictions by confidence ────────────────────────
preds = [
    Prediction("cat",  0.72, 11.2),
    Prediction("dog",  0.91, 12.5),
    Prediction("bird", 0.43, 10.8),
]
ranked = sorted(preds, key=lambda p: p.confidence, reverse=True)
top1   = ranked[0]   # Prediction(label='dog', confidence=0.91, ...)

# Single best: max() is O(n) — faster than sort for top-1
best = max(preds, key=lambda p: p.confidence)

# ── Slicing for train/val/test split ──────────────────────
data = list(range(1000))
n    = len(data)
train = data[:int(0.70 * n)]
val   = data[int(0.70 * n):int(0.85 * n)]
test  = data[int(0.85 * n):]

# ── Batch accumulation with clear() ──────────────────────
BATCH_SIZE = 64
batch: list = []
for record in data_stream:
    batch.append(record)
    if len(batch) >= BATCH_SIZE:
        flush_batch(batch)
        batch.clear()          # O(1) in-place — faster than batch = []
if batch:
    flush_batch(batch)         # handle final partial batch

from collections import Counter

# ── Counter: class distribution analysis ─────────────────
y_train = ["pos"] * 500 + ["neg"] * 150 + ["neutral"] * 350
dist    = Counter(y_train)
print(dist.most_common())   # [('pos', 500), ('neutral', 350), ('neg', 150)]

total = sum(dist.values())
# Inverse-frequency class weights for imbalanced training
class_weights = {k: total / (len(dist) * v) for k, v in dist.items()}
# {'pos': 0.67, 'neutral': 0.95, 'neg': 2.22}

# Token frequency for NLP vocabulary
tokens = "the quick brown fox jumps over the lazy dog".split()
freq   = Counter(tokens)
top3   = freq.most_common(3)   # [('the', 2), ('quick', 1), ...]

# ── Nested dict for model comparison ─────────────────────
results = {
    "random_forest": {"accuracy": 0.924, "f1": 0.891, "train_s": 12.3},
    "xgboost":       {"accuracy": 0.941, "f1": 0.908, "train_s": 45.1},
    "logistic_reg":  {"accuracy": 0.871, "f1": 0.855, "train_s":  0.8},
}
best    = max(results, key=lambda m: results[m]["accuracy"])
best_acc = results[best]["accuracy"]
print(f"Best: {best} ({best_acc:.3f})")

# Safe nested access — no KeyError on missing key
xgb_f1 = results.get("gbm", {}).get("f1", 0.0)  # 0.0 — key missing

# ── Config inheritance (Python 3.9+) ─────────────────────
base_cfg  = {"lr": 0.001, "epochs": 50, "batch": 32, "seed": 42}
overrides = {"lr": 0.01,  "epochs": 10}
final_cfg = base_cfg | overrides   # overrides win; base not mutated
# {'lr': 0.01, 'epochs': 10, 'batch': 32, 'seed': 42}

# Pre-3.9 equivalent: {**base_cfg, **overrides}

# ── Simple in-memory feature store ───────────────────────
_cache: dict = {}
def get_features(user_id: str) -> dict:
    if user_id not in _cache:
        _cache[user_id] = compute_features(user_id)
    return _cache[user_id]

# ── Vocabulary analysis with set algebra ─────────────────
train_vocab = set("the cat sat on the mat".split())
test_vocab  = set("the dog ran on the mat quickly".split())

oov       = test_vocab - train_vocab      # words only in test
overlap   = test_vocab & train_vocab      # words in both
all_vocab = test_vocab | train_vocab      # union

coverage  = len(overlap) / len(test_vocab)
print(f"Test vocab coverage: {coverage:.1%}")   # 57.1%
print(f"OOV words: {oov}")                       # {'dog', 'ran', 'quickly'}

# ── Stop-word filtering — O(1) per token ─────────────────
STOP_WORDS = {"the", "a", "an", "is", "are", "was", "in", "on", "of"}

tokens  = "the cat sat on the mat in the garden".split()
clean   = [t for t in tokens if t not in STOP_WORDS]
# ['cat', 'sat', 'mat', 'garden']

# ── Deduplication — preserving insertion order ────────────
urls = ["a.com", "b.com", "a.com", "c.com"]
unique = list(dict.fromkeys(urls))    # preserves order (Python 3.7+)
# ['a.com', 'b.com', 'c.com']

# ── Data leakage detection ────────────────────────────────
train_ids = {101, 102, 103, 104, 105}
test_ids  = {104, 105, 106, 107, 108}
leakage   = train_ids & test_ids

if leakage:
    raise ValueError(f"Data leakage: {len(leakage)} IDs in both splits: {leakage}")

# ── frozenset as hashable dict key ────────────────────────
feature_cache: dict = {}

def score_feature_combo(features: frozenset[str]) -> float:
    if features not in feature_cache:
        feature_cache[features] = run_ablation(features)
    return feature_cache[features]

result = score_feature_combo(frozenset({"age", "income", "education"}))

from collections import Counter, defaultdict, deque
import heapq

# ── Counter: class distribution ───────────────────────────
y_train = ["pos"] * 500 + ["neg"] * 150 + ["neutral"] * 350
dist    = Counter(y_train)
print(dist.most_common())   # [('pos', 500), ('neutral', 350), ('neg', 150)]

total = sum(dist.values())
class_weight = {k: total / (len(dist) * v) for k, v in dist.items()}
# {'pos': 0.67, 'neutral': 0.95, 'neg': 2.22}

# ── defaultdict: group-by ─────────────────────────────────
records = [("alice", "nlp"), ("bob", "cv"), ("alice", "mlops")]
by_person = defaultdict(list)
for person, topic in records:
    by_person[person].append(topic)
# {'alice': ['nlp', 'mlops'], 'bob': ['cv']}

# ── defaultdict: inverted index ───────────────────────────
corpus  = {0: "cat sat mat", 1: "dog ran away", 2: "cat and dog"}
inverted = defaultdict(set)
for doc_id, text in corpus.items():
    for token in text.split():
        inverted[token].add(doc_id)
# {'cat': {0, 2}, 'dog': {1, 2}, 'sat': {0}, ...}

# Boolean AND retrieval: docs containing "cat" AND "dog"
cat_and_dog = inverted["cat"] & inverted["dog"]   # {2}

# ── deque(maxlen): sliding window feature ─────────────────
WINDOW = 5
window = deque(maxlen=WINDOW)   # auto-evicts oldest when full
rolling_means = []
for price in stock_prices:
    window.append(price)        # old item auto-removed when maxlen exceeded
    if len(window) == WINDOW:
        rolling_means.append(sum(window) / WINDOW)

# ── heapq: top-k — O(n log k) vs sort O(n log n) ─────────
scores = [(0.72, "cat"), (0.91, "dog"), (0.43, "bird"), (0.88, "fox")]
top2   = heapq.nlargest(2, scores)    # [(0.91, 'dog'), (0.88, 'fox')]
bot2   = heapq.nsmallest(2, scores)   # [(0.43, 'bird'), (0.72, 'cat')]

import pandas as pd, numpy as np

class FeatureEngineer:
    # ── Class attribute: shared constant ─────────────────
    NUMERIC_COLS    = ["age", "income", "score"]
    CATEGORICAL_COLS = ["city", "category"]

    def __init__(self, clip_outliers: bool = True):
        # ── Instance attributes: per-object state ────────
        self.clip_outliers = clip_outliers
        self._stats: dict  = {}     # filled during fit()
        self._is_fitted    = False

    def fit(self, df: pd.DataFrame) -> "FeatureEngineer":
        FeatureEngineer.validate_columns(df, self.NUMERIC_COLS)
        for col in self.NUMERIC_COLS:
            self._stats[col] = {"mean": df[col].mean(), "std": df[col].std()}
        self._is_fitted = True
        return self     # enables: fe.fit(train).transform(test)

    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        if not self._is_fitted:
            raise RuntimeError("Call fit() before transform()")
        df = df.copy()
        for col in self.NUMERIC_COLS:
            mu, sd = self._stats[col]["mean"], self._stats[col]["std"]
            df[col] = (df[col] - mu) / sd
        return df

    def fit_transform(self, df: pd.DataFrame) -> pd.DataFrame:
        return self.fit(df).transform(df)

    @classmethod
    def from_config(cls, config: dict) -> "FeatureEngineer":
        """Alternative constructor — build from a JSON config dict."""
        return cls(clip_outliers=config.get("clip_outliers", True))

    @staticmethod
    def validate_columns(df: pd.DataFrame, required: list) -> None:
        """Pure validator — needs neither self nor cls."""
        missing = set(required) - set(df.columns)
        if missing:
            raise ValueError(f"Missing required columns: {missing}")

# Usage
fe = FeatureEngineer.from_config({"clip_outliers": False})
train_out = fe.fit_transform(train_df)    # fit on train
test_out  = fe.transform(test_df)         # apply same stats to test

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import GridSearchCV
import numpy as np

class LogTransformer(BaseEstimator, TransformerMixin):
    """Log(x + offset) — sklearn-compatible transformer."""

    def __init__(self, offset: float = 1.0):
        self.offset = offset      # name MUST match exactly — get_params uses it

    def fit(self, X, y=None):
        self.min_val_ = float(np.min(X))   # trailing _ = fitted attribute
        return self

    def transform(self, X):
        return np.log(np.abs(X) + self.offset)

# Inherits for free: get_params, set_params, fit_transform, clone
lt = LogTransformer(offset=0.5)
print(lt.get_params())    # {'offset': 0.5}

# ── sklearn Pipeline ───────────────────────────────────────
from sklearn.preprocessing import StandardScaler

pipe = Pipeline([
    ("log",   LogTransformer()),
    ("scale", StandardScaler()),
    ("clf",   LogisticRegression(max_iter=1000)),
])

pipe.fit(X_train, y_train)          # fit() on each step with training data
preds = pipe.predict(X_test)         # transform() then predict on test data

# ── GridSearchCV tunes transformer params inside pipeline ──
grid = GridSearchCV(
    pipe,
    param_grid={"log__offset": [0.0, 0.5, 1.0, 2.0]},  # step__param syntax
    cv=5,
    scoring="f1",
)
grid.fit(X_train, y_train)
print(grid.best_params_)    # {'log__offset': 1.0}
print(grid.best_score_)     # 0.912

# ── Inspect MRO ───────────────────────────────────────────
print(LogTransformer.__mro__)
# (<class 'LogTransformer'>, <class 'BaseEstimator'>,
#  <class 'TransformerMixin'>, <class 'object'>)

import pandas as pd
from functools import cached_property

# ── Lazy loading: load on first access, cache result ──────
class Dataset:
    def __init__(self, filepath: str):
        self._filepath = filepath
        self._df: pd.DataFrame | None = None   # not loaded yet

    @property
    def df(self) -> pd.DataFrame:
        if self._df is None:
            print(f"Loading {self._filepath}...")
            self._df = pd.read_csv(self._filepath)
        return self._df    # subsequent accesses return cached result instantly

    @property
    def n_rows(self) -> int:
        return len(self.df)    # triggers load if not yet loaded

# Alternative: functools.cached_property — auto-caches after first call
class Dataset2:
    def __init__(self, filepath: str):
        self._filepath = filepath

    @cached_property
    def df(self) -> pd.DataFrame:
        return pd.read_csv(self._filepath)    # called once, stored in __dict__

ds = Dataset("train.csv")
# Nothing loaded yet — object creation is instant
print(ds.n_rows)   # loads CSV here, prints count
print(ds.n_rows)   # instant — already in self._df

# ── @property setter: validate hyperparameters early ──────
class ModelConfig:
    def __init__(self, n_estimators: int = 100, learning_rate: float = 0.001):
        self.n_estimators  = n_estimators    # routes through setter
        self.learning_rate = learning_rate

    @property
    def n_estimators(self) -> int:
        return self._n_estimators

    @n_estimators.setter
    def n_estimators(self, value: int) -> None:
        if not isinstance(value, int) or value < 1:
            raise ValueError(f"n_estimators must be int >= 1, got {value!r}")
        self._n_estimators = value

    @property
    def learning_rate(self) -> float:
        return self._learning_rate

    @learning_rate.setter
    def learning_rate(self, value: float) -> None:
        if value <= 0:
            raise ValueError(f"learning_rate must be > 0, got {value}")
        self._learning_rate = value

cfg = ModelConfig(n_estimators=200)
cfg.n_estimators = -5   # raises ValueError immediately — not 6 hours later

from dataclasses import dataclass, field, asdict
import json

@dataclass
class TrainingConfig:
    """Typed training configuration — replaces an untyped dict."""
    model_name:    str
    n_estimators:  int   = 100
    max_depth:     int   = 10
    learning_rate: float = 0.001
    random_state:  int   = 42
    feature_cols:  list  = field(default_factory=list)  # NOT =[] !

    def __post_init__(self):
        if self.learning_rate <= 0:
            raise ValueError(f"learning_rate must be > 0, got {self.learning_rate}")
        if self.n_estimators < 1:
            raise ValueError(f"n_estimators must be >= 1")

@dataclass(frozen=True)
class ExperimentKey:
    """Immutable, hashable key for caching experiment results."""
    model:        str
    dataset:      str
    n_estimators: int
    max_depth:    int
    # frozen=True → __hash__ auto-generated → usable as dict key

@dataclass
class ModelResults:
    model_name:         str
    accuracy:           float
    f1_score:           float
    train_time_s:       float
    feature_importance: dict = field(default_factory=dict)

    def to_json(self) -> str:
        return json.dumps(asdict(self), indent=2)  # recursively dict

# Usage
cfg = TrainingConfig("xgboost", n_estimators=200, max_depth=6)
print(cfg)
# TrainingConfig(model_name='xgboost', n_estimators=200, max_depth=6, ...)

key   = ExperimentKey("xgboost", "train_v2", 200, 6)
cache: dict[ExperimentKey, ModelResults] = {}
cache[key] = ModelResults("xgboost", 0.941, 0.908, 45.1)

# Serialize results to disk
print(cache[key].to_json())

# "Copy with changes" for frozen dataclasses
from dataclasses import replace
key_v2 = replace(key, n_estimators=300)   # new key, original unchanged

import pandas as pd
import numpy as np
import numpy.typing as npt
from typing import Optional, Callable

# TypeAlias — give complex types a readable name
Features = npt.NDArray[np.float64]   # type alias
Labels   = npt.NDArray[np.int64]
TextList = list[str]

# ── Annotate preprocessing functions ─────────────────────
def clean_text(text: str) -> str:
    return text.lower().strip()

def extract_features(texts: TextList) -> Features:
    return np.array([[len(t), t.count(" ")] for t in texts], dtype=np.float64)

# ── Optional: value may be absent ────────────────────────
def load_labels(path: str, default: Optional[Labels] = None) -> Optional[Labels]:
    try:
        return np.load(path)
    except FileNotFoundError:
        return default

result = load_labels("missing.npy")    # return type: Optional[Labels] = None

# ── Callable: typed function arguments ───────────────────
def apply_transform(
    series: pd.Series,
    fn: Callable[[pd.Series], pd.Series],
) -> pd.Series:
    return fn(series)

apply_transform(df["age"], np.log1p)   # passes a numpy ufunc as Callable

# ── Union / 3.10+ X | Y syntax ───────────────────────────
def encode_label(label: str | int) -> int:
    return label if isinstance(label, int) else LABEL2ID.get(label, -1)

# ── Annotate class methods ────────────────────────────────
class Scaler:
    def fit(self, X: Features) -> "Scaler": ...
    def transform(self, X: Features) -> Features: ...
    def fit_transform(self, X: Features) -> Features:
        return self.fit(X).transform(X)

from typing import TypeVar, Generic, Sequence, Callable, Iterator
import pandas as pd, numpy as np

T  = TypeVar("T")
DF = TypeVar("DF", pd.DataFrame, pd.Series)   # constrained TypeVar

# ── TypeVar: return type mirrors input type ───────────────
def first(items: Sequence[T]) -> T:
    """Works on any sequence — returns the same element type."""
    return items[0]

x: int  = first([1, 2, 3])        # mypy infers int
s: str  = first(("a", "b", "c"))  # mypy infers str

# ── Constrained TypeVar: accept DataFrame or Series ───────
def clip_values(data: DF, lo: float, hi: float) -> DF:
    return data.clip(lo, hi)   # return type matches input

clipped_df = clip_values(train_df, 0.0, 1.0)   # returns DataFrame
clipped_s  = clip_values(series,   0.0, 1.0)   # returns Series

# ── Generic class: typed pipeline ────────────────────────
class DataPipeline(Generic[T]):
    def __init__(self) -> None:
        self._steps: list[Callable[[T], T]] = []

    def add(self, step: Callable[[T], T]) -> "DataPipeline[T]":
        self._steps.append(step)
        return self

    def run(self, data: T) -> T:
        for step in self._steps:
            data = step(data)
        return data

    def __iter__(self) -> Iterator[Callable[[T], T]]:
        return iter(self._steps)

# Typed as DataPipeline[str] — each step must be str→str
text_pipe: DataPipeline[str] = DataPipeline()
text_pipe.add(str.lower).add(str.strip)
result = text_pipe.run("  Hello World  ")   # "hello world"

# ── Sequence vs List ──────────────────────────────────────
# Sequence[str]: accepts list, tuple, ndarray — read-only
def count_tokens(texts: Sequence[str]) -> list[int]:
    return [len(t.split()) for t in texts]

count_tokens(["hello world", "hi"])   # list — OK
count_tokens(("hi there",))          # tuple — OK (if List, this would fail mypy)

from typing import Protocol, runtime_checkable
import numpy as np

# ── Define structural interfaces ──────────────────────────
@runtime_checkable
class Predictor(Protocol):
    """Any class with fit() and predict() satisfies this — including all sklearn classifiers."""
    def fit(self, X: np.ndarray, y: np.ndarray) -> "Predictor": ...
    def predict(self, X: np.ndarray) -> np.ndarray: ...

@runtime_checkable
class Transformer(Protocol):
    def fit(self, X: np.ndarray, y=None) -> "Transformer": ...
    def transform(self, X: np.ndarray) -> np.ndarray: ...

# ── sklearn models satisfy these without any changes ──────
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler

rf     = RandomForestClassifier()
scaler = StandardScaler()

print(isinstance(rf,     Predictor))    # True — rf has fit() and predict()
print(isinstance(scaler, Transformer))  # True — scaler has fit() and transform()

# ── Write functions that accept any conforming model ──────
def cross_validate(model: Predictor, X: np.ndarray,
                   y: np.ndarray, n_folds: int = 5) -> float:
    """Works with ANY class that has fit() and predict()."""
    from sklearn.model_selection import cross_val_score
    return cross_val_score(model, X, y, cv=n_folds).mean()

# Works with sklearn, XGBoost, custom models — anything that fits
cross_validate(rf, X_train, y_train)

# ── Custom model satisfies Protocol without inheriting ────
class NaiveBayes:
    def fit(self, X: np.ndarray, y: np.ndarray) -> "NaiveBayes":
        self.class_priors_ = np.bincount(y) / len(y)
        return self
    def predict(self, X: np.ndarray) -> np.ndarray:
        return np.zeros(len(X), dtype=int)   # stub

print(isinstance(NaiveBayes(), Predictor))  # True — structurally compatible
cross_validate(NaiveBayes(), X_train, y_train)  # accepted by mypy

from abc import ABC, abstractmethod
import pandas as pd

class BaseDataLoader(ABC):
    """Abstract interface — all loaders must implement load() and preprocess()."""

    def __init__(self, source: str, batch_size: int = 256):
        self.source     = source
        self.batch_size = batch_size

    @abstractmethod
    def load(self) -> pd.DataFrame:
        """Load raw data from source (CSV, API, DB, S3...)."""
        ...

    @abstractmethod
    def preprocess(self, df: pd.DataFrame) -> pd.DataFrame:
        """Clean, validate, and transform raw data."""
        ...

    def load_and_process(self) -> pd.DataFrame:
        """Template method — orchestrates abstract steps."""
        raw = self.load()
        return self.preprocess(raw)

# Trying to instantiate the ABC directly raises TypeError
# BaseDataLoader("path.csv")  →  TypeError: Can't instantiate abstract class

class CsvLoader(BaseDataLoader):
    def load(self) -> pd.DataFrame:
        return pd.read_csv(self.source)

    def preprocess(self, df: pd.DataFrame) -> pd.DataFrame:
        return df.dropna().reset_index(drop=True)

class S3Loader(BaseDataLoader):
    def load(self) -> pd.DataFrame:
        import boto3, io
        obj = boto3.client("s3").get_object(Bucket="ml-data", Key=self.source)
        return pd.read_parquet(io.BytesIO(obj["Body"].read()))

    def preprocess(self, df: pd.DataFrame) -> pd.DataFrame:
        df["text"] = df["text"].str.lower().str.strip()
        return df.dropna(subset=["text", "label"])

# Both loaders share the same orchestration
for Loader in [CsvLoader, S3Loader]:
    data = Loader("train.csv").load_and_process()
    train_model(data)

# isinstance works without @runtime_checkable on ABC
print(isinstance(CsvLoader("f.csv"), BaseDataLoader))  # True

import math, pandas as pd

class BatchDataset:
    """Iterable dataset that yields mini-batches; resets on each epoch."""

    def __init__(self, df: pd.DataFrame, batch_size: int = 64):
        self._df         = df.reset_index(drop=True)
        self._batch_size = batch_size
        self._pos        = 0

    def __iter__(self) -> "BatchDataset":
        self._pos = 0          # reset position on every new for loop
        return self

    def __next__(self) -> pd.DataFrame:
        if self._pos >= len(self._df):
            raise StopIteration
        batch     = self._df.iloc[self._pos : self._pos + self._batch_size]
        self._pos += self._batch_size
        return batch

    def __len__(self) -> int:
        return math.ceil(len(self._df) / self._batch_size)

dataset = BatchDataset(train_df, batch_size=32)

# Works seamlessly with for, enumerate, tqdm
for i, batch in enumerate(dataset):
    loss = train_step(batch)
    if i % 10 == 0:
        print(f"batch {i}/{len(dataset)} loss={loss:.4f}")

# Multi-epoch training — __iter__ resets each time
for epoch in range(10):
    for batch in dataset:      # calls __iter__() → self._pos = 0
        train_step(batch)

# Can also use next() manually
it = iter(dataset)
first_batch  = next(it)
second_batch = next(it)

# ── Separating container (iterable) from iterator ─────────
class DataFolder:
    """Iterable container — creating multiple iterators is safe."""
    def __init__(self, items): self._items = items
    def __iter__(self):
        return iter(self._items)   # returns a NEW iterator each time

from pathlib import Path
from typing import Iterator
import csv

# ── yield from: delegate to sub-generator ────────────────
def read_csv(path: Path) -> Iterator[dict]:
    with open(path, encoding="utf-8") as f:
        yield from csv.DictReader(f)   # delegates all DictReader yields

def load_many(paths: list[Path]) -> Iterator[dict]:
    for path in paths:
        yield from read_csv(path)      # concatenate files lazily

# ── Composable pipeline stages ────────────────────────────
def filter_valid(rows: Iterator[dict]) -> Iterator[dict]:
    for row in rows:
        if row.get("text") and row.get("label"):
            yield row                  # only pass through rows with content

def normalise(rows: Iterator[dict]) -> Iterator[dict]:
    for row in rows:
        row["text"] = row["text"].lower().strip()
        yield row

def to_batches(rows: Iterator[dict], size: int = 64) -> Iterator[list[dict]]:
    batch: list[dict] = []
    for row in rows:
        batch.append(row)
        if len(batch) == size:
            yield batch
            batch.clear()
    if batch:
        yield batch            # last partial batch

# ── Wire stages together — O(batch_size) memory always ────
csv_files = list(Path("data").glob("**/*.csv"))   # all CSVs in tree

pipeline  = to_batches(
    normalise(
        filter_valid(
            load_many(csv_files)
        )
    ),
    size=256,
)

for batch in pipeline:
    features = extract_features(batch)
    model.partial_fit(features)        # online / incremental learning

import time
from contextlib import contextmanager, suppress
from pathlib import Path

# ── @contextmanager: code block timer ─────────────────────
@contextmanager
def timer(label: str):
    t0 = time.perf_counter()
    try:
        yield                  # block runs here
    finally:
        elapsed = time.perf_counter() - t0
        print(f"{label}: {elapsed:.3f}s")   # always prints, even on exception

with timer("preprocessing"):
    df = preprocess(raw_df)    # timed regardless of exceptions

with timer("batch inference"):
    preds = model.predict(X_test)

# ── suppress: silence expected exceptions ─────────────────
with suppress(FileNotFoundError):
    Path("cache/embeddings.npy").unlink()   # delete if exists, ignore if not

with suppress(KeyError):
    val = config["optional_param"]          # use a default instead

# ── DB / vector store connection ──────────────────────────
@contextmanager
def get_connection(dsn: str):
    conn = connect(dsn)
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise                  # re-raise — do not swallow errors
    else:
        conn.commit()          # only commits if no exception
    finally:
        conn.close()           # always closes

with get_connection(DB_URL) as conn:
    upsert_embeddings(conn, embeddings)

# ── GPU memory management ─────────────────────────────────
@contextmanager
def inference_mode(model_path: str):
    import torch
    model = torch.load(model_path, map_location="cuda")
    model.eval()
    try:
        with torch.no_grad():
            yield model
    finally:
        del model
        torch.cuda.empty_cache()   # free GPU memory even on exception

with inference_mode("best_model.pt") as model:
    preds = model(batch_tensor)

import itertools
from itertools import chain, islice, product, groupby, chain

# ── chain: combine datasets lazily ───────────────────────
train_iter  = read_csv(Path("train.csv"))
augment_iter = read_csv(Path("augmented.csv"))
all_data    = chain(train_iter, augment_iter)   # single lazy iterator

# chain.from_iterable: flatten nested token lists
doc_tokens  = [["cat", "sat", "mat"], ["dog", "ran", "away"]]
all_tokens  = list(chain.from_iterable(doc_tokens))
# ['cat', 'sat', 'mat', 'dog', 'ran', 'away']

# ── islice: sample first N from a large generator ─────────
huge_rows   = read_csv(Path("100M_rows.csv"))    # lazy — nothing loaded yet
sample_500  = list(islice(huge_rows, 500))       # only 500 rows read from disk

# ── product: hyperparameter grid — no nested loops ────────
n_est_vals  = [100, 200, 500]
depth_vals  = [5, 10, 20]
lr_vals     = [0.01, 0.001, 0.0001]

for n_est, depth, lr in product(n_est_vals, depth_vals, lr_vals):
    # 3 × 3 × 3 = 27 combinations
    model  = train(n_estimators=n_est, max_depth=depth, lr=lr)
    result = evaluate(model)
    print(f"n={n_est} d={depth} lr={lr}: {result:.3f}")

# Equivalent to np.meshgrid but works with any iterables, not just numbers
all_combos = list(product(n_est_vals, depth_vals))
# [(100, 5), (100, 10), (100, 20), (200, 5), ...]

# ── groupby: aggregate predictions by category ────────────
# IMPORTANT: input must be sorted by the key
records = sorted([
    {"category": "pos", "score": 0.91},
    {"category": "neg", "score": 0.23},
    {"category": "pos", "score": 0.85},
    {"category": "neg", "score": 0.31},
], key=lambda r: r["category"])

for cat, group in groupby(records, key=lambda r: r["category"]):
    scores = [r["score"] for r in group]
    print(f"{cat}: n={len(scores)} mean={sum(scores)/len(scores):.2f}")

class DataValidationError(ValueError):
    """Raised when input data fails schema or type checks."""
    def __init__(self, column: str, issue: str, n_bad: int = 0) -> None:
        self.column = column
        self.n_bad = n_bad
        msg = f"[{column}] {issue}"
        if n_bad:
            msg += f" ({n_bad} rows affected)"
        super().__init__(msg)

class ModelNotFittedError(RuntimeError):
    """Raised when predict() is called before fit()."""

class PipelineError(Exception):
    """Top-level pipeline failure — wraps the upstream cause."""

# Chaining: raise X from Y preserves root cause
try:
    validate_schema(df)
except DataValidationError as e:
    raise PipelineError("Ingestion stage failed") from e
# PipelineError.__cause__ → DataValidationError → full traceback

# Raise early, catch narrow
def predict(self, X: pd.DataFrame) -> npt.NDArray[np.float64]:
    if not hasattr(self, "model_"):          # trailing _ = fitted state
        raise ModelNotFittedError("Call fit() before predict()")
    missing = set(self.feature_names_) - set(X.columns)
    if missing:
        raise DataValidationError("features", f"Missing columns: {missing}")
    return self.model_.predict(X)

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Literal

class TrainingConfig(BaseModel, frozen=True):
    model_type: Literal["xgb", "lgbm", "rf"]
    n_estimators: int   = Field(100,  ge=1,  le=5_000)
    max_depth:    int   = Field(6,    ge=1,  le=50)
    learning_rate: float = Field(0.1, gt=0.0, le=1.0)
    feature_cols: tuple[str, ...]   # tuple for hashability (frozen)
    target_col: str

    @field_validator("feature_cols")
    @classmethod
    def features_not_empty(cls, v: tuple[str, ...]) -> tuple[str, ...]:
        if not v:
            raise ValueError("feature_cols must not be empty")
        return v

    @model_validator(mode="after")
    def target_not_in_features(self) -> "TrainingConfig":
        if self.target_col in self.feature_cols:
            raise ValueError(
                f"target_col '{self.target_col}' must not be in feature_cols"
            )
        return self

# Load from YAML/env — validated at the boundary
cfg = TrainingConfig.model_validate({
    "model_type": "xgb",
    "feature_cols": ["age", "tenure", "spend"],
    "target_col": "churn",
})
# frozen=True → cfg is hashable → usable as cache key
results_cache: dict[TrainingConfig, float] = {}

# Serialize for experiment tracking (MLflow, W&B)
json_str = cfg.model_dump_json()        # deterministic, round-trips

# API response schema
class EmbeddingResponse(BaseModel):
    vector: list[float]
    model:  str
    tokens: int

import logging
from contextlib import suppress

logger = logging.getLogger(__name__)

def load_and_validate(path: str) -> pd.DataFrame:
    try:
        df = pd.read_parquet(path)
    except FileNotFoundError:
        raise DataValidationError("path", f"File not found: {path}")
    except pa.ArrowInvalid as exc:           # Parquet-specific parse error
        raise DataValidationError("schema", "Parquet schema mismatch") from exc
    except Exception as exc:                 # catch-all only as last resort
        logger.exception("Unexpected error loading %s", path)
        raise PipelineError(f"Failed to load {path}") from exc

    if df.empty:
        raise DataValidationError("rows", "File is empty", n_bad=0)
    return df

# Python 3.11: ExceptionGroup for concurrent failures
import asyncio

async def load_all(paths: list[str]) -> list[pd.DataFrame]:
    tasks = [asyncio.create_task(async_load(p)) for p in paths]
    try:
        return await asyncio.gather(*tasks, return_exceptions=False)
    except* DataValidationError as eg:       # except* matches group members
        failed = [str(e) for e in eg.exceptions]
        logger.error("%d files failed validation: %s", len(failed), failed)
        raise

# suppress: only for truly optional cleanup operations
with suppress(FileNotFoundError):
    cache_path.unlink()                      # clean up stale cache; ignore if absent

# Structured logging in handlers (not in raisers)
try:
    result = model.predict(features)
except DataValidationError as e:
    logger.error(
        "Validation failed",
        extra={"column": e.column, "n_bad": e.n_bad, "path": path}
    )
    return fallback_prediction()

from typing import TypeGuard, assert_never, cast, Never
import numpy as np
import numpy.typing as npt

# TypeGuard: custom narrowing function
def is_float64_array(arr: object) -> TypeGuard[npt.NDArray[np.float64]]:
    """Narrows object → float64 NDArray — validates before consuming embeddings."""
    return (
        isinstance(arr, np.ndarray)
        and arr.dtype == np.float64
        and arr.ndim == 2
    )

embeddings = model_registry.get_embeddings()
if is_float64_array(embeddings):
    # mypy knows: embeddings is NDArray[np.float64] here
    similarity = embeddings @ embeddings.T

# assert_never: exhaustiveness checking
ModelOutput = str | float | None

def to_float(result: ModelOutput) -> float:
    match result:
        case float(v):
            return v
        case str(s):
            return float(s)
        case None:
            return 0.0
        case _:
            assert_never(result)    # mypy error if ModelOutput gains a new arm

# cast(): tell mypy what you verified manually
raw = config_registry["embedding_model"]          # type: object
model = cast(EmbeddingModel, raw)                 # you checked isinstance() above

# isinstance narrowing — works automatically with mypy
def process(data: pd.DataFrame | np.ndarray) -> pd.DataFrame:
    if isinstance(data, np.ndarray):
        # mypy knows: data is np.ndarray here
        return pd.DataFrame(data)
    # mypy knows: data is pd.DataFrame here
    return data

# conftest.py — shared fixtures across the test suite
import pytest
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

@pytest.fixture(scope="session")
def sample_df() -> pd.DataFrame:
    """Synthetic DataFrame built once per test session."""
    rng = np.random.default_rng(42)
    return pd.DataFrame({
        "age":    rng.integers(18, 80, size=1_000),
        "tenure": rng.integers(0, 120, size=1_000),
        "spend":  rng.exponential(scale=200, size=1_000),
        "churn":  rng.integers(0, 2, size=1_000),
    })

@pytest.fixture(scope="session")
def fitted_pipeline(sample_df: pd.DataFrame) -> Pipeline:
    pipe = Pipeline([("scaler", StandardScaler()), ("clf", LogisticRegression())])
    X = sample_df[["age", "tenure", "spend"]]
    return pipe.fit(X, sample_df["churn"])

# test_model.py
def test_predict_returns_binary(fitted_pipeline, sample_df):
    X = sample_df[["age", "tenure", "spend"]]
    preds = fitted_pipeline.predict(X)
    assert set(preds).issubset({0, 1})

def test_predict_proba_sums_to_one(fitted_pipeline, sample_df):
    X = sample_df[["age", "tenure", "spend"]]
    proba = fitted_pipeline.predict_proba(X)
    # pytest.approx handles float: checks within abs=1e-6
    assert proba.sum(axis=1) == pytest.approx(np.ones(len(X)), abs=1e-6)

def test_model_roundtrips_via_joblib(fitted_pipeline, sample_df, tmp_path):
    import joblib
    path = tmp_path / "model.joblib"
    joblib.dump(fitted_pipeline, path)
    loaded = joblib.load(path)
    X = sample_df[["age", "tenure", "spend"]]
    np.testing.assert_array_equal(loaded.predict(X), fitted_pipeline.predict(X))

import pytest
import numpy as np
import numpy.typing as npt
from hypothesis import given, settings
from hypothesis import strategies as st
from hypothesis.extra.numpy import arrays

# parametrize: multiple input shapes, one test function
@pytest.mark.parametrize("n_rows,n_cols", [
    (100, 5),
    (1_000, 50),
    (1, 10),        # single-row edge case
    (500, 1),       # single-feature edge case
])
def test_feature_engineer_preserves_rows(n_rows: int, n_cols: int) -> None:
    X = np.random.rand(n_rows, n_cols)
    result = feature_engineer(X)
    assert result.shape[0] == n_rows

# parametrize over model types
@pytest.mark.parametrize("model_type", ["xgb", "lgbm", "rf"])
def test_training_config_valid_model_types(model_type: str) -> None:
    cfg = TrainingConfig(
        model_type=model_type,
        feature_cols=("age", "tenure"),
        target_col="churn",
    )
    assert cfg.model_type == model_type

# hypothesis: generates adversarial edge cases automatically
@given(
    arrays(
        dtype=np.float64,
        shape=st.tuples(st.integers(1, 200), st.just(5)),
        elements=st.floats(min_value=-1e6, max_value=1e6, allow_nan=False),
    )
)
@settings(max_examples=300)
def test_normalise_output_in_range(arr: npt.NDArray[np.float64]) -> None:
    result = min_max_normalise(arr)
    assert float(result.min()) >= 0.0 - 1e-9
    assert float(result.max()) <= 1.0 + 1e-9

# hypothesis finds the minimal failing example automatically
@given(st.lists(st.text(min_size=1), min_size=1, max_size=100))
def test_tokeniser_never_returns_empty_tokens(texts: list[str]) -> None:
    tokens = batch_tokenise(texts)
    assert len(tokens) == len(texts)
    assert all(isinstance(t, list) for t in tokens)

from unittest.mock import AsyncMock, MagicMock, patch, call
import pytest

# Mock async OpenAI embeddings call
@pytest.mark.asyncio
async def test_embedder_batches_correctly() -> None:
    mock_resp = MagicMock()
    mock_resp.data = [MagicMock(embedding=[0.1] * 1_536)] * 10

    with patch("myapp.embedder.AsyncOpenAI") as mock_cls:
        mock_client = mock_cls.return_value
        mock_client.embeddings.create = AsyncMock(return_value=mock_resp)

        embedder = OpenAIEmbedder(client=mock_client, batch_size=10)
        result = await embedder.embed(["text"] * 25)     # 25 docs → 3 batches

    assert mock_client.embeddings.create.call_count == 3
    assert len(result) == 25

# MagicMock(spec=) catches attribute typos against the real API
mock_s3 = MagicMock(spec=boto3.client("s3"))
mock_s3.upload_file("local.parquet", "bucket", "key.parquet")   # OK
# mock_s3.uplod_file(...)                                        # → AttributeError

# side_effect: simulate rate-limit then success (retry testing)
@pytest.mark.asyncio
async def test_llm_client_retries_on_rate_limit() -> None:
    from openai import RateLimitError
    attempts = 0

    async def flaky_create(*args, **kwargs):
        nonlocal attempts
        attempts += 1
        if attempts < 3:
            raise RateLimitError("rate limited", response=MagicMock(), body={})
        return MagicMock(choices=[MagicMock(message=MagicMock(content="answer"))])

    with patch.object(llm_client.chat.completions, "create", side_effect=flaky_create):
        result = await call_with_retry(llm_client, prompt="hello")

    assert result == "answer"
    assert attempts == 3         # failed twice, succeeded on third

# Sequence of return values
mock_fn = MagicMock(side_effect=[10, 20, ValueError("done")])
mock_fn()   # → 10
mock_fn()   # → 20
mock_fn()   # → raises ValueError

import pandas as pd
import numpy as np
from pandas.testing import assert_frame_equal, assert_series_equal
from sklearn.pipeline import Pipeline
import pytest

# Contract test: feature pipeline output schema
def test_feature_pipeline_output_schema(sample_df: pd.DataFrame) -> None:
    result = feature_pipeline(sample_df)

    expected_cols = {"age_norm", "tenure_log", "spend_per_day"}
    assert set(result.columns) == expected_cols, (
        f"Missing: {expected_cols - set(result.columns)}, "
        f"Extra: {set(result.columns) - expected_cols}"
    )
    assert result.dtypes["age_norm"] == np.float64
    assert result.isna().sum().sum() == 0       # pipeline must not introduce NaNs
    assert len(result) == len(sample_df)        # rows preserved

# assert_frame_equal: element-wise with float tolerance
def test_normalise_transform(sample_df: pd.DataFrame) -> None:
    result = log_normalise(sample_df[["spend"]].copy())
    expected = pd.DataFrame(
        {"spend": np.log1p(sample_df["spend"])},
        index=sample_df.index,
    )
    assert_frame_equal(
        result, expected,
        check_exact=False,
        atol=1e-9,          # absolute tolerance for float comparison
        check_dtype=True,   # dtype must match exactly
        check_names=True,   # column names must match
    )

# Integration test: raw CSV → prediction
def test_full_pipeline_end_to_end(tmp_path: Path) -> None:
    # Write a minimal test CSV
    csv_path = tmp_path / "test.csv"
    pd.DataFrame({"age": [30, 45], "tenure": [12, 60], "spend": [150.0, 800.0]})       .to_csv(csv_path, index=False)

    predictions = run_pipeline(str(csv_path))   # full pipeline under test

    assert len(predictions) == 2
    assert all(0.0 <= p <= 1.0 for p in predictions)   # probability scores

# Regression test: output must not change from a known-good baseline
def test_pipeline_output_is_deterministic(fitted_pipeline, sample_df):
    X = sample_df[["age", "tenure", "spend"]]
    run1 = fitted_pipeline.predict_proba(X)
    run2 = fitted_pipeline.predict_proba(X)
    np.testing.assert_array_equal(run1, run2)   # exact equality for deterministic model

import numpy as np

# dtype selection — memory footprint matters at scale
f64 = np.array([1.0, 2.0, 3.0])                       # float64 default: 24 bytes
f32 = np.array([1.0, 2.0, 3.0], dtype=np.float32)     # float32: 12 bytes
print(f64.dtype, f64.nbytes)  # float64 24
print(f32.dtype, f32.nbytes)  # float32 12

# shape and strides
arr = np.arange(12).reshape(3, 4)
print(arr.shape)    # (3, 4)
print(arr.strides)  # (32, 8) — 32 bytes/row, 8 bytes/element in float64

# C-order vs F-order
c = np.array([[1, 2], [3, 4]], order='C')  # row-major (default)
f = np.asfortranarray(c)                   # column-major
print(c.flags['C_CONTIGUOUS'])   # True
print(f.flags['F_CONTIGUOUS'])   # True

# View vs copy — critical to understand
arr = np.arange(10, dtype=np.float64)
view = arr[2:7]      # slice → view (shares memory)
view[0] = 99.0       # modifies arr[2] in place!
print(arr[2])        # 99.0

copy = arr[2:7].copy()  # independent copy
copy[0] = 0.0           # does NOT affect arr

# astype always returns a copy
arr_int = np.array([1, 2, 3], dtype=np.int32)
arr_f64 = arr_int.astype(np.float64)   # new array, no shared memory

# np.asarray: avoid copy if already the right dtype
def to_float64(x):
    return np.asarray(x, dtype=np.float64)  # copies only if needed

# Structured dtypes — lightweight typed records
dt = np.dtype([('age', np.int32), ('score', np.float64)])
records = np.array([(25, 0.91), (30, 0.85)], dtype=dt)
print(records['age'])    # [25 30]
print(records['score'])  # [0.91 0.85]

# Memory layout check
X = np.random.randn(1000, 512)
print(X.flags['C_CONTIGUOUS'])  # True — row-wise ops are cache-friendly

import numpy as np

# Basic: (3,) broadcasts over (4, 3)
matrix = np.arange(12).reshape(4, 3).astype(float)   # (4, 3)
bias   = np.array([10.0, 20.0, 30.0])                 # (3,)
result = matrix + bias                                  # each row += bias → (4, 3)

# Row-wise z-score — keepdims preserves the reduction axis
X = np.random.randn(1000, 512)
mu  = X.mean(axis=1, keepdims=True)   # (1000, 1) — not (1000,)
std = X.std(axis=1,  keepdims=True)   # (1000, 1)
Z   = (X - mu) / std                   # (1000, 512) — broadcast (1000,1)

# Without keepdims, mu shape is (1000,) which won't broadcast against (1000, 512)

# np.newaxis: control dimensions explicitly
col_vec = np.array([1, 2, 3])[:, np.newaxis]   # (3, 1)
row_vec = np.array([10, 20, 30])               # (3,)
outer   = col_vec * row_vec                    # (3, 3) — outer product

# Pairwise L2 distance: (N, D) vs (M, D) → (N, M) without a loop
def pairwise_l2(a: np.ndarray, b: np.ndarray) -> np.ndarray:
    # a: (N, D), b: (M, D)
    diff = a[:, np.newaxis, :] - b[np.newaxis, :, :]  # (N, M, D)
    return np.sqrt((diff ** 2).sum(axis=-1))            # (N, M)

A = np.random.randn(100, 128)
B = np.random.randn(50, 128)
dists = pairwise_l2(A, B)   # (100, 50) — zero Python loops

# Numerically stable softmax (subtract max per row)
logits = np.array([[1.0, 2.0, 3.0], [4.0, 1.0, 0.5]])
shifted = logits - logits.max(axis=1, keepdims=True)  # (2, 3) - (2, 1)
exp     = np.exp(shifted)
probs   = exp / exp.sum(axis=1, keepdims=True)         # (2, 3)
print(probs.sum(axis=1))   # [1.0, 1.0] — correct

# Shape error diagnosis — always print shapes
try:
    np.array([1, 2, 3]) + np.array([1, 2])
except ValueError as e:
    print(e)   # operands could not be broadcast together with shapes (3,) (2,)

import numpy as np

# Ufuncs: C-speed element-wise ops — no Python loop
scores = np.array([0.92, 0.45, 0.78, 0.31, 0.87])
log_odds = np.log(scores / (1 - scores))   # logit transform, vectorised

# Boolean masking — returns 1D array of matching elements
high    = scores[scores > 0.75]            # [0.92, 0.78, 0.87]
indices = np.where(scores > 0.75)[0]       # [0, 2, 4] — indices, not values

# np.where(cond, x, y): vectorised if-else
clamped = np.where(scores > 0.9, 0.9, scores)   # cap at 0.9
labels  = np.where(scores > 0.5, "pos", "neg")  # string labels

# Fancy indexing — select arbitrary elements (always a copy)
order     = np.array([2, 0, 4, 1, 3])
reordered = scores[order]                  # [0.78, 0.92, 0.87, 0.45, 0.31]

# 2D fancy indexing: select specific rows
X     = np.random.randn(1000, 128)
idx   = np.array([0, 5, 99, 500])
batch = X[idx]                             # (4, 128) — selected rows

# argsort: rank all elements
ranks = np.argsort(scores)[::-1]          # descending: [0, 4, 2, 1, 3]
top3_vals = scores[ranks[:3]]             # [0.92, 0.87, 0.78]

# argpartition: O(n) top-k — faster than full sort when k << n
k    = 3
part = np.argpartition(scores, -k)[-k:]  # indices of top-3 (unordered)
top3 = scores[part][np.argsort(scores[part])[::-1]]  # sort only k items

# Vectorised vs apply_along_axis
X = np.random.randn(10_000, 512)

# SLOW: apply_along_axis is a Python-level loop
# result = np.apply_along_axis(lambda row: (row - row.mean()) / row.std(), 1, X)

# FAST: broadcast operations — all in C
mu  = X.mean(axis=1, keepdims=True)
std = X.std(axis=1, keepdims=True)
result = (X - mu) / std   # same result, ~50× faster

import numpy as np

# @ operator: matrix-vector and matrix-matrix products
W = np.random.randn(512, 256)   # weight matrix
x = np.random.randn(512)        # input vector

y = W.T @ x                          # matrix-vector → (256,)
Y = np.random.randn(100, 512) @ W    # batch → (100, 256)

# norms: L1, L2, Frobenius
v = np.array([3.0, 4.0])
print(np.linalg.norm(v))          # L2 = 5.0
print(np.linalg.norm(v, ord=1))   # L1 = 7.0

# Batch row norms — vectorised
X = np.random.randn(1000, 128)
row_norms = np.linalg.norm(X, axis=1)          # (1000,)
X_unit    = X / row_norms[:, np.newaxis]        # unit vectors

# Cosine similarity between two vectors
def cosine(a, b):
    return (a @ b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Solve Ax = b — NEVER use inv for this
A = np.array([[3.0, 1.0], [1.0, 2.0]])
b = np.array([9.0, 8.0])
x = np.linalg.solve(A, b)   # [2.0, 3.0] — numerically stable
# AVOID: x = np.linalg.inv(A) @ b  — slower, less stable

# PCA via eigendecomposition of covariance
X = np.random.randn(500, 50)
cov = np.cov(X.T)                              # (50, 50) — symmetric
eigenvalues, eigenvectors = np.linalg.eigh(cov)  # eigh for symmetric
idx = np.argsort(eigenvalues)[::-1]            # descending by variance
components = eigenvectors[:, idx]              # (50, 50) — sorted PCs

n_components = 10
X_pca = X @ components[:, :n_components]      # (500, 10) — projected

# einsum: versatile tensor contraction
A = np.random.randn(32, 128)
B = np.random.randn(32, 128)

dots   = np.einsum('ij,ij->i', A, B)    # per-row dot product → (32,)
outer  = np.einsum('i,j->ij', A[0], B[0])  # outer product → (128, 128)
matmul = np.einsum('ik,jk->ij', A, B)   # A @ B.T → (32, 32)

# Attention scores: (batch, heads, seq, d_k) — typical transformer op
Q = np.random.randn(2, 8, 64, 64)   # (batch, heads, seq, d_k)
K = np.random.randn(2, 8, 64, 64)
scores = np.einsum('bhqd,bhkd->bhqk', Q, K)  # (2, 8, 64, 64)

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'age':   [25, 30, 35, 40],
    'score': [0.91, 0.75, 0.88, 0.62],
    'label': ['pos', 'neg', 'pos', 'neg'],
}, index=['a', 'b', 'c', 'd'])

# loc: label-based, both ends inclusive
print(df.loc['b':'c', 'age':'score'])    # rows b,c — columns age,score

# iloc: position-based, right end exclusive
print(df.iloc[1:3, 0:2])               # rows 1-2, cols 0-1 (same result)

# at / iat: single-element, O(1) — no full loc overhead
print(df.at['b', 'age'])    # 30  — label lookup
print(df.iat[1, 0])         # 30  — position lookup

# Boolean indexing
high   = df[df['score'] > 0.8]
pos_hi = df[(df['label'] == 'pos') & (df['score'] > 0.8)]

# SettingWithCopyWarning — BAD pattern
subset = df[df['label'] == 'pos']   # may be a view OR copy — ambiguous
# subset['score'] = 0.0             # raises SettingWithCopyWarning — do not do this

# CORRECT: modify via .loc on the original DataFrame
df.loc[df['label'] == 'pos', 'score'] = 0.99

# CORRECT: .copy() when you need an independent subset to modify
subset = df[df['label'] == 'pos'].copy()
subset['score'] = 0.0               # safe — only affects the copy

# dtype inspection and casting
print(df.dtypes)
num_cols = df.select_dtypes(include='number').columns.tolist()

# Nullable integer type — keeps int without float promotion on NaN
s = pd.Series([1, 2, None], dtype=pd.Int64Dtype())
print(s.isna())    # [False, False, True]
print(s.dtype)     # Int64 — not float64

# pd.NA vs np.nan
pd.isna(pd.NA)     # True
pd.isna(np.nan)    # True  — pd.isna handles both uniformly

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'model':   ['rf','rf','xgb','xgb','lgbm','lgbm'],
    'split':   ['train','val','train','val','train','val'],
    'auc':     [0.92, 0.88, 0.95, 0.91, 0.93, 0.90],
    'f1':      [0.87, 0.84, 0.91, 0.87, 0.89, 0.86],
    'fit_sec': [10,   0,    45,   0,    12,   0],
})

# Named aggregation — explicit, no column renaming step
result = df.groupby('model').agg(
    mean_auc  = ('auc', 'mean'),
    best_auc  = ('auc', 'max'),
    total_fit = ('fit_sec', 'sum'),
).reset_index()

# Multiple functions on one column
stats = df.groupby('model')['auc'].agg(['mean', 'std', 'min', 'max'])

# transform: per-group stats broadcast back to original rows
df['auc_dev']     = df.groupby('model')['auc'].transform(lambda x: x - x.mean())
df['auc_rank']    = df.groupby('model')['auc'].transform('rank', ascending=False)
df['group_count'] = df.groupby('model')['auc'].transform('count')

# filter: keep only groups whose mean AUC > threshold
good = df.groupby('model').filter(lambda g: g['auc'].mean() > 0.90)

# Multi-key groupby with unstack → wide format
pivot = df.groupby(['model', 'split'])['auc'].mean().unstack()
# columns are train/val values — each model is a row

# apply: flexible but slow — only for complex multi-column logic
def eval_summary(group: pd.DataFrame) -> pd.Series:
    return pd.Series({
        'n':        len(group),
        'mean_auc': group['auc'].mean(),
        'auc_f1_ratio': group['auc'].mean() / group['f1'].mean(),
    })

summary = df.groupby('model').apply(eval_summary).reset_index()

# value_counts for quick label distribution
df['model'].value_counts(normalize=True, dropna=False)

import pandas as pd
import numpy as np

users  = pd.DataFrame({'user_id': [1, 2, 3], 'name': ['Alice','Bob','Carol']})
orders = pd.DataFrame({'user_id': [1, 1, 2, 4], 'amount': [100, 200, 150, 300]})

# Inner join — only matched rows
inner = users.merge(orders, on='user_id', how='inner')
# user_id 3 (Carol) and 4 (unknown order) are dropped

# Left join — all users, NaN for users without orders
left = users.merge(orders, on='user_id', how='left')

# validate: fail early if cardinality assumption is wrong
users.merge(orders, on='user_id', how='left', validate='one_to_many')
# Raises MergeError if user_id is duplicated in users

# indicator: audit which rows matched
merged = users.merge(orders, on='user_id', how='outer', indicator=True)
left_only  = merged[merged['_merge'] == 'left_only']    # users without orders
right_only = merged[merged['_merge'] == 'right_only']   # orders without users

# merge on multiple keys
df1 = pd.DataFrame({'date': ['2024-01','2024-02'], 'store': ['A','A'], 'sales': [100, 120]})
df2 = pd.DataFrame({'date': ['2024-01','2024-02'], 'store': ['A','A'], 'cost': [40, 50]})
joined = df1.merge(df2, on=['date','store'], how='inner')

# concat: stack DataFrames vertically
parts   = [pd.read_parquet(f'part_{i}.parquet') for i in range(3)]
full_df = pd.concat(parts, ignore_index=True)   # reset to RangeIndex

# concat horizontal: add columns side-by-side
X = pd.concat([df_features, df_extra_features], axis=1)

# merge_asof: nearest-key join for time-series (must be sorted)
events = pd.DataFrame({'time': pd.to_datetime(['09:00','09:05','09:10']),
                        'price': [100, 102, 101]})
trades = pd.DataFrame({'time': pd.to_datetime(['09:02','09:07']), 'qty': [10, 20]})

matched = pd.merge_asof(
    trades.sort_values('time'),
    events.sort_values('time'),
    on='time',
    direction='backward'    # match nearest event at or before trade time
)

import pandas as pd
import numpy as np

# ── category dtype: 10-100× memory reduction ─────────────────
N = 400_000
df = pd.DataFrame({'city': np.random.choice(['London','New York','Paris'], size=N)})
print(df.memory_usage(deep=True)['city'])   # ~26 MB as object

df['city'] = df['city'].astype('category')
print(df.memory_usage(deep=True)['city'])   # ~0.4 MB as category
# groupby on category is faster too
df.groupby('city').size()

# Safe category conversion that handles unseen values at inference
train_cats = pd.Categorical(df['city'])     # fit on training data
# At inference — unknown city becomes NaN, not an error
df_test = pd.DataFrame({'city': ['Tokyo']})
df_test['city'] = pd.Categorical(df_test['city'], categories=train_cats.categories)

# ── chunked reading for files larger than RAM ─────────────────
chunks = []
for chunk in pd.read_csv('large.csv', chunksize=100_000):
    filtered = chunk[chunk['score'] > 0.5]
    chunks.append(filtered)
result = pd.concat(chunks, ignore_index=True)

# ── query() — compiled expression, uses numexpr if installed ──
df_big = pd.DataFrame({'age': np.random.randint(18, 80, 1_000_000),
                        'score': np.random.rand(1_000_000)})

# Slower: boolean indexing creates three temporary arrays
r = df_big[(df_big['age'] > 30) & (df_big['score'] > 0.7)]

# Faster: query parses and evaluates in one compiled pass
r = df_big.query('age > 30 and score > 0.7')

# eval(): compute column without intermediate array
df_big.eval('adjusted = score * (age / 50.0)', inplace=True)

# ── pipe() for readable method chains ─────────────────────────
def clip_outliers(df, col, q_lo=0.01, q_hi=0.99):
    lo, hi = df[col].quantile([q_lo, q_hi])
    return df.assign(**{col: df[col].clip(lo, hi)})

result = (
    df_big
    .query('age >= 18')
    .pipe(clip_outliers, 'score')
    .assign(log_score=lambda d: np.log1p(d['score']))
    .groupby('age')['log_score'].mean()
)

# ── Memory audit ──────────────────────────────────────────────
print(df_big.dtypes)
print(df_big.memory_usage(deep=True).sum() / 1e6, "MB")
df_big.info(memory_usage='deep')

import asyncio
import httpx
import time

# Coroutine: defined with async def, runs when awaited
async def fetch_embedding(client: httpx.AsyncClient, text: str) -> list[float]:
    resp = await client.post(
        "https://api.openai.com/v1/embeddings",
        json={"input": text, "model": "text-embedding-3-small"},
        headers={"Authorization": f"Bearer {OPENAI_KEY}"},
    )
    return resp.json()["data"][0]["embedding"]

# Sequential: 10 calls × 500ms = 5 000ms
async def embed_sequential(texts: list[str]) -> list[list[float]]:
    async with httpx.AsyncClient() as client:
        return [await fetch_embedding(client, t) for t in texts]

# Concurrent: all calls fire simultaneously → ~500ms total
async def embed_concurrent(texts: list[str]) -> list[list[float]]:
    async with httpx.AsyncClient() as client:
        return await asyncio.gather(
            *[fetch_embedding(client, t) for t in texts]
        )

# TaskGroup (Python 3.11+): structured cancellation on first failure
async def embed_with_group(texts: list[str]) -> list[list[float]]:
    results: list[asyncio.Task] = []
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:    # cancels all on any failure
            for text in texts:
                results.append(tg.create_task(fetch_embedding(client, text)))
    return [t.result() for t in results]

# Semaphore: cap concurrent API calls
async def embed_rate_limited(texts: list[str], max_concurrent: int = 5):
    sem = asyncio.Semaphore(max_concurrent)
    async def bounded(text: str):
        async with sem:                          # blocks if max_concurrent running
            async with httpx.AsyncClient() as c:
                return await fetch_embedding(c, text)
    return await asyncio.gather(*[bounded(t) for t in texts])

# run_in_executor: offload blocking calls to a thread pool
async def load_and_embed(path: str) -> list[float]:
    loop = asyncio.get_event_loop()
    df   = await loop.run_in_executor(None, pd.read_csv, path)  # blocking I/O → thread
    text = df["text"].iloc[0]
    return await fetch_embedding_async(text)

# Entry point
if __name__ == "__main__":
    texts = ["RAG retrieval", "LLM prompts", "embeddings"]
    embeddings = asyncio.run(embed_concurrent(texts))

import asyncio
from asyncio import Queue, Semaphore, Lock, Event
from typing import Any

# ── Queue: async producer-consumer with backpressure ─────────
async def producer(queue: Queue, texts: list[str]) -> None:
    for text in texts:
        await queue.put(text)        # blocks if queue full — backpressure
    await queue.put(None)            # sentinel: signal end of stream

async def consumer(queue: Queue, results: list) -> None:
    while True:
        item = await queue.get()     # suspends until item available
        if item is None:
            break
        result = await embed(item)   # async embedding call
        results.append(result)
        queue.task_done()            # signal item processed

async def pipeline(texts: list[str]) -> list:
    queue:   Queue = asyncio.Queue(maxsize=20)   # bounded — producer blocks when full
    results: list  = []
    await asyncio.gather(
        producer(queue, texts),
        consumer(queue, results),
        consumer(queue, results),   # two consumers for throughput
    )
    return results

# ── Semaphore: cap concurrent LLM API calls ───────────────────
RATE_LIMIT = Semaphore(5)   # at most 5 concurrent requests at a time

async def call_llm(prompt: str) -> str:
    async with RATE_LIMIT:          # blocks if 5 already running
        return await _make_api_call(prompt)

async def batch_inference(prompts: list[str]) -> list[str]:
    return await asyncio.gather(*[call_llm(p) for p in prompts])
    # asyncio schedules all but Semaphore ensures max 5 run concurrently

# ── Lock: protect a shared in-memory cache ───────────────────
_cache: dict[str, list[float]] = {}
_lock  = Lock()

async def get_or_embed(text: str) -> list[float]:
    async with _lock:
        if text in _cache:
            return _cache[text]
    embedding = await embed(text)      # compute outside lock — don't hold lock during I/O
    async with _lock:
        _cache[text] = embedding
    return embedding

# ── Event: one-time signal (model loaded, data ready) ─────────
model_ready = Event()

async def load_model() -> None:
    await asyncio.sleep(0)           # simulate async model load
    global model
    model = await async_load_weights("model.pt")
    model_ready.set()               # notify all waiters

async def serve(prompt: str) -> str:
    await model_ready.wait()         # blocks until model_ready.set() called
    return await model.predict(prompt)

import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

# ThreadPoolExecutor: clean pool API
def call_api_sync(url: str) -> dict:
    import requests
    return requests.get(url, timeout=10).json()

urls = [f"https://api.example.com/record/{i}" for i in range(50)]

# Submit all tasks; collect as they complete (not in submission order)
with ThreadPoolExecutor(max_workers=10) as pool:
    futures = {pool.submit(call_api_sync, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            process(data)
        except Exception as e:
            print(f"Failed: {url} — {e}")

# map: simpler when order matters and all must succeed
with ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(call_api_sync, urls, timeout=30))  # preserves order

# Lock: protect shared mutable state
counter = 0
lock    = threading.Lock()

def increment(n: int) -> None:
    global counter
    for _ in range(n):
        with lock:          # only one thread modifies at a time
            counter += 1

threads = [threading.Thread(target=increment, args=(10_000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)   # 100_000 — correct; without lock: unpredictable

# run_in_executor: offload blocking work inside asyncio
import asyncio

async def async_pipeline(paths: list[str]) -> list:
    loop = asyncio.get_event_loop()
    results = []
    for path in paths:
        # pd.read_csv is blocking — offload to thread pool, don't block event loop
        df = await loop.run_in_executor(None, pd.read_csv, path)
        results.append(df)
    return results

# Thread-local storage: per-thread DB connections
_local = threading.local()

def get_connection():
    """Each thread gets its own DB connection — no sharing needed."""
    if not hasattr(_local, 'conn'):
        _local.conn = create_db_connection()
    return _local.conn

import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
import numpy as np

# Must be inside if __name__ == "__main__" on Windows (spawn start method)
if __name__ == "__main__":

    # ProcessPoolExecutor: clean API, prefers over mp.Pool for new code
    def softmax_chunk(chunk: np.ndarray) -> np.ndarray:
        exp = np.exp(chunk - chunk.max(axis=1, keepdims=True))
        return exp / exp.sum(axis=1, keepdims=True)

    X      = np.random.randn(10_000, 512)
    chunks = np.array_split(X, 8)              # split for 8 workers

    with ProcessPoolExecutor(max_workers=8) as pool:
        results = list(pool.map(softmax_chunk, chunks))
    output = np.vstack(results)                # (10_000, 512)

    # starmap: multiple arguments per call
    def cross_val(n_est: int, depth: int, X, y) -> float:
        from sklearn.ensemble import RandomForestClassifier
        from sklearn.model_selection import cross_val_score
        clf = RandomForestClassifier(n_estimators=n_est, max_depth=depth, n_jobs=1)
        return cross_val_score(clf, X, y, cv=3).mean()

    configs = [(100, 5), (200, 10), (500, 20)]
    with mp.Pool(processes=3) as pool:
        scores = pool.starmap(cross_val, [(n, d, X_train, y_train) for n, d in configs])

    # SharedMemory: zero-copy large array sharing (Python 3.8+)
    from multiprocessing import shared_memory

    arr = np.random.randn(100_000, 512).astype(np.float32)
    shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
    shared = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
    np.copyto(shared, arr)             # copy once into shared memory

    def worker(shm_name: str, shape, dtype):
        existing = shared_memory.SharedMemory(name=shm_name)
        view = np.ndarray(shape, dtype=dtype, buffer=existing.buf)
        result = view.mean(axis=1)    # read without copying — zero-copy
        existing.close()
        return result

    with ProcessPoolExecutor(max_workers=4) as pool:
        futs = [pool.submit(worker, shm.name, arr.shape, arr.dtype) for _ in range(4)]
        results = [f.result() for f in futs]

    shm.close()
    shm.unlink()   # always clean up — leaked shared memory persists across restarts

import cProfile, pstats, io
import timeit
import numpy as np

# ── cProfile: find the hot function ──────────────────────────
def profile(fn, *args, n_shown=20, **kwargs):
    pr = cProfile.Profile()
    pr.enable()
    result = fn(*args, **kwargs)
    pr.disable()
    s  = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
    ps.print_stats(n_shown)
    print(s.getvalue())
    return result

profile(train_model, X_train, y_train)

# CLI: python -m cProfile -s cumtime train.py | head -30

# ── timeit: precise micro-benchmarks ─────────────────────────
data = np.random.randn(10_000)

loop_ms = timeit.timeit(
    '[x**2 for x in data]',
    globals={'data': data}, number=500
) * 2   # ms per call

numpy_ms = timeit.timeit(
    'data**2',
    globals={'data': data, 'np': np}, number=500
) * 2

print(f"List comp: {loop_ms:.2f}ms  NumPy: {numpy_ms:.2f}ms")

# ── line_profiler: line-level timing ─────────────────────────
# pip install line_profiler
# Decorate the function you already know is slow (from cProfile)
# from line_profiler import profile  (or use @profile with kernprof)
#
# @profile
def feature_extraction(df):
    df = df.copy()
    df['log_income'] = np.log1p(df['income'])   # line-level time shown
    df['age_sq']     = df['age'] ** 2
    df['interact']   = df['age'] * df['income']
    return df
# Run: kernprof -l -v script.py

# ── memory_profiler: memory growth per line ───────────────────
# pip install memory_profiler
# from memory_profiler import profile
#
# @profile
def load_and_process(path: str) -> np.ndarray:
    df      = pd.read_csv(path)         # peak shown per line
    df      = df.dropna()
    features= df.select_dtypes('number').values
    return features.astype(np.float32)  # peak after cast
# Run: python -m memory_profiler script.py

# ── py-spy: attach to a running process without code changes ──
# sudo py-spy top --pid <PID>            # live top-like view
# sudo py-spy record -o prof.svg --pid <PID> --duration 30   # flamegraph

# ── tracemalloc: Python-level allocation tracking ─────────────
import tracemalloc

tracemalloc.start()
result = run_feature_pipeline(df)
current, peak = tracemalloc.get_traced_memory()
print(f"Peak: {peak / 1e6:.1f} MB")
tracemalloc.stop()

import threading, time
import numpy as np
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

N = 10_000_000

# ── Pure Python CPU work: threading gives NO speedup (GIL) ───
def count_up(n):
    total = 0
    for i in range(n): total += i
    return total

t0 = time.perf_counter()
count_up(N); count_up(N)
seq_time = time.perf_counter() - t0

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=2) as pool:
    list(pool.map(count_up, [N, N]))
thread_time = time.perf_counter() - t0

print(f"Sequential: {seq_time:.2f}s  Threaded: {thread_time:.2f}s")
# Both ~same — GIL prevents parallel execution

# ── Multiprocessing: true parallelism (separate GIL per process)
t0 = time.perf_counter()
with ProcessPoolExecutor(max_workers=2) as pool:
    list(pool.map(count_up, [N, N]))
mp_time = time.perf_counter() - t0
print(f"Multiprocessing: {mp_time:.2f}s")  # ~2× faster

# ── NumPy releases the GIL: threading DOES help ──────────────
def numpy_svd(matrix):
    return np.linalg.svd(matrix)    # LAPACK call releases GIL

matrices = [np.random.randn(500, 500) for _ in range(4)]

t0 = time.perf_counter()
[numpy_svd(m) for m in matrices]
numpy_seq = time.perf_counter() - t0

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as pool:
    list(pool.map(numpy_svd, matrices))
numpy_threaded = time.perf_counter() - t0

print(f"NumPy sequential: {numpy_seq:.2f}s  Threaded: {numpy_threaded:.2f}s")
# Threaded is genuinely faster — LAPACK runs in parallel!

# ── Cython: type declarations allow GIL-free C code ──────────
# Save as fast_sum.pyx, compile with: cythonize -i fast_sum.pyx
#
# cdef double fast_sum(double[:] arr) nogil:
#     cdef double total = 0.0
#     cdef int i
#     for i in range(arr.shape[0]):
#         total += arr[i]
#     return total
#
# Python 3.13 experimental free-threaded build:
# Build CPython with --disable-gil for truly parallel Python threads

import sys, tracemalloc
from dataclasses import dataclass

# Without __slots__: per-instance __dict__ overhead
class Point:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# With __slots__: compact attribute storage, no __dict__
class PointSlots:
    __slots__ = ('x', 'y')
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

p1 = Point(1.0, 2.0)
p2 = PointSlots(1.0, 2.0)

print(sys.getsizeof(p1))           # ~48 bytes + __dict__ (~200 bytes)
print(sys.getsizeof(p2))           # ~56 bytes — no __dict__
print(hasattr(p1, '__dict__'))     # True
print(hasattr(p2, '__dict__'))     # False

# Measure savings at scale with tracemalloc
N = 1_000_000

tracemalloc.start()
pts_dict  = [Point(float(i), float(i)) for i in range(N)]
_, peak_d = tracemalloc.get_traced_memory()
tracemalloc.clear_traces()

pts_slots = [PointSlots(float(i), float(i)) for i in range(N)]
_, peak_s = tracemalloc.get_traced_memory()
tracemalloc.stop()

print(f"dict:  {peak_d / 1e6:.0f} MB")
print(f"slots: {peak_s / 1e6:.0f} MB")  # typically 50-70% less

# @dataclass with slots=True (Python 3.10+)
@dataclass(slots=True)
class Prediction:
    label:      str
    confidence: float
    latency_ms: float

# __slots__ with inheritance — must redeclare at every level
class Base:
    __slots__ = ('x',)

class Child(Base):
    __slots__ = ('y',)    # omitting this adds __dict__ back to Child

# Weak references: add __weakref__ explicitly if needed
class Cacheable:
    __slots__ = ('value', '__weakref__')   # required for weakref.ref(obj)

# gc: disable for acyclic object graphs (rare, but powerful)
import gc
gc.disable()              # only safe if no reference cycles in your objects
build_many_records()
gc.enable()
gc.collect()

import functools
from functools import lru_cache, cache, cached_property
import numpy as np

# ── functools.cache: unlimited in-process memoisation ────────
@cache                          # equivalent to @lru_cache(maxsize=None)
def expensive(n: int) -> int:
    return sum(range(n))        # pure function — same input → same output

# ── lru_cache: bounded in-process memoisation ────────────────
@lru_cache(maxsize=10_000)      # evicts LRU when full
def get_embedding(text: str) -> tuple[float, ...]:
    vec = embedding_model.encode(text)
    return tuple(vec)           # tuple is hashable; list is not

print(get_embedding.cache_info())
# CacheInfo(hits=42, misses=100, maxsize=10000, currsize=100)
get_embedding.cache_clear()     # call when model version changes

# ── joblib.Memory: persistent disk cache ─────────────────────
from joblib import Memory

mem = Memory(location='.cache', verbose=0)

@mem.cache
def extract_features(data_path: str, feature_version: str) -> np.ndarray:
    """Cached by (data_path, feature_version) — bump version to invalidate."""
    df = pd.read_parquet(data_path)
    return compute_features(df)   # expensive operation, cached to disk

feats = extract_features('data/train.parquet', feature_version='v3')
# Second call: loads from .cache/ instantly

# ── Redis: cross-process, distributed cache ───────────────────
import redis, json, hashlib

r = redis.Redis(host='localhost', port=6379, db=0)

def get_cached_embedding(text: str, model_version: str) -> list[float]:
    # Include model_version in key — invalidates on model upgrade
    key = f"emb:{model_version}:{hashlib.md5(text.encode()).hexdigest()}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    embedding = model.encode(text).tolist()
    r.setex(key, time=3600, value=json.dumps(embedding))  # TTL 1 hour
    return embedding

# Bulk invalidation when model changes
def invalidate_model_cache(old_version: str) -> int:
    pattern = f"emb:{old_version}:*"
    keys    = r.keys(pattern)
    if keys:
        r.delete(*keys)
    return len(keys)

# ── cached_property: memoised instance attribute ──────────────
class Dataset:
    def __init__(self, path: str):
        self._path = path

    @cached_property
    def features(self) -> np.ndarray:
        return pd.read_parquet(self._path).values  # called once, stored in __dict__

    def reload(self) -> None:
        """Invalidate cached_property by deleting it from __dict__."""
        self.__dict__.pop('features', None)