okWarrior_EA/research/sqx_decode/sqxbars.py

"""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}  high<low {bad}")
    np.savez_compressed(out, bars=bars, columns=np.array(COLS))
    return bars, COLS


def _cross_feed_report():
    """The replication check that replaced the lost reference series.

    Six of these markets are carried TWICE, by vendors that share no data path. Agreement on
    the shared minutes is therefore an independent validation of the decoder AND of the
    scale, in the same way matching the tick-derived SP500 once was - and it is the only
    such check that survives the tick files being gone.
    """
    print()
    print(f"{'pair':>19}{'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()