Modeling / Pipeline

建模（包含 Modeling 和 Pipeline）暂时停止功能迭代，现有入口与使用文档保留。只处理必要的正确性、运行修复和上游适配。 Experimental 说明接口成熟度，暂停说明开发投入。你可以结合编程型 AI、MARS 分析报告和自己的模型工具构建定制流程。 核心接口变化时，本模块直接修改调用代码，适配规则见稳定性。

Experimental

Modeling 和 Pipeline 的参数、结果对象与 artifact 结构仍可能在 0.0.x 版本间调整。生产流程应 固定精确版本，并为依赖字段增加契约测试。

适用场景

mars.modeling 负责样本切分、模型调参、trial replay、评估和特征重要性；mars.pipeline 将筛选、 可选 WOE 分箱和建模串成一个可复用流程。

支持 LightGBM、XGBoost、CatBoost 和 Logistic Regression。使用前安装：

pip install "mars-risk[ml,tuning] @ git+https://github.com/leeesq/mars-risk.git"

Pipeline 完整调用

from datetime import date, timedelta

import numpy as np
import polars as pl

from mars.pipeline import MarsModelingPipeline, MarsModelingStep

rng = np.random.default_rng(1206)
row_count = 180
income = rng.normal(5200, 1300, row_count)
utilization = rng.uniform(0.05, 0.95, row_count)
logit = -0.00045 * (income - 5200) + 3.2 * (utilization - 0.5)
probability = 1.0 / (1.0 + np.exp(-logit))

development_df = pl.DataFrame(
    {
        "apply_dt": [date(2025, 1, 1) + timedelta(days=index) for index in range(row_count)],
        "income": income,
        "utilization": utilization,
        "target": rng.binomial(1, probability),
    }
)

pipeline = MarsModelingPipeline(
    target="target",
    features=["income", "utilization"],
    steps=[
        MarsModelingStep(
            name="modeling",
            model_type="lgb",
            time_col="apply_dt",
            split_ratios={"train": 0.6, "val": 0.2, "oot": 0.2},
            tune_params={
                "n_trials": 1,
                "startup_trials": 1,
                "num_boost_round": 20,
                "early_stopping_rounds": 5,
                "artifact_dir": None,
            },
        )
    ],
)

pipeline_result = pipeline.fit(development_df)
scored_df = pipeline.predict(development_df, pred_col="model_score")

示例使用时间列严格切分 train/val/oot，只运行一次轻量 LightGBM trial，并设置 artifact_dir=None 避免写文件。

Session 工作流

不需要 Pipeline 时，可以直接使用 MarsModelingSession：

阶段	方法	输出
时间切分	slice()	带 dataset_flag 的样本
调参	tune()	MarsModelTuningResult
指定 trial 复盘	replay() / MarsModelReplayRunner	MarsModelReplayResult
建模评估	evaluate()	MarsModelingReport
特征增长	tune_incrementally()	MarsFeatureGrowthResult

构造 Session 时固定模型类型、特征、target、优化指标和随机种子；数据、时间列、切分比例和输出目录 属于单次方法调用。

Pipeline 约束

• MarsSelectionStep 可以出现多次，每步只消费上一阶段 active features。
• MarsWOEBinningStep 主要服务 LR/评分卡；树模型通常直接使用筛选后的原始特征。
• MarsModelingStep 最多出现一次且必须位于最后。
• 任一筛选步骤筛空特征时立即抛出 ValueError。

Artifact 与 Replay

artifact_dir=None 表示完全不落盘。指定目录时，每次调参生成独立运行目录，保存 history、配置、 元数据、重要性和保留模型。Replay 可以按 Top-K 或 trial_nums 复现候选模型；未保留模型需要设置 retrain=True。

候选排序与外部选择政策

tune() 主要根据 val 指标优化模型。默认 replay() 按以下顺序准备候选排序：

1. 保留 history 中 trial_state=COMPLETE 且 is_valid 的行。
2. 寻找列名含 oot、且以 _<sort_metric> 结尾的列。
3. include_val=True 时，加入名称为 val_<sort_metric> 的列。
4. 用 Pandas 按行计算这些列的均值，跳过缺失值，得到 custom_mean_score。

没有找到排序列时会报错；即使显式提供 trial_nums，history 也必须包含这些列。

参数	实际作用
top_k	未提供 trial_nums 时取均值排序前 K 行
sort_metric	选择排序列的指标后缀，名称小写规范化
include_val	是否加入识别到的 val 列，不会自动加入 TEST
metric_directions	本次逐项覆盖 > 历史配置 > 项目默认；排序和 backend 共用解析结果
trial_nums	回放显式候选，保持给定顺序，top_k 不参与选择

metric_directions=None 是未提供覆盖，{} 是无覆盖项；都保留历史。部分覆盖不丢失其他方向。 结果的 metric_directions 和 replay artifact 会记录实际使用的方向。 均值排序只提供候选顺序，业务流程仍须制定最终模型选择规则。 出现相同均值时，也应由外部流程明确并列处理方式。

外部流程先取得候选的真实 TEST 评估，声明方向、候选数量、资格条件和并列处理方式，再生成 trial_nums。 下面的示例使用外部表，其中的字段由调用方准备，MARS history 不会自动增加这些字段。 缺少 TEST 结果时，必须先评估候选，不能将 val 重命名为 test。示例无需模型 SDK、LLM 或重训练：

