56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""
|
|
Resamples HistData 1-minute CSV zips to 5-minute OHLCV CSVs.
|
|
Place all downloaded zip files in the data/ folder then run:
|
|
python3 scripts/resample_to_5m.py
|
|
"""
|
|
import zipfile
|
|
import pandas as pd
|
|
from pathlib import Path
|
|
|
|
INPUT_DIR = Path(__file__).parent.parent / "data"
|
|
OUTPUT_DIR = INPUT_DIR
|
|
PAIRS = ["USDJPY", "EURUSD", "GBPUSD"]
|
|
|
|
def load_histdata_zip(zip_path: Path) -> pd.DataFrame:
|
|
with zipfile.ZipFile(zip_path) as z:
|
|
csv_name = [f for f in z.namelist() if f.endswith(".csv")][0]
|
|
with z.open(csv_name) as f:
|
|
df = pd.read_csv(
|
|
f, header=None,
|
|
names=["date","time","open","high","low","close","volume"],
|
|
)
|
|
df["datetime"] = pd.to_datetime(
|
|
df["date"].astype(str) + " " + df["time"].astype(str),
|
|
format="%Y.%m.%d %H:%M")
|
|
df = df.drop(columns=["date","time"])
|
|
df.set_index("datetime", inplace=True)
|
|
df.sort_index(inplace=True)
|
|
return df
|
|
|
|
def resample_5m(df: pd.DataFrame) -> pd.DataFrame:
|
|
return df.resample("5min").agg({
|
|
"open":"first","high":"max","low":"min",
|
|
"close":"last","volume":"sum"
|
|
}).dropna()
|
|
|
|
for pair in PAIRS:
|
|
print(f"\nProcessing {pair}...")
|
|
zips = sorted(INPUT_DIR.glob(f"HISTDATA_COM_MT_{pair}_M1*.zip"))
|
|
if not zips:
|
|
print(f" No files found for {pair}")
|
|
continue
|
|
frames = []
|
|
for z in zips:
|
|
print(f" Loading {z.name}")
|
|
frames.append(load_histdata_zip(z))
|
|
combined = pd.concat(frames).sort_index()
|
|
combined = combined[~combined.index.duplicated(keep="first")]
|
|
print(f" 1m rows: {len(combined):,} | {combined.index[0]} → {combined.index[-1]}")
|
|
df_5m = resample_5m(combined)
|
|
print(f" 5m rows: {len(df_5m):,}")
|
|
out = OUTPUT_DIR / f"{pair}_5M.csv"
|
|
df_5m.to_csv(out)
|
|
print(f" Saved → {out}")
|
|
|
|
print("\nDone. Place CSV files in data/ folder before running training.")
|