gavindiaz/qsforex
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved streaming.py into data directory. Modified how trading.py and backtest.py behave so that the price streaming is fixed.

Browse Source
This commit is contained in:
Michael Halls-Moore
2015-04-23 12:45:40 +01:00
parent e84512e1e7
commit 8e74edb4f7
6 changed files with 139 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/price.py
+53 -75
View File
@@ -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)
data/streaming.py
+80
View File
@@ -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)
execution/execution.py
+2 -1
View File
@@ -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
streaming/__init__.py
View File
streaming/streaming.py
-57
View File
@@ -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)
trading/trading.py
+4 -4
View File
@@ -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