本文整理汇总了Python中zipline.utils.simfactory.create_test_zipline函数的典型用法代码示例。如果您正苦于以下问题:Python create_test_zipline函数的具体用法?Python create_test_zipline怎么用?Python create_test_zipline使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create_test_zipline函数的19个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_volshare_slippage
def test_volshare_slippage(self):
# verify order -> transaction -> portfolio position.
# --------------
test_algo = TradingAlgorithm(
script="""
from zipline.api import *
def initialize(context):
model = slippage.VolumeShareSlippage(
volume_limit=.3,
price_impact=0.05
)
set_slippage(model)
set_commission(commission.PerShare(0.02))
context.count = 2
context.incr = 0
def handle_data(context, data):
if context.incr < context.count:
# order small lots to be sure the
# order will fill in a single transaction
order(0, 5000)
record(price=data[0].price)
record(volume=data[0].volume)
record(incr=context.incr)
context.incr += 1
""",
sim_params=self.sim_params,
)
set_algo_instance(test_algo)
self.zipline_test_config['algorithm'] = test_algo
self.zipline_test_config['trade_count'] = 100
# 67 will be used inside assert_single_position
# to confirm we have as many transactions as expected.
# The algo places 2 trades of 5000 shares each. The trade
# events have volume ranging from 100 to 950. The volume cap
# of 0.3 limits the trade volume to a range of 30 - 316 shares.
# The spreadsheet linked below calculates the total position
# size over each bar, and predicts 67 txns will be required
# to fill the two orders. The number of bars and transactions
# differ because some bars result in multiple txns. See
# spreadsheet for details:
# https://www.dropbox.com/s/ulrk2qt0nrtrigb/Volume%20Share%20Worksheet.xlsx
self.zipline_test_config['expected_transactions'] = 67
zipline = simfactory.create_test_zipline(
**self.zipline_test_config)
output, _ = assert_single_position(self, zipline)
# confirm the slippage and commission on a sample
# transaction
per_share_commish = 0.02
perf = output[1]
transaction = perf['daily_perf']['transactions'][0]
commish = transaction['amount'] * per_share_commish
self.assertEqual(commish, transaction['commission'])
self.assertEqual(2.029, transaction['price'])
开发者ID:1TTT9,项目名称:zipline,代码行数:59,代码来源:test_algorithm.py
示例2: test_datasource_exception
def test_datasource_exception(self):
self.zipline_test_config["trade_source"] = ExceptionSource()
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
with self.assertRaises(ZeroDivisionError) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message, "integer division or modulo by zero")
开发者ID:goodchina,项目名称:zipline,代码行数:8,代码来源:test_exception_handling.py
示例3: test_datasource_exception
def test_datasource_exception(self):
self.zipline_test_config['trade_source'] = ExceptionSource()
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(ZeroDivisionError):
output, _ = drain_zipline(self, zipline)
开发者ID:99plus2,项目名称:zipline,代码行数:8,代码来源:test_exception_handling.py
示例4: test_tranform_exception
def test_tranform_exception(self):
exc_tnfm = StatefulTransform(ExceptionTransform)
self.zipline_test_config["transforms"] = [exc_tnfm]
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
with self.assertRaises(AssertionError) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message, "An assertion message")
开发者ID:goodchina,项目名称:zipline,代码行数:10,代码来源:test_exception_handling.py
示例5: test_fixed_slippage
def test_fixed_slippage(self):
# verify order -> transaction -> portfolio position.
# --------------
test_algo = TradingAlgorithm(
script="""
from zipline.api import (slippage,
commission,
set_slippage,
set_commission,
order,
record)
def initialize(context):
model = slippage.FixedSlippage(spread=0.10)
set_slippage(model)
set_commission(commission.PerTrade(100.00))
context.count = 1
context.incr = 0
def handle_data(context, data):
if context.incr < context.count:
order(0, -1000)
record(price=data[0].price)
context.incr += 1""",
sim_params=self.sim_params,
)
set_algo_instance(test_algo)
self.zipline_test_config['algorithm'] = test_algo
self.zipline_test_config['trade_count'] = 200
# this matches the value in the algotext initialize
# method, and will be used inside assert_single_position
# to confirm we have as many transactions as orders we
# placed.
self.zipline_test_config['order_count'] = 1
# self.zipline_test_config['transforms'] = \
# test_algo.transform_visitor.transforms.values()
zipline = simfactory.create_test_zipline(
**self.zipline_test_config)
output, _ = assert_single_position(self, zipline)
# confirm the slippage and commission on a sample
# transaction
recorded_price = output[1]['daily_perf']['recorded_vars']['price']
transaction = output[1]['daily_perf']['transactions'][0]
self.assertEqual(100.0, transaction['commission'])
expected_spread = 0.05
expected_commish = 0.10
expected_price = recorded_price - expected_spread - expected_commish
self.assertEqual(expected_price, transaction['price'])
开发者ID:erain,项目名称:zipline,代码行数:55,代码来源:test_algorithm.py
示例6: test_full_zipline
def test_full_zipline(self):
# provide enough trades to ensure all orders are filled.
self.zipline_test_config['order_count'] = 100
# making a small order amount, so that each order is filled
# in a single transaction, and txn_count == order_count.
self.zipline_test_config['order_amount'] = 25
# No transactions can be filled on the first trade, so
# we have one extra trade to ensure all orders are filled.
self.zipline_test_config['trade_count'] = 101
full_zipline = simfactory.create_test_zipline(
**self.zipline_test_config)
assert_single_position(self, full_zipline)
开发者ID:litespeeddi2,项目名称:zipline,代码行数:12,代码来源:test_finance.py
示例7: test_order_methods
def test_order_methods(self):
"""Only test that order methods can be called without error.
Correct filling of orders is tested in zipline.
"""
test_algo = TradingAlgorithm(script=call_all_order_methods, sim_params=self.sim_params)
set_algo_instance(test_algo)
self.zipline_test_config["algorithm"] = test_algo
self.zipline_test_config["trade_count"] = 200
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
output, _ = drain_zipline(self, zipline)
开发者ID:ChrisBg,项目名称:zipline,代码行数:13,代码来源:test_algorithm.py
示例8: test_exception_in_handle_data
def test_exception_in_handle_data(self):
# Simulation
# ----------
self.zipline_test_config["algorithm"] = ExceptionAlgorithm(
"handle_data", self.zipline_test_config["sid"], sim_params=factory.create_simulation_parameters()
)
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
with self.assertRaises(Exception) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message, "Algo exception in handle_data")
开发者ID:goodchina,项目名称:zipline,代码行数:13,代码来源:test_exception_handling.py
示例9: _algo_record_float_magic_should_pass
def _algo_record_float_magic_should_pass(self, var_type):
test_algo = TradingAlgorithm(script=record_float_magic % var_type, sim_params=self.sim_params)
set_algo_instance(test_algo)
self.zipline_test_config["algorithm"] = test_algo
self.zipline_test_config["trade_count"] = 200
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
output, _ = drain_zipline(self, zipline)
self.assertEqual(len(output), 252)
incr = []
for o in output[:200]:
incr.append(o["daily_perf"]["recorded_vars"]["data"])
np.testing.assert_array_equal(incr, [np.nan] * 200)
开发者ID:ChrisBg,项目名称:zipline,代码行数:14,代码来源:test_algorithm.py
示例10: test_zerodivision_exception_in_handle_data
def test_zerodivision_exception_in_handle_data(self):
# Simulation
# ----------
self.zipline_test_config["algorithm"] = DivByZeroAlgorithm(
self.zipline_test_config["sid"], sim_params=factory.create_simulation_parameters()
)
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
with self.assertRaises(ZeroDivisionError) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message, "integer division or modulo by zero")
开发者ID:goodchina,项目名称:zipline,代码行数:14,代码来源:test_exception_handling.py
示例11: test_algo_record_vars
def test_algo_record_vars(self):
test_algo = TradingAlgorithm(script=record_variables, sim_params=self.sim_params)
set_algo_instance(test_algo)
self.zipline_test_config["algorithm"] = test_algo
self.zipline_test_config["trade_count"] = 200
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
output, _ = drain_zipline(self, zipline)
self.assertEqual(len(output), 252)
incr = []
for o in output[:200]:
incr.append(o["daily_perf"]["recorded_vars"]["incr"])
np.testing.assert_array_equal(incr, range(1, 201))
开发者ID:ChrisBg,项目名称:zipline,代码行数:15,代码来源:test_algorithm.py
示例12: test_zerodivision_exception_in_handle_data
def test_zerodivision_exception_in_handle_data(self):
# Simulation
# ----------
self.zipline_test_config['algorithm'] = \
DivByZeroAlgorithm(
self.zipline_test_config['sid'],
sim_params=factory.create_simulation_parameters()
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(ZeroDivisionError):
output, _ = drain_zipline(self, zipline)
开发者ID:99plus2,项目名称:zipline,代码行数:16,代码来源:test_exception_handling.py
示例13: test_account_in_init
def test_account_in_init(self):
"""
Test that accessing account in init doesn't break.
"""
test_algo = TradingAlgorithm(
script=access_account_in_init,
sim_params=self.sim_params,
)
set_algo_instance(test_algo)
self.zipline_test_config['algorithm'] = test_algo
self.zipline_test_config['trade_count'] = 1
zipline = simfactory.create_test_zipline(
**self.zipline_test_config)
output, _ = drain_zipline(self, zipline)
开发者ID:WQGit,项目名称:zipline,代码行数:17,代码来源:test_algorithm.py
示例14: test_set_portfolio
def test_set_portfolio(self):
"""
Are we protected against overwriting an algo's portfolio?
"""
# Simulation
# ----------
self.zipline_test_config["algorithm"] = SetPortfolioAlgorithm(
self.zipline_test_config["sid"], sim_params=factory.create_simulation_parameters()
)
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
with self.assertRaises(AttributeError) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message, "can't set attribute")
开发者ID:goodchina,项目名称:zipline,代码行数:17,代码来源:test_exception_handling.py
示例15: test_exception_in_handle_data
def test_exception_in_handle_data(self):
# Simulation
# ----------
self.zipline_test_config['algorithm'] = \
ExceptionAlgorithm(
'handle_data',
self.zipline_test_config['sid']
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(Exception) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message,
'Algo exception in handle_data')
开发者ID:Elektra58,项目名称:zipline,代码行数:18,代码来源:test_exception_handling.py
示例16: test_zerodivision_exception_in_handle_data
def test_zerodivision_exception_in_handle_data(self):
# Simulation
# ----------
self.zipline_test_config['algorithm'] = \
DivByZeroAlgorithm(
self.zipline_test_config['sid']
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(ZeroDivisionError) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message,
'integer division or modulo by zero')
开发者ID:Elektra58,项目名称:zipline,代码行数:18,代码来源:test_exception_handling.py
示例17: test_exception_in_handle_data
def test_exception_in_handle_data(self):
# Simulation
# ----------
self.zipline_test_config['algorithm'] = \
ExceptionAlgorithm(
'handle_data',
self.zipline_test_config['sid'],
sim_params=factory.create_simulation_parameters(),
env=self.env
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(Exception) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(str(ctx.exception), 'Algo exception in handle_data')
开发者ID:HectorZarate,项目名称:zipline,代码行数:19,代码来源:test_exception_handling.py
示例18: test_set_portfolio
def test_set_portfolio(self):
"""
Are we protected against overwriting an algo's portfolio?
"""
# Simulation
# ----------
self.zipline_test_config['algorithm'] = \
SetPortfolioAlgorithm(
self.zipline_test_config['sid'],
sim_params=factory.create_simulation_parameters()
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(AttributeError):
output, _ = drain_zipline(self, zipline)
开发者ID:99plus2,项目名称:zipline,代码行数:19,代码来源:test_exception_handling.py
示例19: test_full_zipline
def test_full_zipline(self):
#provide enough trades to ensure all orders are filled.
self.zipline_test_config['order_count'] = 100
self.zipline_test_config['trade_count'] = 200
zipline = simfactory.create_test_zipline(**self.zipline_test_config)
assert_single_position(self, zipline)
开发者ID:aeppert,项目名称:zipline,代码行数:6,代码来源:test_finance.py
注:本文中的zipline.utils.simfactory.create_test_zipline函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论