jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a56052e768e3e0d3cccf38714d5ab421904228fc
Chan/ChanLun.py
T
jackyu66git a56052e768 Add BB
2025-07-06 17:45:31 +08:00

2562 lines
97 KiB
Python
Raw Blame History

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
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
class ChanLun():
timeframes = ["5m", "15m", "30m", "60m", "4h"]
times = {
"5m": 5,
"15m": 15,
"30m": 30,
"60m": 60,
"4h": 240
}
time5 = 5
time15 = 15
time30 = 30
time60 = 60
time4h = 240
last_peak = {'high': 0, 'low': float('inf')}
def create_all_data(self, dataframe, ticker_indicator):
all_data = dict()
all_data['1m'] = dataframe
for timeframe in self.timeframes:
df = resample_to_interval(dataframe, ticker_indicator*self.times[timeframe])
all_data[timeframe] = df
return all_data
def print_zs(self, zs_list):
for zs in zs_list:
if zs.end_klc:
print(zs.start_klc.start_time, zs.end_klc.end_time, zs.sure_time, zs.zg, zs.zd, zs.bi_out_count)
else:
print(zs.start_klc.start_time, zs.zg, zs.zd, zs.bi_out_count)
def print_seg(self, seg_list):
for seg in seg_list:
if seg.is_sure:
print(seg.start_bi.start_time, seg.end_bi.end_time, seg.dir, seg.sure_time, "SEG")
else:
print(seg.start_bi.start_time, seg.dir, "SEG")
def print_bsp_list(self, bsp_list):
for bsp in bsp_list:
if bsp.is_sure:
print(bsp.klc.start_time, bsp.type, bsp.dir, bsp.sure_time, bsp.zs_count, len(bsp.zs.bi_out_list), bsp.dir, bsp.seg.dir, bsp.bi.dir)
else:
print(bsp.klc.start_time, bsp.type, bsp.dir, bsp.zs_count, len(bsp.zs.bi_out_list), bsp.dir, bsp.seg.dir, bsp.bi.dir)
def print_bi(self, bi_list):
for bi in bi_list:
if bi.end_klc:
if bi.sure_time:
print(bi.start_klc.end_time, bi.end_klc.end_time, bi.dir)
else:
print(bi.start_klc.end_time, bi.end_klc.end_time, bi.dir)
else:
print(bi.start_klc.end_time, bi.dir, bi.is_sure)
def check_fx(self, klc):
if klc.pre and klc.next:
if klc.high > klc.pre.high and klc.high > klc.next.high:
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
if klc.pre and klc.next:
if klc.low < klc.pre.low and klc.low < klc.next.low:
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 get_macd(self, df):
fast = 8
slow = 15
period = 2
macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period)
df['macd'] = macd['macd']
df['macdsignal'] = macd['macdsignal']
df['macdhist'] = macd['macdhist']
return df
def plot_dataframe(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list= self.cal_bi_list(klc_list)
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.TOP3:
state_list.append("10")
print(klc.start_time, klc.end_time, klc.klc_fx_type)
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM3:
state_list.append("-10")
print(klc.start_time, klc.end_time, klc.klc_fx_type)
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 print_data(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list = self.cal_bi_list(klc_list)
seg_list = self.get_seg_list(bi_list)
bsp_list, zs_list = self.calculate_zs(bi_list, seg_list)
bi_macd_div_list = self.get_bi_macd_div_list(bi_list, dataframe)
seg_macd_div_list = self.get_seg_macd_div_list(seg_list, dataframe)
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
]
if h > self.last_peak['high']:
self.last_peak['high'] = h
if l < self.last_peak['low']:
self.last_peak['low'] = l
#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_full_klc_list(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list = self.cal_bi_list(klc_list)
return klc_list
def print_bi_klc(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list = self.cal_bi_list(klc_list)
rsi_list = dataframe['rsi']
fx_list = []
for klc in klc_list:
if klc.klc_fx_type != Chan_KLC_FX.UNKNOWN:
if klc.bi.dir == Chan_BI_DIR.UP and klc.volume_ratio > 2:
print(klc.start_time, klc.end_time, klc.klc_fx_type, klc.rsi, klc.volume_ratio)
fx_list.append(klc)
elif klc.bi.dir == Chan_BI_DIR.DOWN and klc.volume_ratio > 2:
print(klc.start_time, klc.end_time, klc.klc_fx_type, klc.rsi, klc.volume_ratio)
fx_list.append(klc)
bi_start_index_list = []
for bi in bi_list:
bi_start_index_list.append(bi.start_klc.index)
if bi.end_klc:
bi.cal_macdhist()
bi.cal_macd_div()
print("Bi:", bi.start_time, bi.end_time, bi.dir, bi.macd_hist, bi.macd_div)
klc_index_count = 0
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 get_bi_zs_list(self, bi_list):
"""识别笔中枢列表
与线段中枢不同，笔中枢是由连续的同向笔构成，是更细粒度的中枢结构
Args:
bi_list: 笔列表
Returns:
bi_zs_list: 笔中枢列表
"""
bi_zs_list = []
if len(bi_list) < 3: # 至少需要3个笔才能形成中枢
print("笔数量不足，无法形成中枢")
return bi_zs_list
last_zs = None
first_bi_out = None
in_again = False
bi_out_count = 0
# 遍历所有笔，识别中枢
for i in range(2, len(bi_list)):
# 确保当前笔和前两个笔都是完成的
if not bi_list[i].end_klc or not bi_list[i-1].end_klc or not bi_list[i-2].end_klc:
continue
current_bi = bi_list[i]
prev_bi = bi_list[i-1]
prev_prev_bi = bi_list[i-2]
# 如果没有中枢或上一个中枢已完成
if len(bi_zs_list) == 0 or (last_zs and last_zs.is_sure):
# 检查是否是三个连续同向笔
if (current_bi.dir == prev_bi.dir == prev_prev_bi.dir):
# 创建潜在中枢
if current_bi.dir == Chan_BI_DIR.UP:
# 向上的三笔区间定义中枢
# 中枢的上沿：取三个笔的终点的最小值
# 中枢的下沿：取三个笔的起点的最大值
zd = max(prev_prev_bi.start_klc.low, prev_bi.start_klc.low, current_bi.start_klc.low)
zg = min(prev_prev_bi.end_klc.high, prev_bi.end_klc.high, current_bi.end_klc.high)
# 确保中枢有效(上沿大于下沿)
if zg > zd:
print(f"发现向上笔中枢: 起始时间={prev_prev_bi.start_klc.start_time}, ZG={zg}, ZD={zd}")
zs = ChanZS(prev_prev_bi.start_klc, zg, zd)
zs.start_bi = prev_prev_bi
zs.start_idx = i-2
zs.end_bi = current_bi
zs.end_idx = i
zs.end_klc = current_bi.end_klc
zs.type = "BI_ZS"
zs.direction = Chan_ZS_DIR.UP
zs.sure_time = None # 中枢尚未确认完成
bi_zs_list.append(zs)
last_zs = zs
else:
# 向下的三笔区间定义中枢
# 中枢的上沿：取三个笔的起点的最小值
# 中枢的下沿：取三个笔的终点的最大值
zg = min(prev_prev_bi.start_klc.high, prev_bi.start_klc.high, current_bi.start_klc.high)
zd = max(prev_prev_bi.end_klc.low, prev_bi.end_klc.low, current_bi.end_klc.low)
# 确保中枢有效(上沿大于下沿)
if zg > zd:
print(f"发现向下笔中枢: 起始时间={prev_prev_bi.start_klc.start_time}, ZG={zg}, ZD={zd}")
zs = ChanZS(prev_prev_bi.start_klc, zg, zd)
zs.start_bi = prev_prev_bi
zs.start_idx = i-2
zs.end_bi = current_bi
zs.end_idx = i
zs.end_klc = current_bi.end_klc
zs.type = "BI_ZS"
zs.direction = Chan_ZS_DIR.DOWN
zs.sure_time = None # 中枢尚未确认完成
bi_zs_list.append(zs)
last_zs = zs
# 处理已有的未完成中枢
elif last_zs and not last_zs.is_sure:
# 当前笔与中枢最后一笔方向相同，可能延伸中枢
if current_bi.dir == prev_bi.dir:
if last_zs.direction == Chan_ZS_DIR.UP and current_bi.dir == Chan_BI_DIR.UP:
# 检查是否仍在中枢内：向上时终点高价在中枢区间内
if current_bi.end_klc.high >= last_zs.zd and current_bi.end_klc.high <= last_zs.zg:
print(f"延伸向上笔中枢: 终点时间={current_bi.end_klc.end_time}")
# 延伸中枢
last_zs.end_klc = current_bi.end_klc
last_zs.end_bi = current_bi
last_zs.end_idx = i
else:
# 笔离开中枢，记录第一个离开的笔
if not first_bi_out:
first_bi_out = current_bi
bi_out_count += 1
print(f"笔离开向上中枢: 时间={current_bi.end_klc.end_time}, 价格={current_bi.end_klc.high}, 中枢上沿={last_zs.zg}")
else:
if not in_again:
# 第二次离开，确认中枢完成
print(f"确认向上笔中枢完成: 时间={current_bi.end_klc.end_time}")
last_zs.is_sure = True
last_zs.sure_bi = current_bi
last_zs.sure_time = current_bi.end_klc.end_time
elif last_zs.direction == Chan_ZS_DIR.DOWN and current_bi.dir == Chan_BI_DIR.DOWN:
# 检查是否仍在中枢内：向下时终点低价在中枢区间内
if current_bi.end_klc.low <= last_zs.zg and current_bi.end_klc.low >= last_zs.zd:
print(f"延伸向下笔中枢: 终点时间={current_bi.end_klc.end_time}")
# 延伸中枢
last_zs.end_klc = current_bi.end_klc
last_zs.end_bi = current_bi
last_zs.end_idx = i
else:
# 笔离开中枢，记录第一个离开的笔
if not first_bi_out:
first_bi_out = current_bi
bi_out_count += 1
print(f"笔离开向下中枢: 时间={current_bi.end_klc.end_time}, 价格={current_bi.end_klc.low}, 中枢下沿={last_zs.zd}")
else:
if not in_again:
# 第二次离开，确认中枢完成
print(f"确认向下笔中枢完成: 时间={current_bi.end_klc.end_time}")
last_zs.is_sure = True
last_zs.sure_bi = current_bi
last_zs.sure_time = current_bi.end_klc.end_time
# 方向改变，判断是否破坏中枢
else:
# 方向改变可能导致重新进入中枢或破坏中枢
# 向上中枢被向下笔破坏：低点低于中枢下沿
# 向下中枢被向上笔破坏：高点高于中枢上沿
if (last_zs.direction == Chan_ZS_DIR.UP and current_bi.end_klc.low < last_zs.zd) or \
(last_zs.direction == Chan_ZS_DIR.DOWN and current_bi.end_klc.high > last_zs.zg):
# 破坏中枢
print(f"笔中枢被破坏: 方向={current_bi.dir}, 时间={current_bi.end_klc.end_time}")
last_zs.is_sure = True
last_zs.sure_bi = current_bi
last_zs.sure_time = current_bi.end_klc.end_time
elif first_bi_out:
# 重新进入中枢
print(f"笔重新进入中枢: 时间={current_bi.end_klc.end_time}")
in_again = True
first_bi_out = None
# 延伸中枢
last_zs.end_klc = current_bi.end_klc
last_zs.end_bi = current_bi
last_zs.end_idx = i
# 打印识别结果
print(f"笔中枢识别完成，共找到 {len(bi_zs_list)} 个笔中枢")
return bi_zs_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:
if len(bi_list) > 0:
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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 = 8
slow = 16
period = 6
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
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
Reference in New Issue View Git Blame Copy Permalink