本文整理汇总了Python中zipline.data.minute_bars.BcolzMinuteBarReader类的典型用法代码示例。如果您正苦于以下问题:Python BcolzMinuteBarReader类的具体用法?Python BcolzMinuteBarReader怎么用?Python BcolzMinuteBarReader使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了BcolzMinuteBarReader类的15个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_unadjusted_minutes_early_close
def test_unadjusted_minutes_early_close(self):
"""
Test unadjusted minute window, ensuring that early closes are filtered
out.
"""
day_before_thanksgiving = Timestamp('2015-11-25', tz='UTC')
xmas_eve = Timestamp('2015-12-24', tz='UTC')
market_day_after_xmas = Timestamp('2015-12-28', tz='UTC')
minutes = [self.market_closes[day_before_thanksgiving] -
Timedelta('2 min'),
self.market_closes[xmas_eve] - Timedelta('1 min'),
self.market_opens[market_day_after_xmas] +
Timedelta('1 min')]
sids = [1, 2]
data_1 = DataFrame(
data={
'open': [
15.0, 15.1, 15.2],
'high': [17.0, 17.1, 17.2],
'low': [11.0, 11.1, 11.3],
'close': [14.0, 14.1, 14.2],
'volume': [1000, 1001, 1002],
},
index=minutes)
self.writer.write_sid(sids[0], data_1)
data_2 = DataFrame(
data={
'open': [25.0, 25.1, 25.2],
'high': [27.0, 27.1, 27.2],
'low': [21.0, 21.1, 21.2],
'close': [24.0, 24.1, 24.2],
'volume': [2000, 2001, 2002],
},
index=minutes)
self.writer.write_sid(sids[1], data_2)
reader = BcolzMinuteBarReader(self.dest)
columns = ['open', 'high', 'low', 'close', 'volume']
sids = [sids[0], sids[1]]
arrays = list(map(transpose, reader.load_raw_arrays(
columns, minutes[0], minutes[-1], sids,
)))
data = {sids[0]: data_1, sids[1]: data_2}
start_minute_loc = \
self.trading_calendar.all_minutes.get_loc(minutes[0])
minute_locs = [
self.trading_calendar.all_minutes.get_loc(minute)
- start_minute_loc
for minute in minutes
]
for i, col in enumerate(columns):
for j, sid in enumerate(sids):
assert_almost_equal(data[sid].loc[minutes, col],
arrays[i][j][minute_locs])
开发者ID:jeyoor,项目名称:zipline,代码行数:60,代码来源:test_minute_bars.py
示例2: test_minute_updates
def test_minute_updates(self):
"""
Test minute updates.
"""
start_minute = self.market_opens[TEST_CALENDAR_START]
minutes = [start_minute,
start_minute + Timedelta('1 min'),
start_minute + Timedelta('2 min')]
sids = [1, 2]
data_1 = DataFrame(
data={
'open': [15.0, nan, 15.1],
'high': [17.0, nan, 17.1],
'low': [11.0, nan, 11.1],
'close': [14.0, nan, 14.1],
'volume': [1000, 0, 1001]
},
index=minutes)
data_2 = DataFrame(
data={
'open': [25.0, nan, 25.1],
'high': [27.0, nan, 27.1],
'low': [21.0, nan, 21.1],
'close': [24.0, nan, 24.1],
'volume': [2000, 0, 2001]
},
index=minutes)
frames = {1: data_1, 2: data_2}
update_path = self.instance_tmpdir.getpath('updates.h5')
update_writer = H5MinuteBarUpdateWriter(update_path)
update_writer.write(frames)
update_reader = H5MinuteBarUpdateReader(update_path)
self.writer.write(update_reader.read(minutes, sids))
# Refresh the reader since truncate update the metadata.
reader = BcolzMinuteBarReader(self.dest)
columns = ['open', 'high', 'low', 'close', 'volume']
sids = [sids[0], sids[1]]
arrays = list(map(transpose, reader.load_raw_arrays(
columns, minutes[0], minutes[-1], sids,
)))
data = {sids[0]: data_1, sids[1]: data_2}
for i, col in enumerate(columns):
for j, sid in enumerate(sids):
assert_almost_equal(data[sid][col], arrays[i][j])
开发者ID:FranSal,项目名称:zipline,代码行数:51,代码来源:test_minute_bars.py
示例3: test_truncate_all_data_points
def test_truncate_all_data_points(self):
tds = self.market_opens.index
days = tds[tds.slice_indexer(start=self.test_calendar_start + 1, end=self.test_calendar_start + 3)]
minutes = DatetimeIndex(
[self.market_opens[days[0]] + timedelta(minutes=60), self.market_opens[days[1]] + timedelta(minutes=120)]
)
sid = 1
data = DataFrame(
data={
"open": [10.0, 11.0],
"high": [20.0, 21.0],
"low": [30.0, 31.0],
"close": [40.0, 41.0],
"volume": [50.0, 51.0],
},
index=minutes,
)
self.writer.write_sid(sid, data)
# Truncate to first day in the calendar, a day before the first
# day with minute data.
self.writer.truncate(self.test_calendar_start)
# Refresh the reader since truncate update the metadata.
self.reader = BcolzMinuteBarReader(self.dest)
self.assertEqual(self.writer.last_date_in_output_for_sid(sid), self.test_calendar_start)
cal = self.trading_calendar
_, last_close = cal.open_and_close_for_session(self.test_calendar_start)
self.assertEqual(self.reader.last_available_dt, last_close)
开发者ID:quantopian,项目名称:zipline,代码行数:32,代码来源:test_minute_bars.py
示例4: test_truncate_between_data_points
def test_truncate_between_data_points(self):
tds = self.market_opens.index
days = tds[tds.slice_indexer(
start=self.test_calendar_start + 1,
end=self.test_calendar_start + 3
)]
minutes = DatetimeIndex([
self.market_opens[days[0]] + timedelta(minutes=60),
self.market_opens[days[1]] + timedelta(minutes=120),
])
sid = 1
data = DataFrame(
data={
'open': [10.0, 11.0],
'high': [20.0, 21.0],
'low': [30.0, 31.0],
'close': [40.0, 41.0],
'volume': [50.0, 51.0]
},
index=minutes)
self.writer.write_sid(sid, data)
# Open a new writer to cover `open` method, also truncating only
# applies to an existing directory.
writer = BcolzMinuteBarWriter.open(self.dest)
# Truncate to first day with data.
writer.truncate(days[0])
# Refresh the reader since truncate update the metadata.
self.reader = BcolzMinuteBarReader(self.dest)
self.assertEqual(self.writer.last_date_in_output_for_sid(sid), days[0])
cal = self.trading_calendar
_, last_close = cal.open_and_close_for_session(days[0])
self.assertEqual(self.reader.last_available_dt, last_close)
minute = minutes[0]
open_price = self.reader.get_value(sid, minute, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute, 'volume')
self.assertEquals(50.0, volume_price)
开发者ID:FranSal,项目名称:zipline,代码行数:60,代码来源:test_minute_bars.py
示例5: test_append_on_new_day
def test_append_on_new_day(self):
sid = 1
ohlcv = {
'open': [2.0],
'high': [3.0],
'low': [1.0],
'close': [2.0],
'volume': [10.0]
}
dt = self.market_opens[TEST_CALENDAR_STOP]
data = DataFrame(
data=ohlcv,
index=[dt])
self.writer.write_sid(sid, data)
# Open a new writer to cover `open` method, also a common usage
# of appending new days will be writing to an existing directory.
cday = self.trading_calendar.schedule.index.freq
new_end_session = TEST_CALENDAR_STOP + cday
writer = BcolzMinuteBarWriter.open(self.dest, new_end_session)
next_day_minute = dt + cday
new_data = DataFrame(
data=ohlcv,
index=[next_day_minute])
writer.write_sid(sid, new_data)
# Get a new reader to test updated calendar.
reader = BcolzMinuteBarReader(self.dest)
second_minute = dt + Timedelta(minutes=1)
# The second minute should have been padded with zeros
for col in ('open', 'high', 'low', 'close'):
assert_almost_equal(
nan, reader.get_value(sid, second_minute, col)
)
self.assertEqual(
0, reader.get_value(sid, second_minute, 'volume')
)
# The next day minute should have data.
for col in ('open', 'high', 'low', 'close', 'volume'):
assert_almost_equal(
ohlcv[col], reader.get_value(sid, next_day_minute, col)
)
开发者ID:FranSal,项目名称:zipline,代码行数:47,代码来源:test_minute_bars.py
示例6: init_instance_fixtures
def init_instance_fixtures(self):
super(BcolzMinuteBarTestCase, self).init_instance_fixtures()
self.dest = self.instance_tmpdir.getpath("minute_bars")
os.makedirs(self.dest)
self.writer = BcolzMinuteBarWriter(
self.dest, self.trading_calendar, TEST_CALENDAR_START, TEST_CALENDAR_STOP, US_EQUITIES_MINUTES_PER_DAY
)
self.reader = BcolzMinuteBarReader(self.dest)
开发者ID:quantopian,项目名称:zipline,代码行数:9,代码来源:test_minute_bars.py
示例7: test_unadjusted_minutes
def test_unadjusted_minutes(self):
"""
Test unadjusted minutes.
"""
start_minute = self.market_opens[TEST_CALENDAR_START]
minutes = [start_minute,
start_minute + Timedelta('1 min'),
start_minute + Timedelta('2 min')]
sids = [1, 2]
data_1 = DataFrame(
data={
'open': [15.0, nan, 15.1],
'high': [17.0, nan, 17.1],
'low': [11.0, nan, 11.1],
'close': [14.0, nan, 14.1],
'volume': [1000, 0, 1001]
},
index=minutes)
self.writer.write_sid(sids[0], data_1)
data_2 = DataFrame(
data={
'open': [25.0, nan, 25.1],
'high': [27.0, nan, 27.1],
'low': [21.0, nan, 21.1],
'close': [24.0, nan, 24.1],
'volume': [2000, 0, 2001]
},
index=minutes)
self.writer.write_sid(sids[1], data_2)
reader = BcolzMinuteBarReader(self.dest)
columns = ['open', 'high', 'low', 'close', 'volume']
sids = [sids[0], sids[1]]
arrays = list(map(transpose, reader.load_raw_arrays(
columns, minutes[0], minutes[-1], sids,
)))
data = {sids[0]: data_1, sids[1]: data_2}
for i, col in enumerate(columns):
for j, sid in enumerate(sids):
assert_almost_equal(data[sid][col], arrays[i][j])
开发者ID:jeyoor,项目名称:zipline,代码行数:44,代码来源:test_minute_bars.py
示例8: setUp
def setUp(self):
self.dir_ = TempDirectory()
self.dir_.create()
self.dest = self.dir_.getpath("minute_bars")
os.makedirs(self.dest)
self.writer = BcolzMinuteBarWriter(
TEST_CALENDAR_START, self.dest, self.market_opens, US_EQUITIES_MINUTES_PER_DAY
)
self.reader = BcolzMinuteBarReader(self.dest)
开发者ID:xiechun,项目名称:zipline,代码行数:10,代码来源:test_minute_bars.py
示例9: test_truncate_between_data_points
def test_truncate_between_data_points(self):
tds = self.market_opens.index
days = tds[tds.slice_indexer(start=self.test_calendar_start + 1, end=self.test_calendar_start + 3)]
minutes = DatetimeIndex(
[self.market_opens[days[0]] + timedelta(minutes=60), self.market_opens[days[1]] + timedelta(minutes=120)]
)
sid = 1
data = DataFrame(
data={
"open": [10.0, 11.0],
"high": [20.0, 21.0],
"low": [30.0, 31.0],
"close": [40.0, 41.0],
"volume": [50.0, 51.0],
},
index=minutes,
)
self.writer.write_sid(sid, data)
# Truncate to first day with data.
self.writer.truncate(days[0])
# Refresh the reader since truncate update the metadata.
self.reader = BcolzMinuteBarReader(self.dest)
self.assertEqual(self.writer.last_date_in_output_for_sid(sid), days[0])
cal = self.trading_calendar
_, last_close = cal.open_and_close_for_session(days[0])
self.assertEqual(self.reader.last_available_dt, last_close)
minute = minutes[0]
open_price = self.reader.get_value(sid, minute, "open")
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute, "high")
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute, "low")
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute, "close")
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute, "volume")
self.assertEquals(50.0, volume_price)
开发者ID:quantopian,项目名称:zipline,代码行数:53,代码来源:test_minute_bars.py
示例10: test_write_one_ohlcv_with_ratios
def test_write_one_ohlcv_with_ratios(self):
minute = self.market_opens[self.test_calendar_start]
sid = 1
data = DataFrame(
data={"open": [10.0], "high": [20.0], "low": [30.0], "close": [40.0], "volume": [50.0]}, index=[minute]
)
# Create a new writer with `ohlc_ratios_per_sid` defined.
writer_with_ratios = BcolzMinuteBarWriter(
self.dest,
self.trading_calendar,
TEST_CALENDAR_START,
TEST_CALENDAR_STOP,
US_EQUITIES_MINUTES_PER_DAY,
ohlc_ratios_per_sid={sid: 25},
)
writer_with_ratios.write_sid(sid, data)
reader = BcolzMinuteBarReader(self.dest)
open_price = reader.get_value(sid, minute, "open")
self.assertEquals(10.0, open_price)
high_price = reader.get_value(sid, minute, "high")
self.assertEquals(20.0, high_price)
low_price = reader.get_value(sid, minute, "low")
self.assertEquals(30.0, low_price)
close_price = reader.get_value(sid, minute, "close")
self.assertEquals(40.0, close_price)
volume_price = reader.get_value(sid, minute, "volume")
self.assertEquals(50.0, volume_price)
开发者ID:quantopian,项目名称:zipline,代码行数:33,代码来源:test_minute_bars.py
示例11: init_instance_fixtures
def init_instance_fixtures(self):
super(BcolzMinuteBarTestCase, self).init_instance_fixtures()
self.dest = self.instance_tmpdir.getpath('minute_bars')
os.makedirs(self.dest)
self.writer = BcolzMinuteBarWriter(
TEST_CALENDAR_START,
self.dest,
self.market_opens,
self.market_closes,
US_EQUITIES_MINUTES_PER_DAY,
)
self.reader = BcolzMinuteBarReader(self.dest)
开发者ID:JasonGiedymin,项目名称:zipline,代码行数:13,代码来源:test_minute_bars.py
示例12: BcolzMinuteBarTestCase
class BcolzMinuteBarTestCase(WithTradingCalendars,
WithInstanceTmpDir,
ZiplineTestCase):
@classmethod
def init_class_fixtures(cls):
super(BcolzMinuteBarTestCase, cls).init_class_fixtures()
cal = cls.trading_calendar.schedule.loc[
TEST_CALENDAR_START:TEST_CALENDAR_STOP
]
cls.market_opens = cal.market_open
cls.market_closes = cal.market_close
cls.test_calendar_start = cls.market_opens.index[0]
cls.test_calendar_stop = cls.market_opens.index[-1]
def init_instance_fixtures(self):
super(BcolzMinuteBarTestCase, self).init_instance_fixtures()
self.dest = self.instance_tmpdir.getpath('minute_bars')
os.makedirs(self.dest)
self.writer = BcolzMinuteBarWriter(
self.dest,
self.trading_calendar,
TEST_CALENDAR_START,
TEST_CALENDAR_STOP,
US_EQUITIES_MINUTES_PER_DAY,
)
self.reader = BcolzMinuteBarReader(self.dest)
def test_version(self):
metadata = self.reader._get_metadata()
self.assertEquals(
metadata.version,
BcolzMinuteBarMetadata.FORMAT_VERSION,
)
def test_write_one_ohlcv(self):
minute = self.market_opens[self.test_calendar_start]
sid = 1
data = DataFrame(
data={
'open': [10.0],
'high': [20.0],
'low': [30.0],
'close': [40.0],
'volume': [50.0]
},
index=[minute])
self.writer.write_sid(sid, data)
open_price = self.reader.get_value(sid, minute, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute, 'volume')
self.assertEquals(50.0, volume_price)
def test_write_two_bars(self):
minute_0 = self.market_opens[self.test_calendar_start]
minute_1 = minute_0 + timedelta(minutes=1)
sid = 1
data = DataFrame(
data={
'open': [10.0, 11.0],
'high': [20.0, 21.0],
'low': [30.0, 31.0],
'close': [40.0, 41.0],
'volume': [50.0, 51.0]
},
index=[minute_0, minute_1])
self.writer.write_sid(sid, data)
open_price = self.reader.get_value(sid, minute_0, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute_0, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute_0, 'low')
self.assertEquals(30.0, low_price)
#.........这里部分代码省略.........
开发者ID:jeyoor,项目名称:zipline,代码行数:101,代码来源:test_minute_bars.py
示例13: BcolzMinuteBarTestCase
class BcolzMinuteBarTestCase(TestCase):
@classmethod
def setUpClass(cls):
cls.env = TradingEnvironment()
all_market_opens = cls.env.open_and_closes.market_open
all_market_closes = cls.env.open_and_closes.market_close
indexer = all_market_opens.index.slice_indexer(
start=TEST_CALENDAR_START,
end=TEST_CALENDAR_STOP
)
cls.market_opens = all_market_opens[indexer]
cls.market_closes = all_market_closes[indexer]
cls.test_calendar_start = cls.market_opens.index[0]
cls.test_calendar_stop = cls.market_opens.index[-1]
def setUp(self):
self.dir_ = TempDirectory()
self.dir_.create()
self.dest = self.dir_.getpath('minute_bars')
os.makedirs(self.dest)
self.writer = BcolzMinuteBarWriter(
TEST_CALENDAR_START,
self.dest,
self.market_opens,
self.market_closes,
US_EQUITIES_MINUTES_PER_DAY,
)
self.reader = BcolzMinuteBarReader(self.dest)
def tearDown(self):
self.dir_.cleanup()
def test_write_one_ohlcv(self):
minute = self.market_opens[self.test_calendar_start]
sid = 1
data = DataFrame(
data={
'open': [10.0],
'high': [20.0],
'low': [30.0],
'close': [40.0],
'volume': [50.0]
},
index=[minute])
self.writer.write(sid, data)
open_price = self.reader.get_value(sid, minute, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute, 'volume')
self.assertEquals(50.0, volume_price)
def test_write_two_bars(self):
minute_0 = self.market_opens[self.test_calendar_start]
minute_1 = minute_0 + timedelta(minutes=1)
sid = 1
data = DataFrame(
data={
'open': [10.0, 11.0],
'high': [20.0, 21.0],
'low': [30.0, 31.0],
'close': [40.0, 41.0],
'volume': [50.0, 51.0]
},
index=[minute_0, minute_1])
self.writer.write(sid, data)
open_price = self.reader.get_value(sid, minute_0, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute_0, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute_0, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute_0, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute_0, 'volume')
#.........这里部分代码省略.........
开发者ID:Retord,项目名称:zipline,代码行数:101,代码来源:test_minute_bars.py
示例14: BcolzMinuteBarTestCase
class BcolzMinuteBarTestCase(TestCase):
@classmethod
def setUpClass(cls):
cls.env = TradingEnvironment()
all_market_opens = cls.env.open_and_closes.market_open
indexer = all_market_opens.index.slice_indexer(
start=TEST_CALENDAR_START,
end=TEST_CALENDAR_STOP
)
cls.market_opens = all_market_opens[indexer]
cls.test_calendar_start = cls.market_opens.index[0]
cls.test_calendar_stop = cls.market_opens.index[-1]
def setUp(self):
self.dir_ = TempDirectory()
self.dir_.create()
self.dest = self.dir_.getpath('minute_bars')
os.makedirs(self.dest)
self.writer = BcolzMinuteBarWriter(
TEST_CALENDAR_START,
self.dest,
self.market_opens,
US_EQUITIES_MINUTES_PER_DAY,
)
self.reader = BcolzMinuteBarReader(self.dest)
def tearDown(self):
self.dir_.cleanup()
def test_write_one_ohlcv(self):
minute = self.market_opens[self.test_calendar_start]
sid = 1
data = DataFrame(
data={
'open': [10.0],
'high': [20.0],
'low': [30.0],
'close': [40.0],
'volume': [50.0]
},
index=[minute])
self.writer.write(sid, data)
open_price = self.reader.get_value(sid, minute, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute, 'volume')
self.assertEquals(50.0, volume_price)
def test_write_two_bars(self):
minute_0 = self.market_opens[self.test_calendar_start]
minute_1 = minute_0 + timedelta(minutes=1)
sid = 1
data = DataFrame(
data={
'open': [10.0, 11.0],
'high': [20.0, 21.0],
'low': [30.0, 31.0],
'close': [40.0, 41.0],
'volume': [50.0, 51.0]
},
index=[minute_0, minute_1])
self.writer.write(sid, data)
open_price = self.reader.get_value(sid, minute_0, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute_0, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute_0, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute_0, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute_0, 'volume')
self.assertEquals(50.0, volume_price)
#.........这里部分代码省略.........
开发者ID:exsanguinator,项目名称:zipline,代码行数:101,代码来源:test_minute_bars.py
示例15: BcolzMinuteBarTestCase
class BcolzMinuteBarTestCase(TestCase):
@classmethod
def setUpClass(cls):
cls.env = TradingEnvironment()
all_market_opens = cls.env.open_and_closes.market_open
indexer = all_market_opens.index.slice_indexer(
start=TEST_CALENDAR_START,
end=TEST_CALENDAR_STOP
)
cls.market_opens = all_market_opens[indexer]
cls.test_calendar_start = cls.market_opens.index[0]
cls.test_calendar_stop = cls.market_opens.index[-1]
def setUp(self):
self.dir_ = TempDirectory()
self.dir_.create()
self.dest = self.dir_.getpath('minute_bars')
os.makedirs(self.dest)
self.writer = BcolzMinuteBarWriter(
TEST_CALENDAR_START,
self.dest,
self.market_opens,
US_EQUITIES_MINUTES_PER_DAY,
)
self.reader = BcolzMinuteBarReader(self.dest)
def tearDown(self):
self.dir_.cleanup()
def test_write_one_ohlcv(self):
minute = self.market_opens[self.test_calendar_start]
sid = 1
data = DataFrame(
data={
'open': [10.0],
'high': [20.0],
'low': [30.0],
'close': [40.0],
'volume': [50.0]
},
index=[minute])
self.writer.write(sid, data)
open_price = self.reader.get_value(sid, minute, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute, 'volume')
self.assertEquals(50.0, volume_price)
def test_write_two_bars(self):
minute_0 = self.market_opens[self.test_calendar_start]
minute_1 = minute_0 + timedelta(minutes=1)
sid = 1
data = DataFrame(
data={
'open': [10.0, 11.0],
'high': [20.0, 21.0],
'low': [30.0, 31.0],
'close': [40.0, 41.0],
'volume': [50.0, 51.0]
},
index=[minute_0, minute_1])
self.writer.write(sid, data)
open_price = self.reader.get_value(sid, minute_0, 'open')
self.assertEquals(10.0, open_price)
high_price = self.reader.get_value(sid, minute_0, 'high')
self.assertEquals(20.0, high_price)
low_price = self.reader.get_value(sid, minute_0, 'low')
self.assertEquals(30.0, low_price)
close_price = self.reader.get_value(sid, minute_0, 'close')
self.assertEquals(40.0, close_price)
volume_price = self.reader.get_value(sid, minute_0, 'volume')
self.assertEquals(50.0, volume_price)
#.........这里部分代码省略.........
开发者ID:AlexanderAA,项目名称:zipline,代码行数:101,代码来源:test_minute_bars.py
注:本文中的zipline.data.minute_bars.BcolzMinuteBarReader类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论