Warrior_EA/research/dukas.py

"""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<b).sum())}")
    print(f"  ask vol median {np.median(av):.3f}   bid vol median {np.median(bv):.3f}   "
          f"zero-vol ticks {int(((av==0)&(bv==0)).sum()):,}")
    print(f"  time monotonic: {bool((np.diff(t)>=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}")