forked from Vespa314/chan.py
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathccxt.py
More file actions
97 lines (81 loc) · 2.94 KB
/
Copy pathccxt.py
File metadata and controls
97 lines (81 loc) · 2.94 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
87
88
89
90
91
92
93
94
95
96
97
from datetime import datetime
import ccxt
from Common.CEnum import AUTYPE, DATA_FIELD, KL_TYPE
from Common.CTime import CTime
from Common.func_util import kltype_lt_day, str2float
from KLine.KLine_Unit import CKLine_Unit
from .CommonStockAPI import CCommonStockApi
def GetColumnNameFromFieldList(fileds: str):
_dict = {
"time": DATA_FIELD.FIELD_TIME,
"open": DATA_FIELD.FIELD_OPEN,
"high": DATA_FIELD.FIELD_HIGH,
"low": DATA_FIELD.FIELD_LOW,
"close": DATA_FIELD.FIELD_CLOSE,
}
return [_dict[x] for x in fileds.split(",")]
class CCXT(CCommonStockApi):
is_connect = None
def __init__(self, code, k_type=KL_TYPE.K_DAY, begin_date=None, end_date=None, autype=AUTYPE.QFQ):
super(CCXT, self).__init__(code, k_type, begin_date, end_date, autype)
def get_kl_data(self):
fields = "time,open,high,low,close"
exchange = ccxt.binance()
timeframe = self.__convert_type()
since_date = exchange.parse8601(f'{self.begin_date}T00:00:00')
data = exchange.fetch_ohlcv(self.code, timeframe, since=since_date)
for item in data:
time_obj = datetime.fromtimestamp(item[0] / 1000)
time_str = time_obj.strftime('%Y-%m-%d %H:%M:%S')
item_data = [
time_str,
item[1],
item[2],
item[3],
item[4]
]
yield CKLine_Unit(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields)), autofix=True)
def SetBasciInfo(self):
pass
@classmethod
def do_init(cls):
pass
@classmethod
def do_close(cls):
pass
def __convert_type(self):
_dict = {
KL_TYPE.K_DAY: '1d',
KL_TYPE.K_WEEK: '1w',
KL_TYPE.K_MON: '1M',
KL_TYPE.K_5M: '5m',
KL_TYPE.K_15M: '15m',
KL_TYPE.K_30M: '30m',
KL_TYPE.K_60M: '1h',
}
return _dict[self.k_type]
def parse_time_column(self, inp):
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 TradingView:{inp}")
return CTime(year, month, day, hour, minute, auto=not kltype_lt_day(self.k_type))
def create_item_dict(self, data, column_name):
for i in range(len(data)):
data[i] = self.parse_time_column(data[i]) if i == 0 else str2float(data[i])
return dict(zip(column_name, data))