"""外部候选选择示例；模拟评估表不是 MARS 自动生成的 TEST 字段。"""

from __future__ import annotations

import math
from typing import Literal

import pandas as pd


def select_candidates(
    evaluation: pd.DataFrame,
    *,
    metric: str,
    direction: Literal["maximize", "minimize"],
    candidate_count: int = 10,
    slice_name: str = "TEST",
) -> tuple[list[int], pd.DataFrame]:
    """在明确切片选取合格候选，并列按 trial_num 升序打破。

    Parameters
    ----------
    evaluation : pd.DataFrame
        外部已计算表，含 trial_num、slice、eligible 和指标；每 trial/切片一行。
    metric : str
        已计算指标列，不能将 val 指标重命名为 TEST 指标。
    direction : Literal["maximize", "minimize"]
        本次候选选择的指标方向。
    candidate_count : int
        候选数，默认示例配置为 10。
    slice_name : str
        本次选择使用的真实评估切片。

    Returns
    -------
    tuple[list[int], pd.DataFrame]
        确定顺序的编号和包括选择口径的审计表。

    Raises
    ------
    ValueError
        配置、列、重复候选或评估状态不满足要求时抛出。
    """
    if (
        direction not in {"maximize", "minimize"}
        or type(candidate_count) is not int
        or candidate_count < 1
    ):
        raise ValueError("provide direction and positive candidate_count")
    required = {"trial_num", "slice", "eligible", metric}
    if not required.issubset(evaluation.columns):
        raise ValueError("first obtain actual candidate evaluation on the requested slice")
    scope = evaluation.loc[evaluation["slice"] == slice_name].copy()
    if scope.empty or scope["trial_num"].duplicated().any():
        raise ValueError("require one evaluated row per trial in the requested slice")
    if not scope["eligible"].isin([True, False]).all():
        raise ValueError("eligibility must be explicitly recorded")
    eligible = scope.loc[scope["eligible"] & scope[metric].notna()].copy()
    eligible = eligible.loc[eligible[metric].map(math.isfinite)]
    ranked = eligible.sort_values(
        [metric, "trial_num"],
        ascending=[direction == "minimize", True],
        kind="stable",
    ).head(candidate_count)
    ranked["selection_slice"] = slice_name
    ranked["selection_metric"] = metric
    ranked["selection_direction"] = direction
    ranked["tie_policy"] = "trial_num ascending"
    return ranked["trial_num"].astype(int).tolist(), ranked


# 已取得候选在 TEST 上的评估；eligible 由外部预先声明的资格规则计算。
evaluation = pd.DataFrame(
    {
        "trial_num": [1, 2, 3, 4],
        "slice": ["TEST"] * 4,
        "ks": [35.0, 40.0, 40.0, 50.0],
        "eligible": [True, True, True, False],
    }
)
trial_nums, selection_record = select_candidates(
    evaluation,
    metric="ks",
    direction="maximize",
    candidate_count=10,
)
assert trial_nums == [2, 3, 1]
print(selection_record.to_string(index=False))

# 实际工作中，已有 tuning_result 和 df 时按上面的顺序调用：
# replay = runner.replay(tuning_result, df, trial_nums=trial_nums,
#                        metric_directions={"ks": "maximize"}, retrain=False)
# retrain=False 要求 tune 已保留这些模型；否则使用 retrain=True。
# 外部代码再读取 replay.reports 的真实 OOT、客群及成本证据，记录最终选择及理由。

已有 runner、tuning_result 和 df 时调用现有接口：

replay = runner.replay(
    tuning_result, df, trial_nums=trial_nums,
    metric_directions={"ks": "maximize"}, retrain=False,
)

retrain=False 仅适用于 tune 已保留这些模型的情况，否则使用 retrain=True。 最终选择由外部业务代码或 Agent 完成。调用方应预先指定实际 TEST／OOT 切片和指标方向， 并确定如何考虑稳定性、客群表现、复杂度和成本。资格阈值由业务提供，本例只读取显式的 eligible 标记。 读取 replay.reports 的真实评估表后保存最终记录，例如：

final_record = {
    "candidate_trials": trial_nums,
    "selection_policy": {"candidate_slice": "TEST", "metric": "ks",
                         "direction": "maximize", "candidate_count": 10,
                         "tie_policy": "trial_num ascending"},
    "final_trial": chosen_trial,  # 外部业务代码根据已计算证据决定
    "evidence": evidence_references,  # 实际表、样本切片、指标及约束检查记录
    "reason": selection_reason,
}

chosen_trial、证据和理由由外部流程产生，上段是记录结构示意，不能替代决策。 用户可按业务工程需要使用 OOT；MARS 不强制 OOT 权限政策，也不自动依据 OOT 无限扩大搜索。

常见失败

• 缺少 ml,tuning extra：按本页安装命令补齐模型和 Optuna 依赖。
• 时间切分没有足够的 train/val/oot 样本：扩大确定性示例或调整切分比例。
• Pipeline 提供 split_ratios 却没有 time_col：两者必须同时配置。
• 升级后读取旧 artifact 失败：按稳定性与兼容性固定版本并检查 Release Notes。

下一步

• 查看完整端到端流程：LightGBM 建模与监控示例。
• 对模型分和入模特征做周期监控：特征与模型监控。
• 查询全部结果对象：Modeling / Pipeline API。