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

added simulation style to transation simulator, to facilitate tests. Fixed roll-over bug in max cap and max leverage calculation.

Browse Source
This commit is contained in:
fawce
2012-04-13 09:53:20 -04:00
parent aea2e1189c
commit 1559a84c7b
6 changed files with 155 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/component.py
+2 -1
View File
@@ -75,7 +75,8 @@ class Component(object):
self.out_socket = None
self.killed = False
self.controller = None
self.heartbeat_timeout = 2000
# timeout after a full minute
self.heartbeat_timeout = 60 *1000
self.state_flag = COMPONENT_STATE.OK
self.error_state = COMPONENT_FAILURE.NOFAILURE
self.on_done = None
zipline/finance/performance.py
+36 -25
View File
@@ -160,8 +160,6 @@ class PerformanceTracker():
self.total_days = self.trading_environment.days_in_period
# one indexed so that we reach 100%
self.day_count = 0.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
@@ -219,8 +217,8 @@ class PerformanceTracker():
'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,
'cumulative_captial_used' : self.cumulative_perf.cumulative_capital_used,
'max_capital_used' : self.cumulative_perf.max_capital_used,
'last_close' : self.market_close,
'last_open' : self.market_open,
'capital_base' : self.capital_base,
@@ -232,36 +230,21 @@ class PerformanceTracker():
}
def process_event(self, event):
assert isinstance(event, zp.namedict)
self.event_count += 1
if(event.dt >= self.market_close):
self.handle_market_close()
if not pandas.isnull(event.TRANSACTION):
if 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)
def handle_market_close(self):
#calculate performance as of last trade
@@ -338,9 +321,6 @@ class PerformanceTracker():
# this signals that the simulation is complete.
self.result_stream.send("DONE")
def round_to_nearest(self, x, base=5):
return int(base * round(float(x)/base))
class Position():
@@ -409,6 +389,8 @@ class PerformancePeriod():
self.starting_cash = starting_cash
self.ending_cash = starting_cash
self.processed_transactions = []
self.cumulative_capital_used = 0.0
self.max_capital_used = 0.0
self.calculate_performance()
@@ -426,11 +408,40 @@ class PerformancePeriod():
self.returns = 0.0
def execute_transaction(self, txn):
# Update Position
# ----------------
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
# Max Leverage
# ---------------
# Calculate the maximum capital used and maximum leverage
transaction_cost = txn.price * txn.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)
# We want to conveye a level, rather than a precise figure.
# round to the nearest 5,000 to keep the number easy on the eyes
self.max_capital_used = self.round_to_nearest(
self.max_capital_used,
base=5000
)
# we're adding a 10% cushion to the capital used.
self.max_leverage = 1.1 * self.max_capital_used / self.starting_cash
# add transaction to the list of processed transactions
self.processed_transactions.append(txn)
def round_to_nearest(self, x, base=5):
return int(base * round(float(x)/base))
def calculate_positions_value(self):
mktValue = 0.0
zipline/finance/trading.py
+75 -7
View File
@@ -11,6 +11,17 @@ import zipline.util as qutil
import zipline.protocol as zp
import zipline.finance.performance as perf
from zipline.protocol_utils import Enum
# the simulation style enumerates the available transaction simulation
# strategies.
SIMULATION_STYLE = Enum(
'PARTIAL_VOLUME',
'BUY_ALL',
'FIXED_SLIPPAGE',
'NOOP'
)
class TradeSimulationClient(qmsg.Component):
def __init__(self, trading_environment):
@@ -19,6 +30,7 @@ class TradeSimulationClient(qmsg.Component):
self.prev_dt = None
self.event_queue = None
self.txn_count = 0
self.order_count = 0
self.trading_environment = trading_environment
self.current_dt = trading_environment.period_start
self.last_iteration_dur = datetime.timedelta(seconds=0)
@@ -132,13 +144,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
def signal_order_done(self):
self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE))
@@ -211,6 +224,8 @@ class OrderDataSource(qmsg.DataSource):
# 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.
# TODO: shouldn't this block until we receive a message?
socks = dict(self.poll.poll(self.heartbeat_timeout/2))
# see if the poller has results for the result_feed
@@ -232,19 +247,20 @@ class OrderDataSource(qmsg.DataSource):
count += 1
self.sent_count += 1
else:
# no orders, break out
break
# TODO: why didn't any unit tests catch this bug????
#else:
# # no orders, break out
# break
#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(zp.namedict({}))
class TransactionSimulator(qmsg.BaseTransform):
def __init__(self):
def __init__(self, style):
qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION)
self.open_orders = {}
self.order_count = 0
@@ -252,6 +268,15 @@ class TransactionSimulator(qmsg.BaseTransform):
self.trade_window = datetime.timedelta(seconds=30)
self.orderTTL = datetime.timedelta(days=1)
self.commission = 0.03
if not style or style == SIMULATION_STYLE.PARTIAL_VOLUME:
self.apply_trade_to_open_orders = self.simulate_with_partial_volume
elif style == SIMULATION_STYLE.BUY_ALL:
self.apply_trade_to_open_orders = self.simulate_buy_all
elif style == SIMULATION_STYLE.FIXED_SLIPPAGE:
self.apply_trade_to_open_orders = self.simulate_with_fixed_cost
elif style == SIMULATION_STYLE.NOOP:
self.apply_trade_to_open_orders = self.simulate_noop
def transform(self, event):
"""
@@ -296,8 +321,51 @@ class TransactionSimulator(qmsg.BaseTransform):
event.filled = 0
self.open_orders[event.sid].append(event)
def apply_trade_to_open_orders(self, event):
#def apply_trade_to_open_orders(self, event):
# return self.simulate_with_fixed_cost(event)
def simulate_buy_all(self, event):
txn = self.create_transaction(
event.sid,
event.volume,
event.price,
event.dt,
1
)
return txn
def simulate_noop(self, event):
return None
def simulate_with_fixed_cost(self, event):
if self.open_orders.has_key(event.sid):
orders = self.open_orders[event.sid]
orders = sorted(orders, key=lambda o: o.dt)
else:
return None
amount = 0
for order in orders:
amount += order.amount
if(amount != 0):
direction = amount / math.fabs(amount)
else:
direction = 1
txn = self.create_transaction(
event.sid,
amount,
event.price + 0.10,
event.dt,
direction
)
self.open_orders[event.sid] = []
return txn
def simulate_with_partial_volume(self, event):
if(event.volume == 0):
#there are zero volume events bc some stocks trade
#less frequently than once per minute.
zipline/lines.py
+8 -5
View File
@@ -90,7 +90,7 @@ from zipline.finance.trading import TransactionSimulator, OrderDataSource, \
TradeSimulationClient
from zipline.simulator import AddressAllocator, Simulator
from zipline.monitor import Controller
from zipline.finance.trading import SIMULATION_STYLE
class SimulatedTrading(object):
@@ -130,6 +130,7 @@ class SimulatedTrading(object):
self.algorithm = config['algorithm']
self.allocator = config['allocator']
self.trading_environment = config['trading_environment']
self.sim_style = config.get('simulation_style')
self.leased_sockets = []
self.sim_context = None
@@ -169,7 +170,7 @@ class SimulatedTrading(object):
self.add_source(self.order_source)
#setup transforms
self.transaction_sim = TransactionSimulator()
self.transaction_sim = TransactionSimulator(self.sim_style)
self.transforms = {}
self.add_transform(self.transaction_sim)
@@ -192,7 +193,8 @@ class SimulatedTrading(object):
- order_count - the number of orders the test algo will place,
defaults to 100
- order_amount - the number of shares per order, defaults to 100
- trade_count - the number of trades to simulate, defaults to 100
- trade_count - the number of trades to simulate, defaults to 101
to ensure all orders are processed.
- simulator_class - optional parameter that provides an alternative
subclass of ComponentHost to hold the whole zipline. Defaults to
:py:class:`zipline.simulator.Simulator`
@@ -228,7 +230,7 @@ class SimulatedTrading(object):
if config.has_key('trade_count'):
trade_count = config['trade_count']
else:
trade_count = 100
trade_count = 101
if config.has_key('simulator_class'):
simulator_class = config['simulator_class']
@@ -266,7 +268,8 @@ class SimulatedTrading(object):
'algorithm':test_algo,
'trading_environment':trading_environment,
'allocator':allocator,
'simulator_class':simulator_class
'simulator_class':simulator_class,
'simulation_style':SIMULATION_STYLE.FIXED_SLIPPAGE
})
#-------------------
zipline/messaging.py
+2 -2
View File
@@ -37,7 +37,7 @@ class ComponentHost(Component):
# ----------------------
self.sync_register = {}
self.timeout = datetime.timedelta(seconds=5)
self.timeout = datetime.timedelta(seconds=60)
self.feed = Feed()
self.merge = Merge()
@@ -214,7 +214,7 @@ class Feed(Component):
def do_work(self):
# wait for synchronization reply from the host
socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds.
socks = dict(self.poll.poll(self.heartbeat_timeout))
# TODO: Abstract this out, maybe on base component
if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
zipline/test/test_finance.py
+32 -5
View File
@@ -21,6 +21,7 @@ TradeSimulationClient, TradingEnvironment
from zipline.simulator import AddressAllocator, Simulator
from zipline.monitor import Controller
from zipline.lines import SimulatedTrading
from zipline.finance.performance import PerformanceTracker
from zipline.protocol_utils import namedict
DEFAULT_TIMEOUT = 15 # seconds
@@ -120,7 +121,22 @@ class FinanceTestCase(TestCase):
self.assertEqual(zipline.sim.feed.pending_messages(), 0, \
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages()))
# the trading client should receive one transaction for every
# order placed.
self.assertEqual(
zipline.trading_client.txn_count,
zipline.trading_client.order_count
)
# the number of transactions in the performance tracker's cumulative
# period should be the same as the number of orders place by the
# algorithm.
self.assertEqual(
zipline.trading_client.order_count,
len(zipline.trading_client.perf.cumulative_performance.processed_transactions)
)
@timed(EXTENDED_TIMEOUT)
@@ -152,8 +168,8 @@ class FinanceTestCase(TestCase):
self.assertEqual(zipline.sim.feed.pending_messages(), 0, \
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages()))
@timed(DEFAULT_TIMEOUT)
def test_performance(self):
#provide enough trades to ensure all orders are filled.
@@ -358,6 +374,7 @@ class FinanceTestCase(TestCase):
def transaction_sim(self, **params):
trade_count = params['trade_count']
trade_amount = params['trade_amount']
trade_interval = params['trade_interval']
@@ -409,6 +426,9 @@ class FinanceTestCase(TestCase):
self.assertEqual(order.sid, sid)
self.assertEqual(order.amount, order_amount)
tracker = PerformanceTracker(trading_environment)
transactions = []
for trade in generated_trades:
if trade_delay:
@@ -418,10 +438,14 @@ class FinanceTestCase(TestCase):
self.assertEqual(sim_state['name'], trade_sim.get_id)
txn = None
if sim_state['value']:
transactions.append(sim_state['value'])
txn = sim_state['value']
transactions.append(txn)
trade[sim_state['name']] = txn
tracker.process_event(trade)
total_volume = 0
for txn in transactions:
total_volume += txn.amount
@@ -429,6 +453,9 @@ class FinanceTestCase(TestCase):
self.assertEqual(total_volume, expected_txn_volume)
self.assertEqual(len(transactions), expected_txn_count)
cumulative_pos = tracker.cumulative_performance.positions[sid]
self.assertEqual(total_volume, cumulative_pos.amount)
# the open orders should now be empty
oo = trade_sim.open_orders
self.assertTrue(oo.has_key(sid))