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

zipline, now a cold, heartless, http://open.spotify.com/track/1xshgoh575otNXRfeYgh9D

Browse Source 
Pretty fast too...
This commit is contained in:
fawce
2012-04-20 12:21:03 -04:00
parent 28c380c245
commit bc14e7e3b7
5 changed files with 70 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/finance/performance.py
+2 -2
View File
@@ -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
zipline/finance/trading.py
+43 -155
View File
@@ -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] = []
zipline/lines.py
+9 -7
View File
@@ -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()
zipline/test/test_finance.py
+10 -22
View File
@@ -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
zipline/test/test_perf_tracking.py
+6 -14
View File
@@ -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)