-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_check_missing_rows.py
More file actions
36 lines (29 loc) · 1.15 KB
/
test_check_missing_rows.py
File metadata and controls
36 lines (29 loc) · 1.15 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import pandas as pd
from datetime import timedelta
import os
def check_missing_timestamps(csv_path, interval="15min", log_path="missing_timestamps.log"):
# Load CSV
df = pd.read_csv(csv_path, parse_dates=['timestamp'], index_col='timestamp')
# Ensure sorted index
df = df.sort_index()
# Infer interval and generate expected timestamp range
start = df.index.min()
end = df.index.max()
expected = pd.date_range(start=start, end=end, freq=interval)
# Find missing timestamps
missing = expected.difference(df.index)
# Write missing timestamps to log
if missing.empty:
print("✅ No missing timestamps found.")
else:
with open(log_path, "w") as log:
for ts in missing:
log.write(f"{ts.isoformat()}\n")
print(f"❌ Found {len(missing)} missing timestamps.")
print(f"📝 Log written to {log_path}")
if __name__ == "__main__":
# Example usage
symbol = "ADAUSDT"
interval = "15min" # or '1h', '1d', etc.
filename = f"datasets/raw/full_scope/15m/{symbol}_15m_historical_data.csv"
check_missing_timestamps(csv_path=filename, interval=interval)