高级数据科学家

面向生产级AI/ML/数据系统的世界级高级数据科学家技能。

核心工作流程

1. 设计A/B测试

python
import numpy as np
from scipy import stats

def calculatesamplesize(baseline_rate, mde, alpha=0.05, power=0.8):

计算每个变体所需的样本量。
baseline_rate: 当前转化率（例如 0.10）
mde: 最小可检测效应（相对值，例如 0.05 = 5%提升）

p1 = baseline_rate
p2 = baseline_rate * (1 + mde)
effect_size = abs(p2 - p1) / np.sqrt((p1 (1 - p1) + p2 (1 - p2)) / 2)
z_alpha = stats.norm.ppf(1 - alpha / 2)
z_beta = stats.norm.ppf(power)
n = ((zalpha + zbeta) / effect_size) 2
return int(np.ceil(n))

def analyze_experiment(control, treatment, alpha=0.05):

运行双比例z检验并返回结构化结果。
control/treatment: 包含conversions和visitors的字典。

p_c = control[conversions] / control[visitors]
p_t = treatment[conversions] / treatment[visitors]
pooled = (control[conversions] + treatment[conversions]) / (control[visitors] + treatment[visitors])
se = np.sqrt(pooled (1 - pooled) (1 / control[visitors] + 1 / treatment[visitors]))
z = (pt - pc) / se
p_value = 2 * (1 - stats.norm.cdf(abs(z)))
cilow = (pt - p_c) - stats.norm.ppf(1 - alpha / 2) * se
cihigh = (pt - p_c) + stats.norm.ppf(1 - alpha / 2) * se
return {
lift: (pt - pc) / p_c,
pvalue: pvalue,
significant: p_value < alpha,
ci95: (cilow, ci_high),
}

--- 实验检查清单 ---

1. 定义一个主要指标并预先注册次要指标。

2. 在开始前计算样本量：calculatesamplesize(0.10, 0.05)

3. 在用户级别（而非会话级别）进行随机化，避免信息泄露。

4. 至少运行一个完整的业务周期（通常为2周）。

5. 检查样本比例偏差：abs(ncontrol - ntreatment) / expected < 0.01

6. 使用analyze_experiment()进行分析，报告提升率和置信区间，而不仅仅是p值。

7. 如果测试多个指标，应用Bonferroni校正：alpha / n_metrics

2. 构建特征工程流水线

python
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer

def buildfeaturepipeline(numericcols, categoricalcols, date_cols=None):

返回一个适用于结构化表格数据的、可拟合的ColumnTransformer。

numeric_pipeline = Pipeline([
(impute, SimpleImputer(strategy=median)),
(scale, StandardScaler()),
])
categorical_pipeline = Pipeline([
(impute, SimpleImputer(strategy=most_frequent)),
(encode, OneHotEncoder(handleunknown=ignore, sparseoutput=False)),
])
transformers = [
(num, numericpipeline, numericcols),
(cat, categoricalpipeline, categoricalcols),
]
return ColumnTransformer(transformers, remainder=drop)

def addtimefeatures(df, date_col):
从日期时间列中提取周期性和滞后特征。
df = df.copy()
df[datecol] = pd.todatetime(df[date_col])
df[dowsin] = np.sin(2 np.pi df[datecol].dt.dayofweek / 7)
df[dowcos] = np.cos(2 np.pi df[datecol].dt.dayofweek / 7)
df[monthsin] = np.sin(2 np.pi df[datecol].dt.month / 12)
df[monthcos] = np.cos(2 np.pi df[datecol].dt.month / 12)
df[isweekend] = (df[datecol].dt.dayofweek >= 5).astype(int)
return df

--- 特征工程检查清单 ---

1. 切勿在整个数据集上拟合转换器——在训练集上拟合，在测试集上转换。

2. 在缩放之前，对右偏的数值特征进行对数变换。

3. 对于高基数类别特征（>50个级别），使用目标编码或嵌入。

4. 在训练/测试集划分之前生成滞后/滚动特征，以避免信息泄露。

5. 记录每个特征的业务含义及其对应代码。

3. 训练、评估和选择预测模型

python
from sklearn.modelselection import StratifiedKFold, crossvalidate
from sklearn.metrics import makescorer, rocaucscore, averageprecision_score
import xgboost as xgb
import mlflow

SCORERS = {
rocauc: makescorer(rocaucscore, needs_proba=True),
avgprec: makescorer(averageprecisionscore, needs_proba=True),
}

def evaluate_model(model, X, y, cv=5):

交叉验证并返回每个评分器的均值±标准差。
对于分类问题使用StratifiedKFold以保持类别平衡。

cvresults = crossvalidate(
model, X, y,
cv=StratifiedKFold(nsplits=cv, shuffle=True, randomstate=42),
scoring=SCORERS,
returntrainscore=True,
)
summary = {}
for metric in SCORERS:
testscores = cvresults[ftest_{metric}]
summary[metric] = {mean: testscores.mean(), std: testscores.std()}
# 标记过拟合：训练集和测试集分数差距过大
trainmean = cvresults[ftrain_{metric}].mean()
summary[metric][overfitgap] = trainmean - test_scores.mean()
return summary

def trainandlog(model, Xtrain, ytrain, Xtest, ytest, run_name):
训练模型并将所有工件记录到MLflow。
with mlflow.startrun(runname=run_name):
model.fit(Xtrain, ytrain)
proba = model.predictproba(Xtest)[:, 1]
metrics = {
rocauc: rocaucscore(ytest, proba),
avgprec: averageprecisionscore(ytest, proba),
}
mlflow.logparams(model.getparams())
mlflow.log_metrics(metrics)
mlflow.sklearn.log_model(model, model)
return metrics

--- 模型评估检查清单 ---

1. 对于不平衡数据集，始终同时报告AUC-PR和AUC-ROC。

2. 将overfit_gap > 0.05视为过拟合的警告信号。

3. 在生产使用前校准概率（Platt缩放/保序回归）。

4. 计算SHAP值以验证特征重要性是否符合业务逻辑。

5. 运行基线模型（例如DummyClassifier）并验证模型优于基线。

6. 将每次运行记录到MLflow——切勿依赖笔记本输出进行比较。

4. 因果推断：双重差分法

python
import statsmodels.formula.api as smf

def diffindiff(df, outcome, treatmentcol, postcol, controls=None):

通过带可选协变量的OLS DiD估计ATT。
df必须包含：结果变量、treatmentcol（0/1）、postcol（0/1）。
返回交互项系数（treatment × post）及其p值。

covariates = + .join(controls) if controls else
formula = (
f{outcome} ~ {treatmentcol} * {postcol}
+ (f + {covariates} if covariates else )
)
result = smf.ols(formula, data=df).fit(cov_type=HC3)
interaction = f{treatmentcol}:{postcol}
return {
att: result.params[interaction],
p_value: result.pvalues[interaction],