212 lines
7.4 KiB
Python
212 lines
7.4 KiB
Python
import copy
|
|
from typing import Dict, Optional
|
|
|
|
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_KLC_FX, Chan_K_DIR, Chan_MACD_STATE, Chan_PRICE_TREND
|
|
import ChanKLU
|
|
import ChanCTime
|
|
|
|
# 根据结合律合并K线后的K线
|
|
class ChanKLC():
|
|
def __init__(self, klu: ChanKLU, index, ddir=Chan_KLINE_DIR.UP):
|
|
self.start_time = klu.time
|
|
self.end_time = None
|
|
self.high = klu.high
|
|
self.low = klu.low
|
|
self.dir = ddir
|
|
self.index = index
|
|
self.klus = []
|
|
self.add_klu(klu)
|
|
self.fx = Chan_FX_TYPE.UNKNOWN
|
|
self.next = None
|
|
self.pre = None
|
|
self.start_klu = klu
|
|
self.end_klu = None
|
|
self.state = "00"
|
|
self.open = klu.open
|
|
self.close = klu.close
|
|
self.volume = klu.volume
|
|
self.bi = None
|
|
self.distance = 0
|
|
self.klc_fx_type = Chan_KLC_FX.UNKNOWN
|
|
self.rsi = klu.rsi
|
|
self.volume_ratio = klu.volume_ratio
|
|
self.macdhist = klu.macdhist
|
|
self.body = klu.body
|
|
self.upper_shadow = klu.upper_shadow
|
|
self.lower_shadow = klu.lower_shadow
|
|
self.body_ratio = klu.body_ratio
|
|
self.upper_shadow_ratio = klu.upper_shadow_ratio
|
|
self.lower_shadow_ratio = klu.lower_shadow_ratio
|
|
self.candle_dir = klu.candle_dir
|
|
self.range = klu.range
|
|
self.bb_out = True
|
|
self.macd = 0
|
|
self.signal = 0
|
|
self.state = Chan_MACD_STATE.UNKNOWN
|
|
self.continue_div = False
|
|
self.separate_div = False
|
|
self.ema52 = klu.ema52
|
|
self.ema24 = klu.ema24
|
|
self.trend = Chan_PRICE_TREND.UNKNOWN
|
|
def set_trend(self, trend):
|
|
self.trend = trend
|
|
def to_string(self):
|
|
out = ""
|
|
start = self.start_time if self.start_time is not None else ""
|
|
end = self.end_time if self.end_time is not None else ""
|
|
price_diff = getattr(self, 'price_diff', None)
|
|
out += str(start) + " " + str(end) + " " + str(self.close) + " " + str(self.ema24) + " " + str(self.ema52) + " " + str(self.trend) + " " + str(self.close - self.ema52)
|
|
return out
|
|
def set_klc_fx_type(self, klc_fx_type):
|
|
#print(self.start_time, klc_fx_type, self.get_feature_data()['klu_macd'], self.get_feature_data()['klu_macdhist'], self.get_feature_data()['klu_rsi'])
|
|
self.klc_fx_type = klc_fx_type
|
|
#self.cal_fx()
|
|
#self.cal_bb_out()
|
|
def add_klu(self, klu):
|
|
self.klus.append(klu)
|
|
def set_end_klu(self, klu):
|
|
self.end_klu = klu
|
|
self.end_time = klu.time
|
|
self.close = klu.close
|
|
for klu in self.klus:
|
|
if klu.separate_div > 0:
|
|
self.separate_div = True
|
|
if klu.continue_div:
|
|
self.continue_div = klu.continue_div
|
|
if klu.macd_state != Chan_MACD_STATE.UNKNOWN:
|
|
self.state = klu.macd_state
|
|
self.cal_indicators()
|
|
def cal_fx(self):
|
|
if self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2:
|
|
#print(self.end_time, self.fx, self.macd, self.macdhist, len(self.klus))
|
|
if self.state == Chan_MACD_STATE.HIGH_EMPTY and self.macd > 0:
|
|
#print(self.end_time, self.state, self.macd, self.klc_fx_type)
|
|
self.klc_fx_type = Chan_KLC_FX.TOP6
|
|
if self.separate_div or self.continue_div:
|
|
self.klc_fx_type = Chan_KLC_FX.TOP7
|
|
if self.signal > 0 and self.macd > self.signal:
|
|
self.klc_fx_type = Chan_KLC_FX.TOP8
|
|
else:
|
|
if self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2:
|
|
if self.macdhist > 0 and self.macd < 0:
|
|
self.klc_fx_type = Chan_KLC_FX.BOTTOM5
|
|
return
|
|
if self.state == Chan_MACD_STATE.HIGH_EMPTY and self.macd < 0:
|
|
self.klc_fx_type = Chan_KLC_FX.BOTTOM6
|
|
#print(self.end_time, self.state, self.macd, self.klc_fx_type)
|
|
if self.separate_div or self.continue_div:
|
|
self.klc_fx_type = Chan_KLC_FX.BOTTOM7
|
|
if self.signal < 0 and self.macd < self.signal:
|
|
self.klc_fx_type = Chan_KLC_FX.BOTTOM8
|
|
def cal_bb_out(self):
|
|
for klu in self.klus:
|
|
if self.high >= klu.bbup302 and klu.bbup302 > 0 and (self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2):
|
|
#print(self.end_time, self.high, klu.bbup302, self.klc_fx_type)
|
|
if self.high >= klu.bbup30 and klu.bbup30 > 0 and self.next and (self.next.macd - self.macd) < 0:
|
|
self.klc_fx_type = Chan_KLC_FX.TOP4
|
|
#self.bb_out = True
|
|
if self.low <= klu.bblow302 and klu.bblow302 > 0 and (self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2):
|
|
if self.low <= klu.bblow30 and klu.bblow30 > 0 and self.next and (self.macd - self.next.macd) < 0:
|
|
self.klc_fx_type = Chan_KLC_FX.BOTTOM4
|
|
#self.bb_out = True
|
|
def cal_indicators(self):
|
|
for index in range(1, len(self.klus)):
|
|
self.volume += self.klus[index].volume
|
|
self.rsi += self.klus[index].rsi
|
|
self.volume_ratio += self.klus[index].volume_ratio
|
|
self.macdhist += self.klus[index].macdhist
|
|
self.ema52 += self.klus[index].ema52
|
|
self.ema24 += self.klus[index].ema24
|
|
self.rsi = self.rsi / len(self.klus)
|
|
self.volume_ratio = self.volume_ratio / len(self.klus)
|
|
self.volume = self.volume / len(self.klus)
|
|
self.macdhist = self.macdhist / len(self.klus)
|
|
self.ema52 = self.ema52 / len(self.klus)
|
|
self.ema24 = self.ema24 / len(self.klus)
|
|
if len(self.klus) > 0:
|
|
self.macd = self.klus[-1].macd
|
|
self.signal = self.klus[-1].signal
|
|
|
|
self.body = abs(self.close - self.open)
|
|
self.upper_shadow = self.high - max(self.close, self.open)
|
|
self.lower_shadow = min(self.close, self.open) - self.low
|
|
self.body_ratio = self.body / self.open
|
|
self.upper_shadow_ratio = self.upper_shadow / self.open
|
|
self.lower_shadow_ratio = self.lower_shadow / self.open
|
|
self.candle_dir = Chan_K_DIR.CROSS if self.close == self.open else Chan_K_DIR.BULL if self.close > self.open else Chan_K_DIR.BEAR
|
|
self.range = self.high - self.low
|
|
def set_next(self, klc):
|
|
self.next = klc
|
|
def set_pre(self, klc):
|
|
self.pre = klc
|
|
def set_state(self, state):
|
|
self.state = state
|
|
def check_klu_included(self, klu):
|
|
if self.high >= klu.high:
|
|
# high大于，low小于，左包含
|
|
if self.low <= klu.low:
|
|
self.add_klu(klu=klu)
|
|
# gn>gn-1
|
|
if self.dir == Chan_KLINE_DIR.UP:
|
|
# UP -> max(dn)
|
|
self.low = klu.low
|
|
else:
|
|
# DOWN -> min(gn)
|
|
self.high = klu.high
|
|
#self.print(klu, "Z")
|
|
return True
|
|
# high大于，low大于，不包含
|
|
else:
|
|
# if self.low > klu.low
|
|
# high相等，右包含
|
|
if self.high == klu.high:
|
|
self.add_klu(klu=klu)
|
|
# UP -> max(gn)
|
|
if self.dir == Chan_KLINE_DIR.UP:
|
|
self.high = klu.high
|
|
else:
|
|
# DOWN -> min(dn)
|
|
self.low = klu.low
|
|
return True
|
|
else:
|
|
return False
|
|
else:
|
|
# high小于，low大于，右包含
|
|
if self.low >= klu.low:
|
|
self.add_klu(klu=klu)
|
|
# gn>gn-1
|
|
if self.dir == Chan_KLINE_DIR.UP:
|
|
# UP -> max(gn)
|
|
self.high = klu.high
|
|
else:
|
|
# DOWN -> min(dn)
|
|
self.low = klu.low
|
|
#self.print(klu, "Y")
|
|
return True
|
|
else:
|
|
# high小于，low小于，不包含
|
|
return False
|
|
def set_fx(self, fx: Chan_FX_TYPE):
|
|
self.fx = fx
|
|
def cal_invisible(self):
|
|
if self.fx == Chan_FX_TYPE.TOP:
|
|
if self.macdhist < 0 and self.macd > 0:
|
|
self.klc_fx_type = Chan_KLC_FX.TOP5
|
|
else:
|
|
if self.fx == Chan_FX_TYPE.BOTTOM:
|
|
if self.macdhist > 0 and self.macd < 0:
|
|
self.klc_fx_type = Chan_KLC_FX.BOTTOM5
|
|
def set_pre_fx(self):
|
|
if self.pre and self.pre.pre:
|
|
self.pre.fx = self.check_fx(self.pre.pre, self.pre)
|
|
def check_fx(self, k1, k2):
|
|
if k2.high > k1.high and k2.high > self.high:
|
|
return Chan_FX_TYPE.TOP
|
|
elif k2.low < k1.low and k2.low < self.low:
|
|
return Chan_FX_TYPE.BOTTOM
|
|
else:
|
|
return Chan_FX_TYPE.UNKNOWN
|
|
def set_bi(self, bi):
|
|
self.bi = bi
|
|
self.distance = self.index - bi.start_klc.index
|
|
#print(self.start_time, self.distance, bi.index, bi.dir) |