ohlcv-processing Checked

Sort and deduplicate

df = df.sort_index()

dupes = df.index.duplicated(keep="last")

if dupes.any():

print(f"Warning: Removed {dupes.sum()} duplicate timestamps")

df = df[~dupes]

Type enforcement

for col in ["open", "high", "low", "close", "volume"]:

df[col] = pd.to_numeric(df[col], errors="coerce")

return df

### Impossible Candle Detection

def find_impossible_candles(df: pd.DataFrame) -> pd.DataFrame:

"""Find candles that violate OHLC constraints."""

issues = pd.DataFrame(index=df.index)

issues["high_lt_low"] = df["high"] < df["low"]

issues["high_lt_open"] = df["high"] < df["open"]

issues["high_lt_close"] = df["high"] < df["close"]

issues["low_gt_open"] = df["low"] > df["open"]

issues["low_gt_close"] = df["low"] > df["close"]

issues["negative_price"] = (df[["open", "high", "low", "close"]] < 0).any(axis=1)

issues["negative_volume"] = df["volume"] < 0

issues["any_issue"] = issues.any(axis=1)

return issues[issues["any_issue"]]

## Gap Handling

Crypto trades 24/7, but gaps still occur from API outages, low liquidity, or aggregator downtime.

### Detect Gaps

def detect_gaps(df: pd.DataFrame, expected_freq: str = "1min") -> pd.Series:

"""Find missing timestamps based on expected frequency."""

full_index = pd.date_range(

start=df.index.min(), end=df.index.max(), freq=expected_freq, tz="UTC"

)

missing = full_index.difference(df.index)

return missing

### Fill Gaps

def handle_gaps(

df: pd.DataFrame,

freq: str = "1min",

method: str = "ffill",

max_gap: int = 5,

) -> pd.DataFrame:

"""Fill gaps in OHLCV data.

Args:

df: OHLCV DataFrame with DatetimeIndex.

freq: Expected bar frequency.

method: 'ffill' (forward fill) or 'interpolate'.

max_gap: Maximum consecutive bars to fill. Larger gaps are left as NaN.

"""

full_index = pd.date_range(

start=df.index.min(), end=df.index.max(), freq=freq, tz="UTC"

)

df = df.reindex(full_index)

df.index.name = "timestamp"

Mark which bars were filled

df["is_filled"] = df["close"].isna()

if method == "ffill":

Forward fill OHLC (flat candle), zero volume

df[["open", "high", "low", "close"]] = (

df[["open", "high", "low", "close"]].ffill(limit=max_gap)

)

df["volume"] = df["volume"].fillna(0)

elif method == "interpolate":

df[["open", "high", "low", "close"]] = (

df[["open", "high", "low", "close"]].interpolate(

method="time", limit=max_gap

)

)

df["volume"] = df["volume"].fillna(0)

return df

## Anomaly Detection

See `references/data_quality.md` for the complete anomaly taxonomy.

### Price Spike Detection

def detect_price_spikes(

df: pd.DataFrame, window: int = 20, threshold: float = 3.0

) -> pd.Series:

"""Flag bars where return exceeds threshold * rolling std."""

returns = df["close"].pct_change()

rolling_std = returns.rolling(window, min_periods=5).std()

spike = returns.abs() > (threshold * rolling_std)

return spike.fillna(False)

### Zero Volume Detection

def detect_zero_volume(df: pd.DataFrame, min_volume: float = 0) -> pd.Series:

"""Flag bars with zero or below-minimum volume."""

return df["volume"] <= min_volume

### Composite Anomaly Flagging

def flag_anomalies(df: pd.DataFrame) -> pd.DataFrame:

"""Add anomaly flag columns to DataFrame."""

df["anomaly_spike"] = detect_price_spikes(df)

df["anomaly_zero_vol"] = detect_zero_volume(df)

impossible = find_impossible_candles(df)

df["anomaly_impossible"] = False

if not impossible.empty:

df.loc[impossible.index, "anomaly_impossible"] = True

df["anomaly_any"] = (

df["anomaly_spike"] | df["anomaly_zero_vol"] | df["anomaly_impossible"]

)

return df

## Resampling

See `references/resampling_guide.md` for detailed guidance.

### Standard Resample

OHLCV_RESAMPLE_RULES = {

"open": "first",

"high": "max",

"low": "min",

"close": "last",

"volume": "sum",

}

def resample_ohlcv(df: pd.DataFrame, target_freq: str) -> pd.DataFrame:

"""Resample OHLCV to a coarser timeframe.

Args:

df: OHLCV DataFrame (must be finer than target_freq).

target_freq: Pandas frequency string ('5min', '15min', '1h', '4h', '1D').

Returns:

Resampled OHLCV DataFrame with no NaN rows.

"""

ohlcv_cols = ["open", "high", "low", "close", "volume"]

resampled = df[ohlcv_cols].resample(target_freq).agg(OHLCV_RESAMPLE_RULES)

return resampled.dropna(subset=["close"])

### Common Timeframe Ladder

TIMEFRAME_LADDER = ["1min", "5min", "15min", "1h", "4h", "1D"]

def resample_ladder(df: pd.DataFrame) -> dict[str, pd.DataFrame]:

"""Resample 1-minute data to all standard timeframes."""

results = {"1min": df.copy()}

for tf in TIMEFRAME_LADDER[1:]:

results[tf] = resample_ohlcv(df, tf)

return results

### VWAP Calculation

def compute_vwap(df: pd.DataFrame) -> pd.Series:

"""Compute