diff --git a/zipline/component.py b/zipline/component.py index b644eb5a..7a8b2cc4 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -302,7 +302,6 @@ class Component(object): socket.gethostname() , os.getpid() , hex(id(self)) , - self.sockets , ) def __repr__(self): diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 2402e553..ee13d432 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -28,7 +28,7 @@ class TradeSimulationClient(qmsg.Component): (rlist, wlist, xlist) = select([self.result_feed], [], [self.result_feed], - timeout=self.heartbeat_timeout/1000) #select timeout is in sec + timeout=self.heartbeat_timeout/100) #select timeout is in sec, use 10x # #no more orders, should be an error condition if len(rlist) == 0 or len(xlist) > 0: @@ -46,7 +46,7 @@ class TradeSimulationClient(qmsg.Component): def _handle_event(self, event): self.event_queue.append(event) - if event.ALGO_TIME <= event.dt: + 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)) @@ -57,75 +57,45 @@ class TradeSimulationClient(qmsg.Component): def order(self, sid, amount): self.order_socket.send(zp.ORDER_FRAME(sid, amount)) - +class OrderDataSource(qmsg.DataSource): + """DataSource that relays orders from the client""" -class TradeSimulator(qmsg.BaseTransform): - - def __init__(self, expected_orders): - qmsg.BaseTransform.__init__(self, "") - self.open_orders = {} - self.algo_time = None - self.event_start = None - self.last_event_time = None - self.last_iteration_duration = None - self.expected_orders = expected_orders - self.order_count = 0 - self.trade_count = 0 + def __init__(self, simulation_dt): + """ + :param simulation_time: datetime in UTC timezone, sets the start time of simulation. orders + will be timestamped relative to this datetime. + event = { + 'sid' : an integer for security id, + 'dt' : datetime object, + 'price' : float for price, + 'volume' : integer for volume + } + """ + zm.DataSource.__init__(self, "ORDER_SIM") + self.simulation_dt = simulation_dt + self.last_iteration_duration = datetime.timedelta(seconds=0) + + def get_type(self): + return 'ORDER_SIM' - @property - def get_id(self): - return "ALGO_TIME" - def open(self): - qmsg.BaseTransform.open(self) + qmsg.DataSource.open(self) self.order_socket = self.bind_order() def bind_order(self): return self.bind_pull_socket(self.addresses['order_address']) def do_work(self): - """ - Pulls one message from the event feed, then - loops on orders until client sends DONE message. - """ - - #next feed event - (rlist, wlist, xlist) = select([self.feed_socket], - [], - [self.feed_socket], - timeout=self.heartbeat_timeout/1000) #select timeout is in sec - self.trade_count += 1 - #no more orders, should be an error condition - if len(rlist) == 0 or len(xlist) > 0: - raise Exception("unexpected end of feed stream") - message = rlist[0].recv() - if message == str(zp.CONTROL_PROTOCOL.DONE): - self.signal_done() - if(self.expected_orders > 0): - assert self.expected_orders == self.order_count - return #leave open orders hanging? client requests for orders? - - event = zp.FEED_UNFRAME(message) - - if self.last_iteration_duration != None: - self.algo_time = self.last_event_time + self.last_iteration_duration - else: - self.algo_time = event.dt #base case, first event we're transporting. - - self.last_event_time = event.dt - - if self.algo_time < self.last_event_time: - #compress time, move algo's clock to the time of this event - self.algo_time = self.last_event_time - - #self.process_orders(event) - #mark the start time for client's processing of this event. self.event_start = datetime.datetime.utcnow() - self.result_socket.send(zp.TRANSFORM_FRAME('ALGO_TIME', self.algo_time), self.zmq.NOBLOCK) + self.result_socket.send(zp.TRANSFORM_FRAME('ORDER_SIM', self.simulation_dt), self.zmq.NOBLOCK) - - while True: #this loop should also poll for portfolio state req/rep + self.simulation_dt = self.simulation_dt + self.last_iteration_duration + + #pull all orders from client. + orders = [] + order_dt = None + while True: (rlist, wlist, xlist) = select([self.order_socket], [], [self.order_socket], @@ -141,15 +111,112 @@ class TradeSimulator(qmsg.BaseTransform): break sid, amount = zp.ORDER_UNFRAME(order_msg) - self.add_open_order(sid, amount) + #send the order along - #end of order processing loop - self.last_iteration_duration = datetime.datetime.utcnow() - self.event_start + self.last_iteration_duration = datetime.datetime.utcnow() - self.event_start + dt = self.simulation_dt + self.last_iteration_duration + order_event = zp.namedict({"sid":sid, "amount":amount, "dt":dt, source_id=self.get_id}) + + message = zp.DATASOURCE_FRAME(event) + self.data_socket.send(message) + + + + +class TransactionSimulator(qmsg.BaseTransform): + + def __init__(self): + qmsg.BaseTransform.__init__(self, "TRANSACTION_SIM") + self.open_orders = {} + self.order_count = 0 + self.tradeWindow = datetime.timedelta(seconds=30) + self.orderTTL = datetime.timedelta(days=1) + self.volume_share = 0.05 + self.commission = 0.03 + + def transform(self, event): + """ + Pulls one message from the event feed, then + loops on orders until client sends DONE message. + """ + if(event.type == "ORDER_SIM"): + self.add_open_order(event.sid, event.amount) + self.state['value'] = self.average + elif(event.type == "EQUITY_TRADE"): + txn = apply_trade_to_open_orders(event) + self.state['value'] = txn + return self.state def add_open_order(self, sid, amount): - self.order_count = self.order_count + 1 + """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("{title}:{id} requested to trade zero shares of {sid}".format(sid=sid, + title=self.hostedAlgo.algo.title, + id=self.hostedAlgo.algo.id)) + return + + self.order_count += 1 + order = zp.namedict({'sid' : sid, + 'amount' : amount, + 'dt' : self.algo_time}, + 'filled': 0, + 'direction': math.fabs(amount) / amount) - def process_orders(self, event): - #TODO put real fill logic here, return a list of fills - return [{'sid':133, 'amount':-100}] \ No newline at end of file + if(not self.open_orders.has_key(sid)): + self.open_orders[sid] = [] + self.open_orders[sid].append(order) + + 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. + continue + 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.tradeWindow): + 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) + + + + def create_transaction(self, sid, amount, price, dt, direction): + if(amount != 0): + txn = {'sid' : sid, + 'amount' : amount, + 'dt' : dt, + 'price' : price, + 'back_test_run_id' : self.btRun.id, + 'transaction_cost' : -1*(price * amount), + 'commision' : self.commission * amount * direction} + + return namedict(txn) + + + + diff --git a/zipline/protocol.py b/zipline/protocol.py index 9367b703..dc3fb408 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -267,7 +267,7 @@ def TRANSFORM_FRAME(name, value): assert isinstance(name, basestring) assert value != None - if(name == 'ALGO_TIME'): + if(name == 'SIM_DT'): value = PACK_ALGO_DT(value) return msgpack.dumps(tuple([name, value])) @@ -282,7 +282,7 @@ def TRANSFORM_UNFRAME(msg): assert isinstance(name, basestring) if(name == "PASSTHROUGH"): value = FEED_UNFRAME(value) - elif(name == "ALGO_TIME"): + elif(name == "SIM_DT"): value = UNPACK_ALGO_DT(value) return namedict({name : value}) except TypeError: @@ -314,8 +314,8 @@ def MERGE_FRAME(event): assert isinstance(event, namedict) assert isinstance(event.dt, datetime.datetime) PACK_DATE(event) - if(event.has_attr('ALGO_TIME')): - event.ALGO_TIME = PACK_ALGO_DT(event.ALGO_TIME) + if(event.has_attr('SIM_DT')): + event.SIM_DT = PACK_ALGO_DT(event.SIM_DT) payload = event.__dict__ return msgpack.dumps(payload) @@ -325,8 +325,8 @@ 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('ALGO_TIME')): - payload.ALGO_TIME = UNPACK_ALGO_DT(payload.ALGO_TIME) + if(payload.has_attr('SIM_DT')): + payload.SIM_DT = UNPACK_ALGO_DT(payload.SIM_DT) assert isinstance(payload.epoch, numbers.Integral) assert isinstance(payload.micros, numbers.Integral) UNPACK_DATE(payload) @@ -340,7 +340,6 @@ def MERGE_UNFRAME(msg): # ================== # Finance Protocol # ================== - INVALID_ORDER_FRAME = FrameExceptionFactory('ORDER') INVALID_TRADE_FRAME = FrameExceptionFactory('TRADE') @@ -426,4 +425,11 @@ def UNPACK_DATE(payload): del(payload.__dict__['epoch']) del(payload.__dict__['micros']) payload['dt'] = dt - return payload \ No newline at end of file + return payload + + +FINANCE_PROTOCOL = Enum( + 'ORDER' , # 0 - req + 'TRANSACTION' , # 1 - req + 'STATUS' , # 2 - req + ) diff --git a/zipline/sources.py b/zipline/sources.py index ea2edf73..a0ba21e2 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -33,7 +33,7 @@ class RandomEquityTrades(TradeDataSource): def get_type(self): - return 'equity_trade' + return 'EQUITY_TRADE' def do_work(self): @@ -67,7 +67,7 @@ class SpecificEquityTrades(TradeDataSource): self.event_list = event_list def get_type(self): - return 'equity_trade' + return 'EQUITY_TRADE' def do_work(self): if(len(self.event_list) == 0): diff --git a/zipline/test/client.py b/zipline/test/client.py index 59e4d785..e9649977 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -33,6 +33,7 @@ class TestClient(qmsg.Component): self.signal_done() if(self.expected_msg_count > 0): assert self.received_count == self.expected_msg_count + return self.received_count += 1 diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index fb8b263c..32384c97 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -45,7 +45,7 @@ class SimulatorTestCase(ThreadPoolExecutorMixin): ret1 = RandomEquityTrades(133, "ret1", 1) ret2 = RandomEquityTrades(134, "ret2", 1) - client = TestClient(self, expected_msg_count=ret1.count + ret2.count) + client = TestClient(expected_msg_count=(ret1.count + ret2.count)) sim.register_controller( con ) sim.register_components([ret1, ret2, client]) @@ -91,7 +91,7 @@ class SimulatorTestCase(ThreadPoolExecutorMixin): ret1 = RandomEquityTrades(133, "ret1", 400) ret2 = RandomEquityTrades(134, "ret2", 400) - client = TestClient(self, expected_msg_count=ret1.count + ret2.count) + client = TestClient(expected_msg_count=ret1.count + ret2.count) sim.register_controller( con ) sim.register_components([ret1, ret2, client]) @@ -138,7 +138,7 @@ class SimulatorTestCase(ThreadPoolExecutorMixin): ret2 = RandomEquityTrades(134, "ret2", 5000) mavg1 = MovingAverage("mavg1", 30) mavg2 = MovingAverage("mavg2", 60) - client = TestClient(self, expected_msg_count=10000) + client = TestClient(expected_msg_count=10000) sim.register_components([ret1, ret2, mavg1, mavg2, client]) sim.register_controller( con ) @@ -168,7 +168,7 @@ class SimulatorTestCase(ThreadPoolExecutorMixin): mavg1 = MovingAverage("mavg1", 30) mavg2 = MovingAverage("mavg2", 60) transforms = {"mavg1":mavg1, "mavg2":mavg2} - client = TestClient(self, expected_msg_count=0) + client = TestClient(expected_msg_count=0) sim = self.get_simulator(sources, transforms, client) # TODO: way too long