vnpy--vnpy/vnpy/alpha/dataset/processor.py

from datetime import datetime

import numpy as np
import polars as pl

from .utility import to_datetime


def process_drop_na(df: pl.DataFrame, names: list[str] | None = None) -> pl.DataFrame:
    """Remove rows with missing values"""
    if names is None:
        names = df.columns[2:-1]

    for name in names:
        df = df.with_columns(
            pl.col(name).fill_nan(None)
        )
    df = df.drop_nulls(subset=names)
    return df


def process_fill_na(df: pl.DataFrame, fill_value: float, fill_label: bool = True) -> pl.DataFrame:
    """Fill missing values"""
    if fill_label:
        df = df.fill_null(fill_value)
        df = df.fill_nan(fill_value)
    else:
        df = df.with_columns(
            [pl.col(col).fill_null(fill_value).fill_nan(fill_value) for col in df.columns[2:-1]]
        )
    return df


def process_cs_norm(
    df: pl.DataFrame,
    method: str,         # robust/zscore
    names: list[str] | None = None
) -> pl.DataFrame:
    """Cross-sectional normalization"""
    if names is None:
        names = df.columns[2:-1]

    _df: pl.DataFrame = df.fill_nan(None)

    # Median method
    if method == "robust":
        for col in names:
            df = df.with_columns(
                _df.select(
                    (pl.col(col) - pl.col(col).median()).over("datetime").alias(col),
                )
            )

            df = df.with_columns(
                df.select(
                    pl.col(col).abs().median().over("datetime").alias("mad"),
                )
            )

            df = df.with_columns(
                (pl.col(col) / pl.col("mad") / 1.4826).clip(-3, 3).alias(col)
            ).drop(["mad"])
    # Z-Score method
    else:
        for col in names:
            df = df.with_columns(
                _df.select(
                    pl.col(col).mean().over("datetime").alias("mean"),
                    pl.col(col).std().over("datetime").alias("std"),
                )
            )

            df = df.with_columns(
                ((pl.col(col) - pl.col("mean")) / pl.col("std")).alias(col)
            ).drop(["mean", "std"])

    return df


def process_replace_inf(df: pl.DataFrame) -> pl.DataFrame:
    """Replace infinite values with per-symbol means"""
    _df: pl.DataFrame = df.fill_nan(None)

    for name in df.columns[2:]:
        mean_values: pl.DataFrame = (
            _df.filter(~pl.col(name).is_infinite()).group_by("vt_symbol").agg(
                [pl.col(name).mean().alias("mean")]
            )
        )

        df_with_mean: pl.DataFrame = df.join(mean_values, on="vt_symbol", how="left")
        df = df_with_mean.with_columns(
            pl.when(pl.col(name).is_infinite())
            .then(pl.col("mean"))
            .otherwise(pl.col(name))
            .alias(name)
        ).drop("mean")

    return df


def process_ts_norm(
    df: pl.DataFrame,
    fit_start_time: datetime | str | None = None,
    fit_end_time: datetime | str | None = None
) -> pl.DataFrame:
    """Time-series normalization"""
    _df: pl.DataFrame = df.fill_nan(None)

    if fit_start_time and fit_end_time:
        fit_start_time = to_datetime(fit_start_time)
        fit_end_time = to_datetime(fit_end_time)
        _df = _df.filter((pl.col("datetime") >= fit_start_time) & (pl.col("datetime") <= fit_end_time))

    for name in df.columns[2:]:
        df = df.with_columns(
            pl.lit(np.nanmean(_df[name])).alias("mean"),
            pl.lit(np.nanstd(_df[name])).alias("std"),
        )

        df = df.with_columns(
            pl.when(pl.col("std") == 0)
            .then(pl.col(name))
            .otherwise(((pl.col(name) - pl.col("mean")) / pl.col("std")).cast(pl.Float64))
            .alias(name)
        ).drop(["mean", "std"])

    return df


def process_drop_feature(df: pl.DataFrame, names: list[str]) -> pl.DataFrame:
    """Drop feature columns"""
    return df.drop(names)


def process_cs_fill_na(df: pl.DataFrame, names: list[str] | None = None) -> pl.DataFrame:
    """Fill missing values with cross-sectional means"""
    _df: pl.DataFrame = df.fill_nan(None)

    if names is None:
        names = _df.columns[2:-1]

    for col in names:
        df = df.with_columns(
            _df.select(
                pl.col(col).fill_null(pl.col(col).mean().over("datetime")).alias(col)
            )
        )

    return df


def process_robust_zscore_norm(
    df: pl.DataFrame,
    fit_start_time: datetime | str | None = None,
    fit_end_time: datetime | str | None = None,
    clip_outlier: bool = True
) -> pl.DataFrame:
    """Robust Z-Score normalization"""
    _df: pl.DataFrame = df.fill_nan(None)

    if fit_start_time and fit_end_time:
        fit_start_time = to_datetime(fit_start_time)
        fit_end_time = to_datetime(fit_end_time)
        _df = _df.filter((pl.col("datetime") >= fit_start_time) & (pl.col("datetime") <= fit_end_time))

    cols = df.columns[2:-1]
    X = _df.select(cols).to_numpy()

    mean_train = np.nanmedian(X, axis=0)
    std_train = np.nanmedian(np.abs(X - mean_train), axis=0)
    std_train += 1e-12
    std_train *= 1.4826

    for name in cols:
        normalized_col = (
            (pl.col(name) - mean_train[cols.index(name)]) / std_train[cols.index(name)]
        ).cast(pl.Float64)

        if clip_outlier:
            normalized_col = normalized_col.clip(-3, 3)

        df = df.with_columns(normalized_col.alias(name))

    return df


def process_cs_rank_norm(df: pl.DataFrame, names: list[str]) -> pl.DataFrame:
    """Cross-sectional rank normalization"""
    _df: pl.DataFrame = df.fill_nan(None)

    _df = _df.with_columns([
        ((pl.col(col).rank("average").over("datetime") / pl.col("datetime").count().over("datetime")) - 0.5) * 3.46
        for col in names
    ])

    df = df.with_columns([
        _df[col].alias(col) for col in names
    ])

    return df