hdc_create_data_v2.py

# create_data_generator
import os
import numpy as np
import polars as pl

EPS = 1e-15
FAST_WINDOW = 5
MEDIUM_WINDOW = 20
SLOW_WINDOW = 100

# Flags
FLAG_COMBINE_FEATURES = True

# Path Constants
TIMESCALE = "1d"
RAW_DIR = f"datasets/raw/{TIMESCALE}"
OUTPUT_DIR = f"datasets/refined/{TIMESCALE}"

def load_dataset(dataset_path: str) -> pl.DataFrame:
    df = pl.read_csv(dataset_path, try_parse_dates=True).sort("timestamp")
    # Drop unused columns
    for col in ["close_time", "ignore"]:
        if col in df.columns:
            df = df.drop(col)
    return df

'''
TARGET CREATION (updated for generative modeling)
'''

def create_targets(df: pl.DataFrame, eps: float = EPS, k: float = 0.2) -> pl.DataFrame:
    """
    Adds generative modeling targets:
    - target_open, target_high, target_low, target_close:
        log(value / close_{t-1})
    - target_volume*, target_quote_asset_volume*, etc.:
        tanh(k * log(value / value_{t-1})) — soft-bounded per row
    """
    prev_close = pl.col("close").shift(1) + eps

    ohlc_fields = ["open", "high", "low", "close"]
    volume_fields = [
        "volume", "quote_asset_volume", "number_of_trades",
        "taker_buy_base_asset_volume", "taker_buy_quote_asset_volume"
    ]

    # OHLC: log(value / close_{t-1})
    targets_ohlc = [
        (((pl.col(field) + eps) / prev_close).log()
         .alias(f"target_{field}"))
        for field in ohlc_fields
    ]

    # Volume: tanh(k * log(value / value_{t-1}))
    targets_volume = [
        (((pl.col(field) + eps).log() - (pl.col(field).shift(1) + eps).log()) * k)
        .tanh()
        .alias(f"target_{field}")
        for field in volume_fields
    ]

    return df.with_columns(targets_ohlc + targets_volume)


'''
Feature Creation
'''

def log_return(col_name: str, epsilon: float = EPS) -> pl.Expr:
    """1–step log return (close → closeₜ₋₁)."""
    return (
        (pl.col(col_name) + epsilon).log()
        - (pl.col(col_name).shift(1) + epsilon).log()
    )

def log_distance(base_col: str, measure_col: str, epsilon: float = EPS) -> pl.Expr:
    """1–step log distance (measure / base)."""
    return (
        ((pl.col(measure_col) + epsilon) / (pl.col(base_col) + epsilon))
        .log()
    )

def create_log_returns(df: pl.DataFrame) -> pl.DataFrame:
    return df.with_columns([
        log_return("close").alias("log_returns_close"),
        log_distance("open",  "high").alias("log_dist_open_to_high"),
        log_distance("open",  "low").alias("log_dist_open_to_low"),
        log_distance("high", "close").alias("log_dist_high_to_close"),
        log_distance("low",  "close").alias("log_dist_low_to_close"),
    ])

def compress_log_returns(log_returns, k: float = 4.0):
    return np.tanh(k * log_returns)

def decompress_log_returns(compressed, k: float = 4.0):
    return np.arctanh(compressed) / k

def copy_original_values(df: pl.DataFrame) -> pl.DataFrame:
    return df.with_columns([
        pl.Series("original_close",  df["close"].to_list()),
        pl.Series("original_open",   df["open"].to_list()),
        pl.Series("original_low",    df["low"].to_list()),
        pl.Series("original_high",   df["high"].to_list()),
        pl.Series("original_volume", df["volume"].to_list()),
        pl.Series("original_quote_asset_volume", df["quote_asset_volume"].to_list()),
        pl.Series("original_number_of_trades", df["number_of_trades"].to_list()),
        pl.Series("original_taker_buy_base_asset_volume", df["taker_buy_base_asset_volume"].to_list()),
        pl.Series("original_taker_buy_quote_asset_volume", df["taker_buy_quote_asset_volume"].to_list()),
    ])


def log_return_base_inputs(df: pl.DataFrame, base_inputs_list: list) -> pl.DataFrame:
    """
    Overwrite each base input column with its 1-step log return.
    """
    exprs = [log_return(col).alias(col) for col in base_inputs_list]
    return df.with_columns(exprs)

