Merge pull request #14 from giuse88/0.2.0

0.2.0

duka/core/csv_dumper.py

@@ -1,14 +1,17 @@
 import csv
 import time
+from os.path import join
 
 from .candle import Candle
 from .utils import TimeFrame, stringify, Logger
 
-from os.path import join
-
 TEMPLATE_FILE_NAME = "{}-{}_{:02d}_{:02d}-{}_{:02d}_{:02d}.csv"
 
 
+def format_float(number):
+    return format(number, '.5f')
+
+
 class CSVFormatter(object):
     COLUMN_TIME = 0
     COLUMN_ASK = 1
@@ -20,19 +23,19 @@ class CSVFormatter(object):
 def write_tick(writer, tick):
     writer.writerow(
         {'time': tick[0],
-         'ask': tick[1],
-         'bid': tick[2],
+         'ask': format_float(tick[1]),
+         'bid': format_float(tick[2]),
          'ask_volume': tick[3],
          'bid_volume': tick[4]})
 
 
 def write_candle(writer, candle):
     writer.writerow(
         {'time': stringify(candle.timestamp),
-         'open': candle.open_price,
-         'close': candle.close_price,
-         'high': candle.high,
-         'low': candle.low})
+         'open': format_float(candle.open_price),
+         'close': format_float(candle.close_price),
+         'high': format_float(candle.high),
+         'low': format_float(candle.low)})
 
 
 class CSVDumper:
@@ -42,6 +45,7 @@ def __init__(self, symbol, timeframe, start, end, folder):
         self.start = start
         self.end = end
         self.folder = folder
+        self.include_header = True
         self.buffer = {}
 
     def get_header(self):
@@ -60,7 +64,10 @@ def append(self, day, ticks):
                 ts = time.mktime(tick[0].timetuple())
                 key = int(ts - (ts % self.timeframe))
                 if previous_key != key and previous_key is not None:
-                    self.buffer[day].append(Candle(self.symbol, previous_key, self.timeframe, current_ticks))
+                    n = int((key - previous_key) / self.timeframe)
+                    for i in range(0, n):
+                        self.buffer[day].append(
+                            Candle(self.symbol, previous_key + i * self.timeframe, self.timeframe, current_ticks))
                     current_ticks = []
                 current_ticks.append(tick[1])
                 previous_key = key
@@ -77,7 +84,8 @@ def dump(self):
 
         with open(join(self.folder, file_name), 'w') as csv_file:
             writer = csv.DictWriter(csv_file, fieldnames=self.get_header())
-            writer.writeheader()
+            if self.include_header:
+                writer.writeheader()
             for day in sorted(self.buffer.keys()):
                 for value in self.buffer[day]:
                     if self.timeframe == TimeFrame.TICK:

duka/core/fetch.py

@@ -1,16 +1,18 @@
 import asyncio
+import datetime
 import threading
 import time
 from functools import reduce
 from io import BytesIO, DEFAULT_BUFFER_SIZE
 
 import requests
 
-from ..core.utils import Logger
+from ..core.utils import Logger, is_dst
 
 URL = "https://www.dukascopy.com/datafeed/{currency}/{year}/{month:02d}/{day:02d}/{hour:02d}h_ticks.bi5"
 ATTEMPTS = 5
 
+
 async def get(url):
     loop = asyncio.get_event_loop()
     buffer = BytesIO()
@@ -24,30 +26,51 @@ async def get(url):
                 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))
+                if len(buffer.getbuffer()) <= 0:
+                    Logger.info("Buffer for {0} is empty ".format(id))
                 return buffer.getbuffer()
             else:
                 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)
+            time.sleep(0.5 * i)
 
     raise Exception("Request failed for {0} after ATTEMPTS attempts".format(url))
 
 
-def fetch_day(symbol, day):
-    local_data = threading.local()
-    loop = getattr(local_data, 'loop', asyncio.new_event_loop())
-    asyncio.set_event_loop(loop)
+def create_tasks(symbol, day):
+
+    start = 0
+
+    if is_dst(day):
+        start = 1
 
     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)]
 
+    # if is_dst(day):
+    #     next_day = day + datetime.timedelta(days=1)
+    #     url_info = {
+    #         'currency': symbol,
+    #         'year': next_day.year,
+    #         'month': next_day.month - 1,
+    #         'day': next_day.day
+    #     }
+    #     tasks.append(asyncio.ensure_future(get(URL.format(**url_info, hour=0))))
+    return tasks
+
+
+def fetch_day(symbol, day):
+    local_data = threading.local()
+    loop = getattr(local_data, 'loop', asyncio.new_event_loop())
+    asyncio.set_event_loop(loop)
     loop = asyncio.get_event_loop()
-    tasks = [asyncio.ensure_future(get(URL.format(**url_info, hour=i))) for i in range(24)]
+    tasks = create_tasks(symbol, day)
     loop.run_until_complete(asyncio.wait(tasks))
 
     def add(acc, task):

duka/core/processor.py

@@ -1,6 +1,7 @@
 import struct
 from datetime import timedelta, datetime
 from lzma import LZMADecompressor, LZMAError, FORMAT_AUTO
+from .utils import is_dst
 
 
 def decompress_lzma(data):
@@ -39,8 +40,11 @@ def add_hour(ticks):
 
     hour_delta = 0
 
