"""Download tick data straight from Dukascopy's public datafeed. SQX does NOT fetch from Dukascopy directly - it mirrors through its own CDN (CdnCache/CdnDownloadJob in SQTradingLib.jar), so there is nothing reusable there and pointing a scraper at their paid infrastructure would be wrong anyway. Dukascopy publishes the raw feed itself, free for personal use, and the format is simple. URL https://datafeed.dukascopy.com/datafeed/{SYM}/{YYYY}/{MM}/{DD}/{HH}h_ticks.bi5 MONTH IS ZERO-BASED. January is 00. This is the single most common mistake with this feed and it fails silently - you get a valid file for the wrong month. BODY raw LZMA stream (no container/header), decompressing to 20-byte BIG-ENDIAN records: uint32 milliseconds since the hour uint32 ask, in POINTS uint32 bid, in POINTS float32 ask volume float32 bid volume An empty body (0 bytes) means "no ticks that hour" - weekends, holidays - and is normal, not an error. WHY THIS IS WORTH HAVING over the SQX .dat already decoded: those records carry ONE volume field, so signed order flow had to be synthesised from quote dynamics (event-count OFI). This feed carries BID AND ASK VOLUME SEPARATELY, which is true signed flow - the exact limitation recorded in the tick-format notes. It is a different measurement, not more of the same. Politeness: bounded concurrency, retry with backoff, and a disk cache that makes re-runs free. Do not raise WORKERS - this is someone else's free service. """ import os, sys, time, struct, lzma, math import datetime as dt import numpy as np from concurrent.futures import ThreadPoolExecutor import urllib.request, urllib.error sys.stdout.reconfigure(encoding='utf-8', errors='replace') BASE = "https://datafeed.dukascopy.com/datafeed" CACHE = 'c:/Users/admin/Documents/Workspaces/Market Data/dukas_cache' OUT = 'c:/Users/admin/Documents/Workspaces/Market Data/bars/' WORKERS = 6 UA = "Mozilla/5.0 (research; personal use)" #--- Point scale = 10**decimals of the instrument. Getting this wrong does not crash #--- anything, it silently rescales every price - the same trap as the SQX decoder, so it #--- is explicit here rather than guessed, and calibrate() checks it against a known series. DECIMALS = { 'EURUSD': 5, 'GBPUSD': 5, 'AUDUSD': 5, 'NZDUSD': 5, 'USDCAD': 5, 'USDCHF': 5, 'EURGBP': 5, 'EURCHF': 5, 'EURAUD': 5, 'GBPCHF': 5, 'AUDCAD': 5, 'AUDNZD': 5, 'USDJPY': 3, 'EURJPY': 3, 'GBPJPY': 3, 'AUDJPY': 3, 'CHFJPY': 3, 'CADJPY': 3, 'XAUUSD': 3, 'XAGUSD': 3, 'USA500IDXUSD': 3, 'USATECHIDXUSD': 3, 'DEUIDXEUR': 3, 'GBRIDXGBP': 3, } def url_for(sym, when): return (f"{BASE}/{sym}/{when.year:04d}/{when.month - 1:02d}/{when.day:02d}/" f"{when.hour:02d}h_ticks.bi5") def cache_path(sym, when): return os.path.join(CACHE, sym, f"{when.year:04d}{when.month:02d}{when.day:02d}" f"{when.hour:02d}.bi5") def fetch_hour(sym, when, retries=4): """Bytes of one hour (possibly empty). Cached on disk; a cached hour is never refetched.""" cp = cache_path(sym, when) if os.path.exists(cp): with open(cp, 'rb') as f: return f.read() os.makedirs(os.path.dirname(cp), exist_ok=True) delay = 1.0 for attempt in range(retries): try: req = urllib.request.Request(url_for(sym, when), headers={'User-Agent': UA}) with urllib.request.urlopen(req, timeout=30) as r: body = r.read() with open(cp, 'wb') as f: f.write(body) return body except urllib.error.HTTPError as e: if e.code == 404: #--- 404 = no data for that hour. Cache the emptiness so a re-run does not #--- ask again; the feed has many such hours and they are not errors. with open(cp, 'wb') as f: f.write(b'') return b'' if attempt == retries - 1: raise except Exception: if attempt == retries - 1: raise time.sleep(delay) delay *= 2 return b'' def decode_hour(body, hour_start, scale): """bi5 bytes -> (time_ms, ask, bid, ask_vol, bid_vol). Empty body -> empty arrays.""" if not body: return (np.empty(0, np.int64),) + tuple(np.empty(0, np.float64) for _ in range(4)) try: raw = lzma.decompress(body, format=lzma.FORMAT_ALONE) except lzma.LZMAError: try: raw = lzma.decompress(body) except lzma.LZMAError: return (np.empty(0, np.int64),) + tuple(np.empty(0, np.float64) for _ in range(4)) n = len(raw) // 20 if n == 0: return (np.empty(0, np.int64),) + tuple(np.empty(0, np.float64) for _ in range(4)) a = np.frombuffer(raw[:n * 20], dtype='>u4').reshape(n, 5) ms = a[:, 0].astype(np.int64) ask = a[:, 1].astype(np.float64) / scale bid = a[:, 2].astype(np.float64) / scale vols = np.frombuffer(raw[:n * 20], dtype='>f4').reshape(n, 5) av = vols[:, 3].astype(np.float64) bv = vols[:, 4].astype(np.float64) base = int(hour_start.replace(tzinfo=dt.UTC).timestamp() * 1000) return base + ms, ask, bid, av, bv def hours(start, end): h = start while h < end: yield h h += dt.timedelta(hours=1) def download(sym, start, end, workers=WORKERS): """Fetch every hour in [start, end). Returns the hour list in order with their bytes.""" if sym not in DECIMALS: raise ValueError(f"{sym}: add its decimals to DECIMALS - guessing the scale is the " "one error that will not announce itself") hs = list(hours(start, end)) out = [None] * len(hs) done = [0] t0 = time.time() def job(i): out[i] = fetch_hour(sym, hs[i]) done[0] += 1 if done[0] % 500 == 0: el = time.time() - t0 print(f" {done[0]:,}/{len(hs):,} hours {done[0]/max(el,1e-9):.0f} h/s " f"{hs[i]:%Y-%m-%d}", flush=True) with ThreadPoolExecutor(max_workers=workers) as ex: list(ex.map(job, range(len(hs)))) return hs, out if __name__ == '__main__': #--- usage: python dukas.py EURUSD 2024-01-01 2024-02-01 sym = sys.argv[1] if len(sys.argv) > 1 else 'EURUSD' d0 = dt.datetime.fromisoformat(sys.argv[2]) if len(sys.argv) > 2 else dt.datetime(2024, 1, 1) d1 = dt.datetime.fromisoformat(sys.argv[3]) if len(sys.argv) > 3 else dt.datetime(2024, 1, 2) scale = 10.0 ** DECIMALS[sym] hs, bodies = download(sym, d0, d1) T, A, B, AV, BV = [], [], [], [], [] for h, body in zip(hs, bodies): t, a, b, av, bv = decode_hour(body, h, scale) if len(t): T.append(t); A.append(a); B.append(b); AV.append(av); BV.append(bv) if not T: print("no ticks in range"); sys.exit(0) t = np.concatenate(T); a = np.concatenate(A); b = np.concatenate(B) av = np.concatenate(AV); bv = np.concatenate(BV) U = lambda ms: dt.datetime.fromtimestamp(ms / 1000, dt.UTC) print(f"\n{sym} {len(t):,} ticks {U(t[0])} -> {U(t[-1])}") print(f" bid {b.min():.5f}..{b.max():.5f} spread median {np.median(a-b):.5f} " f"negative {int((a=0).all())}") print(" first 3:") for i in range(min(3, len(t))): print(f" {U(t[i])} bid {b[i]:.5f} ask {a[i]:.5f} bvol {bv[i]:.2f} avol {av[i]:.2f}")