mirror of
https://github.com/Arianhgh/fx-quant-research.git
synced 2026-07-28 02:57:50 +00:00
Arian Haghparast
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""Lightweight CSV loading and data-quality helpers.
|
|
|
|
Loads raw 5-minute candles and inspects the gaps between consecutive timestamps.
|
|
"""
|
|
import pandas as pd
|
|
|
|
|
|
def load_ohlcv_csv(file_path, timestamp_col="timestamp"):
|
|
"""Load an OHLCV CSV with a datetime index.
|
|
|
|
Parameters
|
|
----------
|
|
file_path : str
|
|
Path to a CSV containing a timestamp column plus OHLCV columns.
|
|
timestamp_col : str
|
|
Name of the timestamp column to parse and use as the index.
|
|
"""
|
|
data = pd.read_csv(file_path, index_col=timestamp_col, parse_dates=True)
|
|
data = data[~data.index.duplicated(keep="first")]
|
|
return data.sort_index()
|
|
|
|
|
|
def summarize(data):
|
|
"""Print basic information about a loaded OHLCV frame."""
|
|
print("Data Info:")
|
|
print(data.info())
|
|
print("\nFirst few rows:")
|
|
print(data.head())
|
|
print("\nLast few rows:")
|
|
print(data.tail())
|
|
print(f"\nTotal candles: {len(data)}")
|
|
|
|
|
|
def analyze_time_gaps(data, expected_gap=pd.Timedelta(minutes=5),
|
|
big_gap=pd.Timedelta(hours=1), verbose=True):
|
|
"""Report gaps between consecutive timestamps.
|
|
|
|
Returns the Series of gaps larger than ``expected_gap``.
|
|
"""
|
|
time_diffs = data.index.to_series().diff()
|
|
gaps = time_diffs[time_diffs > expected_gap]
|
|
|
|
if verbose:
|
|
print(f"Total gaps larger than {expected_gap}: {len(gaps)}")
|
|
print("\nGap statistics:")
|
|
print(gaps.describe())
|
|
|
|
big_gaps = gaps[gaps > big_gap]
|
|
print(f"\nNumber of gaps > {big_gap}: {len(big_gaps)}")
|
|
print(f"Big gaps (> {big_gap}):")
|
|
for gap_start, gap_size in big_gaps.items():
|
|
gap_end = gap_start + gap_size
|
|
print(f"Gap from {gap_start} to {gap_end} ({gap_size})")
|
|
|
|
return gaps
|