high_grade

lib.prospecting.high_grade — 特高品位处理算法（纯 Python/pandas 实现）

不依赖库内封装的 C++ extra_high_grade_process 接口。

算法流程： 1. 数据清洗（剔除 NaN、零/负样长） 2. 按用户指定维度分组（钻孔 / 勘探线 / 全局） 3. 双策略计算特高品位阈值（平均值倍数法 + 累积频率法） 4. 标记超阈值样品 5. 按六种替换方法之一处理标记样品 6. 返回处理结果 + 完整审计报告

Usage

from dimine_python_sdk.lib.prospecting import HighGradeParams, HighGradeProcessor

params = HighGradeParams(grade_field="TFe") processor = HighGradeProcessor(params) result = processor.process(db.sample, collar_df=db.collar)

或直接通过 DrillDBManager 调用：

result = db.process_high_grade_samples(params)

HighGradeParams dataclass

特高品位处理参数

Attributes:

Name	Type	Description
grade_field	str	品位字段名（如 "CU"、"TFe"）
group_mode	GroupMode	分组维度 - "by_hole": 按工程号分组 - "by_section": 按勘探线分组（需传入 collar_df） - "global": 全矿区统一计算（默认）
use_average_multiple	bool	启用平均值倍数法
average_multiple	Optional[int]	倍数系数。设为 None（默认）时按品位变化系数(CV)自动判定： CV<0.5→6倍, 0.5≤CV<1.0→7倍, CV≥1.0→8倍。 也可手动指定 6/7/8。
use_frequency	bool	启用累积频率法
frequency	float	累积频率阈值（常用 0.95 / 0.975 / 0.977）
replace_method	ReplaceMethod	替换方式 0 - 剔除法（设为 NaN） 1 - 给定值法 2 - 相邻样品平均值法（默认，长度加权） 3 - 矿体平均品位法 4 - 单一工程法 5 - 截止品位法（cap 到阈值）
assign_value	float	方法 1 的给定值
max_adjacent_search	int	方法 2 相邻搜索最大扩展步数（默认 3）
max_depth_gap	float	方法 2 深度连续性容差（m）。候选样品与当前样品的 FROM/TO 间隙超过此值则视为不连续，跳过该候选。 默认 0.5m；设为 0 表示要求严格连续。
average_method	AverageMethod	均值计算方式 - "arithmetic": 算术平均 - "length_weighted": 长度加权平均（默认）
result_field	Optional[str]	结果字段名（None 时默认为 {grade_field}_processed）

Source code in dimine_python_sdk\lib\prospecting\high_grade.py

@dataclass
class HighGradeParams:
    """特高品位处理参数

    Attributes:
        grade_field: 品位字段名（如 "CU"、"TFe"）
        group_mode: 分组维度
            - "by_hole":    按工程号分组
            - "by_section": 按勘探线分组（需传入 collar_df）
            - "global":     全矿区统一计算（默认）
        use_average_multiple: 启用平均值倍数法
        average_multiple: 倍数系数。设为 None（默认）时按品位变化系数(CV)自动判定：
                          CV<0.5→6倍, 0.5≤CV<1.0→7倍, CV≥1.0→8倍。
                          也可手动指定 6/7/8。
        use_frequency: 启用累积频率法
        frequency: 累积频率阈值（常用 0.95 / 0.975 / 0.977）
        replace_method: 替换方式
            0 - 剔除法（设为 NaN）
            1 - 给定值法
            2 - 相邻样品平均值法（默认，长度加权）
            3 - 矿体平均品位法
            4 - 单一工程法
            5 - 截止品位法（cap 到阈值）
        assign_value: 方法 1 的给定值
        max_adjacent_search: 方法 2 相邻搜索最大扩展步数（默认 3）
        max_depth_gap: 方法 2 深度连续性容差（m）。候选样品与当前样品的
                       FROM/TO 间隙超过此值则视为不连续，跳过该候选。
                       默认 0.5m；设为 0 表示要求严格连续。
        average_method: 均值计算方式
            - "arithmetic":      算术平均
            - "length_weighted": 长度加权平均（默认）
        result_field: 结果字段名（None 时默认为 {grade_field}_processed）
    """

    grade_field: str
    group_mode: GroupMode = "global"

    # 阈值策略
    use_average_multiple: bool = True
    average_multiple: Optional[int] = None  # None=按CV自动判定6/7/8
    use_frequency: bool = False
    frequency: float = 0.95

    # 替换策略
    replace_method: ReplaceMethod = 2
    assign_value: float = 0.0
    max_adjacent_search: int = 3
    max_depth_gap: float = 0.5

    # 计算方式
    average_method: AverageMethod = "length_weighted"

    # 输出
    result_field: Optional[str] = None

    def __post_init__(self):
        if self.result_field is None:
            self.result_field = f"{self.grade_field}_processed"

