新米データサイエンティストが守るコードの３つの原則

この記事の背景

多くのデータ分析プロジェクトは Jupyter Notebook の中で生まれます。しかしプロジェクトが「一度きりの分析」から「継続的に更新する予測システム」に育つと、ある日こんな壁にぶつかります。

• 「先月のモデルを再現したいのに、同じ結果にならない」
• 「前任者が書いた 500 行の関数、どこを直せばいいか分からない」
• 「新しい特徴量を足したら、関係ないはずの推論パイプラインが壊れた」

こうした問題の多くは、コードの設計で防げます。本記事では、筆者が実務で痛感した失敗をもとに、分析コードを壊れにくくする ３つの原則 と、それぞれの具体的な実装パターンを紹介します。

３つの原則と対応パターン

原則 1  読み手に優しい名前をつける
   ├── Python のスタイルガイドに従う
   ├── 文脈の重複を削る
   └── 「何をする関数か」が名前だけで伝わるようにする

原則 2  関数は「ひとつの仕事」だけにする
   ├── 巨大パイプラインを分解する
   ├── 特徴量は「1 特徴量 = 1 関数」で作る
   └── 関数のインターフェースを小さく保つ

原則 3  コードに「取扱説明書」を埋め込む
   ├── 型ヒントで入出力を宣言する
   └── docstring で意図と使用例を残す

原則 1：読み手に優しい名前をつける

コードは書く時間より読む時間の方がはるかに長い。

パターン 2-1：Python のスタイルガイドに従う

命名規則はチーム内の「交通ルール」です。PEP 8 に従えば、外部のライブラリとも自然に調和します。

種類	書式	例
関数・変数	snake_case	compute_shap_values, test_ratio
クラス	PascalCase	TimeSeriesSplitter, CategoricalEncoder
定数	UPPER_SNAKE	N_SPLITS, TARGET_COL
内部用	_leading_underscore	_validate_input

# ✅ ルールが統一されている
N_SPLITS = 5
TARGET_COL = "churn"


class CategoricalEncoder:
    def __init__(self, min_frequency: int = 10):
        self.min_frequency = min_frequency
        self._mapping = {}

    def fit(self, series: "pd.Series") -> "CategoricalEncoder":
        counts = series.value_counts()
        self._mapping = {
            cat: idx for idx, cat in enumerate(counts[counts >= self.min_frequency].index)
        }
        return self

パターン 1-2：文脈の重複を削る

属性名はそのクラスの中でしか使わないので、クラス名を繰り返す必要はありません。呼び出し側で pipeline.pipeline_steps と二重になるのを避けます。

# ❌ 冗長
class Pipeline:
    def __init__(self):
        self.pipeline_steps = []       # pipeline.pipeline_steps

# ✅ スッキリ
class Pipeline:
    def __init__(self):
        self.steps = []                # pipeline.steps

パターン 1-3：名前だけで「何をする関数か」を伝える

ルール	理由	Before → After
関数名は動詞で始める	「行動」が伝わる	metrics() → compute_metrics()
クラス名は名詞にする	「モノ」が伝わる	Predict → Predictor
bool 変数は is_/has_	真偽が伝わる	trained → is_trained

同じカテゴリの変数は接頭辞を揃えると、コード補完が利きやすく一覧性も上がります。

# 指標ごとに揃える
metric_accuracy = 0.92
metric_precision = 0.88
metric_recall = 0.91

# データ分割ごとに揃える
split_train = df.iloc[:800]
split_val   = df.iloc[800:900]
split_test  = df.iloc[900:]

原則 2：関数は「ひとつの仕事」だけにする

「この関数は何をするの？」に一文で答えられないなら、分割すべきサイン。

パターン 2-1：巨大パイプラインを分解する

Notebook でよく見る「読み込みから評価まで全部入り」の関数は、テストも差し替えも困難です。1 ステップ = 1 関数に分割すると、各関数が独立に検証・交換できるようになります。

# ❌ 600 行の train_and_evaluate() を…

# ✅ ステップごとに分離する
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.metrics import f1_score


def ingest(path: str) -> pd.DataFrame:
    """CSV を読み込み、基本的な型変換を行う"""
    df = pd.read_csv(path, parse_dates=["signup_date"])
    return df


def engineer_features(
    df: pd.DataFrame,
    reference_date: pd.Timestamp | None = None,
) -> pd.DataFrame:
    """個別の特徴量関数を順に適用する（詳細はパターン 3-2）"""
    if reference_date is None:
        reference_date = pd.Timestamp("2026-04-01")  # 固定日を使い再現性を保つ
    df = df.copy()
    df["tenure_days"] = add_tenure_days(df, reference_date)
    df["log_revenue"] = add_log_revenue(df)
    return df


def split(df: pd.DataFrame, target: str):
    """訓練・検証にデータを分割する"""
    X = df.drop(columns=[target])
    y = df[target]
    return train_test_split(X, y, test_size=0.25, stratify=y, random_state=2026)


def fit_model(X_train, y_train, **hparams):
    """モデルを学習して返す"""
    clf = HistGradientBoostingClassifier(**hparams)
    clf.fit(X_train, y_train)
    return clf


def score(clf, X_test, y_test) -> dict[str, float]:
    """主要指標を辞書で返す"""
    y_pred = clf.predict(X_test)
    return {"f1": f1_score(y_test, y_pred, average="macro")}

分割の恩恵：

• engineer_features だけ差し替えて A/B 比較できる
• fit_model に別のアルゴリズムを渡してもパイプライン全体は壊れない
• 各関数が十数行なので、ユニットテストも簡潔に書ける

