| """
|
| MT5 `.hcc` history decoder (broker M1 bars), rebuilt 2026-09-27.
|
|
|
| Layout (measured on FivePercentOnline-Real SP500 2024):
|
| * 228-byte file header (UTF-16 copyright string)
|
| * index from byte 228: 18-byte records (u32 idx, u32 update time, u16 ?,
|
| u32 chunk size, u32 ABSOLUTE chunk offset), one per day, newest first
|
| * each chunk: 129-byte header (u16 = 129, UTF-16 symbol name, ...), then
|
| N x 60-byte MqlRates (i64 time, 4 x f64 OHLC, i64 tick_volume, i32 spread,
|
| i64 real_volume)
|
| * the first record of a chunk is often a DAILY SUMMARY (time 00:00, tick
|
| volume = the day's total) - dropped when its tick volume is >= 90% of the
|
| rest of the chunk, or when it is not strictly before the next bar.
|
|
|
| Times are the broker clock. Real volume is 0 on CFDs; spread is in points.
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import glob
|
| import os
|
| import struct
|
|
|
| import numpy as np
|
|
|
| REC = np.dtype([("t", "<i8"), ("o", "<f8"), ("h", "<f8"), ("l", "<f8"), ("c", "<f8"),
|
| ("tv", "<i8"), ("sp", "<i4"), ("rv", "<i8")])
|
| assert REC.itemsize == 60
|
|
|
|
|
| def read_hcc(path: str) -> np.ndarray:
|
| b = open(path, "rb").read()
|
| chunks = []
|
| k = 0
|
| while 228 + 18 * (k + 1) <= len(b):
|
| _, _, _, size, off = struct.unpack_from("<IIHII", b, 228 + 18 * k)
|
| if off < 228 or off + size > len(b) or size < 129:
|
| break
|
| hdr = struct.unpack_from("<H", b, off)[0]
|
| body = size - hdr
|
| if hdr == 129 and body > 0 and body % 60 == 0:
|
| a = np.frombuffer(b, REC, body // 60, off + hdr).copy()
|
| if len(a) > 1 and (a["tv"][0] >= 0.9 * a["tv"][1:].sum() or a["t"][0] >= a["t"][1]):
|
| a = a[1:]
|
| chunks.append(a)
|
| k += 1
|
| if not chunks:
|
| return np.zeros(0, REC)
|
| a = np.concatenate(chunks)
|
| a = a[np.argsort(a["t"], kind="stable")]
|
| keep = np.concatenate([[True], np.diff(a["t"]) > 0])
|
| return a[keep]
|
|
|
|
|
| def load_m1(folder: str, years=None) -> np.ndarray:
|
| files = sorted(glob.glob(os.path.join(folder, "*.hcc")))
|
| if years is not None:
|
| files = [f for f in files if int(os.path.basename(f)[:4]) in years]
|
| a = np.concatenate([read_hcc(f) for f in files])
|
| a = a[np.argsort(a["t"], kind="stable")]
|
| a = a[np.concatenate([[True], np.diff(a["t"]) > 0])]
|
| step = np.diff(a["t"])
|
| return a, step
|
5a4b36bb61