def rolling_z_score_rescaled_expr(col_name: str, window: int, eps: float = EPS) -> pl.Expr:
    """
    Numerically stable rolling z-score rescaling.
    Rescales each rolling z-score window into [-1, 1].
    """
    c = pl.col(col_name)
    mean = c.rolling_mean(window_size=window, min_samples=1)
    std = c.rolling_std(window_size=window, min_samples=1)
    z = (c - mean) / (std + eps)

    # Use pl.when/then to avoid propagating NaNs during rescaling
    zmin = z.rolling_min(window_size=window, min_samples=1)
    zmax = z.rolling_max(window_size=window, min_samples=1)
    range_ = zmax - zmin

    # Final expression with strong NaN protection
    return (
        pl.when(range_ > eps)
        .then(((z - zmin) / range_ * 2) - 1)
        .otherwise(0.0)  # default to zero when no variation
        .alias(f"{col_name}_norm_window{window}")
    )

def rolling_z_base_inputs(df: pl.DataFrame, base_inputs_list: list) -> pl.DataFrame:
    """
    Fast/medium/slow rolling-z normalized versions for each base input.
    """
    exprs = []
    for col in base_inputs_list:
        exprs += [
            rolling_z_score_rescaled_expr(col, FAST_WINDOW).alias(f"{col}_fast_norm"),
            rolling_z_score_rescaled_expr(col, MEDIUM_WINDOW).alias(f"{col}_medium_norm"),
            rolling_z_score_rescaled_expr(col, SLOW_WINDOW).alias(f"{col}_slow_norm"),
        ]
    return df.with_columns(exprs).drop(base_inputs_list)

'''
Temporal Feature Creation
'''

def create_temporal_features(df: pl.DataFrame) -> pl.DataFrame:
    """
    Adds two purely timestamp-based features:
      - Hour_of_Day ∈ [-1,1] (0 for daily candles)
      - Day_of_Week ∈ [-1,1]
    """
    # Ensure timestamp is datetime
    df = df.with_columns([
        pl.col("timestamp").cast(pl.Datetime).alias("timestamp")
    ])

    return df.with_columns([
        # Hour (0 for daily timestamps at midnight)
        (pl.col("timestamp").dt.hour().cast(pl.Float64) / 23.0 * 2 - 1).fill_null(0.0).alias("Hour_of_Day"),
        # Day of week
        ((pl.col("timestamp").dt.weekday().cast(pl.Float64) / 6.0 * 2) - 1).alias("Day_of_Week")
    ])

'''
Technical Analysis Feature Creation
'''

def calculate_macd_polars(
    df: pl.DataFrame,
    short_window: int = 12,
    long_window:  int = 26,
    signal_window: int = 9
) -> pl.DataFrame:
    alpha_s  = 2.0 / (short_window + 1)
    alpha_l  = 2.0 / (long_window  + 1)
    alpha_sig= 2.0 / (signal_window + 1)

    df = df.with_columns([
        pl.col("original_close").ewm_mean(alpha=alpha_s,  adjust=False).alias("ema_short"),
        pl.col("original_close").ewm_mean(alpha=alpha_l,  adjust=False).alias("ema_long"),
    ])

    macd_line = (pl.col("ema_short") - pl.col("ema_long")).alias("macd_line")
    df = df.with_columns([
        macd_line,
        macd_line.ewm_mean(alpha=alpha_sig, adjust=False).alias("macd_signal"),
    ])

    hist = (pl.col("macd_line") - pl.col("macd_signal")).alias("macd_hist")
    norm_w = signal_window * 3
    df = df.with_columns([
        hist,
        hist.rolling_min(window_size=norm_w, min_samples=1).alias("_macd_min"),
        hist.rolling_max(window_size=norm_w, min_samples=1).alias("_macd_max"),
    ])

    df = df.with_columns([
        ((pl.col("macd_hist") - pl.col("_macd_min"))
         / (pl.col("_macd_max") - pl.col("_macd_min")) * 2 - 1)
        .alias("macd_norm")
    ])

    return df.drop([
        "ema_short","ema_long",
        "macd_line","macd_signal",
        "macd_hist","_macd_min","_macd_max",
    ])

