Data Transforms API¶

import qldata as qd

# Fluent API (recommended)
df = qd.data("BTCUSDT", source="binance") \
    .last(30) \
    .resolution("1h") \
    .clean() \
    .fill_forward() \
    .resample("4h") \
    .get()

# Standalone functions
from qldata import remove_duplicates, fill_forward, resample
df = remove_duplicates(raw_df)
df = fill_forward(df)
df = resample(df, "4h")

# Basic cleaning
.clean()

# With options
.clean(
    remove_invalid_prices=True,
    validate_ohlc=True,
    remove_outliers=True
)

from qldata import remove_duplicates

# Keep first occurrence
df = remove_duplicates(raw_df)

# Keep last occurrence
df = remove_duplicates(raw_df, keep="last")

from qldata import remove_invalid_prices

df = remove_invalid_prices(raw_df)

# Specify price columns
df = remove_invalid_prices(raw_df, columns=["open", "high", "low", "close"])

from qldata import remove_outliers

# Default: Remove > 3 standard deviations
df = remove_outliers(raw_df)

# Custom threshold
df = remove_outliers(raw_df, sigma=2.5)

# Specific columns only
df = remove_outliers(raw_df, columns=["close", "volume"])

from qldata import fill_forward

df = fill_forward(raw_df)

# Or in fluent API
df = qd.data(...).clean().fill_forward().get()

from qldata import fill_backward

df = fill_backward(raw_df)

from qldata import fill_interpolate

# Linear interpolation (default)
df = fill_interpolate(raw_df)

# Time-weighted interpolation
df = fill_interpolate(raw_df, method="time")

# Polynomial interpolation
df = fill_interpolate(raw_df, method="polynomial", order=2)

from qldata import resample

# Resample 1m to 1h
hourly = resample(minute_df, "1h")

# Resample 1h to 4h
four_hour = resample(hourly_df, "4h")

from qldata import resample_bars

# Properly aggregates OHLCV:
# - open: first value
# - high: max value
# - low: min value
# - close: last value
# - volume: sum
hourly = resample_bars(minute_df, "1h")

from qldata import ticks_to_bars

# Convert ticks to 1-minute bars
bars = ticks_to_bars(tick_df, "1m")

# Convert to hourly
bars = ticks_to_bars(tick_df, "1h")

from qldata import TransformPipeline, remove_duplicates, remove_outliers, fill_forward

# Create a reusable pipeline
pipeline = TransformPipeline() \
    .add(remove_duplicates) \
    .add(remove_outliers, sigma=3) \
    .add(fill_forward)

# Apply to any DataFrame
clean_df = pipeline.apply(raw_df)

# Apply to multiple DataFrames
for symbol, df in data.items():
    data[symbol] = pipeline.apply(df)

import qldata as qd

# Fetch and clean data
df = qd.data("BTCUSDT", source="binance") \
    .last(30) \
    .resolution("1h") \
    .clean() \
    .get()

print(f"Cleaned data: {len(df)} bars")
print(f"Missing values: {df.isna().sum().sum()}")

import qldata as qd

# Production-grade cleaning
df = qd.data("BTCUSDT", source="binance") \
    .last(30) \
    .resolution("1h") \
    .clean(
        remove_invalid_prices=True,  # Remove zero/negative
        validate_ohlc=True,          # Check high >= low, etc.
        remove_outliers=True         # Remove > 3σ outliers
    ) \
    .fill_forward() \
    .get()

# Verify data quality
assert (df[["open", "high", "low", "close"]] > 0).all().all()
assert (df["high"] >= df["low"]).all()
assert df.isna().sum().sum() == 0

import qldata as qd

# Fetch 1-minute data
minute_df = qd.data("BTCUSDT", source="binance") \
    .last(1, "days") \
    .resolution("1m") \
    .clean() \
    .get()

print(f"1-minute bars: {len(minute_df)}")

# Resample to hourly
hourly_df = qd.data("BTCUSDT", source="binance") \
    .last(1, "days") \
    .resolution("1m") \
    .clean() \
    .resample("1h") \
    .get()

print(f"Hourly bars: {len(hourly_df)}")

from qldata import TransformPipeline, remove_duplicates, fill_forward
import pandas as pd

def custom_smooth(df: pd.DataFrame, window: int = 5) -> pd.DataFrame:
    """Custom smoothing transform."""
    df = df.copy()
    df["close_smooth"] = df["close"].rolling(window).mean()
    return df

# Build pipeline with custom transform
pipeline = TransformPipeline() \
    .add(remove_duplicates) \
    .add(fill_forward) \
    .add(custom_smooth, window=10)

# Apply to data
import qldata as qd
raw_df = qd.data("BTCUSDT", source="binance").last(30).resolution("1h").get()
smooth_df = pipeline.apply(raw_df)

print(f"Added column: close_smooth")
print(smooth_df[["close", "close_smooth"]].head())

import qldata as qd
from qldata import TransformPipeline, remove_duplicates, remove_outliers, fill_forward

# Create reusable pipeline
pipeline = TransformPipeline() \
    .add(remove_duplicates) \
    .add(remove_outliers, sigma=3) \
    .add(fill_forward)

# Fetch multiple symbols
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
data = qd.data(symbols, source="binance") \
    .last(30) \
    .resolution("1h") \
    .get(parallel=True)

# Clean each symbol
for symbol in data:
    data[symbol] = pipeline.apply(data[symbol])
    print(f"{symbol}: {len(data[symbol])} bars, no NaN: {data[symbol].isna().sum().sum() == 0}")

# ✓ Good: Clean → Fill → Resample
.clean().fill_forward().resample("1h")

# ✗ Bad: Resample before cleaning may spread bad data
.resample("1h").clean().fill_forward()

# ✓ Recommended
df = qd.data(...).clean().get()

# ✗ Risky in production
df = qd.data(...).get()

# Sanity check after cleaning
assert df.isna().sum().sum() == 0, "Unexpected NaN values"
assert (df["close"] > 0).all(), "Invalid prices found"

# Define once, use everywhere
PRODUCTION_PIPELINE = TransformPipeline() \
    .add(remove_duplicates) \
    .add(remove_outliers) \
    .add(fill_forward)

# Consistent cleaning across all data
df1 = PRODUCTION_PIPELINE.apply(raw_df1)
df2 = PRODUCTION_PIPELINE.apply(raw_df2)