HighGradeProcessor

特高品位处理器

对样品表 DataFrame 执行特高品位识别与替换，支持： - 平均值倍数法 / 累积频率法双阈值策略 - 六种替换方法（剔除、给定值、相邻平均、矿体平均、单一工程、截止品位） - 单次处理，不做迭代收敛

Source code in dimine_python_sdk\lib\prospecting\high_grade.py

class HighGradeProcessor:
    """特高品位处理器

    对样品表 DataFrame 执行特高品位识别与替换，支持：
    - 平均值倍数法 / 累积频率法双阈值策略
    - 六种替换方法（剔除、给定值、相邻平均、矿体平均、单一工程、截止品位）
    - 单次处理，不做迭代收敛
    """

    def __init__(self, params: HighGradeParams):
        self.params = params
        # 运行时状态
        self.thresholds_: Optional[pd.Series] = None   # 各分组阈值
        self.group_labels_: Optional[pd.Series] = None  # 每个样品所属分组标签

    # ------------------------------------------------------------------
    # 公开入口
    # ------------------------------------------------------------------

    def process(
        self,
        sample_df: pd.DataFrame,
        collar_df: Optional[pd.DataFrame] = None,
    ) -> Dict[str, Any]:
        """执行特高品位处理（单次处理，不迭代）

        Args:
            sample_df: 样品表 DataFrame，需包含 [工程号, 样本编号, 起始, 结束] 及品位列。
            collar_df: 孔口表 DataFrame（group_mode="by_section" 时必需）。

        Returns:
            (processed_df, summary) 元组:
                processed_df — 处理后的完整样品 DataFrame（新增 result_field 列）
                summary     — 汇总统计 dict
        """
        self._reset_state()

        df = sample_df.copy()
        df = self._clean_data(df)
        df["_sample_length"] = self._compute_length(df)
        df["_original_grade"] = df[self.params.grade_field].copy()

        # 分组标签
        self.group_labels_ = self._build_group_labels(df, collar_df)

        # 初始化结果列
        result_field = self.params.result_field
        df[result_field] = df[self.params.grade_field].copy()

        # --- 1. 计算阈值 ---
        thresholds = self._compute_thresholds(df, self.group_labels_)
        self.thresholds_ = thresholds

        # --- 2. 标记超阈值样品 ---
        sample_thresholds = self.group_labels_.map(thresholds)
        flagged_mask = df[result_field] > sample_thresholds

        if flagged_mask.any():
            # --- 3. 替换 ---
            replacements = self._compute_replacements(df, flagged_mask, thresholds)
            df.loc[flagged_mask, result_field] = replacements

        # --- 4. 构建输出 ---
        return self._build_output(df, result_field, flagged_mask)

    # ------------------------------------------------------------------
    # [1] 数据清洗
    # ------------------------------------------------------------------

    def _clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """剔除无效样品"""
        required = ["工程号", "样本编号", "起始", "结束"]
        for col in required:
            if col not in df.columns:
                raise ValueError(f"样品表缺少必要列: {col}")

        grade_col = self.params.grade_field
        if grade_col not in df.columns:
            raise ValueError(f"样品表缺少品位列: {grade_col}")

        # 剔除品位为 NaN 的行
        before = len(df)
        df = df.dropna(subset=[grade_col]).copy()
        if len(df) < before:
            print(f"[清洗] 剔除 {before - len(df)} 条品位为 NaN 的样品")

        # 剔除样长 ≤ 0 的样品
        length = df["结束"] - df["起始"]
        invalid_len = length <= 0
        if invalid_len.any():
            print(f"[清洗] 剔除 {invalid_len.sum()} 条样长 ≤ 0 的样品")
            df = df[~invalid_len].copy()

        # 确保工程号为字符串
        df["工程号"] = df["工程号"].astype(str)

        return df.reset_index(drop=True)

    @staticmethod
    def _compute_length(df: pd.DataFrame) -> pd.Series:
        """计算样长"""
        return df["结束"] - df["起始"]

    # ------------------------------------------------------------------
    # [2] 分组
    # ------------------------------------------------------------------

    def _build_group_labels(
        self, df: pd.DataFrame, collar_df: Optional[pd.DataFrame]
    ) -> pd.Series:
        """为每个样品生成分组标签"""
        mode = self.params.group_mode

        if mode == "by_hole":
            return df["工程号"]

        elif mode == "by_section":
            if collar_df is not None:
                section_map = collar_df.set_index("工程号")["勘探线"]
            else:
                raise ValueError("group_mode='by_section' 需要传入 collar_df")
            return df["工程号"].map(section_map).fillna("未分组")

        elif mode == "global":
            return pd.Series("全矿区", index=df.index)

        else:
            raise ValueError(f"未知分组模式: {mode}")

    # ------------------------------------------------------------------
    # [3] 阈值计算
    # ------------------------------------------------------------------

    def _compute_thresholds(self, df: pd.DataFrame, labels: pd.Series) -> pd.Series:
        """计算各分组的特高品位阈值，取双策略中较严格者"""
        source = df["_original_grade"]

        thresholds = None

        if self.params.use_average_multiple:
            t_avg = self._average_multiple_threshold(df, source, labels)
            thresholds = t_avg

        if self.params.use_frequency:
            t_freq = self._frequency_threshold(source, labels)
            if thresholds is None:
                thresholds = t_freq
            else:
                thresholds = pd.concat([thresholds, t_freq], axis=1).min(axis=1)

        if thresholds is None:
            raise ValueError("至少需要启用一种阈值策略")

        return thresholds

    # ------------------------------------------------------------------
    # CV 自动判定倍数
    # ------------------------------------------------------------------

    @staticmethod
    def _detect_multiple(values: pd.Series) -> int:
        """按品位变化系数(CV)自动选择倍数

        规范规定：
          - 品位变化均匀 → 6 倍（CV < 0.5）
          - 品位变化较均匀 → 7 倍（0.5 ≤ CV < 1.0）
          - 品位变化不均匀 → 8 倍（CV ≥ 1.0）
        """
        if len(values) < 2:
            return 6
        mean = values.mean()
        if mean <= 0:
            return 6
        cv = values.std() / mean
        if cv < 0.5:
            return 6
        elif cv < 1.0:
            return 7
        else:
            return 8

    # ------------------------------------------------------------------
    # 阈值计算方法
    # ------------------------------------------------------------------

    def _average_multiple_threshold(
        self,
        df: pd.DataFrame,
        source_values: pd.Series,
        labels: pd.Series,
    ) -> pd.Series:
        """平均值倍数法：阈值 = 分组品位均值 × 倍数（按CV自动判定或手动指定）"""
        if self.params.average_method == "length_weighted":
            lengths = df["_sample_length"]
            grouped = pd.DataFrame({
                "label": labels,
                "value": source_values,
                "length": lengths,
            })
            means = grouped.groupby("label").apply(
                lambda g: np.average(g["value"], weights=g["length"])
                if g["length"].sum() > 0 else np.nan,
                include_groups=False,
            )
        else:
            means = source_values.groupby(labels).mean()

        # 确定各分组的倍数
        if self.params.average_multiple is not None:
            multiples = pd.Series(self.params.average_multiple, index=means.index)
        else:
            multiples = source_values.groupby(labels).apply(
                self._detect_multiple
            )

        result = (means * multiples).dropna()

        # 记录各分组使用的倍数供输出参考
        self._detected_multiples_ = multiples.to_dict()

        return result

    def _frequency_threshold(
        self,
        source_values: pd.Series,
        labels: pd.Series,
    ) -> pd.Series:
        """累积频率法：阈值 = 第 N 百分位数"""
        q = self.params.frequency
        thresholds = source_values.groupby(labels).quantile(q)
        return thresholds.dropna()

    # ------------------------------------------------------------------
    # [4] 替换值计算
    # ------------------------------------------------------------------

    def _compute_replacements(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
        thresholds: pd.Series,
    ) -> pd.Series:
        """对标记样品计算替换值"""
        method = self.params.replace_method
        flagged_indices = df.index[flagged_mask]

        if method == 0:
            return self._replace_by_discard(flagged_indices)
        elif method == 1:
            return self._replace_by_fixed(flagged_indices)
        elif method == 2:
            return self._replace_by_adjacent(df, flagged_mask)
        elif method == 3:
            return self._replace_by_orebody_average(df, flagged_mask)
        elif method == 4:
            return self._replace_by_hole_average(df, flagged_mask)
        elif method == 5:
            return self._replace_by_cap(df, flagged_mask, thresholds)
        else:
            raise ValueError(f"未知替换方法: {method}")

    def _replace_by_discard(self, indices: pd.Index) -> pd.Series:
        """方法 0：剔除（设为 NaN）"""
        return pd.Series(np.nan, index=indices)

    def _replace_by_fixed(self, indices: pd.Index) -> pd.Series:
        """方法 1：给定固定值"""
        return pd.Series(self.params.assign_value, index=indices)

    def _replace_by_adjacent(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
    ) -> pd.Series:
        """方法 2：相邻未标记样品的长度加权平均

        在同一钻孔内向上/向下搜索未标记样品（跳过已标记），
        最多扩展 max_adjacent_search 步。

        深度连续性校验：候选样品与当前样品的 FROM/TO 间隙超过
        max_depth_gap 时视为不连续，跳过该候选继续向外搜索。
        """
        results = {}
        max_step = self.params.max_adjacent_search
        max_gap = self.params.max_depth_gap

        for hole, hole_df in df.groupby("工程号"):
            hole_indices = hole_df.index

            for idx in hole_indices[flagged_mask.loc[hole_indices]]:
                pos = hole_df.index.get_loc(idx)
                current_from = df.at[idx, "起始"]
                current_to = df.at[idx, "结束"]

                # 向上搜索
                above_values = []
                above_lengths = []
                step = 1
                while step <= max_step and (pos - step) >= 0:
                    candidate_idx = hole_indices[pos - step]
                    candidate = df.loc[candidate_idx]

                    gap = abs(candidate["结束"] - current_from)
                    if gap > max_gap:
                        step += 1
                        continue

                    if not flagged_mask.get(candidate_idx, False):
                        above_values.append(candidate["_original_grade"])
                        above_lengths.append(candidate["_sample_length"])
                        break
                    step += 1

                # 向下搜索
                below_values = []
                below_lengths = []
                step = 1
                while step <= max_step and (pos + step) < len(hole_indices):
                    candidate_idx = hole_indices[pos + step]
                    candidate = df.loc[candidate_idx]

                    gap = abs(candidate["起始"] - current_to)
                    if gap > max_gap:
                        step += 1
                        continue

                    if not flagged_mask.get(candidate_idx, False):
                        below_values.append(candidate["_original_grade"])
                        below_lengths.append(candidate["_sample_length"])
                        break
                    step += 1

                all_vals = above_values + below_values
                all_lens = above_lengths + below_lengths

                if all_vals:
                    if self.params.average_method == "length_weighted" and sum(all_lens) > 0:
                        replacement = np.average(all_vals, weights=all_lens)
                    else:
                        replacement = np.mean(all_vals)
                else:
                    # 回退：该孔内所有未标记样品的均值
                    hole_unflagged_mask = ~flagged_mask.loc[hole_indices]
                    hole_unflagged = hole_indices[hole_unflagged_mask]
                    if len(hole_unflagged) > 0:
                        if self.params.average_method == "length_weighted":
                            vals = df.loc[hole_unflagged, "_original_grade"]
                            lens = df.loc[hole_unflagged, "_sample_length"]
                            replacement = np.average(vals, weights=lens) if lens.sum() > 0 else np.nan
                        else:
                            replacement = df.loc[hole_unflagged, "_original_grade"].mean()
                    else:
                        replacement = np.nan

                results[idx] = replacement

        return pd.Series(results)

    def _replace_by_orebody_average(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
    ) -> pd.Series:
        """方法 3：矿体内未标记样品的长度加权平均"""
        unflagged_mask = ~flagged_mask
        labels = self.group_labels_

        results = {}
        for idx in df.index[flagged_mask]:
            group = labels.at[idx]
            group_mask = (labels == group) & unflagged_mask
            group_df = df.loc[group_mask]
            if len(group_df) == 0:
                results[idx] = np.nan
                continue
            if self.params.average_method == "length_weighted":
                results[idx] = np.average(
                    group_df["_original_grade"],
                    weights=group_df["_sample_length"]
                )
            else:
                results[idx] = group_df["_original_grade"].mean()

        return pd.Series(results)

    def _replace_by_hole_average(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
    ) -> pd.Series:
        """方法 4：同一钻孔内未标记样品的长度加权平均"""
        unflag = ~flagged_mask

        results = {}
        for idx in df.index[flagged_mask]:
            hole = df.at[idx, "工程号"]
            hole_mask = (df["工程号"] == hole) & unflag
            hole_df = df.loc[hole_mask]
            if len(hole_df) == 0:
                results[idx] = np.nan
                continue
            if self.params.average_method == "length_weighted":
                results[idx] = np.average(
                    hole_df["_original_grade"],
                    weights=hole_df["_sample_length"]
                )
            else:
                results[idx] = hole_df["_original_grade"].mean()

        return pd.Series(results)

    def _replace_by_cap(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
        thresholds: pd.Series,
    ) -> pd.Series:
        """方法 5：截止品位法（cap 到分组阈值）"""
        results = {}
        group_thresholds = self.group_labels_.map(thresholds)
        for idx in df.index[flagged_mask]:
            results[idx] = group_thresholds.at[idx]
        return pd.Series(results)

    # ------------------------------------------------------------------
    # [5] 输出构建
    # ------------------------------------------------------------------

    def _build_output(
        self,
        df: pd.DataFrame,
        result_field: str,
        flagged_mask: pd.Series,
    ) -> Dict[str, Any]:
        """构建最终返回字典"""
        out_df = df.drop(columns=["_sample_length", "_original_grade"], errors="ignore")

        # 标记明细
        flagged_indices = df.index[flagged_mask]
        flagged_samples = []
        for idx in flagged_indices:
            flagged_samples.append({
                "工程号": df.at[idx, "工程号"],
                "样本编号": df.at[idx, "样本编号"],
                "起始": df.at[idx, "起始"],
                "结束": df.at[idx, "结束"],
                "原始品位": df.at[idx, "_original_grade"],
                "处理后品位": df.at[idx, result_field],
                "所属分组": str(self.group_labels_.at[idx]) if self.group_labels_ is not None else None,
            })

        # 分组阈值
        final_thresholds = {}
        if self.thresholds_ is not None:
            for label, thr in self.thresholds_.items():
                final_thresholds[str(label)] = float(thr)

        # CV 自动判定的倍数
        detected_multiples = getattr(self, '_detected_multiples_', None) or {}

        # 汇总
        total = len(df)
        flagged_count = int(flagged_mask.sum())
        original_mean = float(df["_original_grade"].mean())
        processed_mean = float(df[result_field].mean())

        summary = {
            "总样品数": total,
            "特高样品数": flagged_count,
            "特高占比": f"{flagged_count / total * 100:.2f}%" if total > 0 else "0%",
            "原始品位均值": round(original_mean, 4),
            "处理后品位均值": round(processed_mean, 4),
            "均值降幅": f"{(original_mean - processed_mean) / original_mean * 100:.2f}%"
            if original_mean > 0 else "N/A",
            "分组模式": self.params.group_mode,
            "替换方法": _REPLACE_METHOD_NAMES.get(self.params.replace_method, str(self.params.replace_method)),
        }

        return out_df, summary

    def _reset_state(self):
        """重置运行时状态"""
        self.thresholds_ = None
        self.group_labels_ = None

