2419 lines
91 KiB
Python
2419 lines
91 KiB
Python
from datetime import timedelta
|
|
from pandas import DataFrame
|
|
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_SEG_DIR, Chan_ZS_DIR, Chan_BSP_DIR, Chan_BSP_TYPE, Chan_KLC_FX, Chan_MACD_STATE
|
|
from ChanKLU import ChanKLU
|
|
from ChanKLC import ChanKLC
|
|
from ChanBI import ChanBI
|
|
from ChanSBI import ChanSBI
|
|
from ChanSEG import ChanSEG
|
|
from ChanZS import ChanZS
|
|
from ChanBSP import ChanBSP
|
|
import talib.abstract as ta
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from matplotlib.dates import DateFormatter, date2num
|
|
import matplotlib.patches as patches
|
|
from technical.util import resample_to_interval
|
|
from decimal import Decimal
|
|
import xgboost as xgb
|
|
import numpy as np
|
|
from ChanMACD import ChanMACD
|
|
|
|
class ChanLun():
|
|
time1 = 1
|
|
time3 = 3
|
|
time5 = 5
|
|
time15 = 15
|
|
time30 = 30
|
|
time60 = 60
|
|
time2h = 120
|
|
time4h = 240
|
|
time6h = 360
|
|
time8h = 480
|
|
time12h = 720
|
|
time1d = 1440
|
|
timeframes = [time1, time3, time5, time15, time30, time60]
|
|
tf_df_dict = {}
|
|
def init_data(self, dataframe, ticker_indicator):
|
|
for timeframe in self.timeframes:
|
|
self.tf_df_dict[timeframe] = TF_DF(timeframe, dataframe, ticker_indicator)
|
|
def calculate_bsp(self, dataframe, ticker_indicator):
|
|
|
|
return dataframe
|
|
def get_list_by_time(self, dataframe, ticker_indicator, time):
|
|
df = None
|
|
if time > 1:
|
|
df = resample_to_interval(dataframe, ticker_indicator)
|
|
else:
|
|
df = dataframe
|
|
klu_list = self.get_klu_list(df)
|
|
klc_list = self.get_klc_list(klc_list)
|
|
bi_list = self.cal_bi_list(klc_list)
|
|
return klu_list, klc_list, bi_list
|
|
def check_fx(self, klc):
|
|
if klc.pre and klc.next:
|
|
if klc.high > klc.pre.high and klc.high > klc.next.high:
|
|
if klc.signal > 0 and klc.macd > klc.signal:
|
|
klc.set_fx(Chan_FX_TYPE.TOP)
|
|
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time,klc.fx, "TOP")
|
|
return Chan_FX_TYPE.TOP
|
|
elif klc.low < klc.pre.low and klc.low < klc.next.low:
|
|
if klc.signal < 0 and klc.macd < klc.signal:
|
|
klc.set_fx(Chan_FX_TYPE.BOTTOM)
|
|
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time,klc.fx, "BOTTOM")
|
|
return Chan_FX_TYPE.BOTTOM
|
|
return Chan_FX_TYPE.UNKNOWN
|
|
def add_indicators(self, df):
|
|
fast = 12
|
|
slow = 26
|
|
period = 9
|
|
macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period)
|
|
bb365 = ta.BBANDS(df, timeperiod=365, nbdevup=3.0, nbdevdn=3.0, matype=0)
|
|
bb120 = ta.BBANDS(df, timeperiod=120, nbdevup=3.0, nbdevdn=3.0, matype=0)
|
|
bb30 = ta.BBANDS(df, timeperiod=41, nbdevup=2.3, nbdevdn=2.3, matype=0)
|
|
bb302 = ta.BBANDS(df, timeperiod=41, nbdevup=2.0, nbdevdn=2.0, matype=0)
|
|
bb30 = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0)
|
|
bb302 = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0)
|
|
# 计算布林带中轨（移动平均线）
|
|
bb30_middle = ta.SMA(df, timeperiod=90)
|
|
|
|
# 手动计算布林带 %B 指标 (BBP)
|
|
# %B = (Price - Lower Band) / (Upper Band - Lower Band)
|
|
bbp365 = (df['close'] - bb365['lowerband']) / (bb365['upperband'] - bb365['lowerband'])
|
|
bbp120 = (df['close'] - bb120['lowerband']) / (bb120['upperband'] - bb120['lowerband'])
|
|
bbp30 = (df['close'] - bb30['lowerband']) / (bb30['upperband'] - bb30['lowerband'])
|
|
bbp302 = (df['close'] - bb302['lowerband']) / (bb302['upperband'] - bb302['lowerband'])
|
|
df['atr'] = ta.ATR(df, timeperiod=14)
|
|
df['bbup365'] = bb365['upperband']
|
|
df['bblow365'] = bb365['lowerband']
|
|
df['bbp365'] = bbp365
|
|
df['bbup120'] = bb120['upperband']
|
|
df['bblow120'] = bb120['lowerband']
|
|
df['bbp120'] = bbp120
|
|
df['bbup30'] = bb30['upperband']
|
|
df['bblow30'] = bb30['lowerband']
|
|
df['bbmiddle30'] = bb30_middle # 添加bb30中轨
|
|
df['bbp30'] = bbp30
|
|
df['bbup302'] = bb302['upperband']
|
|
df['bblow302'] = bb302['lowerband']
|
|
df['bbp302'] = bbp302
|
|
df['macd'] = macd['macd']
|
|
df['macdsignal'] = macd['macdsignal']
|
|
df['macdhist'] = macd['macdhist']
|
|
df['ema5'] = ta.EMA(df, timeperiod=5)
|
|
df['ema10'] = ta.EMA(df, timeperiod=10)
|
|
df['ema26'] = ta.EMA(df, timeperiod=26)
|
|
df['ema52'] = ta.EMA(df, timeperiod=52)
|
|
df['rsi'] = ta.RSI(df, timeperiod=14)
|
|
df['volume_ratio'] = self.cal_volume_ratio(df)
|
|
return df
|
|
def get_klc_state_list(self, dataframe):
|
|
klc_list = self.get_klc_list(dataframe)
|
|
bi_list= self.cal_bi_list(klc_list)
|
|
state_list = []
|
|
if len(klc_list) > 0:
|
|
klc_index = 0
|
|
for index in range(0, len(dataframe)):
|
|
if klc_index == len(klc_list):
|
|
klc_index = len(klc_list) - 1
|
|
klc = klc_list[klc_index]
|
|
if klc.end_klu:
|
|
if klc.end_klu.idx == index:
|
|
klc_index += 1
|
|
if klc.klc_fx_type == Chan_KLC_FX.TOP4:
|
|
state_list.append("10")
|
|
elif klc.klc_fx_type == Chan_KLC_FX.TOP5:
|
|
state_list.append("20")
|
|
#print(klc.start_time, klc.end_time, klc.klc_fx_type)
|
|
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM4:
|
|
state_list.append("-10")
|
|
#print(klc.start_time, klc.end_time, klc.klc_fx_type)
|
|
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM5:
|
|
state_list.append("-20")
|
|
else:
|
|
state_list.append("00")
|
|
else:
|
|
state_list.append("00")
|
|
else:
|
|
state_list.append("00")
|
|
else:
|
|
for index in range(0, len(dataframe)):
|
|
state_list.append("00")
|
|
return state_list
|
|
def get_klc_strength_list(self, dataframe):
|
|
klc_list = self.get_klc_list(dataframe)
|
|
bi_list = self.cal_bi_list(klc_list)
|
|
klc_strength_list = []
|
|
klc_index = 0
|
|
fx_list = []
|
|
for index in range(0, len(dataframe)):
|
|
if klc_index == len(klc_list):
|
|
klc_index = len(klc_list) - 1
|
|
klc = klc_list[klc_index]
|
|
if klc.end_klu and klc.end_klu.idx == index:
|
|
klc_index += 1
|
|
if klc.klc_fx_type == Chan_KLC_FX.TOP1 or klc.klc_fx_type == Chan_KLC_FX.TOP2:
|
|
fx_list.append(1)
|
|
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM1 or klc.klc_fx_type == Chan_KLC_FX.BOTTOM2:
|
|
fx_list.append(-1)
|
|
else:
|
|
fx_list.append(0)
|
|
klc_strength_list.append(klc.cal_fx_strength(2))
|
|
#if klc.klc_fx_type != Chan_KLC_FX.UNKNOWN and klc.cal_fx_strength() > 1:
|
|
#print(klc.start_time, klc.end_time, klc.cal_fx_strength(), klc.klc_fx_type, fx_list[-1], klc_strength_list[-1])
|
|
else:
|
|
klc_strength_list.append(0)
|
|
fx_list.append(0)
|
|
return klc_strength_list, fx_list
|
|
def get_klc_bsp_list(self, dataframe):
|
|
klc_list = self.get_klc_list(dataframe)
|
|
bi_list = self.cal_bi_list(klc_list)
|
|
bsp_list = []
|
|
klc_index = 0
|
|
last_top = None
|
|
last_bottom = None
|
|
for index in range(0, len(dataframe)):
|
|
if klc_index == len(klc_list):
|
|
klc_index = len(klc_list) - 1
|
|
klc = klc_list[klc_index]
|
|
if klc.end_klu and klc.end_klu.idx == index:
|
|
klc_index += 1
|
|
if klc.klc_fx_type == Chan_KLC_FX.TOP1 or klc.klc_fx_type == Chan_KLC_FX.TOP2:
|
|
if klc.cal_fx_strength() > 1.0 and klc.cal_fx_shape() < 4:
|
|
bsp_list.append(1)
|
|
last_top = klc
|
|
last_bottom = None
|
|
|
|
else:
|
|
bsp_list.append(0)
|
|
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM1 or klc.klc_fx_type == Chan_KLC_FX.BOTTOM2:
|
|
if klc.cal_fx_strength() > 1.0 and klc.cal_fx_shape() < 4:
|
|
bsp_list.append(-1)
|
|
last_bottom = klc
|
|
last_top = None
|
|
|
|
else:
|
|
bsp_list.append(0)
|
|
else:
|
|
if last_top:
|
|
klc_offset = klc.index - last_top.index if klc.index - last_top.index > 2 else 2
|
|
last_top_strength = last_top.cal_fx_strength(klc_offset)
|
|
if klc.high > last_top.high or (klc_offset > 2 and last_top_strength < 2):
|
|
bsp_list.append(-1)
|
|
last_top = None
|
|
else:
|
|
bsp_list.append(0)
|
|
elif last_bottom:
|
|
klc_offset = klc.index - last_bottom.index if klc.index - last_bottom.index > 2 else 2
|
|
last_bottom_strength = last_bottom.cal_fx_strength(klc_offset)
|
|
if klc.low < last_bottom.low or (klc_offset > 2 and last_bottom_strength < 2):
|
|
bsp_list.append(1)
|
|
last_bottom = None
|
|
else:
|
|
bsp_list.append(0)
|
|
else:
|
|
bsp_list.append(0)
|
|
else:
|
|
bsp_list.append(0)
|
|
return bsp_list
|
|
def get_all_state(self, df_list):
|
|
state_list = []
|
|
for df in df_list:
|
|
state_list.append(self.get_klc_state_list(df))
|
|
return state_list
|
|
def resample_bsp_list(self, bsp_list, dataframe):
|
|
bsp_index = 0
|
|
resampled_bsp_list = []
|
|
if len(bsp_list) > 0:
|
|
for index in range(0, len(dataframe)):
|
|
if bsp_index == len(bsp_list):
|
|
bsp_index = len(bsp_list) - 1
|
|
bsp = bsp_list[bsp_index]
|
|
if dataframe['date'][index].strftime('%Y-%m-%d %H:%M:%S') == bsp.klc.end_time:
|
|
if bsp.type == Chan_BSP_TYPE.T3E or bsp.type == Chan_BSP_TYPE.T3:
|
|
if bsp.dir == Chan_BSP_DIR.BUY:
|
|
resampled_bsp_list.append("-30")
|
|
#print(bsp.klc.end_time, bsp.dir, bsp.seg.dir, "BUY")
|
|
else:
|
|
if bsp.dir == Chan_BSP_DIR.SELL:
|
|
resampled_bsp_list.append("30")
|
|
#print(bsp.klc.end_time, bsp.dir, bsp.seg.dir, "SELL")
|
|
else:
|
|
resampled_bsp_list.append("00")
|
|
#print(bsp.klc.end_time, bsp.dir, bsp.seg.dir, "00")
|
|
bsp_index += 1
|
|
else:
|
|
resampled_bsp_list.append("00")
|
|
else:
|
|
for index in range(0, len(dataframe)):
|
|
resampled_bsp_list.append("00")
|
|
return resampled_bsp_list
|
|
def cal_klu_state(self, dataframe):
|
|
klc_list = self.get_klc_list(dataframe)
|
|
bi_list = self.cal_bi_list(klc_list)
|
|
klc_index = 0
|
|
state_list = []
|
|
bi_dir_list = []
|
|
for index in range(0, len(dataframe)):
|
|
klc = klc_list[klc_index]
|
|
if klc.bi and klc.bi.dir == Chan_BI_DIR.UP:
|
|
bi_dir_list.append(1)
|
|
else:
|
|
bi_dir_list.append(-1)
|
|
if klc.end_klu and klc.end_klu.idx == index:
|
|
klc.set_state("00")
|
|
if klc.klc_fx_type == Chan_KLC_FX.BOTTOM1:
|
|
klc.set_state("10")
|
|
elif klc.klc_fx_type == Chan_KLC_FX.TOP1:
|
|
klc.set_state("-10")
|
|
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM2:
|
|
klc.set_state("20")
|
|
elif klc.klc_fx_type == Chan_KLC_FX.TOP2:
|
|
klc.set_state("-20")
|
|
state_list.append(klc.state)
|
|
klc_index += 1
|
|
else:
|
|
state_list.append("00")
|
|
return state_list, bi_dir_list
|
|
def get_bi_list(self, dataframe):
|
|
bi_list = self.cal_bi_list(self.get_klc_list(dataframe))
|
|
return bi_list
|
|
def get_kl_data(self, dataframe:DataFrame):
|
|
fields = "time,open,high,low,close,volume"
|
|
klu_list = []
|
|
last_klu = None
|
|
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)
|
|
#date = date + timedelta(hours=8)
|
|
time_str = date.strftime('%Y-%m-%d %H:%M:%S')
|
|
item_data = [
|
|
time_str,
|
|
o,
|
|
h,
|
|
l,
|
|
c,
|
|
v
|
|
]
|
|
#klu = KLU(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields)))
|
|
klu = ChanKLU(time_str, o, h, l, c, v)
|
|
#print(klu.time, klu.open, klu.high, klu.low, klu.close, klu.volume)
|
|
klu.set_idx(i)
|
|
klu_list.append(klu)
|
|
if last_klu:
|
|
last_klu.set_next(klu)
|
|
klu.set_pre(last_klu)
|
|
last_klu = klu
|
|
if 'macd' in item:
|
|
klu.set_indicators(item)
|
|
return klu_list
|
|
def cal_volume_ratio(self, dataframe, window=10):
|
|
df = dataframe.copy()
|
|
# 计算过去N根K线的平均成交量
|
|
df['avg_volume'] = df['volume'].rolling(window=window).mean()
|
|
# 计算量比
|
|
df['volume_ratio'] = df['volume'] / df['avg_volume']
|
|
# 填充缺失值（前N根K线）
|
|
df['volume_ratio'] = df['volume_ratio'].fillna(1.0)
|
|
return df['volume_ratio']
|
|
def calculate_zs(self, bi_list, seg_list):
|
|
return self.get_zs_list(bi_list, seg_list)
|
|
def get_seg_list(self, bi_list):
|
|
seg_list = []
|
|
up_bi_list = []
|
|
down_bi_list = []
|
|
last_up_bi = None
|
|
last_down_bi = None
|
|
last_up_sbi = None
|
|
last_down_sbi = None
|
|
last_seg = None
|
|
up_sbi_list = []
|
|
down_sbi_list = []
|
|
look_for_bottom = False
|
|
look_for_top = False
|
|
for bi in bi_list:
|
|
#print(len(up_sbi_list), len(down_sbi_list))
|
|
if len(seg_list) > 0:
|
|
# Last seg is up
|
|
if last_seg.dir == Chan_SEG_DIR.UP:
|
|
if bi.dir == Chan_BI_DIR.DOWN:
|
|
if len(down_sbi_list) > 1:
|
|
# Check down sbi inclusion
|
|
included = last_down_sbi.check_bi_included(bi)
|
|
if not included:
|
|
down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir)
|
|
last_down_sbi.set_next(down_sbi)
|
|
last_down_sbi.set_end_bi(last_down_bi)
|
|
down_sbi.set_pre(last_down_sbi)
|
|
down_sbi_list.append(down_sbi)
|
|
fx = last_down_sbi.check_fx()
|
|
# Found top
|
|
if fx == Chan_FX_TYPE.TOP:
|
|
if look_for_top:
|
|
seg_list[-2].set_sure(bi)
|
|
look_for_top = False
|
|
#print(bi.start_time, look_for_top, "UP 1")
|
|
# Has gap and search for bottom fx
|
|
if last_down_sbi.has_fx_gap:
|
|
look_for_bottom = True
|
|
last_seg.pre_set_end_bi(bi_list[last_down_sbi.start_bi.index - 1])
|
|
seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN)
|
|
seg_list.append(seg)
|
|
last_seg.set_next(seg)
|
|
seg.set_pre(last_seg)
|
|
last_seg = seg
|
|
up_sbi_list = []
|
|
last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir)
|
|
up_sbi_list.append(last_up_sbi)
|
|
#up_sbi_list.append(last_up_sbi)
|
|
#print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 1")
|
|
#print(bi.start_time, look_for_top, "UP 2")
|
|
# No gap end SEG
|
|
else:
|
|
if look_for_bottom:
|
|
look_for_bottom = False
|
|
last_seg.set_start_bi(last_down_sbi.start_bi)
|
|
seg_list[-2].set_end_bi(bi_list[last_down_sbi.start_bi.index - 1], bi)
|
|
up_sbi_list = []
|
|
last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir)
|
|
up_sbi_list.append(last_up_sbi)
|
|
last_seg.add_bi(bi)
|
|
#up_sbi_list.append(last_up_sbi)
|
|
#print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 2")
|
|
#print(bi.start_time, look_for_top, "UP 3")
|
|
else:
|
|
last_seg.set_end_bi(bi_list[last_down_sbi.start_bi.index - 1], bi)
|
|
seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN)
|
|
seg_list.append(seg)
|
|
last_seg.set_next(seg)
|
|
seg.set_pre(last_seg)
|
|
last_seg = seg
|
|
#print(last_down_sbi.end_bi.start_time, "Normal UP SEG", last_up_sbi.start_bi.start_time, bi.start_time)
|
|
#l_up_sbi = up_sbi_list[-1]
|
|
up_sbi_list = []
|
|
last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir)
|
|
up_sbi_list.append(last_up_sbi)
|
|
#up_sbi_list.append(last_up_sbi)
|
|
#print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 3")
|
|
last_down_sbi = down_sbi
|
|
last_seg.add_bi(bi)
|
|
else:
|
|
if len(down_sbi_list) == 1:
|
|
included = last_down_sbi.check_bi_included(bi)
|
|
if not included:
|
|
down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir)
|
|
last_down_sbi.set_next(down_sbi)
|
|
last_down_sbi.set_end_bi(last_down_bi)
|
|
down_sbi.set_pre(last_down_sbi)
|
|
down_sbi_list.append(down_sbi)
|
|
last_down_sbi = down_sbi
|
|
#print(bi.start_time, look_for_top, "UP 4")
|
|
last_seg.add_bi(bi)
|
|
|
|
else:
|
|
last_down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir)
|
|
down_sbi_list.append(last_down_sbi)
|
|
last_seg.add_bi(bi)
|
|
#print(bi.start_time, look_for_top, "UP 5")
|
|
else:
|
|
if last_up_sbi:
|
|
included = last_up_sbi.check_bi_included(bi)
|
|
if not included:
|
|
up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir)
|
|
last_up_sbi.set_next(up_sbi)
|
|
last_up_sbi.set_end_bi(last_up_bi)
|
|
up_sbi.set_pre(last_up_sbi)
|
|
up_sbi_list.append(up_sbi)
|
|
last_up_sbi = up_sbi
|
|
#print(bi.start_time, look_for_top, "UP 6")
|
|
last_seg.add_bi(bi)
|
|
|
|
# Last seg is down
|
|
else:
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
if len(up_sbi_list) > 1:
|
|
# Check down sbi inclusion
|
|
included = last_up_sbi.check_bi_included(bi)
|
|
if not included:
|
|
up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir)
|
|
last_up_sbi.set_next(up_sbi)
|
|
last_up_sbi.set_end_bi(last_up_bi)
|
|
up_sbi.set_pre(last_up_sbi)
|
|
up_sbi_list.append(up_sbi)
|
|
fx = last_up_sbi.check_fx()
|
|
# Found bottom
|
|
if fx == Chan_FX_TYPE.BOTTOM:
|
|
if look_for_bottom:
|
|
seg_list[-2].set_sure(bi)
|
|
look_for_bottom = False
|
|
#print(bi.start_time, look_for_top, "DOWN 1")
|
|
# Has gap and search for bottom fx
|
|
if last_up_sbi.has_fx_gap:
|
|
look_for_top = True
|
|
last_seg.pre_set_end_bi(bi_list[last_up_sbi.start_bi.index - 1])
|
|
seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP)
|
|
seg_list.append(seg)
|
|
last_seg.set_next(seg)
|
|
seg.set_pre(last_seg)
|
|
last_seg = seg
|
|
down_sbi_list = []
|
|
last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir)
|
|
down_sbi_list.append(last_down_sbi)
|
|
#down_sbi_list.append(last_down_sbi)
|
|
#print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 1")
|
|
#print(bi.start_time, look_for_top, "DOWN 2")
|
|
# No gap end SEG
|
|
else:
|
|
if look_for_top:
|
|
look_for_top = False
|
|
last_seg.set_start_bi(last_up_sbi.start_bi)
|
|
seg_list[-2].set_end_bi(bi_list[last_up_sbi.start_bi.index - 1], bi)
|
|
down_sbi_list = []
|
|
last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir)
|
|
down_sbi_list.append(last_down_sbi)
|
|
last_seg.add_bi(bi)
|
|
#down_sbi_list.append(last_down_sbi)
|
|
#print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 2")
|
|
#print(bi.start_time, look_for_top, "DOWN 3")
|
|
else:
|
|
last_seg.set_end_bi(bi_list[last_up_sbi.start_bi.index - 1], bi)
|
|
seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP)
|
|
#print(last_up_sbi.start_bi.start_time)
|
|
last_seg.set_next(seg)
|
|
seg.set_pre(last_seg)
|
|
seg_list.append(seg)
|
|
last_seg = seg
|
|
#print(last_up_sbi.end_bi.start_time, "Normal DOWN SEG", last_down_sbi.start_bi.start_time, bi.start_time)
|
|
down_sbi_list = []
|
|
last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir)
|
|
down_sbi_list.append(last_down_sbi)
|
|
#down_sbi_list.append(last_down_sbi)
|
|
#print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 3")
|
|
last_up_sbi = up_sbi
|
|
last_seg.add_bi(bi)
|
|
else:
|
|
if len(up_sbi_list) == 1:
|
|
#last_up_sbi = up_sbi_list[-1]
|
|
included = last_up_sbi.check_bi_included(bi)
|
|
if not included:
|
|
up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir)
|
|
last_up_sbi.set_next(up_sbi)
|
|
last_up_sbi.set_end_bi(last_up_bi)
|
|
up_sbi.set_pre(last_up_sbi)
|
|
up_sbi_list.append(up_sbi)
|
|
last_up_sbi = up_sbi
|
|
last_seg.add_bi(bi)
|
|
#print(bi.start_time, look_for_top, "DOWN 4")
|
|
else:
|
|
last_up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir)
|
|
up_sbi_list.append(last_up_sbi)
|
|
last_seg.add_bi(bi)
|
|
#print(bi.start_time, look_for_top, "DOWN 5")
|
|
else:
|
|
if last_down_sbi:
|
|
included = last_down_sbi.check_bi_included(bi)
|
|
if not included:
|
|
down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir)
|
|
last_down_sbi.set_next(down_sbi)
|
|
last_down_sbi.set_end_bi(last_down_bi)
|
|
down_sbi.set_pre(last_down_sbi)
|
|
down_sbi_list.append(down_sbi)
|
|
last_down_sbi = down_sbi
|
|
last_seg.add_bi(bi)
|
|
#print(bi.start_time, look_for_top, look_for_bottom, "DOWN 6")
|
|
# len(seg_list) = 0
|
|
else:
|
|
if bi.check_overlap():
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.UP)
|
|
last_up_bi = bi
|
|
last_up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir)
|
|
seg_list.append(seg)
|
|
last_seg = seg
|
|
#print(bi.start_time, 'Create first UP SEG')
|
|
else:
|
|
seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.DOWN)
|
|
last_down_bi = bi
|
|
last_down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir)
|
|
seg_list.append(seg)
|
|
last_seg = seg
|
|
#print(bi.start_time, 'Create first DOWN SEG')
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
last_up_bi = bi
|
|
up_bi_list.append(bi)
|
|
else:
|
|
last_down_bi = bi
|
|
down_bi_list.append(bi)
|
|
"""
|
|
if len(seg_list) > 1:
|
|
seg = seg_list[-1]
|
|
last_seg = seg_list[-2]
|
|
last_seg_bi = last_seg.bi_list[-3]
|
|
bi_index = seg.start_bi.index
|
|
for i in range(bi_index, len(bi_list) - 1):
|
|
# last seg is down
|
|
if seg.dir == Chan_SEG_DIR.UP:
|
|
if bi_list[i].dir == Chan_BI_DIR.UP:
|
|
last_seg_peak = last_seg_bi.high
|
|
if bi_list[i].high > last_seg_peak:
|
|
# The confirmed
|
|
print("Last UP seg is broken, create a new seg. 1")
|
|
seg.pre_set_end_bi(bi_list[i])
|
|
seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.DOWN)
|
|
seg_list.append(seg)
|
|
last_seg = seg_list[-2]
|
|
if len(last_seg.bi_list) > 3:
|
|
last_seg_bi = last_seg.bi_list[-3]
|
|
|
|
else:
|
|
if bi_list[i].dir == Chan_BI_DIR.DOWN:
|
|
last_seg_peak = last_seg_bi.low
|
|
if bi_list[i].low < last_seg_peak:
|
|
print("Last DOWN seg is broken, create a new seg. 1")
|
|
seg.pre_set_end_bi(bi_list[i])
|
|
seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.UP)
|
|
seg_list.append(seg)
|
|
last_seg = seg_list[-2]
|
|
if len(last_seg.bi_list) > 3:
|
|
last_seg_bi = last_seg.bi_list[-3]
|
|
else:
|
|
if len(seg_list) == 1:
|
|
last_seg = seg_list[-1]
|
|
bi_index = last_seg.bi_list[0].index
|
|
for i in range(bi_index, len(bi_list) - 1):
|
|
if i > bi_index + 2:
|
|
last_seg_peak = bi_list[i-2].high
|
|
# last seg is down
|
|
if last_seg.dir == Chan_SEG_DIR.DOWN:
|
|
if bi_list[i].dir == Chan_BI_DIR.UP:
|
|
if bi_list[i].high > last_seg_peak:
|
|
print("Last seg is broken, create a new seg. 2")
|
|
last_seg.pre_set_end_bi(bi_list[i-1])
|
|
seg = ChanSEG(bi_list[i], len(seg_list), Chan_SEG_DIR.UP)
|
|
seg_list.append(seg)
|
|
last_seg = seg
|
|
last_seg_bi = bi_list[i]
|
|
break
|
|
"""
|
|
return seg_list
|
|
|
|
def cal_bi_list(self, klc_list):
|
|
bi_list = []
|
|
last_top = None
|
|
last_bottom = None
|
|
for klc in klc_list:
|
|
fx = self.check_fx(klc)
|
|
|
|
# Do nothing
|
|
if fx == Chan_FX_TYPE.UNKNOWN:
|
|
continue
|
|
if len(bi_list) > 0 and klc.end_klu:
|
|
last_bi = bi_list[-1]
|
|
#print(klc.start_time, last_bi.start_time, last_bi.end_time, last_bi.dir, last_bi.high, last_bi.low, last_bottom.end_time, "last bi")
|
|
if last_top and last_bi.dir == Chan_BI_DIR.DOWN:
|
|
if last_bottom and klc.high > last_bi.high:
|
|
last_bi.set_end_klc(last_bottom, klc)
|
|
bi = ChanBI(last_bottom, len(bi_list), Chan_BI_DIR.UP)
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM7)
|
|
#klc.bb_out = True
|
|
last_bi.set_next(bi)
|
|
bi.set_pre(last_bi)
|
|
for klc_index in range(last_bi.end_klc.index, len(klc_list)):
|
|
bi.add_klc(klc_list[klc_index])
|
|
bi_list.append(bi)
|
|
last_top = klc
|
|
klc.set_bi(bi)
|
|
#print(klc.start_time, bi.start_time, bi.end_time, bi.dir, bi.high, bi.low, bi.is_sure)
|
|
else:
|
|
if last_bottom and last_bi.dir == Chan_BI_DIR.UP:
|
|
if last_top and klc.low < last_bi.low:
|
|
last_bi.set_end_klc(last_top, klc)
|
|
bi = ChanBI(last_top, len(bi_list), Chan_BI_DIR.DOWN)
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.TOP6)
|
|
#klc.bb_out = True
|
|
last_bi.set_next(bi)
|
|
bi.set_pre(last_bi)
|
|
for klc_index in range(last_bi.end_klc.index, len(klc_list)):
|
|
bi.add_klc(klc_list[klc_index])
|
|
bi_list.append(bi)
|
|
last_bottom = klc
|
|
klc.set_bi(bi)
|
|
#print(klc.start_time, bi.start_time, bi.end_time, bi.dir, bi.high, bi.low, bi.is_sure)
|
|
else:
|
|
if fx == Chan_FX_TYPE.TOP:
|
|
if last_top:
|
|
if last_bottom:
|
|
#print(klc.start_time, last_bottom.start_time, last_top.start_time)
|
|
if last_bottom.index < last_top.index:
|
|
# Second top lower to be second sell point
|
|
if last_top.high > klc.high:
|
|
#klc.set_fx(Chan_FX_TYPE.TT)
|
|
#klc.set_state("20")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.TOP3)
|
|
klc.set_last_top_klu(last_top)
|
|
#print(klc.start_time, klc.fx, "二类卖点Sell 1")
|
|
else:
|
|
# A new top found
|
|
#last_top.set_fx(Chan_FX_TYPE.UNKNOWN)
|
|
last_top = klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 1")
|
|
klc.set_klc_fx_type(Chan_KLC_FX.TOP1)
|
|
#print(klc.end_time, klc.fx, "一类卖点Sell 1")
|
|
#klc.set_fx(fx)
|
|
#klc.set_state("10")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
else:
|
|
# 不满足结合律的分型
|
|
if last_bottom.index + 4 > klc.index:
|
|
if last_top.high > klc.high:
|
|
#print(klc.start_time, klc.fx, "二类卖点Sell 1")
|
|
#klc.set_fx(Chan_FX_TYPE.PTOP)
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
# New TOP Found replace last top
|
|
else:
|
|
if last_top.index + 4 < klc.index and len(bi_list) > 1:
|
|
pre_last_bi = bi_list[-2]
|
|
last_bi = bi_list[-1]
|
|
if pre_last_bi.is_sure and not last_bi.is_sure and pre_last_bi.dir == Chan_BI_DIR.UP and False:
|
|
pre_last_bi.update_bi(klc)
|
|
bi_list.remove(last_bi)
|
|
pre_last_bi.set_next(None)
|
|
#last_top.set_fx(Chan_FX_TYPE.PTOP)
|
|
last_top = klc
|
|
last_bottom = pre_last_bi.start_klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Bottom Change 1")
|
|
klc.set_klc_fx_type(Chan_KLC_FX.TOP2)
|
|
#print(klc.start_time, last_bi.start_klc.start_time, "New TOP Found reset last bi")
|
|
#klc.set_state("10")
|
|
#print(klc.start_time, klc.fx, "笔卖点Sell 1")
|
|
###klc.set_klc_fx_type(Chan_KLC_FX.TOP2) # when bi is down but the fx is top
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
else:
|
|
klc.set_fx(Chan_FX_TYPE.PTOP)
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "无效分型")
|
|
# 满足结合律
|
|
else:
|
|
# New Temp TOP and last bottom confirmed ***** confirm last down bi(last bottom and last top)
|
|
last_bi = bi_list[-1]
|
|
if not last_bi.is_sure:
|
|
last_bi.set_end_klc(last_bottom, klc)
|
|
bi = ChanBI(last_bottom, len(bi_list), Chan_BI_DIR.UP)
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM7)
|
|
#klc.bb_out = True
|
|
last_bi.set_next(bi)
|
|
bi.set_pre(last_bi)
|
|
bi.add_klc(klc)
|
|
bi_list.append(bi)
|
|
last_top = klc
|
|
#print(klc.end_time, klc.fx, bi_list[-1].dir, "Last Top Change 2")
|
|
klc.set_klc_fx_type(Chan_KLC_FX.TOP2)
|
|
#klc.set_state('30')
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, last_bottom.start_time, "Normal TOP Found, Confirm down bi 4")
|
|
#print(klc.start_time, klc.fx, "笔卖点Sell 2")
|
|
# last bottom = None
|
|
else:
|
|
if last_top.high < klc.high:
|
|
last_bi = bi_list[-1]
|
|
last_bi.set_start_klc(klc, Chan_BI_DIR.DOWN)
|
|
#last_top.set_fx(Chan_FX_TYPE.UNKNOWN)
|
|
last_top = klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 3")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "笔卖点Sell 3")
|
|
else:
|
|
klc.set_fx(Chan_FX_TYPE.TT)
|
|
#klc.set_state('20')
|
|
#print(klc.start_time, klc.fx, "二类卖点Sell 2")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
else:
|
|
if last_bottom:
|
|
# 不满足结合律的分型
|
|
if last_bottom.index + 4 > klc.index:
|
|
#klc.set_fx(Chan_FX_TYPE.PTOP)
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "中枢卖点Sell 1")
|
|
else:
|
|
# First temp top and last bottom confirmed
|
|
last_top = klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 4")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "一类卖点Sell 1")
|
|
# Last top = None, last bottom = None, create first down bi
|
|
else:
|
|
# First temp top
|
|
last_top = klc
|
|
bi = ChanBI(klc, len(bi_list), Chan_BI_DIR.DOWN)
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.TOP6)
|
|
#klc.bb_out = True
|
|
bi_list.append(bi)
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 5")
|
|
#print(klc.start_time, 'Create first top')
|
|
#print(klc.start_time, klc.fx, "笔卖点Sell 1")
|
|
#klc.fx = Bottom ========================
|
|
else:
|
|
if last_bottom:
|
|
if last_top:
|
|
# Bottom after top and find a new bottom
|
|
if last_top.index < last_bottom.index:
|
|
# Second bottom uppper to be second buy point and confirm last bi
|
|
if last_bottom.low < klc.low:
|
|
#klc.set_fx(Chan_FX_TYPE.BB)
|
|
#klc.set_state("-20")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM3)
|
|
klc.set_last_bottom_klc(last_bottom)
|
|
#print(last_bottom.start_time, last_bottom.end_time, "--------------------------------1")
|
|
#print(klc.start_time, klc.fx, "二类买点Buy 1")
|
|
else:
|
|
# A new bottom found
|
|
#last_bottom.set_fx(Chan_FX_TYPE.UNKNOWN)
|
|
last_bottom = klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 1")
|
|
klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM1)
|
|
#print(klc.start_time, klc.fx, "一类买点Buy 1")
|
|
#klc.set_state("-10")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
else:
|
|
# 不满足结合律的分型
|
|
if last_top.index + 4 > klc.index:
|
|
if last_bottom.low < klc.low:
|
|
#klc.set_fx(Chan_FX_TYPE.PBOTTOM)
|
|
#klc.set_fx(Chan_FX_TYPE.BB)
|
|
#klc.set_state("-100")
|
|
#print(klc.start_time, klc.fx, "中枢买点Buy 1")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
# Found new bottom
|
|
else:
|
|
if last_bottom.index + 4 < klc.index and len(bi_list) > 1:
|
|
pre_last_bi = bi_list[-2]
|
|
last_bi = bi_list[-1]
|
|
if pre_last_bi.is_sure and not last_bi.is_sure and pre_last_bi.dir == Chan_BI_DIR.DOWN and False:
|
|
pre_last_bi.update_bi(klc)
|
|
bi_list.remove(last_bi)
|
|
pre_last_bi.set_next(None)
|
|
#last_bottom.set_fx(Chan_FX_TYPE.PBOTTOM)
|
|
last_bottom = klc
|
|
last_top = pre_last_bi.start_klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Bottom Change 2")
|
|
klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM2)
|
|
#print(klc.start_time, last_bi.start_klc.start_time, "New BOTTOM Found reset last bi")
|
|
#klc.set_state("-10")
|
|
#print(klc.start_time, klc.fx, "笔买点Buy 1")
|
|
###klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM2) # when bi is up but the fx is bottom
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
else:
|
|
#klc.set_fx(Chan_FX_TYPE.UNKNOWN)
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "无效分型")
|
|
# 满足结合律的分型
|
|
else:
|
|
# New Temp Bottom and last top confirmed ***** confirm last up bi(last bottom and last top)
|
|
last_bi = bi_list[-1]
|
|
if not last_bi.is_sure:
|
|
last_bi.set_end_klc(last_top, klc)
|
|
bi = ChanBI(last_top, len(bi_list), Chan_BI_DIR.DOWN)
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.TOP6)
|
|
#klc.bb_out = True
|
|
last_bi.set_next(bi)
|
|
bi.set_pre(last_bi)
|
|
bi.add_klc(klc)
|
|
bi_list.append(bi)
|
|
last_bottom = klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 2")
|
|
klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM2)
|
|
#klc.set_state('-30')
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "笔买点Buy 2")
|
|
#print(klc.start_time, last_top.start_time, "Normal Bottom Found, Confirm up bi 6")
|
|
# last_top = None
|
|
else:
|
|
if last_bottom.low > klc.low:
|
|
last_bi = bi_list[-1]
|
|
last_bi.set_start_klc(klc, Chan_BI_DIR.UP)
|
|
#last_bottom.set_fx(Chan_FX_TYPE.UNKNOWN)
|
|
last_bottom = klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 3")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "笔买点Buy 3")
|
|
else:
|
|
klc.set_fx(Chan_FX_TYPE.BB)
|
|
#klc.set_state('-20')
|
|
#print(klc.start_time, klc.fx, "二类买点Buy 2")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
# last_bottom = None
|
|
else:
|
|
if last_top:
|
|
# 不满足结合律的分型
|
|
if last_top.index + 4 > klc.index:
|
|
#klc.set_fx(Chan_FX_TYPE.PBOTTOM)
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "中枢买点Buy 1")
|
|
else:
|
|
# First temp bottom and last top confirmed
|
|
last_bottom = klc
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 4")
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, "一类买点Buy 1")
|
|
# Last top = None, last bottom = None, create first up bi
|
|
else:
|
|
# First temp bottom and no top yet
|
|
last_bottom = klc
|
|
bi = ChanBI(klc, len(bi_list), Chan_BI_DIR.UP)
|
|
#klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM7)
|
|
#klc.bb_out = True
|
|
bi_list.append(bi)
|
|
bi_list[-1].add_klc(klc)
|
|
klc.set_bi(bi_list[-1])
|
|
#print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 5")
|
|
#print(klc.start_time, klc.fx, "笔买点Buy 4")
|
|
#if klc.fx != Chan_FX_TYPE.UNKNOWN:
|
|
#print(klc.start_time, klc.fx, klc.index)
|
|
"""
|
|
for klc in klc_list:
|
|
if klc.fx == Chan_FX_TYPE.TOP:
|
|
klc.state = "10"
|
|
#print(klc.time, klc.state)
|
|
if klc.fx == Chan_FX_TYPE.BOTTOM:
|
|
klc.state = "-10"
|
|
#print(klc.time, klc.state)
|
|
"""
|
|
#for index in range(0, 10):
|
|
#print(bi_list[index].start_time, bi_list[index].start_klc.start_time, bi_list[index].dir)
|
|
return bi_list
|
|
|
|
def get_zs_list(self, bi_list, seg_list):
|
|
zs_list = []
|
|
bsp_list = []
|
|
if len(seg_list) > 3:
|
|
last_zs = None
|
|
first_bi_out = None
|
|
in_again = False
|
|
bi_out_count = 0
|
|
zs_count = 0
|
|
for seg in seg_list:
|
|
# No zs or Last ZS is completed
|
|
if len(zs_list) == 0 or (last_zs and last_zs.is_sure):
|
|
# Has three completed segments
|
|
if seg.next and seg.next.next:
|
|
if seg.next.next.is_sure:
|
|
zg = min(seg.high, seg.next.high, seg.next.next.high)
|
|
zd = max(seg.low, seg.next.low, seg.next.next.low)
|
|
gg = max(seg.high, seg.next.high, seg.next.next.high)
|
|
dd = min(seg.low, seg.next.low, seg.next.next.low)
|
|
ddir = Chan_ZS_DIR.UP
|
|
ddir = None
|
|
if last_zs:
|
|
if zg < last_zs.zd:
|
|
ddir = Chan_ZS_DIR.DOWN
|
|
else:
|
|
if zd > last_zs.zg:
|
|
ddir = Chan_ZS_DIR.UP
|
|
else:
|
|
ddir = None
|
|
else:
|
|
if seg.dir == Chan_SEG_DIR.UP:
|
|
ddir = Chan_ZS_DIR.DOWN
|
|
else:
|
|
ddir = Chan_ZS_DIR.UP
|
|
if (seg.dir == Chan_SEG_DIR.DOWN and ddir == Chan_ZS_DIR.DOWN) or (seg.dir == Chan_SEG_DIR.UP and ddir == Chan_ZS_DIR.UP):
|
|
ddir = None
|
|
if ddir and zg > zd:
|
|
# New ZS
|
|
zs = ChanZS(seg, len(zs_list), ddir)
|
|
zs.set_zg(zg)
|
|
zs.set_zd(zd)
|
|
zs.set_gg(gg)
|
|
zs.set_dd(dd)
|
|
if last_zs:
|
|
last_zs.set_next(zs)
|
|
zs.set_pre(last_zs)
|
|
zs_list.append(zs)
|
|
if last_zs and last_zs.dir == zs.dir:
|
|
zs_count += 1
|
|
else:
|
|
zs_count = 1
|
|
last_zs = zs
|
|
# Last ZS is not completed
|
|
else:
|
|
# Last ZS is not completed
|
|
if last_zs and not last_zs.is_sure:
|
|
if first_bi_out:
|
|
# SEG is not in ZS
|
|
if seg.is_sure:
|
|
if ((seg.low > last_zs.zg and seg.high > last_zs.zg) or (seg.high < last_zs.zd and seg.low < last_zs.zd)):
|
|
last_zs.set_end_klc(last_zs.last_bi_in.end_klc, seg.sure_time, bi_out_count, seg)
|
|
bi_out_count = 0
|
|
#print(seg.start_bi.start_klc.start_time)
|
|
first_bi_out = None
|
|
# Last ZS is completed and look for new ZS
|
|
if seg.next and seg.next.next:
|
|
if seg.next.next.is_sure:
|
|
zg = min(seg.high, seg.next.high, seg.next.next.high)
|
|
zd = max(seg.low, seg.next.low, seg.next.next.low)
|
|
gg = max(seg.high, seg.next.high, seg.next.next.high)
|
|
dd = min(seg.low, seg.next.low, seg.next.next.low)
|
|
ddir = None
|
|
if last_zs:
|
|
if zg < last_zs.zd:
|
|
ddir = Chan_ZS_DIR.DOWN
|
|
else:
|
|
if zd > last_zs.zg:
|
|
ddir = Chan_ZS_DIR.UP
|
|
else:
|
|
ddir = None
|
|
else:
|
|
if seg.dir == Chan_SEG_DIR.UP:
|
|
ddir = Chan_ZS_DIR.DOWN
|
|
else:
|
|
ddir = Chan_ZS_DIR.UP
|
|
if (seg.dir == Chan_SEG_DIR.DOWN and ddir == Chan_ZS_DIR.DOWN) or (seg.dir == Chan_SEG_DIR.UP and ddir == Chan_ZS_DIR.UP):
|
|
ddir = None
|
|
if ddir and zg > zd:
|
|
# New ZS
|
|
zs = ChanZS(seg, len(zs_list), ddir)
|
|
zs.set_zg(zg)
|
|
zs.set_zd(zd)
|
|
zs.set_gg(gg)
|
|
zs.set_dd(dd)
|
|
last_zs.set_next(zs)
|
|
zs.set_pre(last_zs)
|
|
zs_list.append(zs)
|
|
if last_zs and last_zs.dir == zs.dir:
|
|
zs_count += 1
|
|
else:
|
|
zs_count = 1
|
|
last_zs = zs
|
|
# Last SEG is in ZS
|
|
else:
|
|
# SEG is inside ZS
|
|
if seg.end_bi:
|
|
for index in range(seg.start_bi.index, seg.end_bi.index+1):
|
|
bi = bi_list[index]
|
|
if (bi.high >= last_zs.zd and bi.high <= last_zs.zg) or (bi.low >= last_zs.zd and bi.low <= last_zs.zg) or (bi.high >= last_zs.zg and bi.low <= last_zs.zd):
|
|
in_again = True
|
|
last_zs.set_bi_out(None, None)
|
|
last_zs.set_last_bi_in(None)
|
|
last_zs.set_end_seg(None)
|
|
first_bi_out = None
|
|
#print("Bi in again 3", bi.start_klc.start_time)
|
|
if in_again and (bi.low > last_zs.zg or bi.high < last_zs.zd):
|
|
last_zs.set_bi_out(bi, seg)
|
|
last_zs.set_last_bi_in(bi_list[index - 1])
|
|
#last_zs.set_end_seg(seg.next.next)
|
|
bi_out_count += 1
|
|
first_bi_out = bi
|
|
if (bi.dir == Chan_BI_DIR.UP and seg.dir == Chan_SEG_DIR.DOWN) or (bi.dir == Chan_BI_DIR.DOWN and seg.dir == Chan_SEG_DIR.UP):
|
|
bsp = ChanBSP(first_bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if first_bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, first_bi_out.sure_time, zs_count, zs, seg)
|
|
bsp_list.append(bsp)
|
|
#print("First bi out 3", first_bi_out.start_klc.start_time)
|
|
in_again = False
|
|
""""
|
|
if first_bi_out:
|
|
if seg.dir == Chan_SEG_DIR.UP and bi.dir == Chan_BI_DIR.UP:
|
|
#print(bi.start_klc.start_time, bi.high, seg.high)
|
|
if bi.high == seg.high:
|
|
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.SELL if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.BUY, bi.sure_time, zs_count, zs, seg)
|
|
bsp_list.append(bsp)
|
|
else:
|
|
if seg.dir == Chan_SEG_DIR.DOWN and bi.dir == Chan_BI_DIR.DOWN:
|
|
if bi.low == seg.low:
|
|
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.BUY if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi.sure_time, zs_count, zs, seg)
|
|
bsp_list.append(bsp)
|
|
"""
|
|
else:
|
|
# SEG in ZS and not out and find first bi out
|
|
if seg.end_bi:
|
|
for index in range(seg.start_bi.index, seg.end_bi.index+1):
|
|
bi = bi_list[index]
|
|
if (bi.high >= last_zs.zd and bi.high <= last_zs.zg) or (bi.low >= last_zs.zd and bi.low <= last_zs.zg) or (bi.high >= last_zs.zg and bi.low <= last_zs.zd):
|
|
in_again = True
|
|
last_zs.set_bi_out(None, None)
|
|
last_zs.set_last_bi_in(None)
|
|
last_zs.set_end_seg(None)
|
|
first_bi_out = None
|
|
#print("Bi in again 4", bi.start_klc.start_time)
|
|
if in_again and (bi.low > last_zs.zg or bi.high < last_zs.zd):
|
|
last_zs.set_bi_out(bi, seg)
|
|
last_zs.set_last_bi_in(bi_list[index - 1])
|
|
#last_zs.set_end_seg(seg.next.next)
|
|
bi_out_count += 1
|
|
first_bi_out = bi
|
|
if (bi.dir == Chan_BI_DIR.UP and seg.dir == Chan_SEG_DIR.DOWN) or (bi.dir == Chan_BI_DIR.DOWN and seg.dir == Chan_SEG_DIR.UP):
|
|
bsp = ChanBSP(first_bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if first_bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, first_bi_out.sure_time, zs_count, zs, seg)
|
|
bsp_list.append(bsp)
|
|
#print("First bi out 4", first_bi_out.start_klc.start_time)
|
|
in_again = False
|
|
if first_bi_out:
|
|
if seg.dir == Chan_SEG_DIR.UP and bi.dir == Chan_BI_DIR.UP:
|
|
#print(bi.start_klc.start_time, bi.high, seg.high)
|
|
if bi.high == seg.high:
|
|
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.SELL if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.BUY, bi.sure_time, zs_count, zs, seg)
|
|
bsp_list.append(bsp)
|
|
else:
|
|
if seg.dir == Chan_SEG_DIR.DOWN and bi.dir == Chan_BI_DIR.DOWN:
|
|
if bi.low == seg.low:
|
|
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.BUY if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi.sure_time, zs_count, zs, seg)
|
|
bsp_list.append(bsp)
|
|
#self.print_zs(zs_list)
|
|
return zs_list
|
|
|
|
def get_bi_macdhist_list(self, bi_list, dataframe):
|
|
bi_macdhist_list = []
|
|
for bi in bi_list:
|
|
start_index = bi.start_klc.start_klu.index
|
|
if bi.end_klc:
|
|
end_index = bi.end_klc.end_klu.index
|
|
else:
|
|
end_index = len(dataframe) - 1
|
|
total_macd_hist = 0
|
|
for index in range(start_index, end_index+1):
|
|
macd_hist = dataframe['macdhist'][index]
|
|
if bi.dir == Chan_BI_DIR.UP and macd_hist > 0:
|
|
total_macd_hist += macd_hist
|
|
if bi.dir == Chan_BI_DIR.DOWN and macd_hist < 0:
|
|
total_macd_hist -= macd_hist
|
|
bi_macdhist_list.append(abs(total_macd_hist))
|
|
bi.set_macdhist(total_macd_hist)
|
|
return bi_macdhist_list, bi_list
|
|
|
|
def get_seg_macdhist_list(self, seg_list, dataframe):
|
|
seg_macdhist_list = []
|
|
for seg in seg_list:
|
|
start_index = seg.start_bi.start_klc.start_klu.index
|
|
if seg.end_bi:
|
|
end_index = seg.end_bi.end_klc.end_klu.index
|
|
else:
|
|
end_index = len(dataframe) - 1
|
|
total_macd_hist = 0
|
|
for index in range(start_index, end_index+1):
|
|
macd_hist = dataframe['macdhist'][index]
|
|
if seg.dir == Chan_SEG_DIR.UP and macd_hist > 0:
|
|
total_macd_hist += macd_hist
|
|
if seg.dir == Chan_SEG_DIR.DOWN and macd_hist < 0:
|
|
total_macd_hist -= macd_hist
|
|
seg_macdhist_list.append(abs(total_macd_hist))
|
|
seg.set_macdhist(total_macd_hist)
|
|
return seg_macdhist_list, seg_list
|
|
|
|
def get_bi_macd_div_list(self, bi_list, dataframe):
|
|
bi_macd_div_list = []
|
|
bi_macdhist_list, bi_list = self.get_bi_macdhist_list(bi_list, dataframe)
|
|
for index in range(2, len(bi_list)):
|
|
if bi_macdhist_list[index-2] == 0:
|
|
bi_macd_div = 0.0
|
|
if index > 3 and bi_macdhist_list[index-4] > 0.0:
|
|
bi_macd_div = bi_macdhist_list[index]/bi_macdhist_list[index-4]
|
|
else:
|
|
bi_macd_div = bi_macdhist_list[index]/bi_macdhist_list[index-2]
|
|
if bi_macd_div < 0.01:
|
|
if index > 3 and bi_macdhist_list[index-4] > 0.0:
|
|
bi_macd_div = bi_macdhist_list[index]/bi_macdhist_list[index-4]
|
|
bi_macd_div = self.get_decimal(bi_macd_div)
|
|
bi_macd_div_list.append(bi_macd_div)
|
|
bi_list[index].set_macd_div(bi_macd_div)
|
|
#print(bi_list[index].start_klc.start_time, self.get_decimal(bi_macdhist_list[index]), self.get_decimal(bi_macdhist_list[index - 1]), self.get_decimal(bi_macd_div))
|
|
return bi_macd_div_list, bi_list
|
|
|
|
def get_seg_macd_div_list(self, seg_list, dataframe):
|
|
seg_macd_div_list = []
|
|
seg_macdhist_list, seg_list = self.get_seg_macdhist_list(seg_list, dataframe)
|
|
for index in range(2, len(seg_list)):
|
|
if seg_macdhist_list[index-2] == 0:
|
|
seg_macd_div = 0.0
|
|
if index > 3 and seg_macdhist_list[index-4] > 0.0:
|
|
seg_macd_div = seg_macdhist_list[index]/seg_macdhist_list[index - 4]
|
|
else:
|
|
seg_macd_div = seg_macdhist_list[index]/seg_macdhist_list[index - 2]
|
|
if seg_macd_div < 0.01:
|
|
if index > 3 and seg_macdhist_list[index-4] > 0.0:
|
|
seg_macd_div = seg_macdhist_list[index]/seg_macdhist_list[index - 4]
|
|
seg_macd_div = self.get_decimal(seg_macd_div)
|
|
seg_macd_div_list.append(seg_macd_div)
|
|
seg_list[index].set_macd_div(seg_macd_div)
|
|
#print(seg_list[index].start_bi.start_klc.start_time, self.get_decimal(seg_macdhist_list[index]), self.get_decimal(seg_macdhist_list[index - 1]), self.get_decimal(seg_macd_div))
|
|
return seg_macd_div_list, seg_list
|
|
|
|
def get_macd_div_list(self, dataframe):
|
|
bi_list = self.get_bi_list(dataframe)
|
|
seg_list = self.get_seg_list(bi_list)
|
|
bi_macd_div_list, bi_list = self.get_bi_macd_div_list(bi_list, dataframe)
|
|
seg_macd_div_list, seg_list = self.get_seg_macd_div_list(seg_list, dataframe)
|
|
return bi_macd_div_list, bi_list, seg_macd_div_list, seg_list
|
|
|
|
def get_decimal(self, value):
|
|
return Decimal("{:.2f}".format(value))
|
|
def add_indicators(self, df):
|
|
fast = 12
|
|
slow = 26
|
|
period = 9
|
|
macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period)
|
|
bb365 = ta.BBANDS(df, timeperiod=365, nbdevup=3.0, nbdevdn=3.0, matype=0)
|
|
bbp365 = ta.BBP(df, timeperiod=365)
|
|
bb120 = ta.BBANDS(df, timeperiod=120, nbdevup=3.0, nbdevdn=3.0, matype=0)
|
|
bbp120 = ta.BBP(df, timeperiod=120)
|
|
df['bb365'] = bb365['upperband']
|
|
df['bbp365'] = bbp365
|
|
df['bb120'] = bb120['upperband']
|
|
df['bbp120'] = bbp120
|
|
df['macd'] = macd['macd']
|
|
df['macdsignal'] = macd['macdsignal']
|
|
df['macdhist'] = macd['macdhist']
|
|
df['ma5'] = ta.MA(df, timeperiod=5)
|
|
df['ma10'] = ta.MA(df, timeperiod=10)
|
|
df['ma30'] = ta.EMA(df, timeperiod=30)
|
|
df['ma250'] = ta.MA(df, timeperiod=250)
|
|
df['rsi'] = ta.RSI(df, timeperiod=14)
|
|
return df
|
|
def get_klc_list(self, dataframe):
|
|
klu_list = self.get_klu_list(dataframe)
|
|
klc_list = []
|
|
last_klu = None
|
|
macd = ChanMACD(klu_list)
|
|
klu_list = macd.cal_macd_state()
|
|
for klu in klu_list:
|
|
if len(klc_list) > 0:
|
|
last_klc = klc_list[-1]
|
|
included = last_klc.check_klu_included(klu)
|
|
if not included:
|
|
ddir = Chan_KLINE_DIR.DOWN
|
|
if last_klc.high < klu.high:
|
|
ddir = Chan_KLINE_DIR.UP
|
|
klc = ChanKLC(klu, index=len(klc_list), ddir=ddir)
|
|
klc_list.append(klc)
|
|
last_klc.set_next(klc)
|
|
klc.set_pre(last_klc)
|
|
last_klc.set_end_klu(last_klu)
|
|
klc.set_pre_fx()
|
|
else:
|
|
last_klc.add_klu(klu)
|
|
else:
|
|
ddir = Chan_KLINE_DIR.UP
|
|
if klu.open > klu.close:
|
|
ddir = Chan_KLINE_DIR.DOWN
|
|
klc = ChanKLC(klu, 0, ddir)
|
|
klc_list.append(klc)
|
|
last_klu = klu
|
|
return klc_list
|
|
|
|
def get_klu_list(self, dataframe):
|
|
return self.get_kl_data(dataframe)
|
|
|
|
def copy_klu_to_klc(self, klu_list):
|
|
klc_list = []
|
|
for klu in klu_list:
|
|
if len(klc_list) > 0:
|
|
last_klc = klc_list[-1]
|
|
ddir = Chan_KLINE_DIR.DOWN
|
|
if last_klc.high < klu.high:
|
|
ddir = Chan_KLINE_DIR.UP
|
|
klc = ChanKLC(klu, index=len(klc_list), ddir=ddir)
|
|
klc.set_end_klu(klu)
|
|
klc_list.append(klc)
|
|
last_klc.set_next(klc)
|
|
klc.set_pre(last_klc)
|
|
else:
|
|
klc = ChanKLC(klu, 0)
|
|
klc_list.append(klc)
|
|
klc.set_end_klu(klu)
|
|
return klc_list
|
|
|
|
# ================================================
|
|
# 计算BSP列表
|
|
# ================================================
|
|
def get_bsp_list(self, big_df):
|
|
big_bi_list = self.get_bi_list(big_df)
|
|
big_seg_list = self.get_seg_list(big_bi_list)
|
|
big_zs_list = self.calculate_zs(big_bi_list, big_seg_list)
|
|
big_bsp_list = self.find_third_bsp(big_zs_list)
|
|
big_bi_macd_div_list, big_bi_list = self.get_bi_macd_div_list(big_bi_list, big_df)
|
|
for index in range(0, len(big_bsp_list)-1):
|
|
bsp = big_bsp_list[index]
|
|
last_zs = bsp.zs
|
|
bsp_next = big_bsp_list[index + 1]
|
|
if bsp.zs.index != bsp_next.zs.index:
|
|
end_index = bsp.seg.end_bi.index
|
|
else:
|
|
end_index = bsp_next.bi.index
|
|
# Down trend
|
|
if bsp.bi.dir == Chan_BI_DIR.UP:
|
|
last_up_bi = bsp.bi
|
|
last_down_bi = bsp.bi.pre
|
|
for bi_index in range(bsp.bi.index + 1, end_index + 1):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.is_sure:
|
|
if bi.dir == Chan_BI_DIR.DOWN:
|
|
if bi.low < last_down_bi.low:
|
|
print("背驰点1，第一类买点", bi.start_klc.end_time, bi.macd_div)
|
|
else:
|
|
if bi.macd_div > 1.5:
|
|
print("快速下跌，等待背驰:", bi.start_klc.end_time, bi.macd_div)
|
|
last_down_bi = bi
|
|
else:
|
|
if last_up_bi:
|
|
if (bi.high > last_up_bi.high and bi.macd_div > 1.2) or bi.high > last_zs.zd:
|
|
print("回中枢或者快速拉升，止损点:", bi.start_klc.end_time, bi.macd_div)
|
|
last_up_bi = bi
|
|
# Up trend
|
|
else:
|
|
last_down_bi = bsp.bi
|
|
last_up_bi = bsp.bi.pre
|
|
for bi_index in range(bsp.bi.index + 1, end_index + 1):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.is_sure:
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
if last_up_bi:
|
|
if bi.high < last_up_bi.high:
|
|
if bi.macd_div < 0.8 and bi.macd_div > 0.1:
|
|
print("背驰点2，第一类卖点", bi.start_klc.end_time, bi.macd_div)
|
|
else:
|
|
if bi.macd_div > 1.5:
|
|
print("快速上涨，等待背驰:", bi.start_klc.end_time, bi.macd_div)
|
|
last_up_bi = bi
|
|
else:
|
|
if last_down_bi:
|
|
if (bi.low < last_down_bi.low and bi.macd_div > 1.2) or bi.low < last_zs.zg:
|
|
print("回中枢或者快速下跌，止损点:", bi.start_klc.end_time, bi.macd_div)
|
|
last_down_bi = bi
|
|
bsp_bi = big_bi_list[-1]
|
|
last_zs = big_zs_list[-1]
|
|
# Down trend
|
|
if bsp_bi.dir == Chan_BI_DIR.UP:
|
|
last_down_bi = bsp_bi.pre
|
|
last_up_bi = bsp_bi
|
|
for bi_index in range(bsp_bi.index + 1, bsp.seg.end_bi.index + 1):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.is_sure:
|
|
if bi.dir == Chan_BI_DIR.DOWN:
|
|
if last_down_bi:
|
|
if bi.low < last_down_bi.low:
|
|
if bi.macd_div < 0.8 and bi.macd_div > 0.1:
|
|
print("背驰点3，第一类买点", bi.start_klc.end_time, bi.macd_div)
|
|
else:
|
|
if bi.macd_div > 1.5:
|
|
print("快速下跌，等待背驰:", bi.start_klc.end_time, bi.macd_div)
|
|
last_down_bi = bi
|
|
else:
|
|
if last_up_bi:
|
|
if (bi.high > last_up_bi.high and bi.macd_div > 1.2) or bi.high > last_zs.zd:
|
|
print("回中枢或者快速拉升，止损点:", bi.start_klc.end_time, bi.macd_div)
|
|
last_up_bi = bi
|
|
# Up trend
|
|
else:
|
|
last_down_bi = bsp_bi.pre
|
|
last_up_bi = bsp_bi
|
|
for bi_index in range(bsp_bi.index + 1, bsp.seg.end_bi.index + 1):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.is_sure:
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
if last_up_bi:
|
|
if bi.high < last_up_bi.high:
|
|
if bi.macd_div < 0.8 and bi.macd_div > 0.1:
|
|
print("背驰点4，第一类卖点", bi.start_klc.end_time, bi.macd_div)
|
|
else:
|
|
if bi.macd_div > 1.5:
|
|
print("快速上涨，等待背驰:", bi.start_klc.end_time, bi.macd_div)
|
|
last_up_bi = bi
|
|
else:
|
|
if last_down_bi:
|
|
if (bi.low < last_down_bi.low and bi.macd_div > 1.2) or bi.low < last_zs.zg:
|
|
print("回中枢或者快速下跌，止损点:", bi.start_klc.end_time, bi.macd_div)
|
|
last_down_bi = bi
|
|
return big_bsp_list
|
|
|
|
def cal_qjt(self, small_df, big_df):
|
|
big_bi_list = self.get_bi_list(big_df)
|
|
big_seg_list = self.get_seg_list(big_bi_list)
|
|
big_zs_list = self.calculate_zs(big_bi_list, big_seg_list)
|
|
|
|
small_bi_list = self.get_bi_list(small_df)
|
|
small_seg_list = self.get_seg_list(small_bi_list)
|
|
small_zs_list = self.calculate_zs(small_bi_list, small_seg_list)
|
|
|
|
big_bsp_list = self.find_third_bsp(big_zs_list)
|
|
small_bsp_list = self.find_third_bsp(small_zs_list)
|
|
|
|
#self.print_bsp_list(big_bsp_list)
|
|
|
|
self.print_bsp_list(small_bsp_list)
|
|
|
|
def get_seg_bsp_list(self, big_df):
|
|
big_bi_list = self.get_bi_list(big_df)
|
|
big_seg_list = self.get_seg_list(big_bi_list)
|
|
big_bsp_list = []
|
|
seg = big_seg_list[-1]
|
|
bi = big_bi_list[-1]
|
|
if seg.dir == Chan_SEG_DIR.UP:
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
if bi.high > seg.high:
|
|
if bi.macd_div < 0.8 and bi.macd_div > 0.1:
|
|
bi_bsp = ChanBSP(bi, len(big_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY, bi.sure_time, 0, None, seg)
|
|
big_bsp_list.append(bi_bsp)
|
|
else:
|
|
if bi.dir == Chan_BI_DIR.DOWN:
|
|
if bi.low < seg.low:
|
|
if bi.macd_div < 0.8 and bi.macd_div > 0.1:
|
|
bi_bsp = ChanBSP(bi, len(big_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.SELL, bi.sure_time, 0, None, seg)
|
|
big_bsp_list.append(bi_bsp)
|
|
print("Last SEG: ", seg.start_bi.start_klc.start_time)
|
|
for bsp in big_bsp_list:
|
|
if bsp.bi.end_klc:
|
|
print(bsp.bi.end_klc.end_time, bsp.sure_time, bsp.dir, bsp.bi.macd_div)
|
|
return big_bsp_list
|
|
|
|
def get_bi_bsp_list(self, big_df):
|
|
big_bi_list = self.get_bi_list(big_df)
|
|
big_seg_list = self.get_seg_list(big_bi_list)
|
|
big_bi_macd_div_list, big_bi_list = self.get_bi_macd_div_list(big_bi_list, big_df)
|
|
big_seg_macd_div_list, big_seg_list = self.get_seg_macd_div_list(big_seg_list, big_df)
|
|
bi_bsp_list = []
|
|
for index in range(0, len(big_seg_list)):
|
|
big_seg = big_seg_list[index]
|
|
if big_seg.end_bi:
|
|
if big_seg.dir == Chan_SEG_DIR.UP:
|
|
max_high = big_seg.high
|
|
for bi_index in range(big_seg.start_bi.index, big_seg.end_bi.index+1):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.dir == Chan_BI_DIR.DOWN and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.high > max_high:
|
|
max_high = bi.high
|
|
bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY, bi.sure_time, 0, None, big_seg)
|
|
bi_bsp_list.append(bi_bsp)
|
|
else:
|
|
max_low = big_seg.low
|
|
for bi_index in range(big_seg.start_bi.index, big_seg.end_bi.index+1):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.dir == Chan_BI_DIR.UP and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.low < max_low:
|
|
max_low = bi.low
|
|
bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.SELL, bi.sure_time, 0, None, big_seg)
|
|
bi_bsp_list.append(bi_bsp)
|
|
else:
|
|
print("Not completed segment.", len(big_bi_list) - big_seg.start_bi.index, big_seg.dir)
|
|
if big_seg.dir == Chan_SEG_DIR.UP:
|
|
max_high = big_seg.high
|
|
for bi_index in range(big_seg.start_bi.index, len(big_bi_list)):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.end_klc and bi.dir == Chan_BI_DIR.DOWN and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.high > max_high:
|
|
max_high = bi.high
|
|
bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.SELL, bi.sure_time, 0, None, big_seg)
|
|
bi_bsp_list.append(bi_bsp)
|
|
else:
|
|
max_low = big_seg.low
|
|
for bi_index in range(big_seg.start_bi.index, len(big_bi_list)):
|
|
bi = big_bi_list[bi_index]
|
|
if bi.end_klc and bi.dir == Chan_BI_DIR.UP and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.low < max_low:
|
|
max_low = bi.low
|
|
bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY, bi.sure_time, 0, None, big_seg)
|
|
bi_bsp_list.append(bi_bsp)
|
|
for bsp in bi_bsp_list:
|
|
if bsp.bi.end_klc:
|
|
print(bsp.bi.end_klc.end_time, bsp.sure_time, bsp.dir, bsp.seg.dir, bsp.bi.macd_div)
|
|
return bi_bsp_list
|
|
|
|
# ================================================
|
|
# 第三类买卖点
|
|
def find_third_bsp(self, zs_list):
|
|
bsp_list = []
|
|
zs_count = 0
|
|
last_zs = None
|
|
for zs in zs_list:
|
|
if last_zs and last_zs.dir == zs.dir:
|
|
zs_count += 1
|
|
else:
|
|
zs_count = 1
|
|
if zs.is_sure and zs.end_klc:
|
|
for index in range(0, len(zs.bi_out_list)):
|
|
bi_out = zs.bi_out_list[index]
|
|
bi_out_seg = zs.bi_out_seg_list[index]
|
|
bsp = ChanBSP(bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi_out.sure_time, zs_count, zs, bi_out_seg)
|
|
bsp_list.append(bsp)
|
|
|
|
elif len(zs.bi_out_list) > 0:
|
|
for index in range(0, len(zs.bi_out_list)):
|
|
bi_out = zs.bi_out_list[index]
|
|
bi_out_seg = zs.bi_out_seg_list[index]
|
|
bsp = ChanBSP(bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi_out.sure_time, zs_count, zs, bi_out_seg)
|
|
bsp_list.append(bsp)
|
|
last_zs = zs
|
|
return bsp_list
|
|
|
|
def find_first_bsp(self, bi_list, seg_list, zs_list, dataframe):
|
|
bsp_list = []
|
|
zs_count = 0
|
|
for index in range(1, len(zs_list)):
|
|
zs = zs_list[index]
|
|
pre_zs = zs_list[index - 1]
|
|
if zs.is_sure:
|
|
if pre_zs.dir == zs.dir:
|
|
zs_count += 1
|
|
continue
|
|
else:
|
|
zs_count = 1
|
|
else:
|
|
current_bi = bi_list[-1]
|
|
current_seg = seg_list[-1]
|
|
if zs.dir == pre_zs.dir and ((current_bi.dir == Chan_BI_DIR.UP and current_seg.dir == Chan_SEG_DIR.UP) or (current_bi.dir == Chan_BI_DIR.DOWN and current_seg.dir == Chan_SEG_DIR.DOWN)):
|
|
if zs.bi_out and zs.bi_out.is_sure and bi_list[-1].is_sure:
|
|
pre_start_index = pre_zs.end_seg.start_klc.end_klu.index
|
|
pre_end_index = zs.start_klc.end_klu.index
|
|
start_index = zs.bi_out_seg.start_bi.start_klc.start_klu.index
|
|
end_index = current_bi.end_klc.end_klu.index
|
|
pre_macd_area = self.cal_macd_area(dataframe, pre_start_index, pre_end_index, pre_zs.dir)
|
|
macd_area = self.cal_macd_area(dataframe, start_index, end_index, zs.dir)
|
|
print(zs.bi_out.start_klc.start_time, pre_macd_area, macd_area, zs_count)
|
|
if pre_macd_area > macd_area:
|
|
bsp = ChanBSP(current_bi, len(bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY if current_bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, current_bi.sure_time, zs.zs_count, zs, current_seg)
|
|
bsp_list.append(bsp)
|
|
return bsp_list
|
|
|
|
def cal_macd_area(self, dataframe, start_idx, end_idx, zs_dir):
|
|
"""
|
|
计算指定区间内的MACD面积
|
|
|
|
:param dataframe: K线数据
|
|
:param start_idx: 开始索引
|
|
:param end_idx: 结束索引
|
|
:param seg_dir: 线段方向(Chan_SEG_DIR.UP或Chan_SEG_DIR.DOWN)
|
|
:return: MACD面积的绝对值
|
|
"""
|
|
# 计算MACD指标
|
|
exp1 = dataframe['close'].ewm(span=12, adjust=False).mean()
|
|
exp2 = dataframe['close'].ewm(span=26, adjust=False).mean()
|
|
macd = exp1 - exp2
|
|
signal = macd.ewm(span=9, adjust=False).mean()
|
|
histogram = macd - signal
|
|
|
|
# 根据线段方向选择计算正面积还是负面积
|
|
if zs_dir == Chan_ZS_DIR.UP:
|
|
# 上升线段计算正面积
|
|
area = histogram[start_idx:end_idx+1][histogram[start_idx:end_idx+1] > 0].sum()
|
|
else:
|
|
# 下降线段计算负面积
|
|
area = histogram[start_idx:end_idx+1][histogram[start_idx:end_idx+1] < 0].sum()
|
|
|
|
return abs(area)
|
|
# --------------------------------------------------------------------
|
|
def plot_dual(self, small_df, big_df):
|
|
"""
|
|
绘制双周期K线图表，包括两个周期的笔、线段、中枢和买卖点
|
|
|
|
:param small_df: 小周期K线数据
|
|
:param big_df: 大周期K线数据
|
|
"""
|
|
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'Microsoft YaHei', 'WenQuanYi Micro Hei']
|
|
plt.rcParams['axes.unicode_minus'] = False
|
|
|
|
# 创建图表和子图
|
|
fig = plt.figure(figsize=(15, 12))
|
|
|
|
# 大周期图表(上方60%)
|
|
ax1 = plt.subplot2grid((10, 1), (0, 0), rowspan=4)
|
|
# 小周期图表(中间40%)
|
|
ax2 = plt.subplot2grid((10, 1), (4, 0), rowspan=4, sharex=ax1)
|
|
# MACD图表(下方20%)
|
|
ax3 = plt.subplot2grid((10, 1), (8, 0), rowspan=2, sharex=ax1)
|
|
|
|
# 计算两个周期的缠论结构
|
|
big_klc = self.get_klc_list(big_df)
|
|
big_bi = self.cal_bi_list(big_klc)
|
|
big_seg = self.get_seg_list(big_bi)
|
|
big_zs = self.calculate_zs(big_bi, big_seg)
|
|
big_buy_sell_points = self.check_top_bottom(big_df, big_bi, big_seg, big_zs)
|
|
big_bi_macd_div, big_bi = self.get_bi_macd_div_list(big_bi, big_df)
|
|
big_seg_macd_div, big_seg = self.get_seg_macd_div_list(big_seg, big_df)
|
|
|
|
small_klc = self.get_klc_list(small_df)
|
|
small_bi = self.cal_bi_list(small_klc)
|
|
small_seg = self.get_seg_list(small_bi)
|
|
small_zs = self.calculate_zs(small_bi, small_seg)
|
|
small_buy_sell_points = self.check_top_bottom(small_df, small_bi, small_seg, small_zs)
|
|
small_bi_macd_div, small_bi = self.get_bi_macd_div_list(small_bi, small_df)
|
|
small_seg_macd_div, small_seg = self.get_seg_macd_div_list(small_seg, small_df)
|
|
|
|
# 绘制大周期K线
|
|
big_dates = pd.to_datetime(big_df['date']).dt.tz_localize(None)
|
|
big_dates_num = [date2num(date) for date in big_dates]
|
|
|
|
# 绘制大周期K线
|
|
for i in range(len(big_df)):
|
|
color = 'red' if big_df['close'][i] > big_df['open'][i] else 'green'
|
|
ax1.bar(big_dates_num[i],
|
|
big_df['close'][i] - big_df['open'][i],
|
|
bottom=big_df['open'][i],
|
|
color=color,
|
|
width=0.0005)
|
|
ax1.plot([big_dates_num[i], big_dates_num[i]],
|
|
[big_df['low'][i], big_df['high'][i]],
|
|
color=color,
|
|
linewidth=1.2)
|
|
|
|
# 绘制大周期笔
|
|
for bi in big_bi:
|
|
if bi.end_klc:
|
|
start_time = pd.to_datetime(bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(bi.end_klc.end_time)
|
|
color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple'
|
|
start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high
|
|
end_price = bi.end_klc.high if bi.dir == Chan_BI_DIR.UP else bi.end_klc.low
|
|
ax1.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=1.5)
|
|
else:
|
|
start_time = pd.to_datetime(bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(big_klc[-1].start_time)
|
|
color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple'
|
|
start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high
|
|
end_price = big_klc[-1].high if bi.dir == Chan_BI_DIR.UP else big_klc[-1].low
|
|
ax1.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=0.5)
|
|
# 绘制大周期线段
|
|
for seg in big_seg:
|
|
if seg.end_bi:
|
|
start_time = pd.to_datetime(seg.start_bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(seg.end_bi.end_klc.end_time)
|
|
color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green'
|
|
start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high
|
|
end_price = seg.end_bi.end_klc.high if seg.dir == Chan_SEG_DIR.UP else seg.end_bi.end_klc.low
|
|
ax1.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=2.5)
|
|
else:
|
|
start_time = pd.to_datetime(seg.start_bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(big_klc[-1].start_time)
|
|
color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green'
|
|
start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high
|
|
end_price = big_klc[-1].high if seg.dir == Chan_SEG_DIR.UP else big_klc[-1].low
|
|
ax1.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=1)
|
|
# 绘制大周期中枢
|
|
for idx, zs in enumerate(big_zs):
|
|
start_time = pd.to_datetime(zs.start_klc.end_time).tz_localize(None)
|
|
color = ['orange', 'cyan', 'magenta', 'yellow', 'lime'][idx % 5]
|
|
|
|
if zs.end_klc:
|
|
end_time = pd.to_datetime(zs.end_klc.end_time).tz_localize(None)
|
|
width = date2num(end_time) - date2num(start_time)
|
|
rect = patches.Rectangle(
|
|
(date2num(start_time), zs.zd),
|
|
width,
|
|
zs.zg - zs.zd,
|
|
linewidth=1,
|
|
edgecolor=color,
|
|
facecolor=color,
|
|
alpha=0.2
|
|
)
|
|
ax1.add_patch(rect)
|
|
label_text = f"大中枢{idx+1}"
|
|
else:
|
|
end_time = pd.to_datetime(big_dates.iloc[-1]).tz_localize(None)
|
|
width = date2num(end_time) - date2num(start_time)
|
|
rect = patches.Rectangle(
|
|
(date2num(start_time), zs.zd),
|
|
width,
|
|
zs.zg - zs.zd,
|
|
linewidth=1.5,
|
|
edgecolor=color,
|
|
facecolor=color,
|
|
alpha=0.1,
|
|
linestyle='--'
|
|
)
|
|
ax1.add_patch(rect)
|
|
label_text = f"大中枢{idx+1}(未完成)"
|
|
|
|
ax1.text(
|
|
date2num(start_time) + width/2,
|
|
zs.zd + (zs.zg - zs.zd)/2,
|
|
label_text,
|
|
ha='center',
|
|
va='center',
|
|
fontsize=9,
|
|
color='black',
|
|
bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.6)
|
|
)
|
|
for index in range(0, len(big_bi_macd_div)):
|
|
bi_macd_div = big_bi_macd_div[index]
|
|
bi = big_bi[index + 2]
|
|
if bi.end_klc and False:
|
|
text_index = bi.end_klc.end_klu.index
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
ax1.text(big_dates_num[text_index], bi.end_klc.high+1, bi_macd_div, color='red', fontsize=10, alpha=0.6)
|
|
else:
|
|
ax1.text(big_dates_num[text_index], bi.end_klc.low-1, bi_macd_div, color='green', fontsize=10, alpha=0.6)
|
|
for index in range(0, len(big_seg_macd_div)):
|
|
seg_macd_div = big_seg_macd_div[index]
|
|
seg = big_seg[index + 2]
|
|
if seg.end_bi and False:
|
|
text_index = seg.end_bi.end_klc.end_klu.index
|
|
if seg.dir == Chan_SEG_DIR.UP:
|
|
ax1.text(big_dates_num[text_index], seg.end_bi.end_klc.high+1, seg_macd_div, color='red', fontsize=14, alpha=0.6)
|
|
else:
|
|
ax1.text(big_dates_num[text_index], seg.end_bi.end_klc.low-1, seg_macd_div, color='green', fontsize=14, alpha=0.6)
|
|
big_klc = self.get_klc_list(big_df)
|
|
self.cal_bi_list(big_klc)
|
|
for klc in big_klc:
|
|
if klc.klc_fx_type != Chan_KLC_FX.UNKNOWN:
|
|
text_index = klc.end_klu.index
|
|
if klc.fx == Chan_FX_TYPE.BOTTOM:
|
|
ax1.text(big_dates_num[text_index], klc.low, str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""), color='green', fontsize=6, alpha=1)
|
|
else:
|
|
ax1.text(big_dates_num[text_index], klc.high, str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""), color='red', fontsize=6, alpha=1)
|
|
"""
|
|
model = xgb.Booster()
|
|
model.load_model("30m_modelchan_xgb_model.json")
|
|
for klc in big_klc:
|
|
predict = self.predict(klc, model)
|
|
if predict > 0.35 and klc.fx == Chan_FX_TYPE.BOTTOM:
|
|
text_index = klc.end_klu.index
|
|
ax1.text(big_dates_num[text_index], klc.high+1, predict, color='red', fontsize=14, alpha=0.6)
|
|
if predict > 0.35 and klc.fx == Chan_FX_TYPE.TOP:
|
|
text_index = klc.end_klu.index
|
|
ax1.text(big_dates_num[text_index], klc.low-1, predict, color='green', fontsize=14, alpha=0.6)
|
|
"""
|
|
|
|
"""
|
|
# 绘制大周期买卖点
|
|
marker_styles = {
|
|
'第一类买点': {'marker': '^', 'color': 'red', 'size': 10},
|
|
'第一类卖点': {'marker': 'v', 'color': 'green', 'size': 10},
|
|
'2类买点': {'marker': '^', 'color': 'orange', 'size': 10},
|
|
'2类卖点': {'marker': 'v', 'color': 'cyan', 'size': 10},
|
|
'3类买点': {'marker': '^', 'color': 'purple', 'size': 10},
|
|
'3类卖点': {'marker': 'v', 'color': 'magenta', 'size': 10}
|
|
}
|
|
|
|
for idx, point in big_buy_sell_points.items():
|
|
if idx < 0 or idx >= len(big_df):
|
|
continue
|
|
style = marker_styles.get(point['type'], {'marker': 'o', 'color': 'black', 'size': 8})
|
|
ax1.plot(big_dates_num[idx], point['price'], style['marker'],
|
|
color=style['color'],
|
|
markersize=style['size'])
|
|
ax1.annotate(point['type'],
|
|
(big_dates_num[idx], point['price']),
|
|
textcoords="offset points",
|
|
xytext=(0, 10),
|
|
ha='center',
|
|
fontsize=8,
|
|
bbox=dict(boxstyle="round,pad=0.2", fc=style['color'], alpha=0.5))
|
|
"""
|
|
# 绘制小周期K线
|
|
small_dates = pd.to_datetime(small_df['date']).dt.tz_localize(None)
|
|
small_dates_num = [date2num(date) for date in small_dates]
|
|
|
|
for i in range(len(small_df)):
|
|
color = 'red' if small_df['close'][i] > small_df['open'][i] else 'green'
|
|
ax2.bar(small_dates_num[i],
|
|
small_df['close'][i] - small_df['open'][i],
|
|
bottom=small_df['open'][i],
|
|
color=color,
|
|
width=0.0002)
|
|
ax2.plot([small_dates_num[i], small_dates_num[i]],
|
|
[small_df['low'][i], small_df['high'][i]],
|
|
color=color,
|
|
linewidth=0.8)
|
|
|
|
# 绘制小周期笔
|
|
for bi in small_bi:
|
|
if bi.end_klc:
|
|
start_time = pd.to_datetime(bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(bi.end_klc.end_time)
|
|
color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple'
|
|
start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high
|
|
end_price = bi.end_klc.high if bi.dir == Chan_BI_DIR.UP else bi.end_klc.low
|
|
ax2.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=1.2)
|
|
else:
|
|
start_time = pd.to_datetime(bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(small_klc[-1].start_time)
|
|
color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple'
|
|
start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high
|
|
end_price = small_klc[-1].high if bi.dir == Chan_BI_DIR.UP else small_klc[-1].low
|
|
ax2.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=0.6)
|
|
|
|
|
|
# 绘制小周期线段
|
|
for seg in small_seg:
|
|
if seg.end_bi:
|
|
start_time = pd.to_datetime(seg.start_bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(seg.end_bi.end_klc.end_time)
|
|
color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green'
|
|
start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high
|
|
end_price = seg.end_bi.end_klc.high if seg.dir == Chan_SEG_DIR.UP else seg.end_bi.end_klc.low
|
|
ax2.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=1.8)
|
|
else:
|
|
start_time = pd.to_datetime(seg.start_bi.start_klc.end_time)
|
|
end_time = pd.to_datetime(small_klc[-1].start_time)
|
|
color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green'
|
|
start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high
|
|
end_price = small_klc[-1].high if seg.dir == Chan_SEG_DIR.UP else small_klc[-1].low
|
|
ax2.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=0.9)
|
|
|
|
# 绘制小周期中枢
|
|
for idx, zs in enumerate(small_zs):
|
|
start_time = pd.to_datetime(zs.start_klc.end_time).tz_localize(None)
|
|
color = ['orange', 'cyan', 'magenta', 'yellow', 'lime'][idx % 5]
|
|
|
|
if zs.end_klc:
|
|
end_time = pd.to_datetime(zs.end_klc.end_time).tz_localize(None)
|
|
width = date2num(end_time) - date2num(start_time)
|
|
rect = patches.Rectangle(
|
|
(date2num(start_time), zs.zd),
|
|
width,
|
|
zs.zg - zs.zd,
|
|
linewidth=0.8,
|
|
edgecolor=color,
|
|
facecolor=color,
|
|
alpha=0.2
|
|
)
|
|
ax2.add_patch(rect)
|
|
label_text = f"小中枢{idx+1}"
|
|
else:
|
|
end_time = pd.to_datetime(small_dates.iloc[-1]).tz_localize(None)
|
|
width = date2num(end_time) - date2num(start_time)
|
|
rect = patches.Rectangle(
|
|
(date2num(start_time), zs.zd),
|
|
width,
|
|
zs.zg - zs.zd,
|
|
linewidth=1,
|
|
edgecolor=color,
|
|
facecolor=color,
|
|
alpha=0.1,
|
|
linestyle='--'
|
|
)
|
|
ax2.add_patch(rect)
|
|
label_text = f"小中枢{idx+1}(未完成)"
|
|
|
|
ax2.text(
|
|
date2num(start_time) + width/2,
|
|
zs.zd + (zs.zg - zs.zd)/2,
|
|
label_text,
|
|
ha='center',
|
|
va='center',
|
|
fontsize=8,
|
|
color='black',
|
|
bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.6)
|
|
)
|
|
for index in range(0, len(small_bi_macd_div)):
|
|
bi_macd_div = small_bi_macd_div[index]
|
|
bi = small_bi[index + 2]
|
|
if bi.end_klc and False:
|
|
text_index = bi.end_klc.end_klu.index
|
|
if bi.dir == Chan_BI_DIR.UP:
|
|
ax2.text(small_dates_num[text_index], bi.end_klc.high+1, bi_macd_div, color='red', fontsize=10, alpha=0.6)
|
|
else:
|
|
ax2.text(small_dates_num[text_index], bi.end_klc.low-1, bi_macd_div, color='green', fontsize=10, alpha=0.6)
|
|
for index in range(0, len(small_seg_macd_div)):
|
|
seg_macd_div = small_seg_macd_div[index]
|
|
seg = small_seg[index + 2]
|
|
if seg.end_bi and False:
|
|
text_index = seg.end_bi.end_klc.end_klu.index
|
|
if seg.dir == Chan_SEG_DIR.UP:
|
|
ax2.text(small_dates_num[text_index], seg.end_bi.end_klc.high+1, seg_macd_div, color='red', fontsize=14, alpha=0.8)
|
|
else:
|
|
ax2.text(small_dates_num[text_index], seg.end_bi.end_klc.low-1, seg_macd_div, color='green', fontsize=14, alpha=0.8)
|
|
small_klc = self.get_klc_list(small_df)
|
|
self.cal_bi_list(small_klc)
|
|
for klc in small_klc:
|
|
if klc.klc_fx_type != Chan_KLC_FX.UNKNOWN:
|
|
text_index = klc.end_klu.index
|
|
if klc.fx == Chan_FX_TYPE.BOTTOM:
|
|
ax2.text(small_dates_num[text_index], klc.low, str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""), color='green', fontsize=6, alpha=1)
|
|
else:
|
|
ax2.text(small_dates_num[text_index], klc.high, str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""), color='red', fontsize=6, alpha=1)
|
|
"""
|
|
model = xgb.Booster()
|
|
model.load_model("5m_modelchan_xgb_model.json")
|
|
for klc in small_klc:
|
|
predict = self.predict(klc, model)
|
|
if predict > 0.35 and klc.fx == Chan_FX_TYPE.BOTTOM:
|
|
text_index = klc.end_klu.index
|
|
ax2.text(small_dates_num[text_index], klc.high+1, predict, color='red', fontsize=14, alpha=0.6)
|
|
if predict > 0.35 and klc.fx == Chan_FX_TYPE.TOP:
|
|
text_index = klc.end_klu.index
|
|
ax2.text(small_dates_num[text_index], klc.low-1, predict, color='green', fontsize=14, alpha=0.6)
|
|
"""
|
|
|
|
"""
|
|
# 绘制小周期买卖点
|
|
for idx, point in small_buy_sell_points.items():
|
|
if idx < 0 or idx >= len(small_df):
|
|
continue
|
|
style = marker_styles.get(point['type'], {'marker': 'o', 'color': 'black', 'size': 6})
|
|
ax2.plot(small_dates_num[idx], point['price'], style['marker'],
|
|
color=style['color'],
|
|
markersize=style['size'])
|
|
ax2.annotate(point['type'],
|
|
(small_dates_num[idx], point['price']),
|
|
textcoords="offset points",
|
|
xytext=(0, 8),
|
|
ha='center',
|
|
fontsize=7,
|
|
bbox=dict(boxstyle="round,pad=0.2", fc=style['color'], alpha=0.5))
|
|
"""
|
|
# 绘制MACD(使用小周期数据)
|
|
exp1 = small_df['close'].ewm(span=12, adjust=False).mean()
|
|
exp2 = small_df['close'].ewm(span=26, adjust=False).mean()
|
|
macd = exp1 - exp2
|
|
signal = macd.ewm(span=9, adjust=False).mean()
|
|
histogram = macd - signal
|
|
|
|
ax3.bar(small_dates_num, histogram, width=0.0002, color=['red' if h > 0 else 'green' for h in histogram])
|
|
ax3.plot(small_dates_num, macd, color='blue', linewidth=0.8, label='MACD')
|
|
ax3.plot(small_dates_num, signal, color='orange', linewidth=0.8, label='Signal')
|
|
ax3.axhline(y=0, color='black', linestyle='-', linewidth=0.5)
|
|
ax3.legend(loc='upper left')
|
|
|
|
# 设置图表标题和标签
|
|
ax1.set_title('大周期图表', fontsize=12)
|
|
ax2.set_title('小周期图表', fontsize=12)
|
|
ax3.set_title('MACD指标(小周期)', fontsize=10)
|
|
|
|
ax1.grid(True, linestyle='--', alpha=0.3)
|
|
ax2.grid(True, linestyle='--', alpha=0.3)
|
|
ax3.grid(True, linestyle='--', alpha=0.3)
|
|
|
|
ax1.xaxis.set_major_formatter(DateFormatter('%Y-%m-%d'))
|
|
plt.xticks(rotation=45)
|
|
plt.tight_layout()
|
|
plt.show()
|
|
|
|
|
|
def predict(self, klc, model):
|
|
"""
|
|
使用训练好的模型预测单个KLC
|
|
:param klc: 需要预测的ChanKLC对象
|
|
:return: 预测结果（概率值）
|
|
"""
|
|
# 提取特征
|
|
features = klc.get_feature_data()
|
|
feature_vec = []
|
|
# 与get_feature_data保持一致，只使用相同的特征集
|
|
for key, value in features.items():
|
|
if isinstance(value, (int, float)):
|
|
feature_vec.append(value)
|
|
else:
|
|
feature_vec.append(0)
|
|
|
|
# 转换为模型输入格式
|
|
dtest = xgb.DMatrix(np.array([feature_vec]))
|
|
|
|
# 预测
|
|
return self.get_decimal(model.predict(dtest)[0])
|
|
|
|
def get_decimal(self, value):
|
|
return Decimal("{:.2f}".format(value))
|
|
|
|
def plot(self, dataframe, bi_list, seg_list, zs_list=None, buy_sell_points=None, divergence_points=None):
|
|
"""
|
|
绘制缠论分析图表，包括K线、笔、线段、中枢、买卖点和MACD背驰
|
|
|
|
:param dataframe: K线数据
|
|
:param bi_list: 笔的列表
|
|
:param seg_list: 线段的列表
|
|
:param zs_list: 中枢的列表
|
|
:param buy_sell_points: 买卖点字典
|
|
:param divergence_points: 背驰点字典
|
|
"""
|
|
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'Microsoft YaHei', 'WenQuanYi Micro Hei']
|
|
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题
|
|
bar_line_width = 0.003
|
|
show_sure_time = False
|
|
# 创建具有两个子图的图表
|
|
fig = plt.figure(figsize=(15, 10))
|
|
|
|
# 主图占据上方70%空间
|
|
ax1 = plt.subplot2grid((5, 1), (0, 0), rowspan=3)
|
|
# MACD子图占据下方30%空间
|
|
ax2 = plt.subplot2grid((5, 1), (3, 0), rowspan=2, sharex=ax1)
|
|
|
|
# 转换日期格式 - 确保都是无时区的
|
|
dates = pd.to_datetime(dataframe['date']).dt.tz_localize(None)
|
|
dates_num = [date2num(date) for date in dates]
|
|
|
|
# 绘制K线图
|
|
for i in range(len(dataframe)):
|
|
# 红涨绿跌
|
|
if dataframe['close'][i] > dataframe['open'][i]:
|
|
body_color = 'red'
|
|
else:
|
|
body_color = 'green'
|
|
|
|
# 绘制实体
|
|
ax1.bar(dates_num[i],
|
|
dataframe['close'][i] - dataframe['open'][i],
|
|
bottom=dataframe['open'][i],
|
|
color=body_color,
|
|
width=bar_line_width/len(dataframe))
|
|
|
|
# 绘制上下影线
|
|
ax1.plot([dates_num[i], dates_num[i]],
|
|
[dataframe['low'][i], dataframe['high'][i]],
|
|
color=body_color,
|
|
linewidth=1.2)
|
|
|
|
# 绘制笔
|
|
for bi in bi_list:
|
|
if bi.end_klc: # 确保笔已完成
|
|
start_time = pd.to_datetime(bi.start_klc.start_time)
|
|
end_time = pd.to_datetime(bi.end_klc.end_time)
|
|
|
|
# 上升笔蓝色，下降笔紫色
|
|
color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple'
|
|
start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high
|
|
end_price = bi.end_klc.high if bi.dir == Chan_BI_DIR.UP else bi.end_klc.low
|
|
|
|
# 绘制笔
|
|
ax1.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=1.5)
|
|
|
|
# 绘制线段
|
|
for seg in seg_list:
|
|
if seg.end_bi: # 确保线段已完成
|
|
start_time = pd.to_datetime(seg.start_bi.start_klc.start_time)
|
|
end_time = pd.to_datetime(seg.end_bi.end_klc.end_time)
|
|
|
|
# 上升线段红色，下降线段绿色
|
|
color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green'
|
|
start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high
|
|
end_price = seg.end_bi.end_klc.high if seg.dir == Chan_SEG_DIR.UP else seg.end_bi.end_klc.low
|
|
|
|
# 绘制线段(粗线)
|
|
ax1.plot([date2num(start_time), date2num(end_time)],
|
|
[start_price, end_price],
|
|
color=color,
|
|
linewidth=2.5)
|
|
|
|
# 在线段确认点绘制标记
|
|
if hasattr(seg, 'sure_time') and seg.sure_time and show_sure_time:
|
|
try:
|
|
# 确保sure_time无时区
|
|
sure_time = pd.to_datetime(seg.sure_time).tz_localize(None)
|
|
|
|
# 找到最接近的K线
|
|
closest_idx = (dates - sure_time).abs().argmin()
|
|
|
|
# 获取确认点的价格
|
|
confirm_price = dataframe['close'][closest_idx]
|
|
|
|
# 绘制标记和标签
|
|
ax1.plot(date2num(sure_time), confirm_price, 'D',
|
|
color='black', markersize=6)
|
|
ax1.annotate(sure_time.strftime('%m-%d %H:%M'),
|
|
(date2num(sure_time), confirm_price),
|
|
textcoords="offset points",
|
|
xytext=(0, 10),
|
|
ha='center',
|
|
fontsize=8,
|
|
bbox=dict(boxstyle="round,pad=0.3", fc="yellow", alpha=0.7))
|
|
except Exception as e:
|
|
print(f"处理线段确认时间时出错: {e}")
|
|
continue
|
|
|
|
# 绘制中枢区域
|
|
if zs_list:
|
|
# 定义中枢的颜色和透明度
|
|
zs_colors = ['orange', 'cyan', 'magenta', 'yellow', 'lime']
|
|
|
|
for idx, zs in enumerate(zs_list):
|
|
# 无论中枢是否完成都绘制
|
|
start_time = pd.to_datetime(zs.start_klc.start_time).tz_localize(None)
|
|
|
|
# 选择颜色，循环使用预定义的颜色
|
|
color = zs_colors[idx % len(zs_colors)]
|
|
|
|
if zs.end_klc: # 已完成的中枢
|
|
# 转换结束时间格式
|
|
end_time = pd.to_datetime(zs.end_klc.end_time).tz_localize(None)
|
|
|
|
# 矩形的宽度和高度
|
|
width = date2num(end_time) - date2num(start_time)
|
|
height = zs.zg - zs.zd
|
|
|
|
# 创建实线矩形补丁表示已完成中枢
|
|
rect = patches.Rectangle(
|
|
(date2num(start_time), zs.zd), # 左下角坐标
|
|
width, # 宽度
|
|
height, # 高度
|
|
linewidth=1,
|
|
edgecolor=color,
|
|
facecolor=color,
|
|
alpha=0.2 # 透明度
|
|
)
|
|
ax1.add_patch(rect)
|
|
|
|
# 添加中枢编号标签
|
|
label_text = f"中枢{idx+1}"
|
|
else: # 未完成的中枢
|
|
# 使用最后一根K线的时间作为临时结束时间
|
|
end_time = pd.to_datetime(dates.iloc[-1]).tz_localize(None)
|
|
|
|
# 矩形的宽度和高度
|
|
width = date2num(end_time) - date2num(start_time)
|
|
height = zs.zg - zs.zd
|
|
|
|
# 创建虚线矩形补丁表示未完成中枢
|
|
rect = patches.Rectangle(
|
|
(date2num(start_time), zs.zd), # 左下角坐标
|
|
width, # 宽度
|
|
height, # 高度
|
|
linewidth=1.5,
|
|
edgecolor=color,
|
|
facecolor=color,
|
|
alpha=0.1, # 较低的透明度
|
|
linestyle='--' # 虚线边框
|
|
)
|
|
ax1.add_patch(rect)
|
|
|
|
# 添加中枢编号标签，标明未完成
|
|
label_text = f"中枢{idx+1}(未完成)"
|
|
|
|
# 添加中枢标签
|
|
ax1.text(
|
|
date2num(start_time) + width/2, # x位置（中枢中间）
|
|
zs.zd + height/2, # y位置（中枢中间）
|
|
label_text,
|
|
ha='center',
|
|
va='center',
|
|
fontsize=9,
|
|
color='black',
|
|
bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.6)
|
|
)
|
|
|
|
# 绘制买卖点
|
|
if buy_sell_points:
|
|
marker_styles = {
|
|
'1类买点': {'marker': '^', 'color': 'red', 'size': 10, 'label': '1类买点'},
|
|
'1类卖点': {'marker': 'v', 'color': 'green', 'size': 10, 'label': '1类卖点'},
|
|
'2类买点': {'marker': '^', 'color': 'orange', 'size': 10, 'label': '2类买点'},
|
|
'2类卖点': {'marker': 'v', 'color': 'cyan', 'size': 10, 'label': '2类卖点'},
|
|
'3类买点': {'marker': '^', 'color': 'purple', 'size': 10, 'label': '3类买点'},
|
|
'3类卖点': {'marker': 'v', 'color': 'magenta', 'size': 10, 'label': '3类卖点'}
|
|
}
|
|
|
|
for idx, point in buy_sell_points.items():
|
|
if idx < 0 or idx >= len(dataframe):
|
|
continue
|
|
print("Plot buy sell point: ", point['type'])
|
|
style = marker_styles.get(point['type'], {'marker': 'o', 'color': 'black', 'size': 8, 'label': '其他'})
|
|
|
|
# 绘制买卖点标记
|
|
ax1.plot(dates_num[idx], point['price'], style['marker'],
|
|
color=style['color'],
|
|
markersize=style['size'],
|
|
label=style['label'])
|
|
|
|
# 添加买卖点标签
|
|
ax1.annotate(point['type'],
|
|
(dates_num[idx], point['price']),
|
|
textcoords="offset points",
|
|
xytext=(0, 10),
|
|
ha='center',
|
|
fontsize=8,
|
|
bbox=dict(boxstyle="round,pad=0.2", fc=style['color'], alpha=0.5))
|
|
|
|
# 绘制背驰点
|
|
if divergence_points:
|
|
for idx, point in divergence_points.items():
|
|
if idx < 0 or idx >= len(dataframe) or True:
|
|
continue
|
|
print("Plot divergence point")
|
|
color = 'red' if point['type'] == '底背驰' else 'green'
|
|
marker = '*'
|
|
|
|
# 绘制背驰点标记
|
|
ax1.plot(dates_num[idx], point['price'], marker,
|
|
color=color,
|
|
markersize=12,
|
|
label=point['type'])
|
|
|
|
# 添加背驰点标签
|
|
ax1.annotate(point['type'],
|
|
(dates_num[idx], point['price']),
|
|
textcoords="offset points",
|
|
xytext=(0, -15),
|
|
ha='center',
|
|
fontsize=8,
|
|
bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.5))
|
|
|
|
# 计算MACD指标
|
|
exp1 = dataframe['close'].ewm(span=12, adjust=False).mean()
|
|
exp2 = dataframe['close'].ewm(span=26, adjust=False).mean()
|
|
macd = exp1 - exp2
|
|
signal = macd.ewm(span=9, adjust=False).mean()
|
|
histogram = macd - signal
|
|
|
|
# 绘制MACD
|
|
ax2.bar(dates_num, histogram, width=bar_line_width, color=['red' if h > 0 else 'green' for h in histogram])
|
|
ax2.plot(dates_num, macd, color='blue', linewidth=1.2, label='MACD')
|
|
ax2.plot(dates_num, signal, color='orange', linewidth=1.2, label='Signal')
|
|
ax2.axhline(y=0, color='black', linestyle='-', linewidth=0.5)
|
|
ax2.legend(loc='upper left')
|
|
|
|
# 在MACD图上标记背驰点
|
|
if divergence_points:
|
|
for idx, point in divergence_points.items():
|
|
if idx < 0 or idx >= len(dataframe):
|
|
continue
|
|
|
|
color = 'red' if point['type'] == '底背驰' else 'green'
|
|
|
|
# 在MACD图上标记背驰点
|
|
ax2.plot(dates_num[idx], histogram[idx], '*',
|
|
color=color,
|
|
markersize=12)
|
|
|
|
# 添加简单网格
|
|
ax1.grid(True, linestyle='--', alpha=0.3)
|
|
ax2.grid(True, linestyle='--', alpha=0.3)
|
|
|
|
# 设置坐标轴格式
|
|
ax1.xaxis.set_major_formatter(DateFormatter('%Y-%m-%d'))
|
|
|
|
# 添加简单图例
|
|
from matplotlib.lines import Line2D
|
|
legend_elements = [
|
|
Line2D([0], [0], color='blue', lw=2, label='上升笔'),
|
|
Line2D([0], [0], color='purple', lw=2, label='下降笔'),
|
|
Line2D([0], [0], color='red', lw=2.5, label='上升线段'),
|
|
Line2D([0], [0], color='green', lw=2.5, label='下降线段'),
|
|
patches.Patch(facecolor='orange', alpha=0.2, label='已完成中枢'),
|
|
patches.Patch(facecolor='orange', alpha=0.1, edgecolor='orange', linestyle='--', label='未完成中枢'),
|
|
Line2D([0], [0], marker='^', color='red', label='买点', markersize=10, linestyle='None'),
|
|
Line2D([0], [0], marker='v', color='green', label='卖点', markersize=10, linestyle='None'),
|
|
Line2D([0], [0], marker='*', color='red', label='底背驰', markersize=12, linestyle='None'),
|
|
Line2D([0], [0], marker='*', color='green', label='顶背驰', markersize=12, linestyle='None')
|
|
]
|
|
ax1.legend(handles=legend_elements, loc='upper left')
|
|
|
|
# 设置标题和标签
|
|
ax1.set_title('缠论分析图', fontsize=14)
|
|
ax1.set_ylabel('价格', fontsize=12)
|
|
ax2.set_xlabel('时间', fontsize=12)
|
|
ax2.set_ylabel('MACD', fontsize=12)
|
|
plt.xticks(rotation=45)
|
|
plt.tight_layout()
|
|
|
|
# 显示图表
|
|
plt.show()
|
|
|
|
"""
|
|
for index in range(seg.next.next.start_bi.index, seg.next.next.end_bi.index):
|
|
bi = bi_list[index]
|
|
if (bi.high >= last_zs.zd and bi.high <= last_zs.zg) or (bi.low >= last_zs.zd and bi.low <= last_zs.zg) or (bi.high >= last_zs.zg and bi.low <= last_zs.zd):
|
|
in_again = True
|
|
last_zs.set_bi_out(None)
|
|
last_zs.set_last_bi_in(None)
|
|
last_zs.set_end_seg(None)
|
|
first_bi_out = None
|
|
#print("Bi in again 1", bi.start_klc.start_time)
|
|
if in_again and (bi.low > last_zs.zg or bi.high < last_zs.zd):
|
|
last_zs.set_bi_out(bi)
|
|
last_zs.set_last_bi_in(bi_list[index - 1])
|
|
last_zs.set_end_seg(seg.next.next)
|
|
bi_out_count += 1
|
|
first_bi_out = bi
|
|
print("First bi out 1", first_bi_out.start_klc.start_time)
|
|
in_again = False
|
|
"""
|
|
|
|
def check_top_bottom(self, dataframe, bi_list, seg_list, zs_list):
|
|
"""
|
|
检测新高/新低时的第一类买卖点，结合MACD背驰判断
|
|
|
|
:param dataframe: K线数据
|
|
:param bi_list: 笔的列表
|
|
:param seg_list: 线段的列表
|
|
:param zs_list: 中枢的列表
|
|
:return: 第一类买卖点列表，格式为{index: {'type': 类型, 'price': 价格, 'time': 时间}}
|
|
"""
|
|
buy_sell_points = {}
|
|
|
|
# 计算MACD指标
|
|
exp1 = dataframe['close'].ewm(span=12, adjust=False).mean()
|
|
exp2 = dataframe['close'].ewm(span=26, adjust=False).mean()
|
|
macd = exp1 - exp2
|
|
signal = macd.ewm(span=9, adjust=False).mean()
|
|
histogram = macd - signal
|
|
|
|
# MACD柱状图的面积
|
|
positive_hist = histogram.copy()
|
|
negative_hist = histogram.copy()
|
|
positive_hist[positive_hist < 0] = 0
|
|
negative_hist[negative_hist > 0] = 0
|
|
|
|
# 找到所有底分型和顶分型的笔
|
|
bottom_bi_indices = [] # 底分型的笔索引
|
|
top_bi_indices = [] # 顶分型的笔索引
|
|
|
|
for i, bi in enumerate(bi_list):
|
|
if not bi.end_klc:
|
|
continue
|
|
|
|
if bi.dir == Chan_BI_DIR.UP and i > 0:
|
|
bottom_bi_indices.append(i-1) # 上升笔的前一笔是底分型
|
|
elif bi.dir == Chan_BI_DIR.DOWN and i > 0:
|
|
top_bi_indices.append(i-1) # 下降笔的前一笔是顶分型
|
|
|
|
# 查找创新高的顶分型（第一类卖点）
|
|
for i in range(1, len(top_bi_indices)):
|
|
curr_idx = top_bi_indices[i]
|
|
prev_idx = top_bi_indices[i-1]
|
|
|
|
if curr_idx >= len(bi_list) or prev_idx >= len(bi_list):
|
|
continue
|
|
|
|
curr_bi = bi_list[curr_idx]
|
|
prev_bi = bi_list[prev_idx]
|
|
|
|
if not curr_bi.end_klc or not prev_bi.end_klc:
|
|
continue
|
|
|
|
# 确保是新高：当前高点比前一高点更高
|
|
if curr_bi.high > prev_bi.high:
|
|
# 找到对应的MACD值
|
|
curr_time = curr_bi.end_klc.end_time
|
|
prev_time = prev_bi.end_klc.end_time
|
|
|
|
# 获取对应的dataframe索引
|
|
curr_date_idx = dataframe[dataframe['date'].astype(str).str.contains(curr_time)].index[0] if any(dataframe['date'].astype(str).str.contains(curr_time)) else -1
|
|
prev_date_idx = dataframe[dataframe['date'].astype(str).str.contains(prev_time)].index[0] if any(dataframe['date'].astype(str).str.contains(prev_time)) else -1
|
|
|
|
if curr_date_idx >= 0 and prev_date_idx >= 0:
|
|
# 计算两段走势的MACD柱状图面积（顶分型关注正面积）
|
|
curr_area = positive_hist[prev_date_idx:curr_date_idx+1].sum()
|
|
prev_area = positive_hist[max(0, prev_date_idx-abs(curr_date_idx-prev_date_idx)):prev_date_idx+1].sum()
|
|
|
|
# 检查是否有MACD背驰
|
|
# 新高但MACD力度减弱，形成顶背驰
|
|
if curr_area < prev_area and curr_area > 0:
|
|
# 检查是否在中枢中
|
|
in_zs = False
|
|
for zs in zs_list:
|
|
if zs.zd <= curr_bi.high <= zs.zg:
|
|
in_zs = True
|
|
break
|
|
|
|
if not in_zs: # 不在中枢中的第一类卖点更可靠
|
|
# 检查线段方向，确保是上升趋势
|
|
is_uptrend = False
|
|
for seg in seg_list:
|
|
if seg.end_bi and seg.dir == Chan_SEG_DIR.UP and seg.end_bi.index >= curr_bi.index:
|
|
is_uptrend = True
|
|
break
|
|
|
|
if is_uptrend:
|
|
buy_sell_points[curr_date_idx] = {
|
|
'type': '第一类卖点',
|
|
'price': dataframe.loc[curr_date_idx, 'high'],
|
|
'time': curr_time,
|
|
'reason': f'新高+顶背驰(MACD: {curr_area:.2f}<{prev_area:.2f})',
|
|
'bi_idx': curr_idx,
|
|
'is_sure': curr_bi.is_sure
|
|
}
|
|
|
|
# 查找创新低的底分型（第一类买点）
|
|
for i in range(1, len(bottom_bi_indices)):
|
|
curr_idx = bottom_bi_indices[i]
|
|
prev_idx = bottom_bi_indices[i-1]
|
|
|
|
if curr_idx >= len(bi_list) or prev_idx >= len(bi_list):
|
|
continue
|
|
|
|
curr_bi = bi_list[curr_idx]
|
|
prev_bi = bi_list[prev_idx]
|
|
|
|
if not curr_bi.end_klc or not prev_bi.end_klc:
|
|
continue
|
|
|
|
# 确保是新低：当前低点比前一低点更低
|
|
if curr_bi.low < prev_bi.low:
|
|
# 找到对应的MACD值
|
|
curr_time = curr_bi.end_klc.end_time
|
|
prev_time = prev_bi.end_klc.end_time
|
|
|
|
# 获取对应的dataframe索引
|
|
curr_date_idx = dataframe[dataframe['date'].astype(str).str.contains(curr_time)].index[0] if any(dataframe['date'].astype(str).str.contains(curr_time)) else -1
|
|
prev_date_idx = dataframe[dataframe['date'].astype(str).str.contains(prev_time)].index[0] if any(dataframe['date'].astype(str).str.contains(prev_time)) else -1
|
|
|
|
if curr_date_idx >= 0 and prev_date_idx >= 0:
|
|
# 计算两段走势的MACD柱状图面积（底分型关注负面积）
|
|
curr_area = abs(negative_hist[prev_date_idx:curr_date_idx+1].sum())
|
|
prev_area = abs(negative_hist[max(0, prev_date_idx-abs(curr_date_idx-prev_date_idx)):prev_date_idx+1].sum())
|
|
|
|
# 检查是否有MACD背驰
|
|
# 新低但MACD力度减弱，形成底背驰
|
|
if curr_area < prev_area and curr_area > 0:
|
|
# 检查是否在中枢中
|
|
in_zs = False
|
|
for zs in zs_list:
|
|
if zs.zd <= curr_bi.low <= zs.zg:
|
|
in_zs = True
|
|
break
|
|
|
|
if not in_zs: # 不在中枢中的第一类买点更可靠
|
|
# 检查线段方向，确保是下降趋势
|
|
is_downtrend = False
|
|
for seg in seg_list:
|
|
if seg.end_bi and seg.dir == Chan_SEG_DIR.DOWN and seg.end_bi.index >= curr_bi.index:
|
|
is_downtrend = True
|
|
break
|
|
|
|
if is_downtrend:
|
|
buy_sell_points[curr_date_idx] = {
|
|
'type': '第一类买点',
|
|
'price': dataframe.loc[curr_date_idx, 'low'],
|
|
'time': curr_time,
|
|
'reason': f'新低+底背驰(MACD: {curr_area:.2f}<{prev_area:.2f})',
|
|
'bi_idx': curr_idx,
|
|
'is_sure': curr_bi.is_sure
|
|
}
|
|
|
|
return buy_sell_points |