-    if ticks[0][0].weekday() == 6:
-        hour_delta = 22
+    if ticks[0][0].weekday() == 6 or (ticks[0][0].day == 1 and ticks[0][0].month == 1):
+        if is_dst(ticks[0][0].date()):
+            hour_delta = 21
+        else:
+            hour_delta = 22
 
     for index, v in enumerate(ticks):
         if index != 0:
@@ -56,6 +60,7 @@ def add_hour(ticks):
 def normalize(day, ticks):
     def norm(time, ask, bid, volume_ask, volume_bid):
         date = datetime(day.year, day.month, day.day) + timedelta(milliseconds=time)
+        # date.replace(tzinfo=datetime.tzinfo("UTC"))
         return date, ask / 100000, bid / 100000, round(volume_ask * 1000000), round(volume_bid * 1000000)
 
     return add_hour(list(map(lambda x: norm(*x), ticks)))

duka/core/utils.py

@@ -4,10 +4,12 @@
 import signal
 import sys
 import time
-from datetime import datetime
+from datetime import datetime, timedelta, date
 
 TEMPLATE = '%(asctime)s - %(levelname)s - %(threadName)s [%(thread)d] -  %(message)s'
 
+SUNDAY = 7
+
 
 class TimeFrame(object):
     TICK = 0
@@ -85,3 +87,34 @@ def from_time_string(time_str):
 
 def stringify(timestamp):
     return str(datetime.fromtimestamp(timestamp))
+
+
+def find_sunday(year, month, position):
+    start = date(year, month, 1)
+    day_delta = timedelta(days=1)
+    counter = 0
+
+    while True:
+        if start.isoweekday() == SUNDAY:
+            counter += 1
+        if counter == position:
+            return start
+        start += day_delta
+
+
+def find_dst_begin(year):
+    """
+    DST starts the second sunday of March
+    """
+    return find_sunday(year, 3, 2)
+
+
+def find_dst_end(year):
+    """
+    DST ends the first sunday of November
+    """
+    return find_sunday(year, 11, 1)
+
+
+def is_dst(day):
+    return day >= find_dst_begin(day.year) and day < find_dst_end(day.year)

duka/main.py

@@ -7,9 +7,13 @@
 from duka.core import valid_date, set_up_signals
 from duka.core.utils import valid_timeframe, TimeFrame
 
+VERSION = '0.2.0'
+
 
 def main():
     parser = argparse.ArgumentParser(prog='duka', usage='%(prog)s [options]')
+    parser.add_argument('-v', '--version', action='version',
+                        version='Version: %(prog)s-{version}'.format(version=VERSION))
     parser.add_argument('symbols', metavar='SYMBOLS', type=str, nargs='+',
                         help='symbol list using format EURUSD EURGBP')
     parser.add_argument('-d', '--day', type=valid_date, help='specific day format YYYY-MM-DD (default today)',

duka/tests/test_find_sunday.py

@@ -0,0 +1,58 @@
+import datetime
+import unittest
+
+from duka.core.utils import find_sunday, find_dst_begin, find_dst_end, is_dst
+
+
+class TestFindSunday(unittest.TestCase):
+    def test_find_8_march_2015(self):
+        res = find_sunday(2015, 3, 2)
+        self.assertEqual(res.day, 8)
+
+    def test_find_9_march_2014(self):
+        res = find_sunday(2014, 3, 2)
+        self.assertEqual(res.day, 9)
+
+    def test_find_13_march_2016(self):
+        res = find_sunday(2016, 3, 2)
+        self.assertEqual(res.day, 13)
+
+    def test_find_1_november_2015(self):
+        res = find_sunday(2015, 11, 1)
+        self.assertEqual(res.day, 1)
+
+    def test_find_2_november_2014(self):
+        res = find_sunday(2014, 11, 1)
+        self.assertEqual(res.day, 2)
+
+    def test_find_6_november_2016(self):
+        res = find_sunday(2016, 11, 1)
+        self.assertEqual(res.day, 6)
+
+    def test_dst_2015(self):
+        start = find_dst_begin(2015)
+        end = find_dst_end(2015)
+        self.assertEqual(start.day, 8)
+        self.assertEqual(start.month, 3)
+        self.assertEqual(end.day, 1)
+        self.assertEqual(end.month, 11)
+
+    def test_is_dst(self):
+        day = datetime.datetime(2015, 4, 5)
+        self.assertTrue(is_dst(day))
+
+    def test_is_not_dst(self):
+        day = datetime.datetime(2015, 1, 1)
+        self.assertFalse(is_dst(day))
+
+    def test_day_change_is_dst(self):
+        day = datetime.datetime(2015, 3, 8)
+        self.assertTrue(is_dst(day))
+
+    def test_day_change_back_is_not_dst(self):
+        day = datetime.datetime(2015, 11, 1)
+        self.assertFalse(is_dst(day))
+
+    def test_is_dst(self):
+        day = datetime.datetime(2013, 11, 3)
+        self.assertFalse(is_dst(day))

setup.py

@@ -3,7 +3,7 @@
 from setuptools import setup, find_packages
 
 NAME = "duka"
-VERSION = '0.1.6'
+VERSION = '0.2.0'
 
 setup(
     name=NAME,