274 lines
7.9 KiB
Python
274 lines
7.9 KiB
Python
import sys
|
|
import os
|
|
#sys.path.append(os.path.abspath("/Users/jack/Documents/GitHub/chan.py"))
|
|
sys.path.append(os.path.abspath("/Users/jack/Project/chan.py"))
|
|
from Chan import CChan
|
|
from BuySellPoint.BS_Point import CBS_Point
|
|
from ChanConfig import CChanConfig
|
|
from Common.CEnum import AUTYPE, DATA_SRC, KL_TYPE, DATA_FIELD, BSP_TYPE, FX_TYPE, BI_DIR, KLINE_DIR, SEG_DIR
|
|
from KLine.KLine_Unit import CKLine_Unit
|
|
from Common.CTime import CTime
|
|
from Common.func_util import kltype_lt_day, str2float
|
|
from Bi.Bi import CBi
|
|
from typing import Dict, List
|
|
from functools import reduce
|
|
from pandas import DataFrame
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
|
|
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,
|
|
"volume": DATA_FIELD.FIELD_VOLUME
|
|
}
|
|
return [_dict[x] for x in fileds.split(",")]
|
|
class ChanPY():
|
|
k_type = KL_TYPE.K_5M
|
|
config = CChanConfig({
|
|
"bi_strict": True,
|
|
"bi_algo": "fx",
|
|
"trigger_step": True,
|
|
"skip_step": 0,
|
|
"divergence_rate": 0.9,
|
|
"bsp2_follow_1": False,
|
|
"bsp3_follow_1": False,
|
|
"min_zs_cnt": 1,
|
|
"bs1_peak": False,
|
|
"macd_algo": "peak",
|
|
"bs_type": '1,2,3a,1p,2s,3b',
|
|
"print_warning": True,
|
|
"zs_algo": "normal",
|
|
})
|
|
chan = CChan(
|
|
code="BTC/USDT:USDT",
|
|
data_src=DATA_SRC.CCXT,
|
|
lv_list=[k_type],
|
|
config=config,
|
|
autype=AUTYPE.QFQ,
|
|
)
|
|
klu_list = []
|
|
chanIn = True
|
|
#def __init__(self, dataframe):
|
|
#self.klu_list = self.get_kl_data(dataframe)
|
|
#for klu in self.klu_list:
|
|
#self.chan.trigger_load({self.k_type: [klu]})
|
|
def add_klu(self, klu):
|
|
if klu:
|
|
self.chan.trigger_load({self.k_type: [klu]})
|
|
self.klu_list.append(klu)
|
|
def add_klu_from_dataframe(self, dataframe):
|
|
if len(dataframe) > len(self.klu_list):
|
|
klu = self.get_last_klu(dataframe)
|
|
self.chan.trigger_load({self.k_type: [klu]})
|
|
self.klu_list.append(klu)
|
|
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))
|
|
def get_last_klu(self, dataframe:DataFrame):
|
|
fields = "time,open,high,low,close,volume"
|
|
item = dataframe.iloc[-1]
|
|
date = item['date']
|
|
o = item['open']
|
|
h = item['high']
|
|
l = item['low']
|
|
c = item['close']
|
|
v = item['volume']
|
|
#time_obj = date.fromtimestamp(date)
|
|
time_str = date.strftime('%Y-%m-%d %H:%M:%S')
|
|
item_data = [
|
|
time_str,
|
|
o,
|
|
h,
|
|
l,
|
|
c,
|
|
v
|
|
]
|
|
klu = CKLine_Unit(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields)), autofix=True)
|
|
klu.set_idx(len(dataframe)-1)
|
|
return klu
|
|
def get_kl_data(self, dataframe:DataFrame):
|
|
fields = "time,open,high,low,close,volume"
|
|
klu_list = []
|
|
for i in range(0, len(dataframe)):
|
|
item = dataframe.iloc[i]
|
|
date = item['date']
|
|
o = item['open']
|
|
h = item['high']
|
|
l = item['low']
|
|
c = item['close']
|
|
v = item['volume']
|
|
#time_obj = date.fromtimestamp(date)
|
|
time_str = date.strftime('%Y-%m-%d %H:%M:%S')
|
|
item_data = [
|
|
time_str,
|
|
o,
|
|
h,
|
|
l,
|
|
c,
|
|
v
|
|
]
|
|
klu = CKLine_Unit(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields)), autofix=True)
|
|
klu.set_idx(i)
|
|
klu_list.append(klu)
|
|
return klu_list
|
|
def get_bsp_type(self, bsp_type, is_buy):
|
|
if is_buy:
|
|
if bsp_type == BSP_TYPE.T1:
|
|
return 1
|
|
if bsp_type == BSP_TYPE.T1P:
|
|
return 2
|
|
if bsp_type == BSP_TYPE.T2:
|
|
return 3
|
|
if bsp_type == BSP_TYPE.T2S:
|
|
return 4
|
|
if bsp_type == BSP_TYPE.T3A:
|
|
return 5
|
|
if bsp_type == BSP_TYPE.T3B:
|
|
return 6
|
|
else:
|
|
if bsp_type == BSP_TYPE.T1:
|
|
return -1
|
|
if bsp_type == BSP_TYPE.T1P:
|
|
return -2
|
|
if bsp_type == BSP_TYPE.T2:
|
|
return -3
|
|
if bsp_type == BSP_TYPE.T2S:
|
|
return -4
|
|
if bsp_type == BSP_TYPE.T3A:
|
|
return -5
|
|
if bsp_type == BSP_TYPE.T3B:
|
|
return -6
|
|
def get_bsps(self, dataframe:DataFrame):
|
|
bsps = []
|
|
updown = []
|
|
bi_sure = []
|
|
if self.chanIn:
|
|
kl_data = self.get_kl_data(dataframe)
|
|
bsp_list = []
|
|
bsp_list_pre_len = 0
|
|
last_bsp_value = 0
|
|
last_updown = -1
|
|
bi_list_pre_len = 0
|
|
pre_bi = None
|
|
zs_list_pre_len = 0
|
|
pre_zs = None
|
|
for klu in kl_data: # 获取单根K线
|
|
self.chan.trigger_load({self.k_type: [klu]}) # 喂给CChan新增k线
|
|
self.last_kline = klu
|
|
bsp_list = self.chan.get_bsp()
|
|
kl_datas = self.chan.kl_datas[self.k_type]
|
|
bi_list = kl_datas.bi_list
|
|
lst = kl_datas.lst
|
|
if len(bsp_list) > 0:
|
|
last_bsp = bsp_list[-1]
|
|
#print(klu.time, klu.idx, last_bsp.klu.time, last_bsp.klu.idx, last_bsp_value, lst[-2].fx, bi_list[-1].dir, bi_list[-1].is_sure,klu.close)
|
|
if bsp_list_pre_len > len(bsp_list):
|
|
bsps.append(99)
|
|
#print(klu.time, klu.idx, last_bsp.klu.time, last_bsp.klu.idx, last_bsp_value, 99)
|
|
else:
|
|
if bsp_list_pre_len == len(bsp_list):
|
|
if klu.idx == last_bsp.klu.idx:
|
|
last_bsp_value = self.get_bsp_type(last_bsp.type[0], last_bsp.is_buy)
|
|
bsps.append(last_bsp_value)
|
|
#print(klu.time, klu.idx, last_bsp.klu.time, last_bsp.klu.idx, last_bsp_value)
|
|
else:
|
|
bsps.append(0)
|
|
else:
|
|
last_bsp_value = self.get_bsp_type(last_bsp.type[0], last_bsp.is_buy)
|
|
bsps.append(last_bsp_value)
|
|
#print(klu.time, klu.idx, last_bsp.klu.time, last_bsp.klu.idx, last_bsp_value)
|
|
else:
|
|
bsps.append(0)
|
|
bsp_list_pre_len = len(bsp_list)
|
|
|
|
#Check zs -----------------------------------
|
|
zs_list = kl_datas.zs_list
|
|
if len(zs_list) > 0:
|
|
zs = zs_list[-1]
|
|
#if zs_list_pre_len > len(zs_list):
|
|
#print("No zs", zs.begin.time)
|
|
#if len(zs_list) > zs_list_pre_len:
|
|
#print(zs.begin.time, zs.end.time, zs.end.idx, zs.high, zs.low, zs.peak_high, zs.peak_low)
|
|
zs_list_pre_len = len(zs_list)
|
|
pre_zs = zs
|
|
|
|
#Check Bi -----------------------------------
|
|
if len(bi_list) > 0:
|
|
last_bi = bi_list[-1]
|
|
if len(bi_list) == 1:
|
|
if last_bi.dir == BI_DIR.UP:
|
|
updown.append(1)
|
|
last_updown = 1
|
|
else:
|
|
updown.append(-1)
|
|
last_updown = -1
|
|
else:
|
|
if last_updown == 1:
|
|
if last_bi.dir == BI_DIR.UP:
|
|
updown.append(0)
|
|
else:
|
|
updown.append(-1)
|
|
last_updown = -1
|
|
else:
|
|
if last_bi.dir == BI_DIR.DOWN:
|
|
updown.append(0)
|
|
else:
|
|
updown.append(1)
|
|
last_updown = 1
|
|
else:
|
|
updown.append(0)
|
|
bi_list = kl_datas.bi_list
|
|
if len(bi_list) > 0:
|
|
last_bi = bi_list[-1]
|
|
#if bi_list_pre_len > len(bi_list):
|
|
#print("Bi ", klu.time, pre_bi.idx, pre_bi.is_sure, bi_list[-1].idx, bi_list[-1].is_sure)
|
|
if last_bi.is_sure:
|
|
bi_sure.append(1)
|
|
#print(klu.time, last_bi.is_sure)
|
|
else:
|
|
bi_sure.append(0)
|
|
pre_bi = bi_list[-1]
|
|
bi_list_pre_len = len(bi_list)
|
|
else:
|
|
bi_sure.append(0)
|
|
#if bsps[-1] != 0 or updown[-1] != 0:
|
|
#print(klu.time, bsps[-1], updown[-1], bi_list[-1].is_sure)
|
|
self.chanIn = False
|
|
else:
|
|
klu = CKLine_Unit(self.create_item_dict(self.get_last_item_data(dataframe), GetColumnNameFromFieldList(fields)), autofix=True)
|
|
if self.last_kline.time < klu.time:
|
|
self.chan.trigger_load({self.k_type: [klu]}) # 喂给CChan新增k线
|
|
self.last_kline = klu
|
|
#print(bsps)
|
|
#print(updown)
|
|
kl_datas = self.chan.kl_datas[self.k_type]
|
|
#for zs in kl_datas.zs_list:
|
|
#print(zs.begin.time, zs.end.time)
|
|
return bsps, updown, bi_sure |