本文整理汇总了Python中zipline.utils.factory.create_test_df_source函数的典型用法代码示例。如果您正苦于以下问题:Python create_test_df_source函数的具体用法?Python create_test_df_source怎么用?Python create_test_df_source使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create_test_df_source函数的13个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_asset_date_bounds
def test_asset_date_bounds(self):
# Run the algorithm with a sid that ends far in the future
df_source, _ = factory.create_test_df_source(self.sim_params)
metadata = {0: {'start_date': '1990-01-01',
'end_date': '2020-01-01'}}
asset_finder = AssetFinder()
algo = SetAssetDateBoundsAlgorithm(
asset_finder=asset_finder,
asset_metadata=metadata,
sim_params=self.sim_params,)
algo.run(df_source)
# Run the algorithm with a sid that has already ended
df_source, _ = factory.create_test_df_source(self.sim_params)
metadata = {0: {'start_date': '1989-01-01',
'end_date': '1990-01-01'}}
asset_finder = AssetFinder()
algo = SetAssetDateBoundsAlgorithm(
asset_finder=asset_finder,
asset_metadata=metadata,
sim_params=self.sim_params,)
with self.assertRaises(TradingControlViolation):
algo.run(df_source)
# Run the algorithm with a sid that has not started
df_source, _ = factory.create_test_df_source(self.sim_params)
metadata = {0: {'start_date': '2020-01-01',
'end_date': '2021-01-01'}}
algo = SetAssetDateBoundsAlgorithm(
asset_finder=asset_finder,
asset_metadata=metadata,
sim_params=self.sim_params,)
with self.assertRaises(TradingControlViolation):
algo.run(df_source)
开发者ID:kapil0187,项目名称:zipline,代码行数:35,代码来源:test_algorithm.py
示例2: setUp
def setUp(self):
self.sim_params = factory.create_simulation_parameters(num_days=4)
trade_history = factory.create_trade_history(
133, [10.0, 10.0, 11.0, 11.0], [100, 100, 100, 300], timedelta(days=1), self.sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
self.df_source, self.df = factory.create_test_df_source(self.sim_params)
开发者ID:hackliff,项目名称:zipline,代码行数:8,代码来源:test_algorithm.py
示例3: setUp
def setUp(self):
self.sim_params = factory.create_simulation_parameters(
start=datetime(1990, 1, 1, tzinfo=pytz.utc),
end=datetime(1990, 1, 8, tzinfo=pytz.utc)
)
setup_logger(self)
self.source, self.df = \
factory.create_test_df_source(self.sim_params)
开发者ID:99plus2,项目名称:zipline,代码行数:8,代码来源:test_batchtransform.py
示例4: test_df_source
def test_df_source(self):
source, df = factory.create_test_df_source()
assert isinstance(source.start, pd.lib.Timestamp)
assert isinstance(source.end, pd.lib.Timestamp)
for expected_dt, expected_price in df.iterrows():
sid0 = next(source)
assert expected_dt == sid0.dt
assert expected_price[0] == sid0.price
开发者ID:Coinalytics,项目名称:zipline,代码行数:10,代码来源:test_sources.py
示例5: test_streaming_of_df
def test_streaming_of_df(self):
source, df = factory.create_test_df_source()
for expected_dt, expected_price in df.iterrows():
sid0 = source.next()
sid1 = source.next()
assert expected_dt == sid0.dt == sid1.dt
assert expected_price[0] == sid0.price
assert expected_price[1] == sid1.price
开发者ID:andycwang,项目名称:zipline,代码行数:10,代码来源:test_sources.py
示例6: test_algo_with_rl_violation
def test_algo_with_rl_violation(self):
sim_params = factory.create_simulation_parameters(
start=list(LEVERAGED_ETFS.keys())[0], num_days=4)
trade_history = factory.create_trade_history(
'BZQ',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
self.df_source, self.df = \
factory.create_test_df_source(sim_params)
algo = RestrictedAlgoWithoutCheck(sid='BZQ', sim_params=sim_params)
with self.assertRaises(TradingControlViolation) as ctx:
algo.run(self.source)
self.check_algo_exception(algo, ctx, 0)
# repeat with a symbol from a different lookup date
trade_history = factory.create_trade_history(
'JFT',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
self.df_source, self.df = \
factory.create_test_df_source(sim_params)
algo = RestrictedAlgoWithoutCheck(sid='JFT', sim_params=sim_params)
with self.assertRaises(TradingControlViolation) as ctx:
algo.run(self.source)
self.check_algo_exception(algo, ctx, 0)
开发者ID:1TTT9,项目名称:zipline,代码行数:41,代码来源:test_security_list.py
示例7: setUp
def setUp(self):
self.trading_environment = factory.create_trading_environment()
trade_history = factory.create_trade_history(
133,
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
self.trading_environment
)
self.source = SpecificEquityTrades(event_list=trade_history)
self.df_source, self.df = \
factory.create_test_df_source(self.trading_environment)
开发者ID:aichi,项目名称:zipline,代码行数:12,代码来源:test_algorithm.py
示例8: setUp
def setUp(self):
days = 251
self.sim_params = factory.create_simulation_parameters(num_days=days)
setup_logger(self)
trade_history = factory.create_trade_history(
133, [10.0] * days, [100] * days, timedelta(days=1), self.sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
self.df_source, self.df = factory.create_test_df_source(self.sim_params)
self.zipline_test_config = {"sid": 0}
开发者ID:ChrisBg,项目名称:zipline,代码行数:13,代码来源:test_algorithm.py
示例9: test_history_daily
def test_history_daily(self):
bar_count = 3
algo_text = """
from zipline.api import history, add_history
def initialize(context):
add_history(bar_count={bar_count}, frequency='1d', field='price')
context.history_trace = []
def handle_data(context, data):
prices = history(bar_count={bar_count}, frequency='1d', field='price')
context.history_trace.append(prices)
""".format(bar_count=bar_count).strip()
# March 2006
# Su Mo Tu We Th Fr Sa
# 1 2 3 4
# 5 6 7 8 9 10 11
# 12 13 14 15 16 17 18
# 19 20 21 22 23 24 25
# 26 27 28 29 30 31
start = pd.Timestamp('2006-03-20', tz='UTC')
end = pd.Timestamp('2006-03-30', tz='UTC')
sim_params = factory.create_simulation_parameters(
start=start, end=end, data_frequency='daily', env=self.env,
)
_, df = factory.create_test_df_source(sim_params, self.env)
df = df.astype(np.float64)
source = DataFrameSource(df)
test_algo = TradingAlgorithm(
script=algo_text,
data_frequency='daily',
sim_params=sim_params,
env=TestHistoryAlgo.env,
)
output = test_algo.run(source)
self.assertIsNotNone(output)
df.columns = self.env.asset_finder.retrieve_all(df.columns)
for i, received in enumerate(test_algo.history_trace[bar_count - 1:]):
expected = df.iloc[i:i + bar_count]
assert_frame_equal(expected, received)
开发者ID:AlexanderAA,项目名称:zipline,代码行数:48,代码来源:test_history.py
示例10: run_batchtransform
def run_batchtransform(window_length=10):
sim_params = factory.create_simulation_parameters(
start=datetime(1990, 1, 1, tzinfo=pytz.utc),
end=datetime(1995, 1, 8, tzinfo=pytz.utc)
)
source, df = factory.create_test_df_source(sim_params)
return_price_class = ReturnPriceBatchTransform(
refresh_period=1,
window_length=window_length,
clean_nans=False
)
for raw_event in source:
raw_event['datetime'] = raw_event.dt
event = {0: raw_event}
return_price_class.handle_data(event)
开发者ID:99plus2,项目名称:zipline,代码行数:17,代码来源:test_batchtransform.py
示例11: test_df_sid_filtering
def test_df_sid_filtering(self):
_, df = factory.create_test_df_source()
source = DataFrameSource(df, sids=[0])
assert 1 not in [event.sid for event in source], \
"DataFrameSource should only stream selected sid 0, not sid 1."
开发者ID:Coinalytics,项目名称:zipline,代码行数:5,代码来源:test_sources.py
示例12: setUp
def setUp(self):
setup_logger(self)
self.source, self.df = factory.create_test_df_source()
开发者ID:snth,项目名称:zipline,代码行数:3,代码来源:test_transforms.py
示例13: TestBatchTransformMinutely
self.assertEqual(df[last_elem][last_elem], last_elem)
class TestBatchTransformMinutely(TestCase):
def setUp(self):
start = pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc)
end = pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc)
self.sim_params = factory.create_simulation_parameters(
start=start,
end=end,
)
self.sim_params.emission_rate = 'daily'
self.sim_params.data_frequency = 'minute'
setup_logger(self)
self.source, self.df = \
factory.create_test_df_source(bars='minute')
def test_core(self):
algo = BatchTransformAlgorithmMinute(sim_params=self.sim_params)
algo.run(self.source)
wl = int(algo.window_length * 6.5 * 60)
for bt in algo.history[wl:]:
self.assertEqual(len(bt), wl)
def test_window_length(self):
algo = BatchTransformAlgorithmMinute(sim_params=self.sim_params,
window_length=1, refresh_period=0)
algo.run(self.source)
wl = int(algo.window_length * 6.5 * 60)
np.testing.assert_array_equal(algo.history[:(wl - 1)],
[None] * (wl - 1))
开发者ID:CarterBain,项目名称:AlephNull,代码行数:31,代码来源:test_batchtransform.py
注:本文中的zipline.utils.factory.create_test_df_source函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论