滞后与差分

面板数据的滞后与差分变量生成（含时序结构校验）。

核心代码

def generate_lag_variables(
    *,
    df: pd.DataFrame,
    params: dict[str, Any],
    column_descriptions: dict[str, str],
) -> tuple[pd.DataFrame, dict[str, Any], list[str], dict[str, str]]:
    """按面板结构生成滞后项。"""
    columns = require_columns(df, normalize_string_list(params.get('columns')), label=_('滞后变量'))
    entity_columns = normalize_string_list(params.get('entity_columns'))
    if entity_columns:
        entity_columns = require_columns(df, entity_columns, label=_('个体变量'))
    time_column = require_columns(df, [params.get('time_column')], label=_('时间变量'))[0]
    _validate_temporal_structure(df=df, entity_columns=entity_columns, time_column=time_column, operation_name=_('滞后项'))
    periods = parse_positive_int(params.get('periods'), _('滞后期数'), default=1)
    prefix = str(params.get('prefix') or '').strip()
    suffix = str(params.get('suffix') or f'_lag{periods}').strip()
    mark_missing_periods = normalize_bool(params.get('mark_missing_periods'), default=True)
    result_df = df.copy()
    sort_columns = entity_columns + [time_column] if entity_columns else [time_column]
    result_df = result_df.sort_values(sort_columns).copy()
    descriptions = dict(column_descriptions)
    created_columns = []
    missing_period_report: dict[str, dict[str, Any]] = {}
    leading_mask = _build_leading_period_mask(result_df=result_df, entity_columns=entity_columns, periods=periods)

    for column in columns:
        new_name = ensure_new_column_absent(result_df, f'{prefix}{column}{suffix}')
        if entity_columns:
            grouped = result_df.groupby(entity_columns, dropna=False, sort=False)
            result_df[new_name] = grouped[column].shift(periods)
        else:
            result_df[new_name] = result_df[column].shift(periods)
        descriptions[new_name] = _('%(column)s 的 %(periods)s 期滞后项') % {'column': column, 'periods': periods}
        created_columns.append(new_name)
        flag_column = None
        if mark_missing_periods:
            flag_column = ensure_new_column_absent(result_df, f'{new_name}_initial_missing')
            result_df[flag_column] = leading_mask.astype(int)
            descriptions[flag_column] = _('%(new_name)s 因前 %(periods)s 期缺失形成的标记') % {'new_name': new_name, 'periods': periods}
        missing_period_report[new_name] = {
            'lost_rows': int(leading_mask.sum()),
            'lost_ratio': round(float(leading_mask.mean()) if len(leading_mask) else 0.0, 4),
            'flag_column': flag_column,
        }

    details = {
        'created_columns': created_columns,
        'entity_columns': entity_columns,
        'time_column': time_column,
        'periods': periods,
        'mark_missing_periods': mark_missing_periods,
        'missing_period_report': missing_period_report,
    }
    warnings = [_('滞后项会在每个个体的前 %(periods)s 期产生缺失值') % {'periods': periods}]
    return result_df.reset_index(drop=True), details, warnings, descriptions