def add_bollinger_band_norm_polars(
    df: pl.DataFrame,
    close_col: str = "original_close",
    window: int = 20
) -> pl.DataFrame:
    price  = pl.col(close_col)
    mid    = price.rolling_mean(window_size=window, min_samples=1)
    std    = price.rolling_std(window_size=window, min_samples=1)
    upper  = mid + 2 * std
    lower  = mid - 2 * std
    pos    = (price - lower) / (upper - lower)
    raw_bb = (pos * 2 - 1).alias("bb_norm_raw")

    df = df.with_columns([raw_bb])
    return df.with_columns([
        rolling_z_score_rescaled_expr("bb_norm_raw", 14).alias("bb_norm")
    ]).drop(["bb_norm_raw"])


def add_normed_price_based_indicators(df: pl.DataFrame, k: float = 4.0) -> pl.DataFrame:
    # 1. raw SMAs/EMAs
    df = df.with_columns([
        pl.col("original_close").rolling_mean(window_size=10, min_samples=1).alias("sma_10_raw"),
        pl.col("original_close").rolling_mean(window_size=50, min_samples=1).alias("sma_50_raw"),
        pl.col("original_close").ewm_mean(alpha=2/11, adjust=False).alias("ema_10_raw"),
        pl.col("original_close").ewm_mean(alpha=2/51, adjust=False).alias("ema_50_raw"),
    ])

    # 2. compressed log-returns (still useful if you want directional trend change!)
    def _compressed(col: str) -> pl.Expr:
        lr = (pl.col(col) + EPS).log() - (pl.col(col).shift(1) + EPS).log()
        return (lr * k).tanh()

    df = df.with_columns([
        _compressed("sma_10_raw").alias("SMA_10_compressed"),
        _compressed("sma_50_raw").alias("SMA_50_compressed"),
        _compressed("ema_10_raw").alias("EMA_10_compressed"),
        _compressed("ema_50_raw").alias("EMA_50_compressed"),
    ])

    # 3. roll-norm raw values into [-1, 1] final versions
    mappings = [
        ("sma_10_raw", "SMA_10_z"),
        ("sma_50_raw", "SMA_50_z"),
        ("ema_10_raw", "EMA_10_z"),
        ("ema_50_raw", "EMA_50_z"),
    ]
    exprs = [rolling_z_score_rescaled_expr(src, 14).alias(dst) for src, dst in mappings]

    # drop intermediate columns
    to_drop = [src for src, _ in mappings] + [
        "SMA_10_compressed", "SMA_50_compressed",
        "EMA_10_compressed", "EMA_50_compressed"
    ]

    return df.with_columns(exprs).drop(to_drop)

