Глава 14: Промпты для Machine Learning
9 апреля 2026 г.
Принцип
ML-задачи требуют систематического подхода: от понимания данных до продакшн-деплоя. Промпты должны учитывать весь pipeline.
Exploratory Data Analysis
Пример 1: Обзор данных
"ДАННЫЕ:
- Файл: data.csv
- Задача: бинарная классификация
- Целевая переменная: 'churn'
ЗАДАЧА:
EDA checklist.
\`\`\`python
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
df = pd.read\_csv('data.csv')
\# 1. Базовая информация
print(df.shape)
print(df.dtypes)
print(df.describe())
\# 2. Пропуски
print(df.isnull().sum())
print(df.isnull().mean() \* 100) \# процент
\# 3. Целевая переменная
print(df$'churn'$.value\_counts(normalize=True))
\# 4. Распределения числовых
df.select\_dtypes(include=$np.number$).hist(figsize=(12, 8))
\# 5. Категориальные
for col in df.select\_dtypes(include=$'object'$).columns:
print(f'\{col\}: \{df$col$.nunique()\} unique')
\# 6. Корреляции
plt.figure(figsize=(10, 8))
sns.heatmap(df.corr(), annot=True, cmap='coolwarm')
\# 7. Target vs features
for col in numeric\_cols:
df.groupby('churn')$col$.mean().plot(kind='bar')
\`\`\`"
Пример 2: Выбросы и аномалии
"ЗАДАЧА:
Детекция и обработка выбросов.
\`\`\`python
from scipy import stats
def detect\_outliers\_iqr(df, column):
Q1 = df$column$.quantile(0.25)
Q3 = df$column$.quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 \* IQR
upper = Q3 + 1.5 \* IQR
return df$(df\[column$ \< lower) | (df$column$ \> upper)\]
def detect\_outliers\_zscore(df, column, threshold=3):
z\_scores = np.abs(stats.zscore(df$column$))
return df$z\_scores \> threshold$
\# Визуализация
fig, axes = plt.subplots(1, 2, figsize=(12, 4))
df.boxplot(column='value', ax=axes$0$)
axes$1$.hist(df$'value'$, bins=50)
\# Обработка
\# Вариант 1: удаление
df\_clean = df$~df.index.isin(outliers.index)$
\# Вариант 2: clip
df$'value'$ = df$'value'$.clip(lower, upper)
\# Вариант 3: замена на медиану
df.loc$outliers.index, 'value'$ = df$'value'$.median()
\`\`\`"
Пример 3: Feature importance (предварительная)
"ЗАДАЧА:
Быстрая оценка важности признаков.
\`\`\`python
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature\_selection import mutual\_info\_classif
import pandas as pd
\# 1. Correlation с target
correlations = df.corr()$'target'$.abs().sort\_values(ascending=False)
\# 2. Mutual Information
mi\_scores = mutual\_info\_classif(X, y, random\_state=42)
mi\_df = pd.DataFrame(\{
'feature': X.columns,
'mi\_score': mi\_scores
\}).sort\_values('mi\_score', ascending=False)
\# 3. Random Forest importance
rf = RandomForestClassifier(n\_estimators=100, random\_state=42)
rf.fit(X, y)
importance\_df = pd.DataFrame(\{
'feature': X.columns,
'importance': rf.feature\_importances\_
\}).sort\_values('importance', ascending=False)
\# Визуализация
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
\# plots...
\`\`\`"
Preprocessing
Missing values
"ДАННЫЕ:
- Числовые: age (5% NaN), income (12% NaN)
- Категориальные: city (2% NaN), job (8% NaN)
ЗАДАЧА:
Стратегия заполнения.
\`\`\`python
from sklearn.impute import SimpleImputer, KNNImputer
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
\# Стратегии по типу и проценту пропусков
numeric\_features = $'age', 'income'$
categorical\_features = $'city', 'job'$
\# Числовые: KNN для небольших пропусков, median для больших
numeric\_transformer = Pipeline($('imputer', KNNImputer(n\_neighbors=5)) \# или SimpleImputer(strategy='median')$)
\# Категориальные: mode или 'missing' категория
categorical\_transformer = Pipeline($('imputer', SimpleImputer(strategy='constant', fill\_value='missing'))$)
preprocessor = ColumnTransformer($('num', numeric\_transformer, numeric\_features),
('cat', categorical\_transformer, categorical\_features)$)
\`\`\`"
Feature encoding
"ПРИЗНАКИ:
- city: 500 уникальных значений (высокая кардинальность)
- gender: 2 значения
- education: 5 значений с порядком (школа \< бакалавр \< магистр \< PhD)
- tags: multiple values per row
ЗАДАЧА:
Оптимальное кодирование.
\`\`\`python
from sklearn.preprocessing import OneHotEncoder, OrdinalEncoder
from category\_encoders import TargetEncoder
import pandas as pd
\# 1. Low cardinality → OneHot
gender\_encoder = OneHotEncoder(sparse\_output=False, drop='first')
\# 2. High cardinality → Target Encoding
city\_encoder = TargetEncoder(smoothing=10)
\# 3. Ordinal → OrdinalEncoder
education\_order = $'school', 'bachelor', 'master', 'phd'$
education\_encoder = OrdinalEncoder(categories=$education\_order$)
\# 4. Multi-label → MultiLabelBinarizer
from sklearn.preprocessing import MultiLabelBinarizer
mlb = MultiLabelBinarizer()
tags\_encoded = mlb.fit\_transform(df$'tags'$.str.split(','))
\# Pipeline
from sklearn.compose import ColumnTransformer
preprocessor = ColumnTransformer($('gender', gender\_encoder, \['gender'$),
('city', city\_encoder, $'city'$),
('education', education\_encoder, $'education'$)
\])
\`\`\`"
Scaling
"КОНТЕКСТ:
- Линейная модель (LogisticRegression)
- Признаки разного масштаба: age (18-100), income (0-1M), count (0-10000)
- Есть выбросы в income
ЗАДАЧА:
Выбор и применение scaling.
\`\`\`python
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler
\# StandardScaler: z = (x - μ) / σ
\# Для нормально распределённых, без выбросов
standard = StandardScaler()
\# MinMaxScaler: z = (x - min) / (max - min)
\# Для ограниченных диапазонов, нейронных сетей
minmax = MinMaxScaler()
\# RobustScaler: z = (x - median) / IQR
\# Для данных с выбросами
robust = RobustScaler()
\# Выбор для наших данных:
from sklearn.compose import ColumnTransformer
preprocessor = ColumnTransformer($('age', StandardScaler(), \['age'$), \# нормальное распределение
('income', RobustScaler(), $'income'$), \# выбросы
('count', StandardScaler(), $'count'$)
\])
Формулы:
Standard:
MinMax:
Robust: "
\#\# Model Training
\#\#\# Baseline
"ЗАДАЧА: Бинарная классификация, imbalanced (10:1).
BASELINE МОДЕЛИ:
-
Dummy (most_frequent)
-
LogisticRegression
-
RandomForest
-
XGBoost
from sklearn.dummy import DummyClassifier
from sklearn.linear\_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from xgboost import XGBClassifier
from sklearn.model\_selection import cross\_val\_score
from sklearn.metrics import f1\_score, make\_scorer
models = \{
'dummy': DummyClassifier(strategy='most\_frequent'),
'logreg': LogisticRegression(class\_weight='balanced', max\_iter=1000),
'rf': RandomForestClassifier(class\_weight='balanced', n\_estimators=100),
'xgb': XGBClassifier(scale\_pos\_weight=10, n\_estimators=100)
\}
\# Метрика для imbalanced
scorer = make\_scorer(f1\_score)
results = \{\}
for name, model in models.items():
scores = cross\_val\_score(model, X, y, cv=5, scoring=scorer)
results$name$ = \{
'mean': scores.mean(),
'std': scores.std()
\}
print(f'\{name\}: \{scores.mean():.3f\} ± \{scores.std():.3f\}')
\`\`\`"
Hyperparameter Tuning
"МОДЕЛЬ: XGBoost
БЮДЖЕТ: 100 trials
ЗАДАЧА:
Оптимизация с Optuna.
\`\`\`python
import optuna
from sklearn.model\_selection import cross\_val\_score
from xgboost import XGBClassifier
def objective(trial):
params = \{
'n\_estimators': trial.suggest\_int('n\_estimators', 100, 1000),
'max\_depth': trial.suggest\_int('max\_depth', 3, 10),
'learning\_rate': trial.suggest\_float('learning\_rate', 0.01, 0.3, log=True),
'subsample': trial.suggest\_float('subsample', 0.6, 1.0),
'colsample\_bytree': trial.suggest\_float('colsample\_bytree', 0.6, 1.0),
'reg\_alpha': trial.suggest\_float('reg\_alpha', 1e-8, 10.0, log=True),
'reg\_lambda': trial.suggest\_float('reg\_lambda', 1e-8, 10.0, log=True),
'scale\_pos\_weight': trial.suggest\_float('scale\_pos\_weight', 1, 20)
\}
model = XGBClassifier(\*\*params, random\_state=42, use\_label\_encoder=False)
scores = cross\_val\_score(model, X, y, cv=5, scoring='f1')
return scores.mean()
study = optuna.create\_study(direction='maximize')
study.optimize(objective, n\_trials=100, show\_progress\_bar=True)
print('Best params:', study.best\_params)
print('Best score:', study.best\_value)
\# Визуализация
optuna.visualization.plot\_param\_importances(study)
optuna.visualization.plot\_optimization\_history(study)
\`\`\`"
Model Evaluation
Метрики
"ЗАДАЧА:
Полная оценка классификатора.
\`\`\`python
from sklearn.metrics import (
accuracy\_score, precision\_score, recall\_score, f1\_score,
roc\_auc\_score, average\_precision\_score, confusion\_matrix,
classification\_report, roc\_curve, precision\_recall\_curve
)
import matplotlib.pyplot as plt
\# Предсказания
y\_pred = model.predict(X\_test)
y\_proba = model.predict\_proba(X\_test)$:, 1$
\# Метрики
print('Accuracy:', accuracy\_score(y\_test, y\_pred))
print('Precision:', precision\_score(y\_test, y\_pred))
print('Recall:', recall\_score(y\_test, y\_pred))
print('F1:', f1\_score(y\_test, y\_pred))
print('ROC-AUC:', roc\_auc\_score(y\_test, y\_proba))
print('PR-AUC:', average\_precision\_score(y\_test, y\_proba))
\# Classification report
print(classification\_report(y\_test, y\_pred))
\# Confusion matrix
cm = confusion\_matrix(y\_test, y\_pred)
plt.figure(figsize=(6, 6))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
\# ROC curve
fpr, tpr, \_ = roc\_curve(y\_test, y\_proba)
plt.figure()
plt.plot(fpr, tpr, label=f'AUC = \{roc\_auc\_score(y\_test, y\_proba):.3f\}')
plt.plot($0, 1$, $0, 1$, 'k--')
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
\# Precision-Recall curve
precision, recall, \_ = precision\_recall\_curve(y\_test, y\_proba)
plt.figure()
plt.plot(recall, precision)
plt.xlabel('Recall')
plt.ylabel('Precision')
\`\`\`"
Cross-validation
"КОНТЕКСТ:
Временной ряд — нельзя использовать стандартный CV.
ЗАДАЧА:
Time series cross-validation.
\`\`\`python
from sklearn.model\_selection import TimeSeriesSplit
import numpy as np
\# Time Series Split
tscv = TimeSeriesSplit(n\_splits=5)
scores = $$
for fold, (train\_idx, val\_idx) in enumerate(tscv.split(X)):
X\_train, X\_val = X.iloc$train\_idx$, X.iloc$val\_idx$
y\_train, y\_val = y.iloc$train\_idx$, y.iloc$val\_idx$
model.fit(X\_train, y\_train)
score = model.score(X\_val, y\_val)
scores.append(score)
print(f'Fold \{fold\}: train size=\{len(train\_idx)\}, val size=\{len(val\_idx)\}, score=\{score:.3f\}')
print(f'Mean: \{np.mean(scores):.3f\} ± \{np.std(scores):.3f\}')
\# Визуализация splits
fig, ax = plt.subplots(figsize=(10, 4))
for fold, (train\_idx, val\_idx) in enumerate(tscv.split(X)):
ax.scatter(train\_idx, $fold$ \* len(train\_idx), c='blue', s=1)
ax.scatter(val\_idx, $fold$ \* len(val\_idx), c='red', s=1)
ax.set\_xlabel('Sample index')
ax.set\_ylabel('Fold')
\`\`\`"
Debugging Models
Пример 1: Overfitting
"СИМПТОМЫ:
- Train accuracy: 99%
- Val accuracy: 65%
ЗАДАЧА:
Диагностика и исправление.
\`\`\`python
\# 1. Learning curves
from sklearn.model\_selection import learning\_curve
train\_sizes, train\_scores, val\_scores = learning\_curve(
model, X, y, cv=5,
train\_sizes=np.linspace(0.1, 1.0, 10),
scoring='accuracy'
)
plt.plot(train\_sizes, train\_scores.mean(axis=1), label='Train')
plt.plot(train\_sizes, val\_scores.mean(axis=1), label='Validation')
plt.xlabel('Training size')
plt.ylabel('Score')
plt.legend()
\# 2. Regularization (для дерева)
\# Уменьшить max\_depth, увеличить min\_samples\_leaf
\# 3. Feature selection
from sklearn.feature\_selection import SelectFromModel
selector = SelectFromModel(model, threshold='median')
X\_selected = selector.fit\_transform(X, y)
\# 4. Cross-validation с разными размерами
for train\_size in $0.5, 0.6, 0.7, 0.8$:
X\_train, X\_val, y\_train, y\_val = train\_test\_split(
X, y, train\_size=train\_size, random\_state=42
)
model.fit(X\_train, y\_train)
print(f'\{train\_size\}: train=\{model.score(X\_train, y\_train):.3f\}, val=\{model.score(X\_val, y\_val):.3f\}')
\`\`\`"
Пример 2: Data drift
"СИМПТОМЫ:
Production accuracy упала на 15% за месяц.
ЗАДАЧА:
Диагностика data drift.
\`\`\`python
from scipy import stats
import pandas as pd
def compare\_distributions(train\_df, prod\_df, columns):
'''Сравнение распределений train vs production.'''
results = $$
for col in columns:
if train\_df$col$.dtype in $'int64', 'float64'$:
\# KS test для числовых
stat, p\_value = stats.ks\_2samp(train\_df$col$, prod\_df$col$)
test = 'KS'
else:
\# Chi-square для категориальных
train\_counts = train\_df$col$.value\_counts()
prod\_counts = prod\_df$col$.value\_counts()
\# align categories
all\_cats = set(train\_counts.index) | set(prod\_counts.index)
train\_freq = $train\_counts.get(c, 0) for c in all\_cats$
prod\_freq = $prod\_counts.get(c, 0) for c in all\_cats$
stat, p\_value = stats.chisquare(prod\_freq, train\_freq)
test = 'Chi2'
results.append(\{
'column': col,
'test': test,
'statistic': stat,
'p\_value': p\_value,
'drift': p\_value \< 0.05
\})
return pd.DataFrame(results)
drift\_report = compare\_distributions(X\_train, X\_prod, X\_train.columns)
print(drift\_report$drift\_report\['drift'$\])
\`\`\`"
Production
"ЗАДАЧА:
Сохранение и загрузка модели для production.
\`\`\`python
import joblib
import json
from datetime import datetime
\# 1. Сохранение модели
model\_path = 'models/classifier\_v1.joblib'
joblib.dump(model, model\_path)
\# 2. Сохранение метаданных
metadata = \{
'model\_type': type(model).\_\_name\_\_,
'features': list(X.columns),
'target': 'churn',
'metrics': \{
'accuracy': 0.85,
'f1': 0.72,
'roc\_auc': 0.89
\},
'training\_date': datetime.now().isoformat(),
'training\_samples': len(X),
'version': '1.0.0'
\}
with open('models/classifier\_v1\_metadata.json', 'w') as f:
json.dump(metadata, f, indent=2)
\# 3. Загрузка
model = joblib.load(model\_path)
with open('models/classifier\_v1\_metadata.json') as f:
metadata = json.load(f)
\# 4. Inference endpoint (FastAPI)
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class PredictionRequest(BaseModel):
features: dict
class PredictionResponse(BaseModel):
prediction: int
probability: float
@app.post('/predict', response\_model=PredictionResponse)
def predict(request: PredictionRequest):
X = pd.DataFrame($request.features$)$metadata\['features'$\]
pred = model.predict(X)$0$
proba = model.predict\_proba(X)$0, 1$
return PredictionResponse(prediction=int(pred), probability=float(proba))
\`\`\`"