wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-15 11:22:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

passing basic order and transaction simulation test

Browse Source
This commit is contained in:
fawce
2012-03-01 10:46:50 -05:00
parent 536a1e7fdc
commit 7846a1e1f4
4 changed files with 99 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/finance/trading.py
+61 -51
View File
@@ -1,5 +1,7 @@
import json
import datetime
import pytz
import math
from zmq.core.poll import select
@@ -35,25 +37,26 @@ class TradeSimulationClient(qmsg.Component):
self.signal_done()
return
event = zp.MERGE_UNFRAME(message)
event = zp.MERGE_UNFRAME(msg)
self._handle_event(event)
def connect_order(self):
return self.connect_push_socket(self.addresses['order_address'])
def _handle_event(self, event):
self.event_queue.append(event)
if event.SIM_DT <= event.dt:
#event occurred in the present, send the queue to be processed
self.handle_events(self.event_queue)
self.order_socket.send(str(zp.CONTROL_PROTOCOL.DONE))
self.handle_event(event)
#signal done to order source.
self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
def handle_events(self, event_queue):
def handle_event(self, event):
raise NotImplementedError
def order(self, sid, amount):
self.order_socket.send(zp.ORDER_FRAME(sid, amount))
def signal_order_done(self):
self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE))
class OrderDataSource(qmsg.DataSource):
"""DataSource that relays orders from the client"""
@@ -107,7 +110,11 @@ class OrderDataSource(qmsg.DataSource):
continue
order_msg = rlist[0].recv()
if order_msg == str(zp.CONTROL_PROTOCOL.DONE):
if order_msg == str(zp.ORDER_PROTOCOL.DONE):
self.signal_done()
return
if order_msg == str(zp.ORDER_PROTOCOL.BREAK):
qutil.LOGGER.info("order loop finished")
break
@@ -156,74 +163,77 @@ class TransactionSimulator(qmsg.BaseTransform):
"""
#TODO: need a way to send a placeholder txn, to avoid blocking merge... maybe customize merge to not block on txn?
if(event.type == zp.DATASOURCE_TYPE.ORDER):
self.add_open_order(event.sid, event.amount)
self.state['value'] = self.create_transaction(0, 0, 0.0, event.dt, 1)
self.add_open_order(event)
self.state['value'] = None
elif(event.type == zp.DATASOURCE_TYPE.TRADE):
txn = apply_trade_to_open_orders(event)
txn = self.apply_trade_to_open_orders(event)
self.state['value'] = txn
else:
self.state['value'] = None
qutil.LOGGER.info("unexpected event type in transform: {etype}".format(etype=event.type))
#TODO: what to do if we get another kind of datasource event.type?
return self.state
def add_open_order(self, sid, amount):
def add_open_order(self, event):
"""Orders are captured in a buffer by sid. No calculations are done here.
Amount is explicitly converted to an int.
Orders of amount zero are ignored.
"""
amount = int(amount)
if amount == 0:
qutil.LOGGER.debug("requested to trade zero shares of {sid}".format(sid=sid))
event.amount = int(event.amount)
if event.amount == 0:
qutil.LOGGER.debug("requested to trade zero shares of {sid}".format(sid=event.sid))
return
self.order_count += 1
order = zp.namedict({'sid' : sid,
'amount' : amount,
'dt' : self.algo_time,
'filled': 0,
'direction': math.fabs(amount) / amount
})
if(not self.open_orders.has_key(sid)):
self.open_orders[sid] = []
self.open_orders[sid].append(order)
if(not self.open_orders.has_key(event.sid)):
self.open_orders[event.sid] = []
self.open_orders[event.sid].append(event)
def apply_trade_to_open_orders(self, event):
if(event.volume == 0):
#there are zero volume events bc some stocks trade less frequently than once per minute.
return
return self.create_dummy_txn(event.dt)
if self.open_orders.has_key(event.sid):
orders = self.open_orders[event.sid]
remaining_orders = []
total_order = 0
dt = event.dt
for order in orders:
#we're using minute bars, so allow orders within 30 seconds of the trade
if((order.dt - event.dt) < self.trade_windwo):
total_order += order.amount
if(order.dt > dt):
dt = order.dt
#if the order still has time to live (TTL) keep track
elif((self.algo_time - order.dt) < self.orderTTL):
remaining_orders.append(order)
self.open_orders[event.sid] = remaining_orders
if(total_order != 0):
direction = total_order / math.fabs(total_order)
volume_share = (direction * total_order) / event.volume
if volume_share > .25:
volume_share = .25
amount = volume_share * event.volume * direction
impact = (volShare)**2 * .1 * direction * event.price
return self.create_transaction(event.sid, amount, event.price + impact, dt.replace(tzinfo = pytz.utc), direction)
else:
return None
remaining_orders = []
total_order = 0
dt = event.dt
for order in orders:
#we're using minute bars, so allow orders within 30 seconds of the trade
if((order.dt - event.dt) < self.trade_windwo):
total_order += order.amount
if(order.dt > dt):
dt = order.dt
#if the order still has time to live (TTL) keep track
elif((self.algo_time - order.dt) < self.orderTTL):
remaining_orders.append(order)
self.open_orders[event.sid] = remaining_orders
if(total_order != 0):
direction = total_order / math.fabs(total_order)
else:
direction = 1
volume_share = (direction * total_order) / event.volume
if volume_share > .25:
volume_share = .25
amount = volume_share * event.volume * direction
impact = (volume_share)**2 * .1 * direction * event.price
return self.create_transaction(event.sid, amount, event.price + impact, dt.replace(tzinfo = pytz.utc), direction)
def create_transaction(self, sid, amount, price, dt, direction):
txn = {'sid' : sid,
'amount' : amount,
'amount' : int(amount),
'dt' : dt,
'price' : price,
'commission' : self.commission * amount * direction,
zipline/protocol.py
+26 -8
View File
@@ -106,7 +106,7 @@ class namedict(object):
return "namedict: " + str(self.__dict__)
def __eq__(self, other):
return self.__dict__ == other.__dict__
return other != None and self.__dict__ == other.__dict__
def has_attr(self, name):
return self.__dict__.has_key(name)
@@ -263,13 +263,9 @@ def FEED_UNFRAME(msg):
INVALID_TRANSFORM_FRAME = FrameExceptionFactory('TRANSFORM')
def TRANSFORM_FRAME(name, value):
"""
:event: a nameddict with at least::
- source_id
- type
"""
assert isinstance(name, basestring)
assert value != None
if value == None:
return msgpack.dumps(tuple([name, TRANSFORM_TYPE.EMPTY]))
if(name == TRANSFORM_TYPE.TRANSACTION):
value = TRANSACTION_FRAME(value)
return msgpack.dumps(tuple([name, value]))
@@ -279,13 +275,17 @@ def TRANSFORM_UNFRAME(msg):
:rtype: namedict with <transform_name>:<transform_value>
"""
try:
name, value = msgpack.loads(msg)
if(value == TRANSFORM_TYPE.EMPTY):
return namedict({name : None})
#TODO: anything we can do to assert more about the content of the dict?
assert isinstance(name, basestring)
if(name == TRANSFORM_TYPE.PASSTHROUGH):
value = FEED_UNFRAME(value)
elif(name == TRANSFORM_TYPE.TRANSACTION):
value = TRANSACTION_UNFRAME(value)
return namedict({name : value})
except TypeError:
raise INVALID_TRANSFORM_FRAME(msg)
@@ -305,6 +305,11 @@ def MERGE_FRAME(event):
"""
assert isinstance(event, namedict)
PACK_DATE(event)
if(event.has_attr(TRANSFORM_TYPE.TRANSACTION)):
if(event.TRANSACTION == None):
event.TRANSACTION = TRANSFORM_TYPE.EMPTY
else:
event.TRANSACTION = TRANSACTION_FRAME(event.TRANSACTION)
payload = event.__dict__
return msgpack.dumps(payload)
@@ -314,6 +319,11 @@ def MERGE_UNFRAME(msg):
#TODO: anything we can do to assert more about the content of the dict?
assert isinstance(payload, dict)
payload = namedict(payload)
if(payload.has_attr(TRANSFORM_TYPE.TRANSACTION)):
if(payload.TRANSACTION == TRANSFORM_TYPE.EMPTY):
payload.TRANSACTION = None
else:
payload.TRANSACTION = TRANSACTION_UNFRAME(payload.TRANSACTION)
UNPACK_DATE(payload)
return payload
except TypeError:
@@ -476,9 +486,17 @@ DATASOURCE_TYPE = Enum(
'TRADE' ,
)
ORDER_PROTOCOL = Enum(
'DONE',
'BREAK'
)
#Transform type needs to be a namedict to facilitate merging.
TRANSFORM_TYPE = namedict({
'TRANSACTION':'TRANSACTION', #needed?
'PASSTHROUGH':'PASSTHROUGH'
'PASSTHROUGH':'PASSTHROUGH',
'EMPTY':''
})
zipline/test/client.py
+4 -1
View File
@@ -51,9 +51,12 @@ class TestTradingClient(TradeSimulationClient):
self.amount = amount
self.incr = 0
def handle_events(self, event_queue):
def handle_event(self, event):
#place an order for 100 shares of sid:133
if(self.incr < self.count):
self.order(self.sid, self.amount)
self.incr += 1
else:
self.signal_order_done()
self.signal_done()
zipline/test/test_finance.py
+8 -5
View File
@@ -127,11 +127,14 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase):
# Simulation Components
# ---------------------
set1 = SpecificEquityTrades("flat-133",factory.create_trade_history(133,
[10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0],
[100,100,100,100,100,100,100,100,100,100,100,100,100,100,100,100],
datetime.datetime.strptime("02/1/2012","%m/%d/%Y"),
datetime.timedelta(days=1)))
set1 = SpecificEquityTrades("flat-133",
factory.create_trade_history(
133,
[10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0],
[100,100,100,100,100,100,100,100,100,100,100,100,100,100,100,100],
datetime.datetime.strptime("02/1/2012","%m/%d/%Y"),
datetime.timedelta(days=1)))
#client sill send 10 orders for 100 shares of 133
client = TestTradingClient(133, 100, 10)
order_source = OrderDataSource(datetime.datetime.strptime("02/1/2012","%m/%d/%Y").replace(tzinfo=pytz.utc))