diff --git a/zipline/component.py b/zipline/component.py index 4595c192..db9b7bc8 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -14,7 +14,8 @@ import humanhash from datetime import datetime import zipline.util as qutil -from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \ + COMPONENT_FAILURE, BACKTEST_STATE class Component(object): """ @@ -66,6 +67,7 @@ class Component(object): self.controller = None self.heartbeat_timeout = 2000 self.state_flag = COMPONENT_STATE.OK + self.error_state = COMPONENT_FAILURE.NOFAILURE self.on_done = None self._exception = None @@ -254,8 +256,17 @@ class Component(object): # Internal Maintenance # ---------------------- - def signal_exception(self, exc=None): + def signal_exception(self, exc=None, scope=None): + + if scope == 'algo': + self.error_state = COMPONENT_FAILURE.ALGOEXCEPT + else: + self.error_state = COMPONENT_FAILURE.HOSTEXCEPT + self.state_flag = COMPONENT_STATE.EXCEPTION + # mark the time of failure so we can track the failure + # progogation through the system. + self.stop_tic = time.time() self._exception = exc diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 0f5b7b56..ada03fcd 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -1,58 +1,15 @@ -import datetime -import pytz -import math -import pandas +""" -from zmq.core.poll import select +Performance Tracking +==================== -import zipline.messaging as qmsg -import zipline.util as qutil -import zipline.protocol as zp -import zipline.finance.risk as risk - -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.period_start = self.trading_environment.period_start - self.period_end = self.trading_environment.period_end - self.market_open = self.period_start - self.market_close = self.market_open + self.trading_day - self.progress = 0.0 - self.total_days = (self.period_end - self.period_start).days - self.day_count = 0 - self.cumulative_capital_used= 0.0 - self.max_capital_used = 0.0 - self.capital_base = self.trading_environment.capital_base - self.returns = [] - self.txn_count = 0 - self.event_count = 0 - self.cumulative_performance = PerformancePeriod( - {}, - self.capital_base, - starting_cash = self.capital_base - ) - - self.todays_performance = PerformancePeriod( - {}, - self.capital_base, - starting_cash = self.capital_base - ) - - def to_dict(self): - """ - Creates a dictionary representing the state of this tracker. - Returns a dict object of the form: - +-----------------+----------------------------------------------------+ | key | value | +=================+====================================================+ | period_start | The beginning of the period to be tracked. datetime| | | in pytz.utc timezone. Will always be 0:00 on the | | | date in UTC. The fact that the time may be on the | - | | prior day in the exchange's local time is ignored | + | | prior day in the exchange's local time is ignored | +-----------------+----------------------------------------------------+ | period_end | The end of the period to be tracked. datetime | | | in pytz.utc timezone. Will always be 23:59 on the | @@ -62,7 +19,7 @@ class PerformanceTracker(): | progress | percentage of test completed | +-----------------+----------------------------------------------------+ | cumulative_capti| The net capital used (positive is spent) through | - | al_used | the course of all the events sent to this tracker | + | 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 | @@ -97,160 +54,18 @@ class PerformanceTracker(): | | overkill. | +-----------------+----------------------------------------------------+ | cumulative_risk | A dictionary representing the risk metrics | - | _metrics | calculated based on the positions aggregated | + | _metrics | calculated based on the positions aggregated | | | through all the events delivered to this tracker. | | | For details look at the comments for | | | :py:meth:`zipline.finance.risk.RiskMetrics.to_dict`| +-----------------+----------------------------------------------------+ - - - - """ - returns_list = [x.to_dict() for x in self.returns] - d = { - 'period_start' : self.period_start, - 'period_end' : self.period_end, - 'progress' : self.progress, - 'cumulative_captial_used' : self.cumulative_captial_used, - 'max_capital_used' : self.max_capital_used, - 'last_close' : self.market_close, - 'last_open' : self.market_open, - 'capital_base' : self.capital_base, - 'returns' : returns_list, - 'cumulative_perf' : self.cumulative_perf.to_dict(), - 'todays_perf' : self.todays_perf.to_dict(), - 'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict() - } - - def process_event(self, event): - self.event_count += 1 - if(event.dt >= self.market_close): - self.handle_market_close() - - if not pandas.isnull(event.TRANSACTION): - self.txn_count += 1 - self.cumulative_performance.execute_transaction(event.TRANSACTION) - self.todays_performance.execute_transaction(event.TRANSACTION) - - # we're adding a 10% cushion to the capital used, - # and then rounding to the nearest 5k - transaction_cost = event.TRANSACTION.price * event.TRANSACTION.amount - self.cumulative_capital_used += transaction_cost - if(math.fabs(self.cumulative_capital_used) > self.max_capital_used): - self.max_capital_used = math.fabs(self.cumulative_capital_used) - - cushioned_capital = 1.1 * self.max_capital_used - self.max_capital_used = self.round_to_nearest( - cushioned_capital, - base=5000 - ) - self.max_leverage = self.max_capital_used/self.capital_base - - #update last sale - self.cumulative_performance.update_last_sale(event) - self.todays_performance.update_last_sale(event) - - #calculate performance as of last trade - self.cumulative_performance.calculate_performance() - self.todays_performance.calculate_performance() - - def handle_market_close(self): - #add the return results from today to the list of DailyReturn objects. - todays_date = self.market_close.replace(hour=0, minute=0, second=0) - todays_return_obj = risk.DailyReturn( - todays_date, - self.todays_performance.returns - ) - self.returns.append(todays_return_obj) - - #calculate risk metrics for cumulative performance - self.cumulative_risk_metrics = risk.RiskMetrics( - start_date=self.period_start, - end_date=self.market_close.replace(hour=0, minute=0, second=0), - returns=self.returns, - trading_environment=self.trading_environment - ) - - #move the market day markers forward - self.market_open = self.market_open + self.calendar_day - while not self.trading_environment.is_trading_day(self.market_open): - if self.market_open > self.trading_environment.trading_days[-1]: - raise Exception("Attempt to backtest beyond available history.") - self.market_open = self.market_open + self.calendar_day - self.market_close = self.market_open + self.trading_day - self.day_count += 1.0 - - #calculate progress of test - self.progress = self.day_count / self.total_days - - #################################################################### - #######TODO: relay the results of self.to_dict() ########### - #################################################################### - - #roll over positions to current day. - self.todays_performance.calculate_performance() - self.todays_performance = PerformancePeriod( - self.todays_performance.positions, - self.todays_performance.ending_value, - self.todays_performance.ending_cash - ) + | timestamp | System time evevent occurs in zipilne | + +-----------------+----------------------------------------------------+ - def handle_simulation_end(self): - self.risk_report = risk.RiskReport( - self.returns, - self.trading_environment - ) - - #################################################################### - #######TODO: relay the results of self.risk_report.to_dict() ####### - #################################################################### - - def round_to_nearest(self, x, base=5): - return int(base * round(float(x)/base)) -class Position(): - - def __init__(self, sid): - self.sid = sid - self.amount = 0 - self.cost_basis = 0.0 ##per share - self.last_sale_price = None - self.last_sale_date = None - - def update(self, txn): - if(self.sid != txn.sid): - raise NameError('updating position with txn for a different sid') - #throw exception - - if(self.amount + txn.amount == 0): #we're covering a short or closing a position - self.cost_basis = 0.0 - self.amount = 0 - else: - prev_cost = self.cost_basis*self.amount - txn_cost = txn.amount*txn.price - total_cost = prev_cost + txn_cost - total_shares = self.amount + txn.amount - self.cost_basis = total_cost/total_shares - self.amount = self.amount + txn.amount - - def currentValue(self): - return self.amount * self.last_sale_price - - - def __repr__(self): - template = "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, \ - last_sale_price: {last_sale_price}" - return template.format( - sid=self.sid, - amount=self.amount, - cost_basis=self.cost_basis, - last_sale_price=self.last_sale_price - ) - - def to_dict(self): - """ - Creates a dictionary representing the state of this position. - Returns a dict object of the form: +Position Tracking +================= + +-----------------+----------------------------------------------------+ | key | value | +=================+====================================================+ @@ -264,97 +79,337 @@ class Position(): | last_sale_date | datetime of the last trade of the position's | | | security on the exchange | +-----------------+----------------------------------------------------+ + | timestamp | System time evevent occurs in zipilne | + +-----------------+----------------------------------------------------+ + +Performance Period +================== + + +---------------+------------------------------------------------------+ + | key | value | + +===============+======================================================+ + | ending_value | the total market value of the positions held at the | + | | end of the period | + +---------------+------------------------------------------------------+ + | capital_used | the net capital consumed (positive means spent) by | + | | buying and selling securities in the period | + +---------------+------------------------------------------------------+ + | starting_value| the total market value of the positions held at the | + | | start of the period | + +---------------+------------------------------------------------------+ + | starting_cash | cash on hand at the beginning of the period | + +---------------+------------------------------------------------------+ + | ending_cash | cash on hand at the end of the period | + +---------------+------------------------------------------------------+ + | positions | a list of dicts representing positions, see | + | | :py:meth:`Position.to_dict()` | + | | for details on the contents of the dict | + +---------------+------------------------------------------------------+ + | timestamp | System time evevent occurs in zipilne | + +---------------+------------------------------------------------------+ + +""" +import datetime +import msgpack +import pandas +import math + +import zmq + +import zipline.util as qutil +import zipline.protocol as zp +import zipline.finance.risk as risk + +class PerformanceTracker(): + """ + + Tracks the performance of the zipstream as it is running in + the simulotr, relays this out to the Deluge broker and then + to the client. + + +--------------------+ Result Stream +--------+ + | PerformanceTracker | ----------------> | Deluge | + +--------------------+ +--------+ + + """ + + def __init__(self, period_start, period_end, capital_base, trading_environment): + + self.trading_day = datetime.timedelta(hours = 6, minutes = 30) + self.calendar_day = datetime.timedelta(hours = 24) + + self.period_start = period_start + self.period_end = period_end + self.market_open = self.period_start + self.market_close = self.market_open + self.trading_day + self.progress = 0.0 + self.total_days = (self.period_end - self.period_start).days + self.day_count = 0 + self.cumulative_capital_used = 0.0 + self.max_capital_used = 0.0 + self.capital_base = capital_base + self.trading_environment = trading_environment + self.returns = [] + self.txn_count = 0 + self.event_count = 0 + self.result_stream = None + + self.cumulative_performance = PerformancePeriod( + {}, + capital_base, + starting_cash = capital_base + ) + + self.todays_performance = PerformancePeriod( + {}, + capital_base, + starting_cash = capital_base + ) + + + def publish_to(self, zmq_socket, context=None): """ - state = { - 'sid':self.sid, - 'amount':self.amount, - 'cost_basis':self.cost_basis, - 'last_sale_price':self.last_sale_price, - 'last_sale_date':self.last_sale_date + Publish the performance results asynchronously to a + socket. + """ + ctx = context or zmq.Context.instance() + sock = ctx.socket(zmq.PUSH) + sock.connect(zmq_socket) + + self.result_stream = sock + + def to_dict(self): + """ + Creates a dictionary representing the state of this tracker. + Returns a dict object of the form: + """ + + returns_list = [x.to_dict() for x in self.returns] + + return { + 'period_start' : self.period_start, + 'period_end' : self.period_end, + 'progress' : self.progress, + 'cumulative_captial_used' : self.cumulative_capital_used, + 'max_capital_used' : self.max_capital_used, + 'last_close' : self.market_close, + 'last_open' : self.market_open, + 'capital_base' : self.capital_base, + 'returns' : returns_list, + 'cumulative_perf' : self.cumulative_performance.to_dict(), + 'todays_perf' : self.todays_performance.to_dict(), + 'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict(), + 'timestamp' : datetime.datetime.now(), } - return state - + + def process_event(self, event): + self.event_count += 1 + + if(event.dt >= self.market_close): + self.handle_market_close() + + if not pandas.isnull(event.TRANSACTION): + self.txn_count += 1 + self.cumulative_performance.execute_transaction(event.TRANSACTION) + self.todays_performance.execute_transaction(event.TRANSACTION) + + # we're adding a 10% cushion to the capital used, + # and then rounding to the nearest 5k + transaction_cost = event.TRANSACTION.price * event.TRANSACTION.amount + self.cumulative_capital_used += transaction_cost + + if math.fabs(self.cumulative_capital_used) > self.max_capital_used: + self.max_capital_used = math.fabs(self.cumulative_capital_used) + + cushioned_capital = 1.1 * self.max_capital_used + self.max_capital_used = self.round_to_nearest( + cushioned_capital, + base=5000 + ) + self.max_leverage = self.max_capital_used / self.capital_base + + #update last sale + self.cumulative_performance.update_last_sale(event) + self.todays_performance.update_last_sale(event) + + #calculate performance as of last trade + self.cumulative_performance.calculate_performance() + self.todays_performance.calculate_performance() + + def handle_market_close(self): + #add the return results from today to the list of DailyReturn objects. + todays_date = self.market_close.replace(hour=0, minute=0, second=0) + todays_return_obj = risk.DailyReturn( + todays_date, + self.todays_performance.returns + ) + self.returns.append(todays_return_obj) + + #calculate risk metrics for cumulative performance + self.cumulative_risk_metrics = risk.RiskMetrics( + start_date=self.period_start, + end_date=self.market_close.replace(hour=0, minute=0, second=0), + returns=self.returns, + trading_environment=self.trading_environment + ) + + #move the market day markers forward + self.market_open = self.market_open + self.calendar_day + + while not self.trading_environment.is_trading_day(self.market_open): + if self.market_open > self.trading_environment.trading_days[-1]: + raise Exception("Attempt to backtest beyond available history.") + self.market_open = self.market_open + self.calendar_day + + self.market_close = self.market_open + self.trading_day + self.day_count += 1.0 + + #calculate progress of test + self.progress = self.day_count / self.total_days + + # Output Results + if self.result_stream: + # TODO: proper framing + self.result_stream.send_pyobj(self.to_dict()) + + #roll over positions to current day. + self.todays_performance.calculate_performance() + self.todays_performance = PerformancePeriod( + self.todays_performance.positions, + self.todays_performance.ending_value, + self.todays_performance.ending_cash + ) + + def handle_simulation_end(self): + assert False + + self.risk_report = risk.RiskReport( + self.returns, + self.trading_environment + ) + + # Output Results + if self.result_stream: + # TODO: proper framing + self.result_stream.send_pyobj(self.risk_report.to_dict()) + + self.result_stream.send_pyobj(None) + + def round_to_nearest(self, x, base=5): + return int(base * round(float(x)/base)) + + +class Position(): + + def __init__(self, sid): + self.sid = sid + self.amount = 0 + self.cost_basis = 0.0 ##per share + self.last_sale_price = None + self.last_sale_date = None + + def update(self, txn): + if(self.sid != txn.sid): + raise NameError('updating position with txn for a different sid') + + #we're covering a short or closing a position + if(self.amount + txn.amount == 0): + self.cost_basis = 0.0 + self.amount = 0 + else: + prev_cost = self.cost_basis*self.amount + txn_cost = txn.amount*txn.price + total_cost = prev_cost + txn_cost + total_shares = self.amount + txn.amount + self.cost_basis = total_cost/total_shares + self.amount = self.amount + txn.amount + + def currentValue(self): + return self.amount * self.last_sale_price + + + def __repr__(self): + template = "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, \ + last_sale_price: {last_sale_price}" + return template.format( + sid=self.sid, + amount=self.amount, + cost_basis=self.cost_basis, + last_sale_price=self.last_sale_price + ) + + def to_dict(self): + """ + Creates a dictionary representing the state of this position. + Returns a dict object of the form: + """ + return { + 'sid' : self.sid, + 'amount' : self.amount, + 'cost_basis' : self.cost_basis, + 'last_sale_price' : self.last_sale_price, + 'last_sale_date' : self.last_sale_date, + 'timestamp' : datetime.datetime.now(), + } + + class PerformancePeriod(): - + def __init__(self, initial_positions, starting_value, starting_cash): self.ending_value = 0.0 self.period_capital_used = 0.0 self.pnl = 0.0 #sid => position object - self.positions = initial_positions + 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 - + def calculate_performance(self): self.ending_value = self.calculate_positions_value() - + total_at_start = self.starting_cash + self.starting_value self.ending_cash = self.starting_cash + self.period_capital_used total_at_end = self.ending_cash + self.ending_value - + self.pnl = total_at_end - total_at_start if(total_at_start != 0): self.returns = self.pnl / total_at_start else: self.returns = 0.0 - + def execute_transaction(self, txn): if(not self.positions.has_key(txn.sid)): self.positions[txn.sid] = Position(txn.sid) self.positions[txn.sid].update(txn) self.period_capital_used += -1 * txn.price * txn.amount - + def calculate_positions_value(self): mktValue = 0.0 for key,pos in self.positions.iteritems(): mktValue += pos.currentValue() return mktValue - + def update_last_sale(self, event): is_trade = event.type == zp.DATASOURCE_TYPE.TRADE if self.positions.has_key(event.sid) and is_trade: - self.positions[event.sid].last_sale_price = event.price + self.positions[event.sid].last_sale_price = event.price self.positions[event.sid].last_sale_date = event.dt - + def to_dict(self): """ Creates a dictionary representing the state of this performance period Returns a dict object of the form: - -+---------------+-----------------------------------------------------------+ -| key | value | -+===============+===========================================================+ -| ending_value | the total market value of the positions held at the | -| | end of the period | -+---------------+-----------------------------------------------------------+ -| capital_used | the net capital consumed (positive means spent) by | -| | buying and selling securities in the period | -+---------------+-----------------------------------------------------------+ -| starting_value| the total market value of the positions held at the | -| | start of the period | -+---------------+-----------------------------------------------------------+ -| starting_cash | cash on hand at the beginning of the period | -+---------------+-----------------------------------------------------------+ -| ending_cash | cash on hand at the end of the period | -+---------------+-----------------------------------------------------------+ -| positions | a list of dicts representing positions, see | -| | :py:meth:`Position.to_dict()` | -| | for details on the contents of the dict | -+---------------+-----------------------------------------------------------+ + """ - d = { - 'ending_value':self.ending_value, - 'capital_used':self.capital_used, - 'starting_value':self.starting_value, - 'starting_cash':self.starting_cash, - 'ending_cash':self.ending_cash + + 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, + 'positions' : self.positions, + 'timestamp' : datetime.datetime.now(), } - - position_list = [] - for pos in self.positions: - position_list.append(pos.to_dict()) - - d['positions'] = positions_list - return d \ No newline at end of file diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index cdb3eaa6..28dc35f0 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -45,6 +45,7 @@ class RiskMetrics(): ) raise Exception(messge) + self.trading_days = len(self.benchmark_returns) self.benchmark_volatility = self.calculate_volatility(self.benchmark_returns) self.algorithm_volatility = self.calculate_volatility(self.algorithm_returns) @@ -89,10 +90,10 @@ class RiskMetrics(): | | and self.end_date. | +-----------------+----------------------------------------------------+ """ - d = { + return { 'trading_days' : self.trading_days, 'benchmark_volatility' : self.benchmark_volatility, - 'algo_volatility' : self.algo_volatility, + 'algo_volatility' : self.algorithm_volatility, 'treasury_period_return': self.treasury_period_return, 'sharpe' : self.sharpe, 'beta' : self.beta, @@ -100,7 +101,7 @@ class RiskMetrics(): 'excess_return' : self.excess_return, 'max_drawdown' : self.max_drawdown } - + def __repr__(self): statements = [] for metric in [ diff --git a/zipline/protocol.py b/zipline/protocol.py index 74fbfbba..9fc98238 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -122,95 +122,11 @@ import copy import pandas from collections import namedtuple -import zipline.util as qutil +from protocol_utils import Enum, FrameExceptionFactory, namedict + #import ujson #import ultrajson_numpy -from ctypes import Structure, c_ubyte - -def Enum(*options): - """ - Fast enums are very important when we want really tight zmq - loops. These are probably going to evolve into pure C structs - anyways so might as well get going on that. - """ - class cstruct(Structure): - _fields_ = [(o, c_ubyte) for o in options] - return cstruct(*range(len(options))) - -def FrameExceptionFactory(name): - """ - Exception factory with a closure around the frame class name. - """ - class InvalidFrame(Exception): - def __init__(self, got): - self.got = got - - def __str__(self): - return "Invalid {framecls} Frame: {got}".format( - framecls = name, - got = self.got, - ) - - return InvalidFrame - -class namedict(object): - """ - So that you can use:: - - foo.BAR - -- or -- - foo['BAR'] - - For more complex structs use collections.namedtuple: - """ - - def __init__(self, dct=None): - if dct: - self.__dict__.update(dct) - - def __setitem__(self, key, value): - """ - Required for use by pymongo as_class parameter to find. - """ - if(key == '_id'): - self.__dict__['id'] = value - else: - self.__dict__[key] = value - - def __getitem__(self, key): - return self.__dict__[key] - - def keys(self): - return self.__dict__.keys() - - def as_dict(self): - # shallow copy is O(n) - return copy.copy(self.__dict__) - - def delete(self, key): - del(self.__dict__[key]) - - def merge(self, other_nd): - assert isinstance(other_nd, namedict) - self.__dict__.update(other_nd.__dict__) - - def __repr__(self): - return "namedict: " + str(self.__dict__) - - def __eq__(self, other): - # !!!!!!!!!!!!!!!!!!!! - # !!!! DANGEROUS !!!!! - # !!!!!!!!!!!!!!!!!!!! - return other != None and self.__dict__ == other.__dict__ - - def has_attr(self, name): - return self.__dict__.has_key(name) - - def as_series(self): - s = pandas.Series(self.__dict__, self.__dict__.keys()) - return s - # ================ # Control Protocol # ================ @@ -279,6 +195,27 @@ COMPONENT_STATE = Enum( 'EXCEPTION' , # 2 ) +# NOFAILURE - Component is either not running or has not failed +# ALGOEXCEPT - Exception thrown in the given algorithm +# HOSTEXCEPT - Exception thrown on our end. +# INTERRUPT - Manually interuptted by user + +COMPONENT_FAILURE = Enum( + 'NOFAILURE' , + 'ALGOEXCEPT' , + 'HOSTEXCEPT' , + 'INTERRUPT' , +) + +BACKTEST_STATE = Enum( + 'IDLE' , + 'QUEUED' , + 'INPROGRESS' , + 'CANCELLED' , # cancelled ( before natural completion ) + 'EXCEPTION' , # failure ( due to unnatural causes ) + 'DONE' , # done ( naturally completed ) +) + # ================== # Datasource Protocol # ================== diff --git a/zipline/protocol_utils.py b/zipline/protocol_utils.py new file mode 100644 index 00000000..0e6cb858 --- /dev/null +++ b/zipline/protocol_utils.py @@ -0,0 +1,87 @@ +import copy +from ctypes import Structure, c_ubyte + +def Enum(*options): + """ + Fast enums are very important when we want really tight zmq + loops. These are probably going to evolve into pure C structs + anyways so might as well get going on that. + """ + class cstruct(Structure): + _fields_ = [(o, c_ubyte) for o in options] + return cstruct(*range(len(options))) + +def FrameExceptionFactory(name): + """ + Exception factory with a closure around the frame class name. + """ + class InvalidFrame(Exception): + def __init__(self, got): + self.got = got + + def __str__(self): + return "Invalid {framecls} Frame: {got}".format( + framecls = name, + got = self.got, + ) + + return InvalidFrame + +class namedict(object): + """ + + Namedicts are dict like objects that have fields accessible by attribute lookup + as well as being indexable and iterable:: + + HEARTBEAT_PROTOCOL = namedict({ + 'REQ' : b'\x01', + 'REP' : b'\x02', + }) + + HEARTBEAT_PROTOCOL.REQ # syntactic sugar + HEARTBEAT_PROTOCOL.REP # oh suga suga + + For more complex structs use collections.namedtuple: + """ + + def __init__(self, dct=None): + if(dct): + self.__dict__.update(dct) + + def __setitem__(self, key, value): + """ + Required for use by pymongo as_class parameter to find. + """ + if(key == '_id'): + self.__dict__['id'] = value + else: + self.__dict__[key] = value + + def __getitem__(self, key): + return self.__dict__[key] + + def keys(self): + return self.__dict__.keys() + + def as_dict(self): + # shallow copy is O(n) + return copy.copy(self.__dict__) + + def delete(self, key): + del(self.__dict__[key]) + + def merge(self, other_nd): + assert isinstance(other_nd, namedict) + self.__dict__.update(other_nd.__dict__) + + def __repr__(self): + return "namedict: " + str(self.__dict__) + + def __eq__(self, other): + # !!!!!!!!!!!!!!!!!!!! + # !!!! DANGEROUS !!!!! + # !!!!!!!!!!!!!!!!!!!! + return other != None and self.__dict__ == other.__dict__ + + def has_attr(self, name): + return self.__dict__.has_key(name) diff --git a/zipline/zmq_utils.py b/zipline/zmq_utils.py new file mode 100644 index 00000000..e7c09047 --- /dev/null +++ b/zipline/zmq_utils.py @@ -0,0 +1,16 @@ +import gevent +from gevent_zeromq import zmq + +def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None): + + context = context or zmq.Context.instance() + socket = context.socket(zmq.PULL) + socket.connect('tcp://127.0.0.1:3141') + + def console(): + while True: + msg = socket.recv() + print msg + import pdb; pdb.set_trace() + + return gevent.spawn(console)