def add_rolling_normed_mixed_ta_indicators(df: pl.DataFrame) -> pl.DataFrame:
    # base cols
    H = pl.col("original_high")
    L = pl.col("original_low")
    C = pl.col("original_close")
    O = pl.col("original_open")
    V = pl.col("original_volume")
    P  = C.shift(1)
    eps = EPS

    # raw series
    momentum_10 = (C - C.shift(10)).alias("momentum_10_raw")
    roc_10      = (((C + eps)/(C.shift(10)+eps)) - 1).alias("ROC_10_raw")
    price_gap   = (O - P).alias("price_gap_raw")

    obv_raw = (
      pl.when(C > P).then(V)
        .when(C < P).then(-V)
        .otherwise(0)
        .cum_sum()
        .alias("OBV_raw")
    )

    tr1 = H - L
    tr2 = (H - P).abs()
    tr3 = (L - P).abs()
    true_range = pl.max_horizontal(tr1, tr2, tr3)

    diff = C - P
    gain = pl.when(diff > 0).then(diff).otherwise(0)
    loss = pl.when(diff < 0).then(-diff).otherwise(0)

    # RSI
    ag_sma = gain.rolling_mean(window_size=14, min_samples=1)
    al_sma = loss.rolling_mean(window_size=14, min_samples=1)
    rs_sma = ag_sma / (al_sma + eps)
    rsi_sma= (100 - 100/(1+rs_sma)).alias("RSI_14_SMA_raw")

    ag_ewm = gain.ewm_mean(alpha=1/14, adjust=False)
    al_ewm = loss.ewm_mean(alpha=1/14, adjust=False)
    rs_ewm = ag_ewm / (al_ewm + eps)
    rsi_ewm= (100 - 100/(1+rs_ewm)).alias("RSI_14_EMA_raw")

    # ATR
    atr_sma = true_range.rolling_mean(window_size=14, min_samples=1).alias("ATR_14_SMA_raw")
    atr_ewm = true_range.ewm_mean(alpha=1/14, adjust=False).alias("ATR_14_EMA_raw")

    # Stoch
    lowest_low   = L.rolling_min(window_size=14, min_samples=1)
    highest_high = H.rolling_max(window_size=14, min_samples=1)
    stoch_k = (((C - lowest_low)/(highest_high - lowest_low + eps)*100)
               .alias("Stoch_%K_raw"))
    stoch_d = stoch_k.rolling_mean(window_size=3, min_samples=1).alias("Stoch_%D_raw")

    # CCI
    tp       = (H + L + C)/3
    tp_sma   = tp.rolling_mean(window_size=20, min_samples=1)
    dev      = (tp - tp_sma).abs().rolling_mean(window_size=20, min_samples=1)
    cci_20   = ((tp - tp_sma)/(0.015*dev)).alias("CCI_20_raw")

    # MFI
    mf       = tp * V
    pf       = pl.when(tp > tp.shift(1)).then(mf).otherwise(0)
    nf       = pl.when(tp < tp.shift(1)).then(mf).otherwise(0)
    ps       = pf.rolling_sum(window_size=14, min_samples=1)
    ns       = nf.rolling_sum(window_size=14, min_samples=1)
    mfr      = ps/(ns+eps)
    mfi_14   = (100 - 100/(1+mfr)).alias("MFI_14_raw")

    # Volatility
    vol_14   = (pl.col("log_returns_close")
                 .rolling_std(window_size=14, min_samples=1)
                 * np.sqrt(14)
               ).alias("volatility_14_raw")

    # attach raws
    df = df.with_columns([
        momentum_10, roc_10, price_gap, obv_raw,
        rsi_sma, rsi_ewm, atr_sma, atr_ewm,
        stoch_k, stoch_d, cci_20, mfi_14, vol_14,
    ])

    # normalize raws → finals
    mappings = [
      ("momentum_10_raw","momentum_10"),
      ("ROC_10_raw",      "ROC_10"),
      ("price_gap_raw",   "price_gap"),
      ("OBV_raw",         "OBV"),
      ("RSI_14_SMA_raw",  "RSI_14_SMA"),
      ("RSI_14_EMA_raw",  "RSI_14_EMA"),
      ("ATR_14_SMA_raw",  "ATR_14_SMA"),
      ("ATR_14_EMA_raw",  "ATR_14_EMA"),
      ("Stoch_%K_raw",    "Stoch_%K"),
      ("Stoch_%D_raw",    "Stoch_%D"),
      ("CCI_20_raw",      "CCI_20"),
      ("MFI_14_raw",      "MFI_14"),
      ("volatility_14_raw","volatility_14"),
    ]
    exprs = [rolling_z_score_rescaled_expr(src, 14).alias(dst) for src, dst in mappings]
    raws  = [src for src, _ in mappings]

    return df.with_columns(exprs).drop(raws)