process(sample_df, collar_df=None)

执行特高品位处理（单次处理，不迭代）

Parameters:

Name	Type	Description	Default
sample_df	DataFrame	样品表 DataFrame，需包含 [工程号, 样本编号, 起始, 结束] 及品位列。	required
collar_df	Optional[DataFrame]	孔口表 DataFrame（group_mode="by_section" 时必需）。	None

Returns:

Type	Description
Dict[str, Any]	(processed_df, summary) 元组: processed_df — 处理后的完整样品 DataFrame（新增 result_field 列） summary — 汇总统计 dict

Source code in dimine_python_sdk\lib\prospecting\high_grade.py

def process(
    self,
    sample_df: pd.DataFrame,
    collar_df: Optional[pd.DataFrame] = None,
) -> Dict[str, Any]:
    """执行特高品位处理（单次处理，不迭代）

    Args:
        sample_df: 样品表 DataFrame，需包含 [工程号, 样本编号, 起始, 结束] 及品位列。
        collar_df: 孔口表 DataFrame（group_mode="by_section" 时必需）。

    Returns:
        (processed_df, summary) 元组:
            processed_df — 处理后的完整样品 DataFrame（新增 result_field 列）
            summary     — 汇总统计 dict
    """
    self._reset_state()

    df = sample_df.copy()
    df = self._clean_data(df)
    df["_sample_length"] = self._compute_length(df)
    df["_original_grade"] = df[self.params.grade_field].copy()

    # 分组标签
    self.group_labels_ = self._build_group_labels(df, collar_df)

    # 初始化结果列
    result_field = self.params.result_field
    df[result_field] = df[self.params.grade_field].copy()

    # --- 1. 计算阈值 ---
    thresholds = self._compute_thresholds(df, self.group_labels_)
    self.thresholds_ = thresholds

    # --- 2. 标记超阈值样品 ---
    sample_thresholds = self.group_labels_.map(thresholds)
    flagged_mask = df[result_field] > sample_thresholds

    if flagged_mask.any():
        # --- 3. 替换 ---
        replacements = self._compute_replacements(df, flagged_mask, thresholds)
        df.loc[flagged_mask, result_field] = replacements

    # --- 4. 构建输出 ---
    return self._build_output(df, result_field, flagged_mask)

