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.
326 lines
14 KiB
Python
326 lines
14 KiB
Python
"""Decoder for StrategyQuant X tick .dat files (format 4.2).
|
|
|
|
Derived from SQX's own writer, disassembled out of internal/libs/SQDataLib.jar:
|
|
com.strategyquant.datalib.data.io.newDataFormat.TickDataWriter
|
|
com.strategyquant.datalib.data.io.newDataFormat.NewDataFormat{,Writter}
|
|
|
|
Why bother, when Scripts/ExportTicks.mq5 pulls the same four fields from MT5: DEPTH. The
|
|
broker's MT5 tick history goes back a few years; these files start in 2011. Sample size has
|
|
been the binding constraint on this entire investigation, so 15 years of ticks is worth a
|
|
decoder.
|
|
|
|
FORMAT
|
|
------
|
|
Header: four Java writeUTF strings (version "4.2", a type char, a column string, and an
|
|
empty one), ten zero bytes, then one more writeUTF. Data follows.
|
|
|
|
Records are (time, ASK, BID, volume) - ask before bid, and the writer swaps them if bid>ask,
|
|
so ask is always the larger. Every BLOCK_LENGTH=1000 records a block header is emitted:
|
|
|
|
MAGIC (15 bytes: 0x00..0x0e) + int32 block index + config + four raw int64 values
|
|
|
|
and in between, delta records:
|
|
|
|
config + four variable-width deltas against the previous record
|
|
|
|
CONFIG is two bytes = four nibbles, one per field, laid out high-nibble-first:
|
|
|
|
nibble = (logicType << 2) | dataType
|
|
dataType : 0=BYTE(1) 1=SHORT(2) 2=INT(4) 3=LONG(8) <- chosen by |value| magnitude
|
|
logicType: 0=MINUS 1=PLUS 2=ASIS <- sign carried here, not in the bytes
|
|
|
|
MINUS/PLUS payloads are UNSIGNED magnitudes with the sign supplied by logicType; ASIS is a
|
|
plain signed read (used for the full records at block starts).
|
|
|
|
SCALING
|
|
-------
|
|
Prices are integers scaled by 10**decimals. SQX carries `decimals` in its own metadata, not
|
|
in this file, so it is inferred here by checking which power of ten yields a sane price -
|
|
and validated against the first record, which is uncompressed and therefore unambiguous.
|
|
Volume divides by VOLUME_CONSTANT = 100000.0 (OLD_VOLUME_CONSTANT = 100.0 for old files).
|
|
A volume of Long.MAX_VALUE is the writer's "no volume" sentinel and decodes to NaN.
|
|
"""
|
|
import struct, sys, datetime as dt
|
|
import numpy as np
|
|
|
|
MAGIC = bytes(range(15))
|
|
BLOCK_LENGTH = 1000
|
|
VOLUME_CONSTANT = 100000.0
|
|
_SIZE = (1, 2, 4, 8)
|
|
MINUS, PLUS, ASIS = 0, 1, 2
|
|
NO_VOLUME = (1 << 63) - 1
|
|
|
|
|
|
def read_header(b):
|
|
"""Returns the offset at which the first block MAGIC starts, plus the header strings."""
|
|
p, strs = 0, []
|
|
for _ in range(4):
|
|
n = struct.unpack_from('>H', b, p)[0]
|
|
strs.append(b[p + 2:p + 2 + n].decode('latin1'))
|
|
p += 2 + n
|
|
p += 10 # ten zero bytes
|
|
n = struct.unpack_from('>H', b, p)[0]
|
|
strs.append(b[p + 2:p + 2 + n].decode('latin1'))
|
|
p += 2 + n
|
|
if b[p:p + 15] != MAGIC: # be forgiving: just find it
|
|
p = b.find(MAGIC)
|
|
if p < 0:
|
|
raise ValueError("no block magic found in first chunk")
|
|
return p, strs
|
|
|
|
|
|
def _fields(cfg):
|
|
"""Unpack the two config bytes into four (logicType, dataType) pairs."""
|
|
out = []
|
|
for i in range(4):
|
|
nib = (cfg[i >> 1] >> (4 if i % 2 == 0 else 0)) & 0xF
|
|
out.append((nib >> 2, nib & 3))
|
|
return out
|
|
|
|
|
|
def decode(path, max_records=None, progress=None):
|
|
"""Whole-file decode into four int64 arrays. Convenient, but a full symbol is ~458M
|
|
ticks = ~15 GB of arrays - use decode_iter() for anything beyond a sample."""
|
|
parts = [[], [], [], []]
|
|
for ts, a, b, v in decode_iter(path, max_records=max_records, progress=progress):
|
|
parts[0].append(ts); parts[1].append(a); parts[2].append(b); parts[3].append(v)
|
|
if not parts[0]:
|
|
return tuple(np.empty(0, dtype=np.int64) for _ in range(4))
|
|
return tuple(np.concatenate(p) for p in parts)
|
|
|
|
|
|
def find_block(f, approx, size, window=1 << 20):
|
|
"""Byte offset of the first VALID block header at or after `approx`, or None.
|
|
|
|
This is what makes the format parallel-decodable: a block header restates all four
|
|
fields as absolute int64s, so a decoder can start there cold with no history. The catch
|
|
is that MAGIC is 0x00..0x0e, a byte run that delta payloads produce by coincidence
|
|
fairly often - a small negative delta on a LONG field is a lot of zero bytes. So a
|
|
candidate is only accepted when the NEXT candidate downstream carries the next
|
|
sequential block index, which junk essentially never does.
|
|
"""
|
|
def candidate(off):
|
|
"""(offset, block_index, timestamp) of the first plausible header at/after off."""
|
|
p = off
|
|
while p < size:
|
|
f.seek(p)
|
|
buf = f.read(window + 64)
|
|
if len(buf) < 21 + 32:
|
|
return None
|
|
k = 0
|
|
while True:
|
|
k = buf.find(MAGIC, k)
|
|
if k < 0 or k + 15 + 4 + 2 + 32 > len(buf):
|
|
break
|
|
idx = struct.unpack_from('>i', buf, k + 15)[0]
|
|
t, a, b, _v = struct.unpack_from('>4q', buf, k + 15 + 4 + 2)
|
|
#--- a real header: sane block index, epoch-ms in [2000, 2035], ask >= bid > 0
|
|
if (0 <= idx < (1 << 28) and 946684800000 <= t <= 2051222400000
|
|
and a >= b > 0):
|
|
return (p + k, idx, t)
|
|
k += 1
|
|
p += window
|
|
return None
|
|
|
|
first = candidate(approx)
|
|
while first is not None:
|
|
nxt = candidate(first[0] + 15 + 4 + 2 + 32)
|
|
if nxt is None:
|
|
return first[0] # last block in the file
|
|
if nxt[1] == first[1] + 1 and nxt[2] >= first[2]:
|
|
return first[0]
|
|
first = nxt # the first was noise; try the next
|
|
return None
|
|
|
|
|
|
def split_offsets(path, nchunks):
|
|
"""Byte ranges [(start, stop), ...] on block boundaries covering the whole file.
|
|
|
|
Half-open and contiguous: chunk k's stop IS chunk k+1's start, so every record is
|
|
decoded exactly once with no duplicates and no gaps.
|
|
"""
|
|
import os
|
|
size = os.path.getsize(path)
|
|
with open(path, 'rb') as f:
|
|
head = f.read(1 << 16)
|
|
first, _ = read_header(head)
|
|
if nchunks <= 1:
|
|
return [(first, size)]
|
|
cuts = [first]
|
|
span = (size - first) // nchunks
|
|
for k in range(1, nchunks):
|
|
off = find_block(f, first + k * span, size)
|
|
#--- keep it only if it is strictly after the previous cut
|
|
if off is not None and off > cuts[-1]:
|
|
cuts.append(off)
|
|
cuts.append(size)
|
|
return [(cuts[i], cuts[i + 1]) for i in range(len(cuts) - 1)]
|
|
|
|
|
|
def decode_iter(path, batch=2000000, max_records=None, progress=None, start=None, stop=None):
|
|
"""Streaming decode. Yields (time_ms, ask_i, bid_i, vol_i) int64 arrays in batches.
|
|
|
|
A full symbol here is hundreds of millions of ticks, so nothing that wants the whole
|
|
file can afford to materialise it. Everything downstream aggregates to bars in one pass.
|
|
|
|
start/stop: decode only the byte range [start, stop), which MUST begin on a block
|
|
boundary - see split_offsets(). Used to decode one file across many processes, since
|
|
this per-record loop is pure Python and saturates exactly one core.
|
|
"""
|
|
f = open(path, 'rb')
|
|
if start is None:
|
|
head = f.read(1 << 16)
|
|
start, strs = read_header(head)
|
|
f.seek(start)
|
|
|
|
ts, asks, bids, vols = [], [], [], []
|
|
t = a = bd = v = 0
|
|
buf = b''
|
|
pos = 0
|
|
base = start # absolute file offset of buf[0]
|
|
CH = 1 << 22
|
|
eof = False
|
|
n = 0
|
|
while True:
|
|
if len(buf) - pos < 64 and not eof:
|
|
buf = buf[pos:]
|
|
base += pos
|
|
pos = 0
|
|
more = f.read(CH)
|
|
if not more:
|
|
eof = True
|
|
buf += more
|
|
if len(buf) - pos < 2:
|
|
break
|
|
if stop is not None and base + pos >= stop:
|
|
break
|
|
if buf[pos:pos + 15] == MAGIC:
|
|
pos += 15 + 4 # magic + int32 block index
|
|
if len(buf) - pos < 2 + 32:
|
|
break
|
|
cfg = buf[pos:pos + 2]; pos += 2
|
|
t, a, bd, v = struct.unpack_from('>4q', buf, pos)
|
|
pos += 32
|
|
else:
|
|
cfg = buf[pos:pos + 2]
|
|
fs = _fields(cfg)
|
|
need = 2 + sum(_SIZE[d] for _, d in fs)
|
|
if len(buf) - pos < need:
|
|
if eof:
|
|
break
|
|
buf = buf[pos:]; base += pos; pos = 0
|
|
more = f.read(CH)
|
|
if not more:
|
|
eof = True
|
|
buf += more
|
|
if len(buf) < need:
|
|
break
|
|
continue
|
|
q = pos + 2
|
|
vals = []
|
|
for logic, dtp in fs:
|
|
sz = _SIZE[dtp]
|
|
raw = int.from_bytes(buf[q:q + sz], 'big', signed=(logic == ASIS))
|
|
q += sz
|
|
vals.append(-raw if logic == MINUS else raw)
|
|
pos = q
|
|
t += vals[0]; a += vals[1]; bd += vals[2]; v += vals[3]
|
|
ts.append(t); asks.append(a); bids.append(bd); vols.append(v)
|
|
n += 1
|
|
if progress and n % progress == 0:
|
|
print(f" {n:,} ticks {dt.datetime.fromtimestamp(t/1000, dt.UTC):%Y-%m-%d}", flush=True)
|
|
if len(ts) >= batch:
|
|
yield (np.array(ts, dtype=np.int64), np.array(asks, dtype=np.int64),
|
|
np.array(bids, dtype=np.int64), np.array(vols, dtype=np.int64))
|
|
ts, asks, bids, vols = [], [], [], []
|
|
if max_records and n >= max_records:
|
|
break
|
|
f.close()
|
|
if ts:
|
|
yield (np.array(ts, dtype=np.int64), np.array(asks, dtype=np.int64),
|
|
np.array(bids, dtype=np.int64), np.array(vols, dtype=np.int64))
|
|
|
|
|
|
def calibrate_decimals(ts, ask_i, ref_time_s, ref_close, verbose=True):
|
|
"""Determine the price scale by matching against a KNOWN reference series.
|
|
|
|
SQX keeps `decimals` in metadata outside the .dat, and it cannot be inferred from the
|
|
file alone: 1216010000 is a plausible price at 10^3 (1216010), 10^5 (12160.1) or 10^6
|
|
(1216.01), and nothing in the bytes distinguishes them. Guessing here would be exactly
|
|
the kind of silent, plausible-looking error this project keeps getting burned by - a
|
|
100x price scale error would not crash anything, it would just quietly rescale every
|
|
ATR-normalised feature downstream.
|
|
|
|
So instead: take a reference series for the same instrument (the MT5 rate exports under
|
|
Common\\Files\\Warrior_EA\\Research), find the tick nearest each reference bar time, and
|
|
pick the power of ten minimising median relative error. A correct scale lands at ~0;
|
|
every wrong one is off by a factor of ten or more, so the decision is never marginal.
|
|
"""
|
|
ref_ms = (ref_time_s * 1000).astype(np.int64)
|
|
lo, hi = ts[0], ts[-1]
|
|
m = (ref_ms >= lo) & (ref_ms <= hi)
|
|
if m.sum() < 50:
|
|
raise ValueError(f"only {int(m.sum())} overlapping reference bars - cannot calibrate")
|
|
rt, rc = ref_ms[m], ref_close[m]
|
|
idx = np.searchsorted(ts, rt).clip(1, len(ts) - 1)
|
|
got = ask_i[idx].astype(float)
|
|
best, best_err = None, np.inf
|
|
for d in range(0, 11):
|
|
err = float(np.median(np.abs(got / (10.0 ** d) - rc) / np.maximum(np.abs(rc), 1e-9)))
|
|
if verbose:
|
|
print(f" decimals={d:<3} median rel.err {err:12.6f}")
|
|
if err < best_err:
|
|
best, best_err = d, err
|
|
if best_err > 0.05:
|
|
raise ValueError(f"best scale (decimals={best}) still off by {best_err:.1%} - "
|
|
"reference series may be a different instrument")
|
|
return best, best_err
|
|
|
|
|
|
def to_frame(path, max_records=None, decimals=None, volume_constant=VOLUME_CONSTANT,
|
|
progress=None, ref=None):
|
|
"""decimals: pass explicitly, or pass ref=(time_s, close) to calibrate against a known
|
|
series. There is no safe default - see calibrate_decimals()."""
|
|
ts, ai, bi, vi = decode(path, max_records, progress)
|
|
if decimals is None:
|
|
if ref is None:
|
|
raise ValueError("pass decimals=... or ref=(time_s, close) - see calibrate_decimals()")
|
|
decimals, _ = calibrate_decimals(ts, ai, ref[0], ref[1])
|
|
scale = 10.0 ** decimals
|
|
ask = ai / scale
|
|
bid = bi / scale
|
|
vol = np.where(vi == NO_VOLUME, np.nan, vi / volume_constant)
|
|
return ts, bid, ask, vol, decimals
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# usage: python sqx.py <file.dat> [max_records] [--ref SYMBOL | --decimals N]
|
|
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
|
path = sys.argv[1]
|
|
lim = int(sys.argv[2]) if len(sys.argv) > 2 and not sys.argv[2].startswith('-') else None
|
|
dec_arg, ref_sym = None, None
|
|
if '--decimals' in sys.argv:
|
|
dec_arg = int(sys.argv[sys.argv.index('--decimals') + 1])
|
|
if '--ref' in sys.argv:
|
|
ref_sym = sys.argv[sys.argv.index('--ref') + 1]
|
|
ref = None
|
|
if ref_sym:
|
|
from kit import load_rates
|
|
rt, ro, rh, rl, rc, rv, rs = load_rates(ref_sym, 16385)
|
|
ref = (rt, rc)
|
|
print(f"calibrating scale against {ref_sym} H1 closes:")
|
|
ts, bid, ask, vol, dec = to_frame(path, max_records=lim, decimals=dec_arg,
|
|
progress=5000000, ref=ref)
|
|
print(f"\ndecoded {len(ts):,} ticks decimals={dec}")
|
|
print(f"span {dt.datetime.fromtimestamp(ts[0]/1000, dt.UTC)}"
|
|
f" -> {dt.datetime.fromtimestamp(ts[-1]/1000, dt.UTC)}")
|
|
print(f"bid min {np.nanmin(bid):.5f} max {np.nanmax(bid):.5f}")
|
|
spread = ask - bid
|
|
print(f"spread median {np.nanmedian(spread):.5f} negative: {(spread<0).sum()}")
|
|
print(f"time monotonic: {bool((np.diff(ts)>=0).all())} "
|
|
f"max gap {np.max(np.diff(ts))/1000/3600:.1f} h")
|
|
print(f"volume: median {np.nanmedian(vol)} nan {int(np.isnan(vol).sum()):,}")
|
|
print("\nfirst 5:")
|
|
for i in range(5):
|
|
print(f" {dt.datetime.fromtimestamp(ts[i]/1000, dt.UTC)} "
|
|
f"bid {bid[i]:.5f} ask {ask[i]:.5f} vol {vol[i]}")
|