forked from Vespa314/chan.py
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcsvAPI.py
More file actions
86 lines (73 loc) · 2.9 KB
/
Copy pathcsvAPI.py
File metadata and controls
86 lines (73 loc) · 2.9 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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
import os
from Common.CEnum import DATA_FIELD, KL_TYPE
from Common.ChanException import CChanException, ErrCode
from Common.CTime import CTime
from Common.func_util import str2float
from KLine.KLine_Unit import CKLine_Unit
from .CommonStockAPI import CCommonStockApi
def create_item_dict(data, column_name):
for i in range(len(data)):
data[i] = parse_time_column(data[i]) if column_name[i] == DATA_FIELD.FIELD_TIME else str2float(data[i])
return dict(zip(column_name, data))
def parse_time_column(inp):
# 20210902113000000
# 2021-09-13
if len(inp) == 10:
year = int(inp[:4])
month = int(inp[5:7])
day = int(inp[8:10])
hour = minute = 0
elif len(inp) == 17:
year = int(inp[:4])
month = int(inp[4:6])
day = int(inp[6:8])
hour = int(inp[8:10])
minute = int(inp[10:12])
elif len(inp) == 19:
year = int(inp[:4])
month = int(inp[5:7])
day = int(inp[8:10])
hour = int(inp[11:13])
minute = int(inp[14:16])
else:
raise Exception(f"unknown time column from baostock:{inp}")
return CTime(year, month, day, hour, minute)
class CSV_API(CCommonStockApi):
def __init__(self, code, k_type=KL_TYPE.K_DAY, begin_date=None, end_date=None, autype=None):
self.headers_exist = True # 第一行是否是标题，如果是数据，设置为False
self.columns = [
DATA_FIELD.FIELD_TIME,
DATA_FIELD.FIELD_OPEN,
DATA_FIELD.FIELD_HIGH,
DATA_FIELD.FIELD_LOW,
DATA_FIELD.FIELD_CLOSE,
# DATA_FIELD.FIELD_VOLUME,
# DATA_FIELD.FIELD_TURNOVER,
# DATA_FIELD.FIELD_TURNRATE,
] # 每一列字段
self.time_column_idx = self.columns.index(DATA_FIELD.FIELD_TIME)
super(CSV_API, self).__init__(code, k_type, begin_date, end_date, autype)
def get_kl_data(self):
cur_path = os.path.dirname(os.path.realpath(__file__))
file_path = f"{cur_path}/../{self.code}.csv"
if not os.path.exists(file_path):
raise CChanException(f"file not exist: {file_path}", ErrCode.SRC_DATA_NOT_FOUND)
for line_number, line in enumerate(open(file_path, 'r')):
if self.headers_exist and line_number == 0:
continue
data = line.split(",")
if len(data) != len(self.columns):
raise CChanException(f"file format error: {file_path}", ErrCode.SRC_DATA_FORMAT_ERROR)
if self.begin_date is not None and data[self.time_column_idx] < self.begin_date:
continue
if self.end_date is not None and data[self.time_column_idx] > self.end_date:
continue
yield CKLine_Unit(create_item_dict(data, self.columns))
def SetBasciInfo(self):
pass
@classmethod
def do_init(cls):
pass
@classmethod
def do_close(cls):
pass