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

refactoring orders to discard placeholder messages.

Browse Source
This commit is contained in:
fawce
2012-03-09 01:07:01 -05:00
parent b69ea6b790
commit b44ca63557
7 changed files with 102 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/finance/performance.py
+18 -11
View File
@@ -86,10 +86,11 @@ class PerformanceTracker():
self.returns.append(todays_return_obj)
#calculate risk metrics for cumulative performance
self.cur_period_metrics = risk.RiskMetrics(start_date=self.cumulative_performance.period_start,
end_date=self.cumulative_performance.period_end.replace(hour=0, minute=0, second=0),
returns=self.returns,
trading_environment=self.trading_environment)
self.cur_period_metrics = risk.RiskMetrics(
start_date=self.cumulative_performance.period_start,
end_date=self.cumulative_performance.period_end.replace(hour=0, minute=0, second=0),
returns=self.returns,
trading_environment=self.trading_environment)
######################################################################################################
#######TODO: report/relay metrics out to qexec -- values come from self.cur_period_metrics ###########
@@ -97,14 +98,20 @@ class PerformanceTracker():
######################################################################################################
#roll over positions to current day.
self.todays_performance = PerformancePeriod(self.market_open,
self.market_close,
self.todays_performance.positions,
self.todays_performance.ending_value,
self.capital_base)
self.todays_performance = PerformancePeriod(
self.market_open,
self.market_close,
self.todays_performance.positions,
self.todays_performance.ending_value,
self.capital_base
)
def handle_simulation_end(self):
self.risk_report = risk.RiskReport(self.returns, self.trading_environment)
self.risk_report = risk.RiskReport(
self.returns,
self.trading_environment
)
######################################################################################################
#######TODO: report/relay metrics out to qexec -- values come from self.risk_report ###########
######################################################################################################
zipline/finance/risk.py
+9 -7
View File
@@ -202,13 +202,15 @@ class RiskReport():
cur_end = advance_by_months(cur_start, months_per) - one_day
if(cur_end > the_end):
break
#qutil.LOGGER.debug("start: {start}, end: {end}".format(start=cur_start, end=cur_end))
cur_period_metrics = RiskMetrics(start_date=cur_start,
end_date=cur_end,
returns=self.algorithm_returns,
benchmark_returns=self.bm_returns,
treasury_curves=self.treasury_curves,
trading_calendar=self.trading_calendar)
cur_period_metrics = RiskMetrics(
start_date=cur_start,
end_date=cur_end,
returns=self.algorithm_returns,
benchmark_returns=self.bm_returns,
treasury_curves=self.treasury_curves,
trading_calendar=self.trading_calendar
)
ends.append(cur_period_metrics)
cur_start = advance_by_months(cur_start, 1)
zipline/finance/trading.py
+20 -14
View File
@@ -47,6 +47,7 @@ class TradeSimulationClient(qmsg.Component):
return
event = zp.MERGE_UNFRAME(msg)
for cb in self.event_callbacks:
cb(event)
@@ -57,6 +58,10 @@ class TradeSimulationClient(qmsg.Component):
return self.connect_push_socket(self.addresses['order_address'])
def order(self, sid, amount):
qutil.LOGGER.debug("ordering {amt} of {sid}".format(
amt=amount,
sid=sid
))
self.order_socket.send(zp.ORDER_FRAME(sid, amount))
def signal_order_done(self):
@@ -100,7 +105,7 @@ class OrderDataSource(qmsg.DataSource):
#TODO: if this is the first iteration, break deadlock by sending a dummy order
if(self.sent_count == 0):
self.send_dummy()
self.send(zp.namedict({}))
#pull all orders from client.
orders = []
@@ -127,25 +132,19 @@ class OrderDataSource(qmsg.DataSource):
sid, amount = zp.ORDER_UNFRAME(order_msg)
#send the order along
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})
order = zp.namedict({"dt":dt, 'sid':sid, 'amount':amount})
self.send(order_event)
self.send(order)
count += 1
self.sent_count += 1
#TODO: we have to send at least one dummy order per do_work iteration
# or the feed will block waiting for our messages.
if(count == 0):
self.send_dummy()
self.send(zp.namedict({}))
self.sent_count += 1
def send_dummy(self):
dt = self.simulation_dt + self.last_iteration_duration
dummy_order = zp.namedict({"sid":0, "amount":0, "dt":dt})
self.send(dummy_order)
@@ -166,7 +165,6 @@ class TransactionSimulator(qmsg.BaseTransform):
Pulls one message from the event feed, then
loops on orders until client sends DONE message.
"""
#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)
self.state['value'] = None
@@ -199,7 +197,8 @@ class TransactionSimulator(qmsg.BaseTransform):
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.
#there are zero volume events bc some stocks trade
#less frequently than once per minute.
return self.create_dummy_txn(event.dt)
if self.open_orders.has_key(event.sid):
@@ -212,7 +211,8 @@ class TransactionSimulator(qmsg.BaseTransform):
dt = event.dt
for order in orders:
#we're using minute bars, so allow orders within 30 seconds of the trade
#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):
@@ -233,7 +233,13 @@ class TransactionSimulator(qmsg.BaseTransform):
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)
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):
zipline/messaging.py
+5 -1
View File
@@ -293,7 +293,11 @@ class ParallelBuffer(Component):
continue
cur_source = events
first_in_list = events[0]
if first_in_list.dt == None:
#this is a filler event, discard
events.pop(0)
continue
if (earliest_event == None) or (first_in_list.dt <= earliest_event.dt):
earliest_event = first_in_list
earliest_source = cur_source
zipline/protocol.py
+34 -14
View File
@@ -295,11 +295,27 @@ def DATASOURCE_FRAME(event):
assert isinstance(event.source_id, basestring)
assert isinstance(event.type, int), 'Unexpected type %s' % (event.type)
#datasources will send sometimes send empty msgs to feel gaps
if len(event.keys()) == 2:
return msgpack.dumps(tuple([
event.type,
event.source_id,
DATASOURCE_TYPE.EMPTY
]))
if(event.type == DATASOURCE_TYPE.TRADE):
return msgpack.dumps(tuple([event.type, TRADE_FRAME(event)]))
return msgpack.dumps(tuple([
event.type,
event.source_id,
TRADE_FRAME(event)
]))
elif(event.type == DATASOURCE_TYPE.ORDER):
return msgpack.dumps(tuple([event.type, ORDER_SOURCE_FRAME(event)]))
return msgpack.dumps(tuple([
event.type,
event.source_id,
ORDER_SOURCE_FRAME(event)
]))
else:
raise INVALID_DATASOURCE_FRAME(str(event))
@@ -321,15 +337,21 @@ def DATASOURCE_UNFRAME(msg):
"""
try:
ds_type, payload = msgpack.loads(msg)
ds_type, source_id, payload = msgpack.loads(msg)
assert isinstance(ds_type, int)
if(ds_type == DATASOURCE_TYPE.TRADE):
return TRADE_UNFRAME(payload)
rval = namedict({'source_id':source_id})
if payload == DATASOURCE_TYPE.EMPTY:
child_value = namedict({'dt':None})
elif(ds_type == DATASOURCE_TYPE.TRADE):
child_value = TRADE_UNFRAME(payload)
elif(ds_type == DATASOURCE_TYPE.ORDER):
return ORDER_SOURCE_UNFRAME(payload)
child_value = ORDER_SOURCE_UNFRAME(payload)
else:
raise INVALID_DATASOURCE_FRAME(msg)
rval.merge(child_value)
return rval
except TypeError:
raise INVALID_DATASOURCE_FRAME(msg)
except ValueError:
@@ -461,7 +483,6 @@ def TRADE_FRAME(event):
"""
assert isinstance(event, namedict)
assert isinstance(event.source_id, basestring)
assert event.type == DATASOURCE_TYPE.TRADE
assert isinstance(event.sid, int)
assert isinstance(event.price, numbers.Real)
@@ -474,13 +495,12 @@ def TRADE_FRAME(event):
event.epoch,
event.micros,
event.type,
event.source_id
]))
def TRADE_UNFRAME(msg):
try:
packed = msgpack.loads(msg)
sid, price, volume, epoch, micros, source_type, source_id = packed
sid, price, volume, epoch, micros, source_type = packed
assert isinstance(sid, int)
assert isinstance(price, numbers.Real)
@@ -491,8 +511,7 @@ def TRADE_UNFRAME(msg):
'volume' : volume,
'epoch' : epoch,
'micros' : micros,
'type' : source_type,
'source_id' : source_id
'type' : source_type
})
UNPACK_DATE(rval)
return rval
@@ -663,12 +682,13 @@ def UNPACK_DATE(event):
DATASOURCE_TYPE = Enum(
'ORDER',
'TRADE'
'TRADE',
'EMPTY',
)
ORDER_PROTOCOL = Enum(
'DONE',
'BREAK'
'BREAK',
)
zipline/test/client.py
+5 -3
View File
@@ -77,10 +77,12 @@ class TestAlgorithm():
self.incr = 0
def handle_event(self, event):
qutil.LOGGER.debug(event)
#place an order for 100 shares of sid:133
if(self.incr < self.count):
self.trading_client.order(self.sid, self.amount)
self.incr += 1
if self.incr < self.count:
if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE:
self.trading_client.order(self.sid, self.amount)
self.incr += 1
else:
self.trading_client.signal_order_done()
self.trading_client.signal_done()
zipline/test/test_finance.py
+11 -7
View File
@@ -202,8 +202,7 @@ class FinanceTestCase(TestCase):
.format(n=sim.feed.pending_messages()))
def test_performance(self):
def test_performance(self):
# verify order -> transaction -> portfolio position.
# --------------
@@ -278,12 +277,17 @@ class FinanceTestCase(TestCase):
sim_context = sim.simulate()
sim_context.join()
# TODO: Make more assertions about the final state of the components.
self.assertEqual(sim.feed.pending_messages(), 0, \
self.assertEqual(
sim.feed.pending_messages(),
0,
"The feed should be drained of all messages, found {n} remaining." \
.format(n=sim.feed.pending_messages()))
.format(n=sim.feed.pending_messages())
)
self.assertEqual(
test_algo.count,
test_algo.incr,
"The test algorithm should send as many orders as specified.")
self.assertEqual(
order_source.sent_count,