مهندسی اعتبارسنجی سریهای زمانی مالی و جلوگیری از نشت اطلاعات (Data Leakage)

from typing import Iterator, Tuple
import numpy as np
import pandas as pd
from sklearn.model_selection import KFold

class PurgedEmbargoedKFold(KFold):
    """
    Purged & Embargoed K-Fold Cross-Validator for financial time-series
    where labels span an evaluation horizon [t0, t1].
    """
    def __init__(self, n_splits: int = 5, t1: pd.Series = None, pct_embargo: float = 0.01):
        super().__init__(n_splits=n_splits, shuffle=False, random_state=None)
        if not isinstance(t1, pd.Series):
            raise ValueError('t1 must be a pandas Series mapping index (t0) to label horizon end (t1).')
        self.t1 = t1
        self.pct_embargo = pct_embargo

    def split(self, X: pd.DataFrame, y: pd.Series = None, groups=None) -> Iterator[Tuple[np.ndarray, np.ndarray]]:
        if not X.index.equals(self.t1.index):
            raise ValueError('X index and t1 index must match exactly.')

        indices = np.arange(X.shape[0])
        embargo_step = int(X.shape[0] * self.pct_embargo)
        test_ranges = [(i[0], i[-1] + 1) for i in np.array_split(indices, self.n_splits)]

        for st, end in test_ranges:
            test_indices = indices[st:end]
            t0_test_start = X.index[st]
            t1_test_max = self.t1.iloc[test_indices].max()

            # 1. Left-side Purge: drop train samples whose evaluation horizon t1 overlaps test start
            left_train_mask = self.t1.iloc[:st] < t0_test_start
            left_train_idx = indices[:st][left_train_mask.values]

            # 2. Right-side Purge + Embargo: drop train samples starting before max(t1_test) + embargo
            embargo_bound_idx = min(X.shape[0] - 1, end + embargo_step)
            t0_embargo_cutoff = max(t1_test_max, X.index[embargo_bound_idx] if end < X.shape[0] else t1_test_max)
            right_train_mask = X.index[end:] > t0_embargo_cutoff
            right_train_idx = indices[end:][right_train_mask]

            train_indices = np.concatenate([left_train_idx, right_train_idx])
            yield train_indices, test_indices

def frac_diff_ffd(series: pd.Series, d: float = 0.4, thres: float = 1e-4) -> pd.Series:
    """
    Fixed-Width Window Fractional Differentiation (FFD).
    Achieves stationarity while preserving long-term memory without look-ahead bias.
    """
    weights = [1.0]
    k = 1
    while True:
        w_next = -weights[-1] * (d - k + 1) / k
        if abs(w_next) < thres:
            break
        weights.append(w_next)
        k += 1

    weights_arr = np.array(weights[::-1], dtype=np.float64)
    width = len(weights_arr)
    vals = series.values
    out = np.full(len(vals), np.nan, dtype=np.float64)

    for i in range(width - 1, len(vals)):
        window = vals[i - width + 1 : i + 1]
        if not np.isnan(window).any():
            out[i] = np.dot(weights_arr, window)

    return pd.Series(out, index=series.index, name=f'{series.name}_ffd_{d}')