'''
FEATURE CONSOLIDATION
'''
def combine_and_refine_features(df: pl.DataFrame) -> pl.DataFrame:
    """
    Combine volume and trade-related features across timeframes, and reduce redundancy.
    """

    # Step 1: Create aggregated columns
    df = df.with_columns([
        # Total volume by timeframe
        (pl.sum_horizontal(
            pl.col("volume_fast_norm"),
            pl.col("quote_asset_volume_fast_norm"),
            pl.col("taker_buy_base_asset_volume_fast_norm"),
            pl.col("taker_buy_quote_asset_volume_fast_norm")) / 4).alias("total_volume_fast"),

        (pl.sum_horizontal(
            pl.col("volume_medium_norm"),
            pl.col("quote_asset_volume_medium_norm"),
            pl.col("taker_buy_base_asset_volume_medium_norm"),
            pl.col("taker_buy_quote_asset_volume_medium_norm")) / 4).alias("total_volume_medium"),

        (pl.sum_horizontal(
            pl.col("volume_slow_norm"),
            pl.col("quote_asset_volume_slow_norm"),
            pl.col("taker_buy_base_asset_volume_slow_norm"),
            pl.col("taker_buy_quote_asset_volume_slow_norm")) / 4).alias("total_volume_slow"),

        # Trade intensity by timeframe
        ((pl.col("number_of_trades_fast_norm") + pl.col("taker_buy_base_asset_volume_fast_norm")) / 2).alias("trade_intensity_fast"),
        ((pl.col("number_of_trades_medium_norm") + pl.col("taker_buy_base_asset_volume_medium_norm")) / 2).alias("trade_intensity_medium"),
        ((pl.col("number_of_trades_slow_norm") + pl.col("taker_buy_base_asset_volume_slow_norm")) / 2).alias("trade_intensity_slow"),
    ])

    # Step 2: Compute row-wise mean and std manually
    def row_std_expr(cols: list[str]) -> pl.Expr:
        mean_expr = pl.sum_horizontal(*[pl.col(c) for c in cols]) / len(cols)
        variance_expr = (
            sum([(pl.col(c) - mean_expr) ** 2 for c in cols]) / len(cols)
        )
        return variance_expr.sqrt()

    df = df.with_columns([
        ((pl.col("total_volume_medium") + pl.col("total_volume_slow")) / 2 * 0.99).alias("total_volume_combined"),
        (row_std_expr(["total_volume_fast", "total_volume_medium", "total_volume_slow"]) / 2).alias("total_volume_variation"),
        ((pl.col("trade_intensity_medium") + pl.col("trade_intensity_slow")) / 2 * 0.99).alias("trade_intensity_combined"),
        (row_std_expr(["trade_intensity_fast", "trade_intensity_medium", "trade_intensity_slow"]) / 2).alias("trade_intensity_variation"),
    ])

    # 🔄 Step 2.5: Normalize variation features with trailing window
    df = df.with_columns([
        rolling_z_score_rescaled_expr("total_volume_variation", MEDIUM_WINDOW).alias("total_volume_variation"),
        rolling_z_score_rescaled_expr("trade_intensity_variation", MEDIUM_WINDOW).alias("trade_intensity_variation"),
    ])

    # Step 3: Drop redundant raw inputs
    to_drop = [
        "volume_fast_norm", "quote_asset_volume_fast_norm", "taker_buy_base_asset_volume_fast_norm", "taker_buy_quote_asset_volume_fast_norm",
        "volume_medium_norm", "quote_asset_volume_medium_norm", "taker_buy_base_asset_volume_medium_norm", "taker_buy_quote_asset_volume_medium_norm",
        "volume_slow_norm", "quote_asset_volume_slow_norm", "taker_buy_base_asset_volume_slow_norm", "taker_buy_quote_asset_volume_slow_norm",
        "number_of_trades_fast_norm", "number_of_trades_medium_norm", "number_of_trades_slow_norm",
        "total_volume_medium", "total_volume_slow", "trade_intensity_medium", "trade_intensity_slow"
    ]

    return df.drop([col for col in to_drop if col in df.columns])

def reorder_final_columns(df: pl.DataFrame) -> pl.DataFrame:
    cols = df.columns

    # 1. Timestamp
    timestamp_cols = [col for col in cols if col == "timestamp"]

    # 2. Original OHLCV + volumes/trades
    original_cols = [
        "original_open", "original_high", "original_low",
        "original_close", "original_volume",
        "original_quote_asset_volume", "original_number_of_trades",
        "original_taker_buy_base_asset_volume", "original_taker_buy_quote_asset_volume"
    ]
    original_cols = [col for col in original_cols if col in cols]  # filter for safety

    # 3. Target columns
    target_cols = [col for col in cols if "target" in col]

    # 4. Remaining columns
    already_added = set(timestamp_cols + original_cols + target_cols)
    remaining_cols = [col for col in cols if col not in already_added]

    # Final column order
    ordered_cols = timestamp_cols + original_cols + target_cols + remaining_cols

    return df.select(ordered_cols)




'''
Inspection
'''

def inspect_df(df: pl.DataFrame):
    print(df.head(5))
    print(df.tail(5))

    print("\n=== Data Types ===")
    for name, dtype in zip(df.columns, df.dtypes):
        print(f"  {name}: {dtype}")

    print("\n=== Min / Max (numeric) ===")
    numeric = {
        pl.Int8, pl.Int16, pl.Int32, pl.Int64,
        pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64,
        pl.Float32, pl.Float64
    }
    for name, dtype in zip(df.columns, df.dtypes):
        if dtype in numeric:
            mn, mx = df[name].min(), df[name].max()
            print(f"  {name}: min={mn}, max={mx}")

    print("\n=== ⚠️ Columns With Nulls / NaNs / Infs ===")
    flagged = False
    for name, dtype in zip(df.columns, df.dtypes):
        s = df[name]
        nulls = s.null_count()
        nans = s.is_nan().sum() if dtype in {pl.Float32, pl.Float64} else 0
        infs = s.is_infinite().sum() if dtype in {pl.Float32, pl.Float64} else 0

        if nulls > 0 or nans > 0 or infs > 0:
            flagged = True
            print(f"  ❗ {name}: nulls={nulls}, nans={nans}, infs={infs}")

    if not flagged:
        print("  ✅ No issues detected.")

    print("\n=== Sample Target Columns ===")
    print(df.select([
        "timestamp", "original_close", "original_high",
        "target_open", "target_high", "target_low", "target_close",
        "target_volume", "target_quote_asset_volume",
        "target_number_of_trades",
        "target_taker_buy_base_asset_volume",
        "target_taker_buy_quote_asset_volume"
    ]).head(8))


