Function clean_ohlcv

python/openquant/data.py:170–211

(
    df: pl.DataFrame,
    *,
    dedupe_keep: str = "last",
    return_report: bool = False,
)

def clean_ohlcv(
    df: pl.DataFrame,
    *,
    dedupe_keep: str = "last",
    return_report: bool = False,
) -> pl.DataFrame | tuple[pl.DataFrame, dict[str, Any]]:
    if dedupe_keep not in {"first", "last"}:
        raise ValueError("dedupe_keep must be 'first' or 'last'")

    base_lf = _prepare_ohlcv_lf(df).with_columns(pl.col("ts").dt.timestamp(time_unit="us").alias("ts_us"))
    sorted_lf = base_lf.sort(["symbol", "ts_us"])

    duplicate_key_count = int(
        sorted_lf
        .select(
            (
                ((pl.col("symbol") == pl.col("symbol").shift(1)) & (pl.col("ts_us") == pl.col("ts_us").shift(1)))
                .cast(pl.UInt32)
                .sum()
            ).alias("duplicate_key_count")
        )
        .collect()
        .item(0, 0)
    )

    cleaned = (
        sorted_lf.unique(
            subset=["symbol", "ts_us"],
            keep=dedupe_keep,
            maintain_order=True,
        )
        .sort(["symbol", "ts"])
        .collect()
    )

    frame = cleaned.select(CANONICAL_OHLCV_COLUMNS)
    if not return_report:
        return frame

    report = _build_quality_report(cleaned, rows_removed_by_deduplication=duplicate_key_count)
    report["duplicate_key_count"] = 0
    return frame, report


def data_quality_report(df: pl.DataFrame) -> dict[str, Any]: