mirror of
https://github.com/wassname/catalyst.git
synced 2026-07-19 11:22:06 +08:00
fawce
366405927d
add performance tracker unit tests, made various fixes to perf tracker. still have a hang on exit for zipline/test/test_finance.py:FinanceTestCase.test_orders and zipline/test/test_finance.py:FinanceTestCase.test_performance. pinging realdiehl for help...
525 lines
15 KiB
Python
525 lines
15 KiB
Python
import unittest
|
|
import copy
|
|
import random
|
|
import datetime
|
|
|
|
import zipline.test.factory as factory
|
|
import zipline.util as qutil
|
|
import zipline.finance.performance as perf
|
|
import zipline.finance.risk as risk
|
|
class PerformanceTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
qutil.configure_logging()
|
|
self.benchmark_returns, self.treasury_curves = \
|
|
factory.load_market_data()
|
|
|
|
self.trading_environment = risk.TradingEnvironment(
|
|
self.benchmark_returns,
|
|
self.treasury_curves
|
|
)
|
|
|
|
self.onesec = datetime.timedelta(seconds=1)
|
|
self.oneday = datetime.timedelta(days=1)
|
|
self.tradingday = datetime.timedelta(hours=6, minutes=30)
|
|
random_index = random.randint(
|
|
0,
|
|
len(self.trading_environment.trading_days)
|
|
)
|
|
|
|
self.dt = self.trading_environment.trading_days[random_index]
|
|
|
|
def tearDown(self):
|
|
pass
|
|
|
|
def test_long_position(self):
|
|
"""
|
|
verify that the performance period calculates properly for a \
|
|
single buy transaction
|
|
"""
|
|
#post some trades in the market
|
|
trades = factory.create_trade_history(
|
|
1,
|
|
[10,10,10,11],
|
|
[100,100,100,100],
|
|
self.dt,
|
|
self.onesec,
|
|
self.trading_environment
|
|
)
|
|
|
|
txn = factory.create_txn(1,10.0,100,self.dt + self.onesec)
|
|
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
|
|
|
|
pp.execute_transaction(txn)
|
|
for trade in trades:
|
|
pp.update_last_sale(trade)
|
|
|
|
pp.calculate_performance()
|
|
|
|
self.assertEqual(
|
|
pp.period_capital_used,
|
|
-1 * txn.price * txn.amount,
|
|
"capital used should be equal to the opposite of the transaction \
|
|
cost of sole txn in test"
|
|
)
|
|
|
|
self.assertEqual(len(pp.positions),1,"should be just one position")
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].sid,
|
|
txn.sid,
|
|
"position should be in security with id 1")
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].amount,
|
|
txn.amount,
|
|
"should have a position of {sharecount} shares".format(
|
|
sharecount=txn.amount
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].cost_basis,
|
|
txn.price,
|
|
"should have a cost basis of 10"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].last_sale,
|
|
trades[-1]['price'],
|
|
"last sale should be same as last trade. \
|
|
expected {exp} actual {act}".format(
|
|
exp=trades[-1]['price'],
|
|
act=pp.positions[1].last_sale
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.ending_value,
|
|
1100,
|
|
"ending value should be price of last trade times number of \
|
|
shares in position"
|
|
)
|
|
|
|
self.assertEqual(pp.pnl, 100, "gain of 1 on 100 shares should be 100")
|
|
|
|
def test_short_position(self):
|
|
"""verify that the performance period calculates properly for a \
|
|
single short-sale transaction"""
|
|
trades_1 = factory.create_trade_history(
|
|
1,
|
|
[10,10,10,11],
|
|
[100,100,100,100],
|
|
self.dt,
|
|
self.onesec,
|
|
self.trading_environment
|
|
)
|
|
|
|
txn = factory.create_txn(1, 10.0, -100, self.dt + self.onesec)
|
|
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
|
|
|
|
pp.execute_transaction(txn)
|
|
for trade in trades_1:
|
|
pp.update_last_sale(trade)
|
|
|
|
pp.calculate_performance()
|
|
|
|
self.assertEqual(
|
|
pp.period_capital_used,
|
|
-1 * txn.price * txn.amount,
|
|
"capital used should be equal to the opposite of the transaction\
|
|
cost of sole txn in test"
|
|
)
|
|
|
|
self.assertEqual(
|
|
len(pp.positions),
|
|
1,
|
|
"should be just one position")
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].sid,
|
|
txn.sid,
|
|
"position should be in security from the transaction"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].amount,
|
|
-100,
|
|
"should have a position of -100 shares"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].cost_basis,
|
|
txn.price,
|
|
"should have a cost basis of 10"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].last_sale,
|
|
trades_1[-1]['price'],
|
|
"last sale should be price of last trade"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.ending_value,
|
|
-1100,
|
|
"ending value should be price of last trade times number of \
|
|
shares in position"
|
|
)
|
|
|
|
self.assertEqual(pp.pnl,-100,"gain of 1 on 100 shares should be 100")
|
|
|
|
#simulate additional trades, and ensure that the position value
|
|
#reflects the new price
|
|
trades_2 = factory.create_trade_history(
|
|
1,
|
|
[10,9],
|
|
[100,100],
|
|
trades_1[-1]['dt'] + self.onesec,
|
|
self.onesec,
|
|
self.trading_environment
|
|
)
|
|
|
|
#simulate a rollover to a new period
|
|
pp2 = perf.PerformancePeriod(
|
|
pp.positions,
|
|
pp.ending_value,
|
|
pp.ending_cash
|
|
)
|
|
|
|
for trade in trades_2:
|
|
pp2.update_last_sale(trade)
|
|
|
|
pp2.calculate_performance()
|
|
|
|
self.assertEqual(
|
|
pp2.period_capital_used,
|
|
0,
|
|
"capital used should be zero, there were no transactions in \
|
|
performance period"
|
|
)
|
|
|
|
self.assertEqual(
|
|
len(pp2.positions),
|
|
1,
|
|
"should be just one position"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp2.positions[1].sid,
|
|
txn.sid,
|
|
"position should be in security from the transaction"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp2.positions[1].amount,
|
|
-100,
|
|
"should have a position of -100 shares"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp2.positions[1].cost_basis,
|
|
txn.price,
|
|
"should have a cost basis of 10"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp2.positions[1].last_sale,
|
|
trades_2[-1].price,
|
|
"last sale should be price of last trade"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp2.ending_value,
|
|
-900,
|
|
"ending value should be price of last trade times number of \
|
|
shares in position")
|
|
|
|
self.assertEqual(
|
|
pp2.pnl,
|
|
200,
|
|
"drop of 2 on -100 shares should be 200"
|
|
)
|
|
|
|
#now run a performance period encompassing the entire trade sample.
|
|
ppTotal = perf.PerformancePeriod({}, 0.0, 1000.0)
|
|
|
|
for trade in trades_1:
|
|
ppTotal.update_last_sale(trade)
|
|
|
|
ppTotal.execute_transaction(txn)
|
|
|
|
for trade in trades_2:
|
|
ppTotal.update_last_sale(trade)
|
|
|
|
ppTotal.calculate_performance()
|
|
|
|
self.assertEqual(
|
|
ppTotal.period_capital_used,
|
|
-1 * txn.price * txn.amount,
|
|
"capital used should be equal to the opposite of the transaction \
|
|
cost of sole txn in test"
|
|
)
|
|
|
|
self.assertEqual(
|
|
len(ppTotal.positions),
|
|
1,
|
|
"should be just one position"
|
|
)
|
|
self.assertEqual(
|
|
ppTotal.positions[1].sid,
|
|
txn.sid,
|
|
"position should be in security from the transaction"
|
|
)
|
|
|
|
self.assertEqual(
|
|
ppTotal.positions[1].amount,
|
|
-100,
|
|
"should have a position of -100 shares"
|
|
)
|
|
|
|
self.assertEqual(
|
|
ppTotal.positions[1].cost_basis,
|
|
txn.price,
|
|
"should have a cost basis of 10"
|
|
)
|
|
|
|
self.assertEqual(
|
|
ppTotal.positions[1].last_sale,
|
|
trades_2[-1].price,
|
|
"last sale should be price of last trade"
|
|
)
|
|
|
|
self.assertEqual(
|
|
ppTotal.ending_value,
|
|
-900,
|
|
"ending value should be price of last trade times number of \
|
|
shares in position")
|
|
|
|
self.assertEqual(
|
|
ppTotal.pnl,
|
|
100,
|
|
"drop of 1 on -100 shares should be 100"
|
|
)
|
|
|
|
def test_covering_short(self):
|
|
"""verify performance where short is bought and covered, and shares \
|
|
trade after cover"""
|
|
|
|
trades = factory.create_trade_history(
|
|
1,
|
|
[10,10,10,11,9,8,7,8,9,10],
|
|
[100,100,100,100,100,100,100,100,100,100],
|
|
self.dt,
|
|
self.onesec,
|
|
self.trading_environment
|
|
)
|
|
|
|
short_txn = factory.create_txn(
|
|
1,
|
|
10.0,
|
|
-100,
|
|
self.dt + self.onesec
|
|
)
|
|
|
|
cover_txn = factory.create_txn(1,7.0,100,self.dt + self.onesec * 6)
|
|
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
|
|
|
|
pp.execute_transaction(short_txn)
|
|
pp.execute_transaction(cover_txn)
|
|
|
|
for trade in trades:
|
|
pp.update_last_sale(trade)
|
|
|
|
pp.calculate_performance()
|
|
|
|
short_txn_cost = short_txn.price * short_txn.amount
|
|
cover_txn_cost = cover_txn.price * cover_txn.amount
|
|
|
|
self.assertEqual(
|
|
pp.period_capital_used,
|
|
-1 * short_txn_cost - cover_txn_cost,
|
|
"capital used should be equal to the net transaction costs"
|
|
)
|
|
|
|
self.assertEqual(
|
|
len(pp.positions),
|
|
1,
|
|
"should be just one position"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].sid,
|
|
short_txn.sid,
|
|
"position should be in security from the transaction"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].amount,
|
|
0,
|
|
"should have a position of -100 shares"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].cost_basis,
|
|
0,
|
|
"a covered position should have a cost basis of 0"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].last_sale,
|
|
trades[-1].price,
|
|
"last sale should be price of last trade"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.ending_value,
|
|
0,
|
|
"ending value should be price of last trade times number of \
|
|
shares in position"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.pnl,
|
|
300,
|
|
"gain of 1 on 100 shares should be 300"
|
|
)
|
|
|
|
def test_cost_basis_calc(self):
|
|
trades = factory.create_trade_history(
|
|
1,
|
|
[10,11,11,12],
|
|
[100,100,100,100],
|
|
self.dt,
|
|
self.onesec,
|
|
self.trading_environment
|
|
)
|
|
|
|
transactions = factory.create_txn_history(
|
|
1,
|
|
[10,11,11,12],
|
|
[100,100,100,100],
|
|
self.dt,
|
|
self.onesec,
|
|
self.trading_environment
|
|
)
|
|
|
|
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
|
|
|
|
for txn in transactions:
|
|
pp.execute_transaction(txn)
|
|
|
|
for trade in trades:
|
|
pp.update_last_sale(trade)
|
|
|
|
pp.calculate_performance()
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].last_sale,
|
|
trades[-1].price,
|
|
"should have a last sale of 12, got {val}".format(
|
|
val=pp.positions[1].last_sale
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.positions[1].cost_basis,
|
|
11,
|
|
"should have a cost basis of 11"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp.pnl,
|
|
400
|
|
)
|
|
|
|
saleTxn = factory.create_txn(
|
|
1,
|
|
10.0,
|
|
-100,
|
|
self.dt + self.onesec * 4)
|
|
|
|
down_tick = factory.create_trade(
|
|
1,
|
|
10.0,
|
|
100,
|
|
trades[-1].dt + self.onesec)
|
|
|
|
pp2 = perf.PerformancePeriod(
|
|
copy.deepcopy(pp.positions),
|
|
pp.ending_value,
|
|
pp.ending_cash
|
|
)
|
|
|
|
pp2.execute_transaction(saleTxn)
|
|
pp2.update_last_sale(down_tick)
|
|
|
|
pp2.calculate_performance()
|
|
self.assertEqual(
|
|
pp2.positions[1].last_sale,
|
|
10,
|
|
"should have a last sale of 10, was {val}".format(val=pp2.positions[1].last_sale)
|
|
)
|
|
|
|
self.assertEqual(
|
|
round(pp2.positions[1].cost_basis,2),
|
|
11.33,
|
|
"should have a cost basis of 11.33"
|
|
)
|
|
|
|
#print "second period pnl is {pnl}".format(pnl=pp2.pnl)
|
|
self.assertEqual(pp2.pnl, -800, "this period goes from +400 to -400")
|
|
|
|
pp3 = perf.PerformancePeriod({}, 0.0, 1000.0)
|
|
|
|
transactions.append(saleTxn)
|
|
for txn in transactions:
|
|
pp3.execute_transaction(txn)
|
|
|
|
trades.append(down_tick)
|
|
for trade in trades:
|
|
pp3.update_last_sale(trade)
|
|
|
|
pp3.calculate_performance()
|
|
self.assertEqual(
|
|
pp3.positions[1].last_sale,
|
|
10,
|
|
"should have a last sale of 10"
|
|
)
|
|
|
|
self.assertEqual(
|
|
round(pp3.positions[1].cost_basis,2),
|
|
11.33,
|
|
"should have a cost basis of 11.33"
|
|
)
|
|
|
|
self.assertEqual(
|
|
pp3.pnl,
|
|
-400,
|
|
"should be -400 for all trades and transactions in period"
|
|
)
|
|
|
|
|
|
def dtest_daily_performance_calc(self):
|
|
hostedAlgo = factories.createAlgo("workingAlgo.py")
|
|
btRecord = BackTestRun(duration_unit="Days",duration_count=5,capital_base=25000000)
|
|
bt = BackTest(hostedAlgo,btRecord)
|
|
start = bt.periodStart
|
|
end = bt.periodEnd
|
|
#print "{start} to {end}".format(start=start, end=end)
|
|
|
|
trades = factories.createTradeHistory(1,[10,11,12,11],[100,100,100,100],start, self.oneday)
|
|
#createTransaction(self, sid, amount, price, dt, order_id)
|
|
bt.createTransaction(1, 100, 10.0, trades[0].dt + 30*self.onesec, None)
|
|
curPeriod = start
|
|
bt.positions = {}
|
|
dailyPeriods = []
|
|
bt.initialValue = 0.0
|
|
while (bt.mktClose) <= bt.periodEnd:
|
|
bt.updatePerformance()
|
|
dailyPeriods.append(bt.curPeriod)
|
|
bt.nextMarketDay()
|
|
|
|
self.assertEqual(dailyPeriods[0].pnl,0,"the first day's performance should be zero")
|
|
self.assertEqual(dailyPeriods[1].pnl,100,"the second day's pnl should be 100 but was {pnl}".format(pnl=dailyPeriods[1].pnl))
|
|
|