def decode_target_row(
        prev_close: float,
        prev_values: dict[str, float],
        predicted_targets: dict[str, float],
        volume_k: float = 0.2
) -> dict[str, float]:
    """
    Decode synthetic target values back into real-world OHLCV values.

    Args:
        prev_close: Close price from the last candle in the input window.
        prev_values: Dictionary of raw values for volume-related fields at t-1.
        predicted_targets: Dictionary of predicted target deltas from the model.
        volume_k: Scaling factor used in tanh-compressed log-returns (default: 0.2)

    Returns:
        Dict with decoded raw values: open, high, low, close, and all 5 volume-related fields.
    """
    result = {}

    # Reconstruct OHLC from log distance to prev_close
    for field in ["open", "high", "low", "close"]:
        target = predicted_targets[f"target_{field}"]
        result[field] = prev_close * np.exp(target)

    # Reconstruct volume & trade fields from tanh(log-return)
    volume_fields = [
        "volume",
        "quote_asset_volume",
        "number_of_trades",
        "taker_buy_base_asset_volume",
        "taker_buy_quote_asset_volume"
    ]

    for field in volume_fields:
        target = predicted_targets[f"target_{field}"]
        prev_val = prev_values[field]
        # Invert tanh-compressed log-return
        log_ret = np.arctanh(np.clip(target, -0.999999, 0.999999)) / volume_k
        result[field] = prev_val * np.exp(log_ret)

    return result


def prompt_yes_no(message="Continue? [y/n]: "):
    while True:
        choice = input(message).strip().lower()
        if choice == 'y':
            return True
        elif choice == 'n':
            return False
        else:
            print("❌ Please enter 'y' or 'n'.")

if __name__ == "__main__":
    for filename in os.listdir(RAW_DIR):
        if not filename.endswith(f"_{TIMESCALE}_historical_data.csv"):
            continue

        asset_name = filename.split("_")[0]  # e.g., "ADAUSDT"
        input_path = os.path.join(RAW_DIR, filename)
        output_path = os.path.join(OUTPUT_DIR, f"{asset_name}_{TIMESCALE}_refined.csv")

        print(f"\n🔄 Processing {asset_name}...")

        df = load_dataset(input_path)
        df = create_targets(df)
        df = create_log_returns(df)
        df = copy_original_values(df)

        base_inputs = [
            "open","high","low","close","volume",
            "quote_asset_volume","taker_buy_base_asset_volume",
            "taker_buy_quote_asset_volume","number_of_trades"
        ]
        df = log_return_base_inputs(df, base_inputs)
        df = rolling_z_base_inputs(df, base_inputs)

        df = create_temporal_features(df)
        df = calculate_macd_polars(df)
        df = add_bollinger_band_norm_polars(df)
        df = add_normed_price_based_indicators(df)
        df = add_rolling_normed_mixed_ta_indicators(df)

        # === TRIM ONE: remove non-warmed-up rows ===
        max_head_cut_window = SLOW_WINDOW + 10
        max_tail_cut_window = 12
        df = df.slice(offset=max_head_cut_window)
        df = df.head(df.height - max_tail_cut_window)

        '''
        FEATURE REDUCTION
        '''
        if FLAG_COMBINE_FEATURES:
            df = combine_and_refine_features(df)

        # === TRIM TWO: remove 2 non-warmed-up rows for variation features ===
        max_head_cut_for_variation_features = MEDIUM_WINDOW
        df = df.slice(offset=max_head_cut_for_variation_features)

        '''
        Column reordering and inspection
        '''
        df = reorder_final_columns(df)

        # Preview
        print(f"✅ Finished {asset_name}. Shape: {df.shape}")
        inspect_df(df)

        # Save
        df.write_csv(output_path)
        print("✅ File saved.")