From 8e74edb4f776ef9b95a1214112fd1c841e8d42d4 Mon Sep 17 00:00:00 2001 From: Michael Halls-Moore Date: Thu, 23 Apr 2015 12:45:40 +0100 Subject: [PATCH] Moved streaming.py into data directory. Modified how trading.py and backtest.py behave so that the price streaming is fixed. --- data/price.py | 128 +++++++++++++++++------------------------ data/streaming.py | 80 ++++++++++++++++++++++++++ execution/execution.py | 3 +- streaming/__init__.py | 0 streaming/streaming.py | 57 ------------------ trading/trading.py | 8 +-- 6 files changed, 139 insertions(+), 137 deletions(-) create mode 100644 data/streaming.py delete mode 100644 streaming/__init__.py delete mode 100644 streaming/streaming.py diff --git a/data/price.py b/data/price.py index 1be953e..06a203e 100644 --- a/data/price.py +++ b/data/price.py @@ -1,6 +1,5 @@ -from abc import ABCMeta, abstractmethod import datetime -from decimal import Decimal, getcontext, ROUND_HALF_DOWN +from decimal import Decimal, ROUND_HALF_DOWN import os import os.path import time @@ -26,15 +25,48 @@ class PriceHandler(object): backtesting suite. """ - __metaclass__ = ABCMeta + def _set_up_prices_dict(self): + """ + Due to the way that the Position object handles P&L + calculation, it is necessary to include values for not + only base/quote currencies but also their reciprocals. + This means that this class will contain keys for, e.g. + "GBPUSD" and "USDGBP". - @abstractmethod - def stream_to_queue(self): + At this stage they are calculated in an ad-hoc manner, + but a future TODO is to modify the following code to + be more robust and straightforward to follow. """ - Streams a sequence of tick data events (timestamp, bid, ask) - tuples to the events queue. + prices_dict = dict( + (k, v) for k,v in [ + (p, {"bid": None, "ask": None, "time": None}) for p in self.pairs + ] + ) + inv_prices_dict = dict( + (k, v) for k,v in [ + ( + "%s%s" % (p[3:], p[:3]), + {"bid": None, "ask": None, "time": None} + ) for p in self.pairs + ] + ) + prices_dict.update(inv_prices_dict) + return prices_dict + + def invert_prices(self, pair, bid, ask): """ - raise NotImplementedError("Should implement stream_to_queue()") + Simply inverts the prices for a particular currency pair. + This will turn the bid/ask of "GBPUSD" into bid/ask for + "USDGBP" and place them in the prices dictionary. + """ + inv_pair = "%s%s" % (pair[3:], pair[:3]) + inv_bid = (Decimal("1.0")/bid).quantize( + Decimal("0.00001", ROUND_HALF_DOWN) + ) + inv_ask = (Decimal("1.0")/ask).quantize( + Decimal("0.00001", ROUND_HALF_DOWN) + ) + return inv_pair, inv_bid, inv_ask class HistoricCSVPriceHandler(PriceHandler): @@ -65,34 +97,6 @@ class HistoricCSVPriceHandler(PriceHandler): self.pair_frames = {} self._open_convert_csv_files() - def _set_up_prices_dict(self): - """ - Due to the way that the Position object handles P&L - calculation, it is necessary to include values for not - only base/quote currencies but also their reciprocals. - This means that this class will contain keys for, e.g. - "GBPUSD" and "USDGBP". - - At this stage they are calculated in an ad-hoc manner, - but a future TODO is to modify the following code to - be more robust and straightforward to follow. - """ - prices_dict = dict( - (k, v) for k,v in [ - (p, {"bid": None, "ask": None, "time": None}) for p in self.pairs - ] - ) - inv_prices_dict = dict( - (k, v) for k,v in [ - ( - "%s%s" % (p[3:], p[:3]), - {"bid": None, "ask": None, "time": None} - ) for p in self.pairs - ] - ) - prices_dict.update(inv_prices_dict) - return prices_dict - def _open_convert_csv_files(self): """ Opens the CSV files from the data directory, converting @@ -112,24 +116,6 @@ class HistoricCSVPriceHandler(PriceHandler): self.pair_frames[p]["Pair"] = p self.all_pairs = pd.concat(self.pair_frames.values()).sort().iterrows() - def invert_prices(self, row): - """ - Simply inverts the prices for a particular currency pair. - This will turn the bid/ask of "GBPUSD" into bid/ask for - "USDGBP" and place them in the prices dictionary. - """ - pair = row["Pair"] - bid = row["Bid"] - ask = row["Ask"] - inv_pair = "%s%s" % (pair[3:], pair[:3]) - inv_bid = Decimal(str(1.0/bid)).quantize( - Decimal("0.00001", ROUND_HALF_DOWN) - ) - inv_ask = Decimal(str(1.0/ask)).quantize( - Decimal("0.00001", ROUND_HALF_DOWN) - ) - return inv_pair, inv_bid, inv_ask - def stream_next_tick(self): """ The Backtester has now moved over to a single-threaded @@ -147,33 +133,25 @@ class HistoricCSVPriceHandler(PriceHandler): except StopIteration: return else: - self.prices[row["Pair"]]["bid"] = Decimal(str(row["Bid"])).quantize( + pair = row["Pair"] + bid = Decimal(str(row["Bid"])).quantize( Decimal("0.00001", ROUND_HALF_DOWN) ) - self.prices[row["Pair"]]["ask"] = Decimal(str(row["Ask"])).quantize( + ask = Decimal(str(row["Ask"])).quantize( Decimal("0.00001", ROUND_HALF_DOWN) ) - self.prices[row["Pair"]]["time"] = index - inv_pair, inv_bid, inv_ask = self.invert_prices(row) - self.prices[inv_pair]["bid"] = inv_bid - self.prices[inv_pair]["ask"] = inv_ask - self.prices[inv_pair]["time"] = index - tev = TickEvent(row["Pair"], index, row["Bid"], row["Ask"]) - self.events_queue.put(tev) - def stream_to_queue(self): - self._open_convert_csv_files() - for index, row in self.all_pairs: - self.prices[row["Pair"]]["bid"] = Decimal(str(row["Bid"])).quantize( - Decimal("0.00001", ROUND_HALF_DOWN) - ) - self.prices[row["Pair"]]["ask"] = Decimal(str(row["Ask"])).quantize( - Decimal("0.00001", ROUND_HALF_DOWN) - ) - self.prices[row["Pair"]]["time"] = index - inv_pair, inv_bid, inv_ask = self.invert_prices(row) + # Create decimalised prices for traded pair + self.prices[pair]["bid"] = bid + self.prices[pair]["ask"] = ask + self.prices[pair]["time"] = index + + # Create decimalised prices for inverted pair + inv_pair, inv_bid, inv_ask = self.invert_prices(pair, bid, ask) self.prices[inv_pair]["bid"] = inv_bid self.prices[inv_pair]["ask"] = inv_ask self.prices[inv_pair]["time"] = index - tev = TickEvent(row["Pair"], index, row["Bid"], row["Ask"]) + + # Create the tick event for the queue + tev = TickEvent(pair, index, bid, ask) self.events_queue.put(tev) diff --git a/data/streaming.py b/data/streaming.py new file mode 100644 index 0000000..66d80eb --- /dev/null +++ b/data/streaming.py @@ -0,0 +1,80 @@ +from decimal import Decimal, ROUND_HALF_DOWN +import requests +import json + +from qsforex.event.event import TickEvent +from qsforex.data.price import PriceHandler + + +class StreamingForexPrices(PriceHandler): + def __init__( + self, domain, access_token, + account_id, pairs, events_queue + ): + self.domain = domain + self.access_token = access_token + self.account_id = account_id + self.events_queue = events_queue + self.pairs = pairs + self.prices = self._set_up_prices_dict() + + def invert_prices(self, pair, bid, ask): + """ + Simply inverts the prices for a particular currency pair. + This will turn the bid/ask of "GBPUSD" into bid/ask for + "USDGBP" and place them in the prices dictionary. + """ + inv_pair = "%s%s" % (pair[3:], pair[:3]) + inv_bid = (Decimal("1.0")/bid).quantize( + Decimal("0.00001", ROUND_HALF_DOWN) + ) + inv_ask = (Decimal("1.0")/ask).quantize( + Decimal("0.00001", ROUND_HALF_DOWN) + ) + return inv_pair, inv_bid, inv_ask + + def connect_to_stream(self): + pairs_oanda = ["%s_%s" % (p[:3], p[3:]) for p in self.pairs] + try: + s = requests.Session() + url = "https://" + self.domain + "/v1/prices" + headers = {'Authorization' : 'Bearer ' + self.access_token} + params = {'instruments' : pairs_oanda, 'accountId' : self.account_id} + req = requests.Request('GET', url, headers=headers, params=params) + pre = req.prepare() + resp = s.send(pre, stream=True, verify=False) + return resp + except Exception as e: + s.close() + print "Caught exception when connecting to stream\n" + str(e) + + def stream_to_queue(self): + response = self.connect_to_stream() + if response.status_code != 200: + return + for line in response.iter_lines(1): + if line: + try: + msg = json.loads(line) + except Exception as e: + print "Caught exception when converting message into json\n" + str(e) + return + if msg.has_key("instrument") or msg.has_key("tick"): + print msg + instrument = msg["tick"]["instrument"].replace("_", "") + time = msg["tick"]["time"] + bid = Decimal(str(msg["tick"]["bid"])).quantize( + Decimal("0.00001", ROUND_HALF_DOWN) + ) + ask = Decimal(str(msg["tick"]["ask"])).quantize( + Decimal("0.00001", ROUND_HALF_DOWN) + ) + self.prices[instrument]["bid"] = bid + self.prices[instrument]["ask"] = ask + # Invert the prices (GBP_USD -> USD_GBP) + inv_pair, inv_bid, inv_ask = self.invert_prices(instrument, bid, ask) + self.prices[inv_pair]["bid"] = inv_bid + self.prices[inv_pair]["ask"] = inv_ask + self.prices[inv_pair]["time"] = time + tev = TickEvent(instrument, time, bid, ask) + self.events_queue.put(tev) diff --git a/execution/execution.py b/execution/execution.py index 15dbe4b..6f5c217 100644 --- a/execution/execution.py +++ b/execution/execution.py @@ -42,12 +42,13 @@ class OANDAExecutionHandler(ExecutionHandler): return httplib.HTTPSConnection(self.domain) def execute_order(self, event): + instrument = "%s_%s" % (event.instrument[:3], event.instrument[3:]) headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization": "Bearer " + self.access_token } params = urllib.urlencode({ - "instrument" : event.instrument, + "instrument" : instrument, "units" : event.units, "type" : event.order_type, "side" : event.side diff --git a/streaming/__init__.py b/streaming/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/streaming/streaming.py b/streaming/streaming.py deleted file mode 100644 index eb95e66..0000000 --- a/streaming/streaming.py +++ /dev/null @@ -1,57 +0,0 @@ -import requests -import json - -from qsforex.event.event import TickEvent - - -class StreamingForexPrices(object): - def __init__( - self, domain, access_token, - account_id, instruments, events_queue - ): - self.domain = domain - self.access_token = access_token - self.account_id = account_id - self.instruments = instruments - self.events_queue = events_queue - self.cur_bid = None - self.cur_ask = None - - def connect_to_stream(self): - try: - s = requests.Session() - url = "https://" + self.domain + "/v1/prices" - headers = {'Authorization' : 'Bearer ' + self.access_token} - params = {'instruments' : self.instruments, 'accountId' : self.account_id} - req = requests.Request('GET', url, headers=headers, params=params) - pre = req.prepare() - resp = s.send(pre, stream=True, verify=False) - return resp - except Exception as e: - s.close() - print "Caught exception when connecting to stream\n" + str(e) - - def stream_to_queue(self): - response = self.connect_to_stream() - if response.status_code != 200: - return - for line in response.iter_lines(1): - if line: - try: - msg = json.loads(line) - except Exception as e: - print "Caught exception when converting message into json\n" + str(e) - return - if msg.has_key("instrument") or msg.has_key("tick"): - print msg - instrument = msg["tick"]["instrument"] - time = msg["tick"]["time"] - bid = msg["tick"]["bid"] - ask = msg["tick"]["ask"] - self.cur_bid = bid - self.cur_ask = ask - tev = TickEvent(instrument, time, bid, ask) - self.events_queue.put(tev) - - - diff --git a/trading/trading.py b/trading/trading.py index f38d4f6..f8cc90d 100644 --- a/trading/trading.py +++ b/trading/trading.py @@ -8,7 +8,7 @@ from qsforex.execution.execution import OANDAExecutionHandler from qsforex.portfolio.portfolio import Portfolio from qsforex import settings from qsforex.strategy.strategy import TestStrategy -from qsforex.streaming.streaming import StreamingForexPrices +from qsforex.data.streaming import StreamingForexPrices def trade(events, strategy, portfolio, execution, heartbeat): @@ -44,18 +44,18 @@ if __name__ == "__main__": equity = settings.EQUITY # Trade "Cable" - instrument = "GBP_USD" + pairs = ["GBPUSD"] # Create the OANDA market price streaming class # making sure to provide authentication commands prices = StreamingForexPrices( settings.STREAM_DOMAIN, settings.ACCESS_TOKEN, - settings.ACCOUNT_ID, instrument, events + settings.ACCOUNT_ID, pairs, events ) # Create the strategy/signal generator, passing the # instrument and the events queue - strategy = TestStrategy(instrument, events) + strategy = TestStrategy(pairs, events) # Create the portfolio object that will be used to # compare the OANDA positions with the local, to