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

first edition of leveraged etf list

Browse Source
This commit is contained in:
fawce
2015-02-05 13:57:28 -05:00
parent 536ace94b8
commit 1ab082102a
32 changed files with 321 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_security_list.py
+28 -56
View File
@@ -1,6 +1,4 @@
import pytz
import os.path
import shutil
from datetime import datetime, timedelta
from unittest import TestCase
@@ -8,10 +6,11 @@ from unittest import TestCase
from zipline.algorithm import TradingAlgorithm
from zipline.errors import TradingControlViolation
from zipline.sources import SpecificEquityTrades
from zipline.utils.test_utils import setup_logger
from zipline.utils.test_utils import (
setup_logger, add_security_data, remove_security_data_directory)
from zipline.utils import factory
from zipline.utils.security_list import (
SecurityListSet, load_from_directory, SECURITY_LISTS_DIR)
SecurityListSet, load_from_directory)
LEVERAGED_ETFS = load_from_directory('leveraged_etf_list')
@@ -19,14 +18,14 @@ LEVERAGED_ETFS = load_from_directory('leveraged_etf_list')
class RestrictedAlgoWithCheck(TradingAlgorithm):
def initialize(self, sid):
self.rl = SecurityListSet(self.get_datetime)
self.set_do_not_order_list(self.rl.LEVERAGED_ETF_LIST)
self.set_do_not_order_list(self.rl.leveraged_etf_list)
self.order_count = 0
self.sid = sid
def handle_data(self, data):
if not self.order_count:
if self.sid not in \
self.rl.LEVERAGED_ETF_LIST:
self.rl.leveraged_etf_list:
self.order(self.sid, 100)
self.order_count += 1
@@ -34,7 +33,7 @@ class RestrictedAlgoWithCheck(TradingAlgorithm):
class RestrictedAlgoWithoutCheck(TradingAlgorithm):
def initialize(self, sid):
self.rl = SecurityListSet(self.get_datetime)
self.set_do_not_order_list(self.rl.LEVERAGED_ETF_LIST)
self.set_do_not_order_list(self.rl.leveraged_etf_list)
self.order_count = 0
self.sid = sid
@@ -46,13 +45,13 @@ class RestrictedAlgoWithoutCheck(TradingAlgorithm):
class IterateRLAlgo(TradingAlgorithm):
def initialize(self, sid):
self.rl = SecurityListSet(self.get_datetime)
self.set_do_not_order_list(self.rl.LEVERAGED_ETF_LIST)
self.set_do_not_order_list(self.rl.leveraged_etf_list)
self.order_count = 0
self.sid = sid
self.found = False
def handle_data(self, data):
for stock in self.rl.LEVERAGED_ETF_LIST:
for stock in self.rl.leveraged_etf_list:
if stock == self.sid:
self.found = True
@@ -93,38 +92,38 @@ class SecurityListTestCase(TestCase):
rl = SecurityListSet(get_datetime)
# assert that a sample from the leveraged list are in restricted
self.assertIn("BZQ", rl.LEVERAGED_ETF_LIST)
self.assertIn("URTY", rl.LEVERAGED_ETF_LIST)
self.assertIn("BZQ", rl.leveraged_etf_list)
self.assertIn("URTY", rl.leveraged_etf_list)
# assert that a sample of allowed stocks are not in restricted
# AAPL
self.assertNotIn("AAPL", rl.LEVERAGED_ETF_LIST)
self.assertNotIn("AAPL", rl.leveraged_etf_list)
# GOOG
self.assertNotIn("GOOG", rl.LEVERAGED_ETF_LIST)
self.assertNotIn("GOOG", rl.leveraged_etf_list)
def test_security_add(self):
def get_datetime():
return datetime(2015, 1, 27, tzinfo=pytz.utc)
try:
add_data(['AAPL', 'GOOG'], [])
add_security_data(['AAPL', 'GOOG'], [])
rl = SecurityListSet(get_datetime)
self.assertIn("AAPL", rl.LEVERAGED_ETF_LIST)
self.assertIn("GOOG", rl.LEVERAGED_ETF_LIST)
self.assertIn("BZQ", rl.LEVERAGED_ETF_LIST)
self.assertIn("URTY", rl.LEVERAGED_ETF_LIST)
self.assertIn("AAPL", rl.leveraged_etf_list)
self.assertIn("GOOG", rl.leveraged_etf_list)
self.assertIn("BZQ", rl.leveraged_etf_list)
self.assertIn("URTY", rl.leveraged_etf_list)
finally:
remove_data_directory()
remove_security_data_directory()
def test_security_add_delete(self):
try:
def get_datetime():
return datetime(2015, 1, 27, tzinfo=pytz.utc)
add_data([], ['BZQ', 'URTY'])
add_security_data([], ['BZQ', 'URTY'])
rl = SecurityListSet(get_datetime)
self.assertNotIn("BZQ", rl.LEVERAGED_ETF_LIST)
self.assertNotIn("URTY", rl.LEVERAGED_ETF_LIST)
self.assertNotIn("BZQ", rl.leveraged_etf_list)
self.assertNotIn("URTY", rl.leveraged_etf_list)
finally:
remove_data_directory()
remove_security_data_directory()
def test_algo_without_rl_violation_via_check(self):
sim_params = factory.create_simulation_parameters(
@@ -226,7 +225,7 @@ class SecurityListTestCase(TestCase):
start=LEVERAGED_ETFS.keys()[0] + timedelta(days=7), num_days=4)
try:
add_data(['AAPL'], [])
add_security_data(['AAPL'], [])
trade_history = factory.create_trade_history(
'BZQ',
[10.0, 10.0, 11.0, 11.0],
@@ -242,13 +241,13 @@ class SecurityListTestCase(TestCase):
self.check_algo_exception(algo, ctx, 0)
finally:
remove_data_directory()
remove_security_data_directory()
def test_algo_without_rl_violation_after_delete(self):
try:
# add a delete statement removing bzq
# write a new delete statement file to disk
add_data([], ['BZQ'])
add_security_data([], ['BZQ'])
sim_params = factory.create_simulation_parameters(
start=self.extra_knowledge_date, num_days=3)
@@ -264,11 +263,11 @@ class SecurityListTestCase(TestCase):
sid='BZQ', sim_params=sim_params)
algo.run(self.source)
finally:
remove_data_directory()
remove_security_data_directory()
def test_algo_with_rl_violation_after_add(self):
try:
add_data(['AAPL'], [])
add_security_data(['AAPL'], [])
sim_params = factory.create_simulation_parameters(
start=self.trading_day_before_first_kd, num_days=4)
trade_history = factory.create_trade_history(
@@ -286,7 +285,7 @@ class SecurityListTestCase(TestCase):
self.check_algo_exception(algo, ctx, 2)
finally:
remove_data_directory()
remove_security_data_directory()
def check_algo_exception(self, algo, ctx, expected_order_count):
self.assertEqual(algo.order_count, expected_order_count)
@@ -294,30 +293,3 @@ class SecurityListTestCase(TestCase):
self.assertEqual(TradingControlViolation, type(exc))
exc_msg = str(ctx.exception)
self.assertTrue("RestrictedListOrder" in exc_msg)
def add_data(adds, deletes):
directory = os.path.join(
SECURITY_LISTS_DIR,
"leveraged_etf_list/20150127/20150125"
)
if not os.path.exists(directory):
os.makedirs(directory)
del_path = os.path.join(directory, "delete.txt")
with open(del_path, 'w') as f:
for sym in deletes:
f.write(sym)
f.write('\n')
add_path = os.path.join(directory, "add.txt")
with open(add_path, 'w') as f:
for sym in adds:
f.write(sym)
f.write('\n')
def remove_data_directory():
directory = os.path.join(
SECURITY_LISTS_DIR,
"leveraged_etf_list/20150127/"
)
shutil.rmtree(directory)