forked from QuantFans/quantdigger
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_data.py
More file actions
53 lines (44 loc) · 2.15 KB
/
test_data.py
File metadata and controls
53 lines (44 loc) · 2.15 KB
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# -*- coding: utf-8 -*-
import pandas as pd
import os
import unittest
from logbook import Logger
logger = Logger('test')
from quantdigger import locd, set_config
class TestDataSource(unittest.TestCase):
def test_local_data(self):
old_source = locd.source
if old_source == 'sqlite':
return
logger.info('***** 数据测试开始 *****')
set_config({ 'source': 'csv' })
target = locd.get_data('BB.TEST-1.Minute')
fname = os.path.join(os.getcwd(), 'data', '1MINUTE', 'TEST', 'CC.csv')
source = pd.read_csv(fname, parse_dates='datetime', index_col='datetime')
self.assertFalse(source.equals(target), '本地数据接口负测试失败！')
fname = os.path.join(os.getcwd(), 'data', '1MINUTE', 'TEST', 'BB.csv')
source = pd.read_csv(fname, parse_dates='datetime', index_col='datetime')
self.assertTrue(source.equals(target), '本地数据接口正测试失败！')
logger.info('-- 本地数据接口测试成功 --')
logger.info('***** 数据测试结束 *****\n')
set_config({ 'source': old_source })
def test_local_data2(self):
old_source = locd.source
if old_source == 'csv':
return
logger.info('***** 数据测试开始 *****')
set_config({ 'source': 'sqlite' })
self.assertTrue(locd.source == 'sqlite')
target = locd.get_data('BB.TEST-1.Minute')
fname = os.path.join(os.getcwd(), 'data', '1MINUTE', 'TEST', 'CC.csv')
source = pd.read_csv(fname, parse_dates='datetime', index_col='datetime')
self.assertFalse(source.equals(target), '本地数据接口负测试失败！')
fname = os.path.join(os.getcwd(), 'data', '1MINUTE', 'TEST', 'BB.csv')
source = pd.read_csv(fname, parse_dates='datetime', index_col='datetime')
self.assertTrue(source.equals(target), '本地数据接口正测试失败！')
logger.info('-- 本地数据接口测试成功 --')
logger.info('***** 数据测试结束 *****\n')
# 恢复默认值避免影响其它测试
set_config({ 'source': old_source })
if __name__ == '__main__':
unittest.main()