From 9a6a64ac8a025b42a04bfabbeafa924089381f9f Mon Sep 17 00:00:00 2001 From: Alvin Leong Date: Mon, 2 Dec 2024 13:55:59 +0800 Subject: [PATCH 1/3] Changed datafeed URL --- README.md | 3 +++ duka/core/fetch.py | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 3be844c..d007a15 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,8 @@ # duka - Dukascopy data downloader [![Build Status](https://travis-ci.org/giuse88/duka.svg?branch=master)](https://travis-ci.org/giuse88/duka) +*2024 Dec 2* +The datafeed URL has changed, I have updated it. + Finding good Forex data is difficult or expensive. Dukascopy has made available an excellent [web tool](https://www.dukascopy.com/swiss/english/marketwatch/historical/) to download tick data for a large a variety of Forex, CFD and commodities. This is awesome and extremely useful for people, like me, trying to study the Forex market. However, it takes a lot of time to download a large data set from the website because you can download only one day per time. In order to solve this issue, I created **duka**. diff --git a/duka/core/fetch.py b/duka/core/fetch.py index a10dbe3..c8e9a48 100644 --- a/duka/core/fetch.py +++ b/duka/core/fetch.py @@ -9,7 +9,7 @@ from ..core.utils import Logger, is_dst -URL = "https://www.dukascopy.com/datafeed/{currency}/{year}/{month:02d}/{day:02d}/{hour:02d}h_ticks.bi5" +URL = "https://datafeed.dukascopy.com/datafeed/{currency}/{year}/{month:02d}/{day:02d}/{hour:02d}h_ticks.bi5" ATTEMPTS = 5 From 304c1d79f3c4984a1768343e005140e933552e7d Mon Sep 17 00:00:00 2001 From: Alvin Leong Date: Wed, 29 Apr 2026 11:30:20 +0800 Subject: [PATCH 2/3] Throttle Dukascopy hourly fetches Limit per-day hourly request concurrency and add retry backoff so Colab runs are less likely to trigger transient Dukascopy 503 responses. Made-with: Cursor --- duka/core/fetch.py | 65 ++++++++++++++++++++++++++++++---------------- 1 file changed, 42 insertions(+), 23 deletions(-) diff --git a/duka/core/fetch.py b/duka/core/fetch.py index c8e9a48..b22e5bb 100644 --- a/duka/core/fetch.py +++ b/duka/core/fetch.py @@ -1,28 +1,44 @@ import asyncio -import datetime +import os import threading import time -from functools import reduce from io import BytesIO, DEFAULT_BUFFER_SIZE import requests -from ..core.utils import Logger, is_dst +from ..core.utils import Logger URL = "https://datafeed.dukascopy.com/datafeed/{currency}/{year}/{month:02d}/{day:02d}/{hour:02d}h_ticks.bi5" ATTEMPTS = 5 +REQUEST_TIMEOUT = 30 +DEFAULT_HOUR_CONCURRENCY = 3 +MAX_BACKOFF_SECONDS = 30 + + +def hour_concurrency(): + try: + return max(1, int(os.getenv('DUKA_HOUR_CONCURRENCY', DEFAULT_HOUR_CONCURRENCY))) + except ValueError: + return DEFAULT_HOUR_CONCURRENCY + + +def backoff_seconds(attempt): + return min(MAX_BACKOFF_SECONDS, 2 ** attempt) async def get(url): loop = asyncio.get_event_loop() - buffer = BytesIO() - id = url[35:].replace('/', " ") + id = url.split("/datafeed/", 1)[-1].replace('/', " ") start = time.time() Logger.info("Fetching {0}".format(id)) for i in range(ATTEMPTS): try: - res = await loop.run_in_executor(None, lambda: requests.get(url, stream=True)) + res = await loop.run_in_executor( + None, + lambda: requests.get(url, stream=True, timeout=REQUEST_TIMEOUT) + ) if res.status_code == 200: + buffer = BytesIO() for chunk in res.iter_content(DEFAULT_BUFFER_SIZE): buffer.write(chunk) Logger.info("Fetched {0} completed in {1}s".format(id, time.time() - start)) @@ -33,25 +49,30 @@ async def get(url): Logger.warn("Request to {0} failed with error code : {1} ".format(url, str(res.status_code))) except Exception as e: Logger.warn("Request {0} failed with exception : {1}".format(id, str(e))) - time.sleep(0.5 * i) - - raise Exception("Request failed for {0} after ATTEMPTS attempts".format(url)) + if i < ATTEMPTS - 1: + await asyncio.sleep(backoff_seconds(i)) -def create_tasks(symbol, day): - - start = 0 + raise Exception("Request failed for {0} after ATTEMPTS attempts".format(url)) - if is_dst(day): - start = 1 +async def fetch_hours(symbol, day): url_info = { 'currency': symbol, 'year': day.year, 'month': day.month - 1, 'day': day.day } - tasks = [asyncio.ensure_future(get(URL.format(**url_info, hour=i))) for i in range(0, 24)] + semaphore = asyncio.Semaphore(hour_concurrency()) + + async def fetch_hour(hour): + await semaphore.acquire() + try: + return await get(URL.format(**url_info, hour=hour)) + finally: + semaphore.release() + + tasks = [asyncio.ensure_future(fetch_hour(i)) for i in range(0, 24)] # if is_dst(day): # next_day = day + datetime.timedelta(days=1) @@ -62,7 +83,7 @@ def create_tasks(symbol, day): # 'day': next_day.day # } # tasks.append(asyncio.ensure_future(get(URL.format(**url_info, hour=0)))) - return tasks + return await asyncio.gather(*tasks) def fetch_day(symbol, day): @@ -70,11 +91,9 @@ def fetch_day(symbol, day): loop = getattr(local_data, 'loop', asyncio.new_event_loop()) asyncio.set_event_loop(loop) loop = asyncio.get_event_loop() - tasks = create_tasks(symbol, day) - loop.run_until_complete(asyncio.wait(tasks)) - - def add(acc, task): - acc.write(task.result()) - return acc + buffers = loop.run_until_complete(fetch_hours(symbol, day)) - return reduce(add, tasks, BytesIO()).getbuffer() + output = BytesIO() + for buffer in buffers: + output.write(buffer) + return output.getbuffer() From 963fb7d4629403067f64a026c194a180bda26318 Mon Sep 17 00:00:00 2001 From: Alvin Leong Date: Wed, 29 Apr 2026 11:34:53 +0800 Subject: [PATCH 3/3] Harden Dukascopy retries for Colab Use more conservative defaults and configurable timeout settings to reduce read timeouts from shared Colab networks. Made-with: Cursor --- duka/core/fetch.py | 44 ++++++++++++++++++++++++++++++++------------ 1 file changed, 32 insertions(+), 12 deletions(-) diff --git a/duka/core/fetch.py b/duka/core/fetch.py index b22e5bb..983d26a 100644 --- a/duka/core/fetch.py +++ b/duka/core/fetch.py @@ -1,5 +1,6 @@ import asyncio import os +import random import threading import time from io import BytesIO, DEFAULT_BUFFER_SIZE @@ -9,21 +10,34 @@ from ..core.utils import Logger URL = "https://datafeed.dukascopy.com/datafeed/{currency}/{year}/{month:02d}/{day:02d}/{hour:02d}h_ticks.bi5" -ATTEMPTS = 5 -REQUEST_TIMEOUT = 30 -DEFAULT_HOUR_CONCURRENCY = 3 -MAX_BACKOFF_SECONDS = 30 +DEFAULT_ATTEMPTS = 8 +DEFAULT_REQUEST_TIMEOUT = 120 +DEFAULT_HOUR_CONCURRENCY = 1 +MAX_BACKOFF_SECONDS = 60 +REQUEST_HEADERS = {'User-Agent': 'duka/0.2.1'} -def hour_concurrency(): +def env_int(name, default): try: - return max(1, int(os.getenv('DUKA_HOUR_CONCURRENCY', DEFAULT_HOUR_CONCURRENCY))) + return int(os.getenv(name, default)) except ValueError: - return DEFAULT_HOUR_CONCURRENCY + return default + + +def attempts(): + return max(1, env_int('DUKA_ATTEMPTS', DEFAULT_ATTEMPTS)) + + +def request_timeout(): + return max(1, env_int('DUKA_REQUEST_TIMEOUT', DEFAULT_REQUEST_TIMEOUT)) + + +def hour_concurrency(): + return max(1, env_int('DUKA_HOUR_CONCURRENCY', DEFAULT_HOUR_CONCURRENCY)) def backoff_seconds(attempt): - return min(MAX_BACKOFF_SECONDS, 2 ** attempt) + return min(MAX_BACKOFF_SECONDS, 2 ** attempt) + random.random() async def get(url): @@ -31,11 +45,17 @@ async def get(url): id = url.split("/datafeed/", 1)[-1].replace('/', " ") start = time.time() Logger.info("Fetching {0}".format(id)) - for i in range(ATTEMPTS): + total_attempts = attempts() + for i in range(total_attempts): try: res = await loop.run_in_executor( None, - lambda: requests.get(url, stream=True, timeout=REQUEST_TIMEOUT) + lambda: requests.get( + url, + headers=REQUEST_HEADERS, + stream=True, + timeout=request_timeout() + ) ) if res.status_code == 200: buffer = BytesIO() @@ -50,10 +70,10 @@ async def get(url): except Exception as e: Logger.warn("Request {0} failed with exception : {1}".format(id, str(e))) - if i < ATTEMPTS - 1: + if i < total_attempts - 1: await asyncio.sleep(backoff_seconds(i)) - raise Exception("Request failed for {0} after ATTEMPTS attempts".format(url)) + raise Exception("Request failed for {0} after {1} attempts".format(url, total_attempts)) async def fetch_hours(symbol, day):