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

Merge branch 'master' of github.com:quantopian/zipline

Browse Source 
Conflicts:
	zipline/test/client.py
This commit is contained in:
fawce
2012-02-29 16:47:34 -05:00
parent aa567a1854 62c32e1de7
commit 5247ace0ca
6 changed files with 154 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/component.py
-1
View File
@@ -302,7 +302,6 @@ class Component(object):
socket.gethostname() ,
os.getpid() ,
hex(id(self)) ,
self.sockets ,
)
def __repr__(self):
zipline/finance/trading.py
+133 -66
View File
@@ -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}]
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)
zipline/protocol.py
+14 -8
View File
@@ -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
return payload
FINANCE_PROTOCOL = Enum(
'ORDER' , # 0 - req
'TRANSACTION' , # 1 - req
'STATUS' , # 2 - req
)
zipline/sources.py
+2 -2
View File
@@ -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):
zipline/test/client.py
+1
View File
@@ -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
zipline/test/test_messaging.py
+4 -4
View File
@@ -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