process_high_grade(sample_df, grade_field, group_mode='global', average_multiple=None, frequency=None, replace_method=2, collar_df=None, **kwargs)

特高品位处理便捷函数

对样品表执行特高品位识别与替换，返回处理结果与审计报告。

Parameters:

Name	Type	Description	Default
sample_df	DataFrame	样品表 DataFrame	required
grade_field	str	品位字段名	required
group_mode	GroupMode	分组模式 ("by_hole" / "by_section" / "global")	'global'
average_multiple	Optional[int]	倍数（None=按CV自动判定6/7/8）	None
frequency	Optional[float]	累积频率阈值（None 则不启用频率法）	None
replace_method	ReplaceMethod	替换方法 (0-5)	2
collar_df	Optional[DataFrame]	孔口表（group_mode="by_section" 时需要）	None
**kwargs		其他 HighGradeParams 参数	{}

Returns:

Name	Type	Description
dict	Dict[str, Any]	同 HighGradeProcessor.process()

Source code in dimine_python_sdk\lib\prospecting\high_grade.py

def process_high_grade(
    sample_df: pd.DataFrame,
    grade_field: str,
    group_mode: GroupMode = "global",
    average_multiple: Optional[int] = None,
    frequency: Optional[float] = None,
    replace_method: ReplaceMethod = 2,
    collar_df: Optional[pd.DataFrame] = None,
    **kwargs,
) -> Dict[str, Any]:
    """特高品位处理便捷函数

    对样品表执行特高品位识别与替换，返回处理结果与审计报告。

    Args:
        sample_df: 样品表 DataFrame
        grade_field: 品位字段名
        group_mode: 分组模式 ("by_hole" / "by_section" / "global")
        average_multiple: 倍数（None=按CV自动判定6/7/8）
        frequency: 累积频率阈值（None 则不启用频率法）
        replace_method: 替换方法 (0-5)
        collar_df: 孔口表（group_mode="by_section" 时需要）
        **kwargs: 其他 HighGradeParams 参数

    Returns:
        dict: 同 HighGradeProcessor.process()
    """
    params = HighGradeParams(
        grade_field=grade_field,
        group_mode=group_mode,
        average_multiple=average_multiple,
        use_frequency=frequency is not None,
        frequency=frequency or 0.95,
        replace_method=replace_method,
        **kwargs,
    )
    processor = HighGradeProcessor(params)
    return processor.process(sample_df, collar_df=collar_df)