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, Chan_PRICE_TREND, Chan_KLU_PATTERN 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 from TF_DF import TF_DF class ChanLun(): def __init__(self): self.time3m = 3 self.time5m = 5 self.time10m = 10 self.time15m = 15 self.time30m = 30 self.time_m_intervals = [3, 5, 10, 15, 30] self.time_m_symbols = ['3m', '5m', '10m', '15m', '30m'] self.time2h = 2*60 self.time4h = 4*60 self.time6h = 6*60 self.time8h = 8*60 self.time12h = 12*60 self.time16h = 16*60 self.time_h_intervals = [2*60, 4*60, 6*60, 8*60, 12*60, 16*60] self.time_h_symbols = ['2h', '4h', '6h', '8h', '12h', '16h'] self.time2d = 2*24*60 self.time3d = 3*24*60 self.time1w = 7*24*60 self.time2w = 14*24*60 self.time_d_intervals = [2*24*60, 3*24*60, 7*24*60, 14*24*60] self.time_d_symbols = ['2d', '3d', '1w', '2w'] self.time2M = 2*30*24*60 self.time3M = 3*30*24*60 self.time6M = 6*30*24*60 self.time1y = 12*30*24*60 self.time_M_intervals = [2*30*24*60, 3*30*24*60, 6*30*24*60, 12*30*24*60] self.time_M_symbols = ['2M', '3M', '6M', '1y'] self.time_symbols = ['1m', '3m', '5m', '10m', '15m', '30m', '1h', '2h', '4h', '6h', '8h', '12h', '16h', '1d', '2d', '3d', '1w', '2w', '1M', '3M', '6M', '1y'] self.tf_df_dict = {} self.ema_symbols = ['1m', '3m', '5m', '10m', '15m', '30m', '1h', '2h', '4h', '6h', '8h', '12h', '16h', '1d', '2d', '3d'] self.tf_df = TF_DF() def init_data(self, dataframe, intervals, timeframes): for index in range(0, len(intervals)): timeframe = timeframes[index] interval = intervals[index] self.tf_df_dict[timeframe] = TF_DF(dataframe, interval, timeframe) def init_dataframes(self, dataframe_m, dataframe_h, dataframe_d, dataframe_M): self.tf_df_dict['1m'] = TF_DF(dataframe_m, 1, '1m') self.init_data(dataframe_m, self.time_m_intervals, self.time_m_symbols) self.tf_df_dict['1h'] = TF_DF(dataframe_h, 1, '1h') self.init_data(dataframe_h, self.time_h_intervals, self.time_h_symbols) self.tf_df_dict['1d'] = TF_DF(dataframe_d, 1, '1d') self.init_data(dataframe_d, self.time_d_intervals, self.time_d_symbols) self.tf_df_dict['1M'] = TF_DF(dataframe_M, 1, '1M') self.init_data(dataframe_M, self.time_M_intervals, self.time_M_symbols) def get_ema52_dict(self): if len(self.tf_df_dict) > 0: return {key: self.tf_df_dict[key].get_ema52() for key in self.ema_symbols} return None def get_ema24_dict(self): if len(self.tf_df_dict) > 0: return {key: self.tf_df_dict[key].get_ema24() for key in self.ema_symbols} return None def get_current_klc_dict(self): if len(self.tf_df_dict) > 0: return {key: self.tf_df_dict[key].get_current_klc() for key in self.ema_symbols} return None # TF_DF methods ------------------------------------------ def get_klu_state(self, dataframe): return self.tf_df.get_klu_state(dataframe) def check_fx(self, klc): return self.tf_df.check_fx(klc) def add_indicators1(self, df): return self.tf_df.add_indicators(df) def get_bi_list(self, dataframe): return self.tf_df.get_bi_list(dataframe) def get_kl_data(self, dataframe:DataFrame): return self.tf_df.cal_kl_data(dataframe) def cal_volume_ratio(self, dataframe, window=10): return self.tf_df.cal_volume_ratio(dataframe, window) def calculate_zs(self, bi_list, seg_list): return self.get_zs_list(bi_list, seg_list) def get_seg_list(self, bi_list): return self.tf_df.get_seg_list(bi_list) def cal_trend(self, klc_list): return self.tf_df.cal_trend(klc_list) def check_top_fx(self, last_bottom, klc): return self.tf_df.check_top_fx(last_bottom, klc) def check_bottom_fx(self, last_top, klc): return self.tf_df.check_bottom_fx(last_top, klc) def cal_bi_list(self, klc_list): return self.tf_df.cal_bi_list(klc_list) def get_zs_list(self, bi_list, seg_list): return self.tf_df.get_zs_list(bi_list, seg_list) def cal_bi_zs(self, seg_list): return self.tf_df.cal_bi_zs(seg_list) def get_decimal(self, value): return Decimal("{:.2f}".format(value)) def get_klc_list(self, klu_list): return self.tf_df.get_klc_list(klu_list) def get_klu_list(self, dataframe): return self.tf_df.cal_klu_pattern(self.get_kl_data(dataframe))