パターン 2-2：特徴量は「1 特徴量 = 1 関数」で作る

特徴量エンジニアリングはデータサイエンスの核心です。ここを 1 つの巨大関数にまとめてしまうと、特徴量の追加・削除・修正のたびに関数全体を読み解く必要が生じ、変更が別の特徴量を壊すリスクが高まります。

原則は明快です——1 つの特徴量を生成するロジックは、1 つの関数に閉じ込める。

# src/features/tenure.py
import pandas as pd


def add_tenure_days(
    df: pd.DataFrame,
    reference_date: pd.Timestamp,
    signup_col: str = "signup_date",
) -> pd.Series:
    """基準日からの経過日数を計算する

    reference_date を固定日にすることで、
    実行日によって値が変わる問題を防ぐ（原則 1 との連携）。
    """
    return (reference_date - df[signup_col]).dt.days

# src/features/revenue.py
import numpy as np
import pandas as pd


def add_log_revenue(
    df: pd.DataFrame,
    col: str = "revenue",
) -> pd.Series:
    """収益の対数変換（log1p）を返す

    右裾の重い分布を緩和する。元カラムは残す。
    """
    return np.log1p(df[col])

# src/features/rfm.py
import pandas as pd


def add_purchase_frequency(
    df: pd.DataFrame,
    user_col: str = "user_id",
    date_col: str = "purchase_date",
) -> pd.Series:
    """ユーザーごとの購入頻度（回数）を算出する"""
    freq = df.groupby(user_col)[date_col].transform("count")
    return freq

これらを engineer_features でまとめて呼び出します。

def engineer_features(
    df: pd.DataFrame,
    reference_date: pd.Timestamp,
) -> pd.DataFrame:
    """個別の特徴量関数を順に適用する"""
    df = df.copy()
    df["tenure_days"] = add_tenure_days(df, reference_date)
    df["log_revenue"] = add_log_revenue(df)
    df["purchase_freq"] = add_purchase_frequency(df)
    return df

パターン 2-3：関数のインターフェースを小さく保つ

引数が 4 つ以上になったら、設定値を dataclass または 辞書 にまとめて渡します。

from dataclasses import dataclass


@dataclass
class TrainConfig:
    n_estimators: int = 300
    max_depth: int = 6
    learning_rate: float = 0.05
    subsample: float = 0.8
    seed: int = 2026


def run_training(X, y, config: TrainConfig):
    from sklearn.ensemble import GradientBoostingClassifier

    clf = GradientBoostingClassifier(
        n_estimators=config.n_estimators,
        max_depth=config.max_depth,
        learning_rate=config.learning_rate,
        subsample=config.subsample,
        random_state=config.seed,
    )
    clf.fit(X, y)
    return clf


# 呼び出し側は設定だけ渡せば OK
cfg = TrainConfig(n_estimators=500, learning_rate=0.01)
model = run_training(X_train, y_train, cfg)

辞書のアンパック **dict も便利ですが、dataclass を使うとエディタの補完が利く + Typo を検知できるので、パラメータが固まっている場面では dataclass が優れます。

辞書 vs dataclass の判断基準： 実験の探索段階でパラメータが頻繁に増減するなら辞書が手軽。パラメータが安定してきたら dataclass に移行すると、型チェックと補完の恩恵が得られます。

原則 3：コードに「取扱説明書」を埋め込む

ドキュメントのないコードは、半年後にはレガシーコードになる。

パターン 3-1：型ヒントで入出力を宣言する

型ヒントは「この関数に何を渡して何が返ってくるか」をコードそのもので表現する手段です。mypy や pyright と組み合わせれば、実行前にバグを検出できます。

import numpy as np
import pandas as pd


def detect_anomalies(
    series: pd.Series,
    window: int = 30,
    sigma: float = 3.0,
) -> pd.Series:
    """移動平均 ± nσ を超える時点を True とするブール Series を返す

    戻り値は元の Series と同じインデックスを保持する。
    window 未満の先頭部分は NaN により False になる。
    """
    rolling_mean = series.rolling(window).mean()
    rolling_std = series.rolling(window).std()
    upper = rolling_mean + sigma * rolling_std
    lower = rolling_mean - sigma * rolling_std
    return (series > upper) | (series < lower)

パターン 3-2：docstring で意図と使用例を残す

型ヒントが「仕様」なら、docstring は「なぜそう作ったか」を伝える場所です。Example を書いておくと、新しいメンバが即座に動かせます。

class WinsorizeTransformer:
    """指定パーセンタイルで値をクリップする前処理器

    外れ値の影響を抑えたいが、行そのものは残したい場合に使う。
    sklearn の Pipeline に組み込める（TransformerMixin 準拠）。

    Args:
        lower_pct: 下側パーセンタイル（デフォルト 1%）
        upper_pct: 上側パーセンタイル（デフォルト 99%）

    Example:
        >>> import numpy as np
        >>> X = np.array([[1], [2], [3], [100]])
        >>> wt = WinsorizeTransformer(lower_pct=0, upper_pct=75)
        >>> wt.fit(X).transform(X)
        array([[ 1.  ],
               [ 2.  ],
               [ 3.  ],
               [27.25]])
    """

    def __init__(self, lower_pct: float = 1, upper_pct: float = 99):
        self.lower_pct = lower_pct
        self.upper_pct = upper_pct

あとがき

「きれいなコードを書くこと」自体が目的ではありません。目的は 将来の自分やチームメンバが、安心してコードを変更できる状態を作ること です。最初から全部を完璧にする必要はなく、まずは3原則を意識するだけでも、プロジェクトの健全さは大きく変わります。