forked from mnbvc188199/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()
|