"""Decode SQX .dat BAR files - the free breadth sitting in the SQX install. `sqx.py` decodes the TICK files, where every record has FOUR fields (time, ask, bid, volume). The bar files use the identical container with SIX (time, open, high, low, close, volume), and that is the only reason the tick decoder cannot read them. Everything else - the block magic, the delta encoding, the per-record field map - is the same, so this module is that decoder generalised over the field count rather than a second implementation. THE FORMAT, STATED PLAINLY -------------------------- header as sqx.read_header block anchor MAGIC (15 bytes) + int32 block index + cfg + N ABSOLUTE int64s ordinary record cfg + N values, each a DELTA on the running value cfg ONE NIBBLE PER FIELD, so N fields occupy ceil(N/2) bytes: tick 4 fields -> 2 bytes, bar 6 fields -> 3 bytes. nibble = (logic << 2) | dtype, dtype indexes (1,2,4,8) bytes. The records are delta-encoded, which is why reading at a fixed stride recovers the first record perfectly and then produces noise - a failure mode worth naming, because the first record looking right is exactly what makes it tempting to trust the rest. THE SCALE IS THE DANGEROUS PART ------------------------------- Prices are integers and the decimal count is NOT in the header. A wrong scale does not crash and does not even hurt a correlation - it silently rescales every price, which is recorded twice already in this project's notes as the trap that costs the most time. So it is never inferred here: `catalog` pins it, states the three independent checks that agree on it, and this module ASSERTS it per file against the instrument's known price range rather than trusting the constant. These are M1 bars with no ask, so any trade simulation on them must substitute the5ers' cost from an instrument where the real spread is known. Never Dukascopy's own. """ import numpy as np, os, sys, struct import sqx, catalog from sqx import MAGIC, _SIZE, MINUS, ASIS CACHE = catalog.CACHE COLS = ['time', 'open', 'high', 'low', 'close', 'volume'] def _fields_n(cfg, nf): """Unpack `nf` (logic, dtype) pairs from ceil(nf/2) config bytes, one nibble each.""" out = [] for i in range(nf): nib = (cfg[i >> 1] >> (4 if i % 2 == 0 else 0)) & 0xF out.append((nib >> 2, nib & 3)) return out def field_count(path): """Read the first block's cfg width to learn how many fields the file carries.""" with open(path, 'rb') as fh: head = fh.read(1 << 16) p, _ = sqx.read_header(head) p += 15 + 4 n = 0 while head[p + n] == 0xBB: n += 1 return n * 2 def decode(path, nf=None, max_records=None): """-> (n, nf) int64 array of absolute values: time_ms, open, high, low, close, volume.""" nf = nf or field_count(path) ncfg = (nf + 1) // 2 fh = open(path, 'rb') head = fh.read(1 << 16) start, _ = sqx.read_header(head) fh.seek(start) cur = [0] * nf rows = [] buf = b'' pos = 0 CH = 1 << 22 eof = False n = 0 need_max = ncfg + 8 * nf + 32 while True: if len(buf) - pos < need_max and not eof: buf = buf[pos:]; pos = 0 more = fh.read(CH) if not more: eof = True buf += more if len(buf) - pos < ncfg + 1: break if buf[pos:pos + 15] == MAGIC: pos += 15 + 4 # magic + int32 block index if len(buf) - pos < ncfg + 8 * nf: break pos += ncfg # anchor cfg, values are absolute cur = list(struct.unpack_from(f'>{nf}q', buf, pos)) pos += 8 * nf else: cfg = buf[pos:pos + ncfg] fs = _fields_n(cfg, nf) size = ncfg + sum(_SIZE[d] for _, d in fs) if len(buf) - pos < size: if eof: break buf = buf[pos:]; pos = 0 more = fh.read(CH) if not more: eof = True buf += more if len(buf) < size: break continue q = pos + ncfg for k, (logic, dtp) in enumerate(fs): sz = _SIZE[dtp] raw = int.from_bytes(buf[q:q + sz], 'big', signed=(logic == ASIS)) q += sz cur[k] += (-raw if logic == MINUS else raw) pos = q rows.append(tuple(cur)) n += 1 if max_records and n >= max_records: break fh.close() return np.array(rows, dtype=np.int64), nf def check_scale(close, key): """Refuse a decoded file whose prices do not land in the instrument's known range. The scale is pinned in `catalog`, not fitted, because the reference series it used to be fitted against no longer exists. That makes this check the only thing standing between a silently rescaled price and every ATR-normalised feature downstream, so it is an assertion rather than a warning. The neighbouring exponents are 10x away; nothing about this call is ever marginal. """ lo, hi = catalog.px_range(key) med = float(np.median(close)) if not (lo <= med <= hi): raise ValueError(f"{catalog.nice(key)}: median close {med:,.4f} outside the known " f"range [{lo:,}, {hi:,}] - scale 1e{catalog.DECIMALS} is wrong " f"for this file") return med def load(key, tf='M1', verbose=True): """Decoded, scaled, sorted bars for a catalog key. Cached as .npz after the first call.""" os.makedirs(CACHE, exist_ok=True) out = f"{CACHE}{catalog.nice(key)}_{tf}.npz" if os.path.exists(out): z = np.load(out) return z['bars'], list(z['columns']) path = catalog.path(key, tf) if not os.path.exists(path): raise FileNotFoundError(path) a, nf = decode(path) if nf < 6: raise ValueError(f"{key}: {nf} fields, not a bar file") s = 10.0 ** catalog.DECIMALS bars = np.column_stack([a[:, 0].astype(np.float64), a[:, 1] / s, a[:, 2] / s, a[:, 3] / s, a[:, 4] / s, a[:, 5].astype(np.float64)]) bars = bars[np.argsort(bars[:, 0], kind='stable')] med = check_scale(bars[:, 4], key) #--- structural sanity a scale check cannot do: a high below its low means the FIELD #--- ORDER is wrong, which rescaling would never reveal bad = int((bars[:, 2] < bars[:, 3]).sum()) if bad > len(bars) // 1000: raise ValueError(f"{key}: {bad:,}/{len(bars):,} bars have high < low - field order") if verbose: import datetime as dt print(f" {catalog.nice(key):>9} {tf}: {len(bars):>10,} bars " f"{dt.datetime.fromtimestamp(bars[0,0]/1000, dt.UTC):%Y-%m-%d}.." f"{dt.datetime.fromtimestamp(bars[-1,0]/1000, dt.UTC):%Y-%m-%d}" f" median {med:>11,.4f} high19}{'shared min':>13}{'corr':>10}{'median diff':>14}{'in bp':>9}") for a, b in catalog.PAIRS: try: ba, _ = load(catalog.KEY[a], 'M1', verbose=False) bb, _ = load(catalog.KEY[b], 'M1', verbose=False) except Exception as ex: print(f"{a+'/'+b:>19} SKIPPED: {ex}") continue _, ia, ib = np.intersect1d(ba[:, 0].astype(np.int64), bb[:, 0].astype(np.int64), return_indices=True) if len(ia) < 1000: print(f"{a+'/'+b:>19}{len(ia):>13,} too few shared minutes to compare") continue ca, cb = ba[ia, 4], bb[ib, 4] d = np.median(ca - cb) print(f"{a+'/'+b:>19}{len(ia):>13,}{np.corrcoef(ca, cb)[0,1]:>10.6f}" f"{d:>14.5f}{1e4*d/np.median(cb):>9.2f}") if __name__ == '__main__': sys.stdout.reconfigure(encoding='utf-8', errors='replace') print("=== SQX BAR DECODER: build M1 cache for every catalog instrument ===") for k in catalog.SYMS: try: tf = catalog.base_tf(k) load(k, tf) except Exception as ex: print(f" {catalog.nice(k):>9}: FAILED - {ex}") _cross_feed_report()