From ddb1c5156946d51a1b5c283751c967cf60e825c9 Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 18 Apr 2012 23:14:21 -0400 Subject: [PATCH 1/7] moved epoch_now to date_utils --- zipline/date_utils.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/zipline/date_utils.py b/zipline/date_utils.py index 14a88571..85ff98d3 100644 --- a/zipline/date_utils.py +++ b/zipline/date_utils.py @@ -55,6 +55,10 @@ def UN_EPOCH(ms_since_epoch): def iso8061_to_epoch(datestring): dt = parse_iso8061(datestring) return EPOCH(dt) + +def epoch_now(): + dt = datetime.utcnow().replace(tzinfo=pytz.utc) + return EPOCH(dt) # UTC Datetime Subclasses # ----------------------- From 08a5bd5a0db230362d687e8cd5cea61bc27a9f43 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 19 Apr 2012 23:27:35 -0400 Subject: [PATCH 2/7] re-arranged fields so that history includes daily snapshots for cumulative measures. --- zipline/finance/performance.py | 111 ++++++++++++++++++++------------- zipline/finance/risk.py | 2 +- zipline/protocol.py | 14 +++-- 3 files changed, 79 insertions(+), 48 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 8fd3d3d6..5e9bc1b9 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -20,12 +20,6 @@ Performance Tracking +-----------------+----------------------------------------------------+ | started_at | datetime in utc marking the start of this test | +-----------------+----------------------------------------------------+ - | cumulative_capti| The net capital used (positive is spent) through | - | al_used | the course of all the events sent to this tracker | - +-----------------+----------------------------------------------------+ - | max_capital_used| The maximum amount of capital deployed through the | - | | course of all the events sent to this tracker | - +-----------------+----------------------------------------------------+ | last_close | The most recent close of the market. datetime in | | | pytz.utc timezone. Will always be 23:59 on the | | | date in UTC. The fact that the time may be on the | @@ -106,6 +100,14 @@ Performance Period | returns | percentage returns for the entire portfolio over the | | | period | +---------------+------------------------------------------------------+ + | cumulative_ | The net capital used (positive is spent) during | + | capital_used | the period | + +---------------+------------------------------------------------------+ + | max_capital_ | The maximum amount of capital deployed during the | + | used | period. | + +---------------+------------------------------------------------------+ + | max_leverage | The maximum leverage used during the period. | + +---------------+------------------------------------------------------+ """ @@ -136,10 +138,10 @@ class PerformanceTracker(): def __init__(self, trading_environment): - self.trading_environment = trading_environment - self.trading_day = datetime.timedelta(hours = 6, minutes = 30) - self.calendar_day = datetime.timedelta(hours = 24) - self.started_at = datetime.datetime.utcnow().replace(tzinfo=pytz.utc) + self.trading_environment = trading_environment + self.trading_day = datetime.timedelta(hours = 6, minutes = 30) + self.calendar_day = datetime.timedelta(hours = 24) + self.started_at = datetime.datetime.utcnow().replace(tzinfo=pytz.utc) self.period_start = self.trading_environment.period_start self.period_end = self.trading_environment.period_end @@ -164,7 +166,10 @@ class PerformanceTracker(): # initial portfolio positions have zero value 0, # initial cash is your capital base. - starting_cash = self.capital_base + self.capital_base, + # the cumulative period will be calculated over the entire test. + self.period_start, + self.period_end ) # this performance period will span just the current market day @@ -174,7 +179,10 @@ class PerformanceTracker(): # initial portfolio positions have zero value 0, # initial cash is your capital base. - starting_cash = self.capital_base, + self.capital_base, + # the daily period will be calculated for the market day + self.market_open, + self.market_close, # save the transactions for the daily periods keep_transactions = True ) @@ -206,10 +214,6 @@ class PerformanceTracker(): 'period_start' : self.period_start, 'period_end' : self.period_end, 'progress' : self.progress, - 'cumulative_capital_used' : self.cumulative_performance.cumulative_capital_used, - 'max_capital_used' : self.cumulative_performance.max_capital_used, - 'last_close' : self.market_close, - 'last_open' : self.market_open, 'capital_base' : self.capital_base, 'cumulative_perf' : self.cumulative_performance.to_dict(), 'daily_perf' : self.todays_performance.to_dict(), @@ -283,6 +287,8 @@ class PerformanceTracker(): self.todays_performance.positions, self.todays_performance.ending_value, self.todays_performance.ending_cash, + self.market_open, + self.market_close, keep_transactions = True ) @@ -369,20 +375,32 @@ class Position(): class PerformancePeriod(): - def __init__(self, initial_positions, starting_value, starting_cash, keep_transactions=False): - self.ending_value = 0.0 - self.period_capital_used = 0.0 - self.pnl = 0.0 + def __init__( + self, + initial_positions, + starting_value, + starting_cash, + period_open, + period_close, + keep_transactions=False): + + self.period_open = period_open + self.period_close = period_close + + self.ending_value = 0.0 + self.period_capital_used = 0.0 + self.pnl = 0.0 #sid => position object - self.positions = initial_positions - self.starting_value = starting_value + self.positions = initial_positions + self.starting_value = starting_value #cash balance at start of period - self.starting_cash = starting_cash - self.ending_cash = starting_cash - self.keep_transactions = keep_transactions - self.processed_transactions = [] + self.starting_cash = starting_cash + self.ending_cash = starting_cash + self.keep_transactions = keep_transactions + self.processed_transactions = [] self.cumulative_capital_used = 0.0 self.max_capital_used = 0.0 + self.max_leverage = 0.0 self.calculate_performance() @@ -457,16 +475,21 @@ class PerformancePeriod(): transactions = [x.as_dict() for x in self.processed_transactions] return { - 'ending_value' : self.ending_value, - 'capital_used' : self.period_capital_used, - 'starting_value' : self.starting_value, - 'starting_cash' : self.starting_cash, - 'ending_cash' : self.ending_cash, - 'portfolio_value': self.ending_cash + self.ending_value, - 'positions' : positions, - 'pnl' : self.pnl, - 'returns' : self.returns, - 'transactions' : transactions, + 'ending_value' : self.ending_value, + 'capital_used' : self.period_capital_used, + 'starting_value' : self.starting_value, + 'starting_cash' : self.starting_cash, + 'ending_cash' : self.ending_cash, + 'portfolio_value' : self.ending_cash + self.ending_value, + 'cumulative_capital_used' : self.cumulative_capital_used, + 'max_capital_used' : self.max_capital_used, + 'max_leverage' : self.max_leverage, + 'positions' : positions, + 'pnl' : self.pnl, + 'returns' : self.returns, + 'transactions' : transactions, + 'period_open' : self.period_open, + 'period_close' : self.period_close } def to_namedict(self): @@ -481,12 +504,16 @@ class PerformancePeriod(): positions = zp.namedict(positions) return zp.namedict({ - 'ending_value' : self.ending_value, - 'capital_used' : self.period_capital_used, - 'starting_value' : self.starting_value, - 'starting_cash' : self.starting_cash, - 'ending_cash' : self.ending_cash, - 'positions' : positions + 'ending_value' : self.ending_value, + 'capital_used' : self.period_capital_used, + 'starting_value' : self.starting_value, + 'starting_cash' : self.starting_cash, + 'ending_cash' : self.ending_cash, + 'cumulative_capital_used' : self.cumulative_capital_used, + 'max_capital_used' : self.max_capital_used, + 'max_leverage' : self.max_leverage, + 'positions' : positions, + 'transactions' : self.processed_transactions }) def get_positions(self, namedicted=False): diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index fc4bb076..77cc12ef 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -194,7 +194,7 @@ class RiskMetrics(): http://en.wikipedia.org/wiki/Sharpe_ratio """ if self.algorithm_volatility == 0: - return None + return 0.0 return ( (self.algorithm_period_returns - self.treasury_period_return) / self.algorithm_volatility ) diff --git a/zipline/protocol.py b/zipline/protocol.py index 471238a4..528766fd 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -628,8 +628,6 @@ def PERF_FRAME(perf): assert isinstance(perf['started_at'], datetime.datetime) assert isinstance(perf['period_start'], datetime.datetime) assert isinstance(perf['period_end'], datetime.datetime) - assert isinstance(perf['last_close'], datetime.datetime) - assert isinstance(perf['last_open'], datetime.datetime) assert isinstance(perf['daily_perf'], dict) assert isinstance(perf['cumulative_perf'], dict) @@ -641,16 +639,22 @@ def PERF_FRAME(perf): assert isinstance(cp['transactions'], list) assert isinstance(tp['positions'], list) assert isinstance(cp['positions'], list) + assert isinstance(tp['period_close'], datetime.datetime) + assert isinstance(tp['period_open'], datetime.datetime) + assert isinstance(cp['period_close'], datetime.datetime) + assert isinstance(cp['period_open'], datetime.datetime) perf['started_at'] = EPOCH(perf['started_at']) perf['period_start'] = EPOCH(perf['period_start']) perf['period_end'] = EPOCH(perf['period_end']) - perf['last_close'] = EPOCH(perf['last_close']) - perf['last_open'] = EPOCH(perf['last_open']) + tp['period_close'] = EPOCH(tp['period_close']) + tp['period_open'] = EPOCH(tp['period_open']) + cp['period_close'] = EPOCH(cp['period_close']) + cp['period_open'] = EPOCH(cp['period_open']) tp['transactions'] = convert_transactions(tp['transactions']) cp['transactions'] = convert_transactions(cp['transactions']) - + return BT_UPDATE_FRAME('PERF', perf) def convert_transactions(transactions): From 6f5e9b0559e1f17380d90af211cdb7b1cdefc6f3 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 19 Apr 2012 23:32:37 -0400 Subject: [PATCH 3/7] fixed documentation --- zipline/finance/performance.py | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 5e9bc1b9..770baa45 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -20,16 +20,6 @@ Performance Tracking +-----------------+----------------------------------------------------+ | started_at | datetime in utc marking the start of this test | +-----------------+----------------------------------------------------+ - | last_close | The most recent close of the market. datetime in | - | | pytz.utc timezone. Will always be 23:59 on the | - | | date in UTC. The fact that the time may be on the | - | | next day in the exchange's local time is ignored | - +-----------------+----------------------------------------------------+ - | last_open | The most recent open of the market. datetime in | - | | pytz.utc timezone. Will always be 00:00 on the | - | | date in UTC. The fact that the time may be on the | - | | next day in the exchange's local time is ignored | - +-----------------+----------------------------------------------------+ | capital_base | The initial capital assumed for this tracker. | +-----------------+----------------------------------------------------+ | cumulative_perf | A dictionary representing the cumulative | @@ -108,6 +98,13 @@ Performance Period +---------------+------------------------------------------------------+ | max_leverage | The maximum leverage used during the period. | +---------------+------------------------------------------------------+ + | period_close | The last close of the market in period. datetime in | + | | pytz.utc timezone. | + +---------------+------------------------------------------------------+ + | period_open | The first open of the market in period. datetime in | + | | pytz.utc timezone. | + +---------------+------------------------------------------------------+ + """ From d8c3ab151f88fa86da7d407b9e00b3593ace67cb Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 20 Apr 2012 12:16:42 -0400 Subject: [PATCH 4/7] trying to make jenks install iso8061 --- etc/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/etc/requirements.txt b/etc/requirements.txt index 679b1315..c7090540 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -4,4 +4,4 @@ gevent-zeromq==0.2.2 msgpack-python==0.1.12 humanhash==0.0.1 ujson==1.18 -iso8601==0.1.4 \ No newline at end of file +iso8601==0.1.4 From 28c380c245b5aa6048b7eb11b47b562885e8d7df Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 20 Apr 2012 00:58:24 -0400 Subject: [PATCH 5/7] removed heartbeat on every iteration, resulted in a 50% reduction in processing time per day. However, the big lesson from this experiment is that the major bottleneck is in the feedback loop. By eliminating the feedback loop and instead putting the transaction simulation into the client end of the zipline, I think we could accelerate by 10x. --- zipline/component.py | 12 ++++--- zipline/finance/trading.py | 64 ++++++++++++++++++++------------------ zipline/messaging.py | 21 +++++-------- 3 files changed, 47 insertions(+), 50 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index e4af443c..4e1d6b7f 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -86,6 +86,7 @@ class Component(object): self.start_tic = None self.stop_tic = None self.note = None + self.confirmed = False # Humanhashes make this way easier to debug because they # stick in your mind unlike a 32 byte string of random hex. @@ -235,12 +236,13 @@ class Component(object): """ Send a synchronization request to the host. """ + if not self.confirmed: + # TODO: proper framing + self.sync_socket.send(self.get_id + ":RUN") - # TODO: proper framing - self.sync_socket.send(self.get_id + ":RUN") - - self.receive_sync_ack() # blocking - + self.receive_sync_ack() # blocking + self.confirmed = True + def runtime(self): if self.ready() and self.start_tic and self.stop_tic: return self.stop_tic - self.start_tic diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 9a950e61..228572ec 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -38,7 +38,7 @@ class TradeSimulationClient(qmsg.Component): self.current_dt = trading_environment.period_start self.last_iteration_dur = datetime.timedelta(seconds=0) self.algorithm = None - self.max_wait = datetime.timedelta(seconds=7) + self.max_wait = datetime.timedelta(seconds=60) self.last_msg_dt = datetime.datetime.utcnow() assert self.trading_environment.frame_index != None @@ -207,7 +207,6 @@ class OrderDataSource(qmsg.DataSource): qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE) self.sent_count = 0 self.recv_count = Counter() - self.works = 0 @property def get_type(self): @@ -222,14 +221,14 @@ class OrderDataSource(qmsg.DataSource): def do_work(self): - self.works += 1 + self.recv_count['work_loops'] += 1 #pull all orders from client. count = 0 # one iteration of the client could include several orders # so iterate until the client signals a break or a close. - while True: + # while True: # poll all the sockets # we reduce the timeout here by a factor of 2, because we need # to potentially receive the client's done message before the @@ -237,34 +236,37 @@ class OrderDataSource(qmsg.DataSource): # this will block for timeout/2, and return an empty dict if there # are no messages. - socks = dict(self.poll.poll(self.heartbeat_timeout/2)) - - # see if the poller has results for the result_feed - if self.order_socket in socks and \ - socks[self.order_socket] == self.zmq.POLLIN: - - order_msg = self.order_socket.recv() - if order_msg == str(zp.ORDER_PROTOCOL.DONE): - qutil.LOGGER.info("order source is done") - self.signal_done() - self.recv_count['done'] += 1 - return - - if order_msg == str(zp.ORDER_PROTOCOL.BREAK): - # send a blank message to avoid an empty buffer - # in the feed - self.recv_count['break'] += 1 - if count == 0: - self.send(namedict({})) - break + socks = dict(self.poll.poll()) - order = zp.ORDER_UNFRAME(order_msg) - self.recv_count['order'] += 1 - #send the order along - self.send(order) - count += 1 - self.sent_count += 1 + # see if the poller has results for the result_feed + if self.order_socket in socks and \ + socks[self.order_socket] == self.zmq.POLLIN: + + order_msg = self.order_socket.recv() + + if order_msg == str(zp.ORDER_PROTOCOL.DONE): + qutil.LOGGER.info("order source is done") + self.signal_done() + self.recv_count['done'] += 1 + return + + if order_msg == str(zp.ORDER_PROTOCOL.BREAK): + # send a blank message to avoid an empty buffer + # in the feed + self.recv_count['break'] += 1 + if self.sent_count == 0: + self.send(namedict({})) + self.sent_count = 0 + return + + order = zp.ORDER_UNFRAME(order_msg) + self.recv_count['order'] += 1 + #send the order along + self.send(order) + count += 1 + self.sent_count += 1 + class TransactionSimulator(qmsg.BaseTransform): @@ -435,7 +437,7 @@ class TransactionSimulator(qmsg.BaseTransform): dt.replace(tzinfo = pytz.utc), direction ) - else: + elif len(orders) > 0: warning = """ Calculated a zero volume transaction on trade: {event} diff --git a/zipline/messaging.py b/zipline/messaging.py index 03764807..6cd4154d 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -110,7 +110,7 @@ class ComponentHost(Component): self.launch_component(component) self.launch_controller() - def is_timed_out(self): + def is_running(self): """ DEPRECATED, left in for compatability for now. """ @@ -119,23 +119,16 @@ class ComponentHost(Component): if len(self.components) == 0: qutil.LOGGER.info("Component register is empty.") - return True + return False - for source, last_dt in self.sync_register.iteritems(): - if (cur_time - last_dt) > self.timeout: - qutil.LOGGER.info( - "Time out for {source}. Current component registery: {reg}". - format(source=source, reg=self.components) - ) - return True - - return False + return True def loop(self, lockstep=True): - while not self.is_timed_out(): - # wait for synchronization request - socks = dict(self.sync_poller.poll(self.heartbeat_timeout)) #timeout after 2 seconds. + while self.is_running(): + # wait for synchronization request at start, and DONE at end. + # don't timeout. + socks = dict(self.sync_poller.poll()) if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: msg = self.sync_socket.recv() From bc14e7e3b72f484c206ada7024f73a308ccd3d2e Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 20 Apr 2012 11:45:57 -0400 Subject: [PATCH 6/7] zipline, now a cold, heartless, http://open.spotify.com/track/1xshgoh575otNXRfeYgh9D Pretty fast too... --- zipline/finance/performance.py | 4 +- zipline/finance/trading.py | 198 +++++++---------------------- zipline/lines.py | 16 ++- zipline/test/test_finance.py | 32 ++--- zipline/test/test_perf_tracking.py | 20 +-- 5 files changed, 70 insertions(+), 200 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 770baa45..4f6b3d43 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -377,8 +377,8 @@ class PerformancePeriod(): initial_positions, starting_value, starting_cash, - period_open, - period_close, + period_open=None, + period_close=None, keep_transactions=False): self.period_open = period_open diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 228572ec..9c815156 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -27,7 +27,7 @@ SIMULATION_STYLE = Enum( class TradeSimulationClient(qmsg.Component): - def __init__(self, trading_environment): + def __init__(self, trading_environment, sim_style): qmsg.Component.__init__(self) self.received_count = 0 self.prev_dt = None @@ -40,6 +40,7 @@ class TradeSimulationClient(qmsg.Component): self.algorithm = None self.max_wait = datetime.timedelta(seconds=60) self.last_msg_dt = datetime.datetime.utcnow() + self.txn_sim = TransactionSimulator(sim_style) assert self.trading_environment.frame_index != None self.event_frame = pandas.DataFrame( @@ -63,12 +64,8 @@ class TradeSimulationClient(qmsg.Component): def open(self): self.result_feed = self.connect_result() - self.order_socket = self.connect_order() - # send a wake up call to the order data source. - self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK)) def do_work(self): - # poll all the sockets socks = dict(self.poll.poll(self.heartbeat_timeout)) @@ -99,54 +96,49 @@ class TradeSimulationClient(qmsg.Component): # update performance and relay the event to the algorithm self.process_event(event) - # signal loop is done for order source. - self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK)) - else: - # no events in the sock means the non-order sources are - # drained. Signal the order_source that we're done, and - # the done will cascade through the whole zipline. - # shutdown the feedback loop to the OrderDataSource - wait_time = datetime.datetime.utcnow() - self.last_msg_dt - if wait_time > self.max_wait: - self.signal_order_done() - + def process_event(self, event): - # track the number of transactions, for testing purposes. - if(event.TRANSACTION != None): + + # generate transactions, if applicable + txn = self.txn_sim.apply_trade_to_open_orders(event) + if txn: + event.TRANSACTION = txn + # track the number of transactions, for testing purposes. self.txn_count += 1 + else: + event.TRANSACTION = None + + # the performance class needs to process each event, without + # skipping. Algorithm should wait until the performance has been + # updated, so that down stream components can safely assume that + # performance is up to date. Note that this is done before we + # mark the time for the algorithm's processing, thereby not + # running the algo's clock for performance book keeping. + self.perf.process_event(event) - #filter order flow out of the events sent to callbacks - if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE: - - # the performance class needs to process each event, without - # skipping. Algorithm should wait until the performance has been - # updated, so that down stream components can safely assume that - # performance is up to date. Note that this is done before we - # mark the time for the algorithm's processing, thereby not - # running the algo's clock for performance book keeping. - self.perf.process_event(event) - - # mark the start time for client's processing of this event. - event_start = datetime.datetime.utcnow() - - # queue the event. - self.queue_event(event) - - # if the event is later than our current time, run the algo - # otherwise, the algorithm has fallen behind the feed - # and processing per event is longer than time between events. - if event.dt >= self.current_dt: - # compress time by moving the current_time up to the event - # time. - self.current_dt = event.dt - self.run_algorithm() - - # tally the time spent on this iteration - self.last_iteration_dur = datetime.datetime.utcnow() - event_start - # move the algorithm's clock forward to include iteration time - self.current_dt = self.current_dt + self.last_iteration_dur + # mark the start time for client's processing of this event. + event_start = datetime.datetime.utcnow() + # queue the event. + self.queue_event(event) + + + # if the event is later than our current time, run the algo + # otherwise, the algorithm has fallen behind the feed + # and processing per event is longer than time between events. + if event.dt >= self.current_dt: + # compress time by moving the current_time up to the event + # time. + self.current_dt = event.dt + self.run_algorithm() + + # tally the time spent on this iteration + self.last_iteration_dur = datetime.datetime.utcnow() - event_start + # move the algorithm's clock forward to include iteration time + self.current_dt = self.current_dt + self.last_iteration_dur + + def run_algorithm(self): """ As per the algorithm protocol: @@ -164,15 +156,14 @@ class TradeSimulationClient(qmsg.Component): return self.connect_push_socket(self.addresses['order_address']) def order(self, sid, amount): - order = zp.namedict({ 'dt':self.current_dt, 'sid':sid, 'amount':amount }) - self.order_socket.send(zp.ORDER_FRAME(order)) self.order_count += 1 self.perf.log_order(order) + self.txn_sim.add_open_order(order) def signal_order_done(self): self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE)) @@ -188,91 +179,11 @@ class TradeSimulationClient(qmsg.Component): self.event_frame[event['sid']] = event self.event_queue = [] return self.event_frame - -class OrderDataSource(qmsg.DataSource): - """DataSource that relays orders from the client""" + - def __init__(self): - """ - :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 - } - """ - qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE) - self.sent_count = 0 - self.recv_count = Counter() - - @property - def get_type(self): - return zp.DATASOURCE_TYPE.ORDER - - def 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): - - self.recv_count['work_loops'] += 1 - - #pull all orders from client. - count = 0 - - # one iteration of the client could include several orders - # so iterate until the client signals a break or a close. - # while True: - # poll all the sockets - # we reduce the timeout here by a factor of 2, because we need - # to potentially receive the client's done message before the - # controller or heartbeat times out. - - # this will block for timeout/2, and return an empty dict if there - # are no messages. - - socks = dict(self.poll.poll()) - - # see if the poller has results for the result_feed - if self.order_socket in socks and \ - socks[self.order_socket] == self.zmq.POLLIN: - - order_msg = self.order_socket.recv() - - if order_msg == str(zp.ORDER_PROTOCOL.DONE): - qutil.LOGGER.info("order source is done") - self.signal_done() - self.recv_count['done'] += 1 - return - - if order_msg == str(zp.ORDER_PROTOCOL.BREAK): - # send a blank message to avoid an empty buffer - # in the feed - self.recv_count['break'] += 1 - if self.sent_count == 0: - self.send(namedict({})) - self.sent_count = 0 - return - - order = zp.ORDER_UNFRAME(order_msg) - self.recv_count['order'] += 1 - #send the order along - self.send(order) - count += 1 - self.sent_count += 1 - - - -class TransactionSimulator(qmsg.BaseTransform): +class TransactionSimulator(object): def __init__(self, style=SIMULATION_STYLE.PARTIAL_VOLUME): - qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION) self.open_orders = {} self.order_count = 0 self.txn_count = 0 @@ -289,27 +200,6 @@ class TransactionSimulator(qmsg.BaseTransform): elif style == SIMULATION_STYLE.NOOP: self.apply_trade_to_open_orders = self.simulate_noop - def transform(self, event): - """ - Pulls one message from the event feed, then - loops on orders until client sends DONE message. - """ - if(event.type == zp.DATASOURCE_TYPE.ORDER): - self.add_open_order(event) - self.state['value'] = None - elif(event.type == zp.DATASOURCE_TYPE.TRADE): - txn = self.apply_trade_to_open_orders(event) - self.state['value'] = txn - else: - self.state['value'] = None - log = "unexpected event type in transform: {etype}".format( - etype=event.type - ) - qutil.LOGGER.info(log) - - #TODO: what to do if we get another kind of datasource event.type? - return self.state - 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. @@ -324,8 +214,6 @@ class TransactionSimulator(qmsg.BaseTransform): ) qutil.LOGGER.debug(log) return - - if(not self.open_orders.has_key(event.sid)): self.open_orders[event.sid] = [] diff --git a/zipline/lines.py b/zipline/lines.py index 43c2eac0..56ec2b9a 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -86,8 +86,7 @@ import zipline.messaging as zmsg from zipline.test.algorithms import TestAlgorithm from zipline.sources import SpecificEquityTrades -from zipline.finance.trading import TransactionSimulator, OrderDataSource, \ -TradeSimulationClient +from zipline.finance.trading import TradeSimulationClient from zipline.simulator import AddressAllocator, Simulator from zipline.monitor import Controller from zipline.finance.trading import SIMULATION_STYLE @@ -164,18 +163,21 @@ class SimulatedTrading(object): self.sim = config['simulator_class'](addresses) self.clients = {} - self.trading_client = TradeSimulationClient(self.trading_environment) + self.trading_client = TradeSimulationClient( + self.trading_environment, + self.sim_style + ) self.add_client(self.trading_client) # setup all sources self.sources = {} - self.order_source = OrderDataSource() - self.add_source(self.order_source) + #self.order_source = OrderDataSource() + #self.add_source(self.order_source) #setup transforms - self.transaction_sim = TransactionSimulator(self.sim_style) + #self.transaction_sim = TransactionSimulator(self.sim_style) self.transforms = {} - self.add_transform(self.transaction_sim) + #self.add_transform(self.transaction_sim) self.sim.register_controller( self.con ) self.sim.on_done = self.shutdown() diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index a85f220d..763d8024 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -16,7 +16,7 @@ import zipline.finance.performance as perf from zipline.test.algorithms import TestAlgorithm from zipline.sources import SpecificEquityTrades -from zipline.finance.trading import TransactionSimulator, OrderDataSource, \ +from zipline.finance.trading import TransactionSimulator, \ TradeSimulationClient, TradingEnvironment from zipline.simulator import AddressAllocator, Simulator from zipline.monitor import Controller @@ -214,14 +214,8 @@ class FinanceTestCase(TestCase): zipline.algorithm.incr, "The test algorithm should send as many orders as specified.") - order_source = zipline.sources[zp.FINANCE_COMPONENT.ORDER_SOURCE] - self.assertEqual( - order_source.sent_count, - zipline.algorithm.count, - "The order source should have sent as many orders as the algo." - ) - transaction_sim = zipline.transforms[zp.TRANSFORM_TYPE.TRANSACTION] + transaction_sim = zipline.trading_client.txn_sim self.assertEqual( transaction_sim.txn_count, zipline.trading_client.perf.txn_count, @@ -426,11 +420,7 @@ class FinanceTestCase(TestCase): 'dt' : start_date + i * order_interval }) - sim_state = trade_sim.transform(order) - - # there should not be a new transaction from an order. - self.assertTrue(sim_state['name'] == trade_sim.get_id) - self.assertTrue(sim_state['value'] == None) + trade_sim.add_open_order(order) # there should now be one open order list stored under the sid oo = trade_sim.open_orders @@ -446,21 +436,19 @@ class FinanceTestCase(TestCase): tracker = PerformanceTracker(trading_environment) + # this approximates the loop inside TradingSimulationClient transactions = [] for trade in generated_trades: if trade_delay: trade.dt = trade.dt + trade_delay - sim_state = trade_sim.transform(trade) - - self.assertEqual(sim_state['name'], trade_sim.get_id) - - txn = None - if sim_state['value']: - txn = sim_state['value'] + txn = trade_sim.apply_trade_to_open_orders(trade) + if txn: transactions.append(txn) - trade[sim_state['name']] = txn - + trade.TRANSACTION = txn + else: + trade.TRANSACTION = None + tracker.process_event(trade) total_volume = 0 diff --git a/zipline/test/test_perf_tracking.py b/zipline/test/test_perf_tracking.py index be7c9a25..e952fed4 100644 --- a/zipline/test/test_perf_tracking.py +++ b/zipline/test/test_perf_tracking.py @@ -10,7 +10,8 @@ import zipline.util as qutil import zipline.finance.performance as perf import zipline.finance.risk as risk import zipline.protocol as zp -from zipline.finance.trading import TradeSimulationClient, TradingEnvironment +from zipline.finance.trading import TradeSimulationClient, TradingEnvironment, \ +SIMULATION_STYLE class PerformanceTestCase(unittest.TestCase): def setUp(self): @@ -539,11 +540,7 @@ shares in position" self.trading_environment.capital_base = 1000.0 self.trading_environment.frame_index = ['sid', 'volume', 'dt', \ 'price', 'changed'] - client = TradeSimulationClient(self.trading_environment) - # the client expects an algorithm that fullfills the algorithm - # protocol, so we use the noop algorithm. - test_algo = zipline.test.algorithms.NoopAlgorithm() - client.set_algorithm(test_algo) + perf_tracker = perf.PerformanceTracker(self.trading_environment) for event in trade_history: #create a transaction for all but @@ -559,18 +556,13 @@ shares in position" else: txn = None event[zp.TRANSFORM_TYPE.TRANSACTION] = txn - client.process_event(event) - - df = client.get_frame() - - self.assertEqual(df[133]['price'], price) - self.assertEqual(df[134]['price'], price2) + perf_tracker.process_event(event) #we skip two trades, to test case of None transaction txn_count = len(trade_history) - 2 - self.assertEqual(client.perf.txn_count, txn_count) + self.assertEqual(perf_tracker.txn_count, txn_count) - cumulative_pos = client.perf.cumulative_performance.positions[sid] + cumulative_pos = perf_tracker.cumulative_performance.positions[sid] expected_size = txn_count / 2 * -25 self.assertEqual(cumulative_pos.amount, expected_size) \ No newline at end of file From bab0e2bd197d77e2b606949e05777e6d24353042 Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 20 Apr 2012 15:08:39 -0400 Subject: [PATCH 7/7] made distinction between cumulative behavior and daily behavior a bit more clear. --- zipline/finance/performance.py | 15 ++++++++++----- zipline/finance/risk.py | 2 +- zipline/protocol.py | 9 +++++---- 3 files changed, 16 insertions(+), 10 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 4f6b3d43..351466a8 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -56,9 +56,6 @@ Position Tracking +-----------------+----------------------------------------------------+ | last_sale_price | price at last sale of the security on the exchange | +-----------------+----------------------------------------------------+ - | transactions | all the transactions that were acrued into this | - | | position. | - +-----------------+----------------------------------------------------+ Performance Period @@ -104,7 +101,9 @@ Performance Period | period_open | The first open of the market in period. datetime in | | | pytz.utc timezone. | +---------------+------------------------------------------------------+ - + | transactions | all the transactions that were acrued during this | + | | period. Unset/missing for cumulative periods. | + +---------------+------------------------------------------------------+ """ @@ -471,7 +470,7 @@ class PerformancePeriod(): positions = self.get_positions_list() transactions = [x.as_dict() for x in self.processed_transactions] - return { + rval = { 'ending_value' : self.ending_value, 'capital_used' : self.period_capital_used, 'starting_value' : self.starting_value, @@ -489,6 +488,12 @@ class PerformancePeriod(): 'period_close' : self.period_close } + # we want the key to be absent, not just empty + if not self.keep_transactions: + del(rval['transactions']) + + return rval + def to_namedict(self): """ Creates a namedict representing the state of this perfomance period. diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 77cc12ef..98f7b7e7 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -292,7 +292,7 @@ class RiskMetrics(): curve = None # in case end date is not a trading day, search for the next market # day for an interest rate - for i in range(7): + for i in xrange(7): if(self.treasury_curves.has_key(self.end_date + i * one_day)): curve = self.treasury_curves[self.end_date + i * one_day] break diff --git a/zipline/protocol.py b/zipline/protocol.py index 528766fd..0736d79c 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -636,7 +636,9 @@ def PERF_FRAME(perf): cp = perf['cumulative_perf'] assert isinstance(tp['transactions'], list) - assert isinstance(cp['transactions'], list) + # we never want to send transactions for the cumulative period. + # performance.py should never send them, but just to be safe: + assert not cp.has_key('transactions') assert isinstance(tp['positions'], list) assert isinstance(cp['positions'], list) assert isinstance(tp['period_close'], datetime.datetime) @@ -652,9 +654,8 @@ def PERF_FRAME(perf): cp['period_close'] = EPOCH(cp['period_close']) cp['period_open'] = EPOCH(cp['period_open']) - tp['transactions'] = convert_transactions(tp['transactions']) - cp['transactions'] = convert_transactions(cp['transactions']) - + tp['transactions'] = convert_transactions(tp['transactions']) + return BT_UPDATE_FRAME('PERF', perf) def convert_transactions(transactions):