forked from animatedread/Warrior_EA
f8a15c7b50
- Implemented `catalog.py` to define instrument metadata and price scaling checks. - Developed `sqx.py` for decoding SQX tick .dat files, handling scaling and volume adjustments. - Created `sqxbars.py` to decode SQX bar files, ensuring price integrity against known ranges. - Introduced `volume_gate.py` for relative volume analysis on D1 bars, facilitating backtesting strategies.
214 lines
8.7 KiB
Python
214 lines
8.7 KiB
Python
"""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()
|