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 from technical.util import resample_to_interval from decimal import Decimal import numpy as np from ChanMACD import ChanMACD from TF_DF import TF_DF class ChanLun(): def __init__(self): self.time2m = 2 self.time3m = 3 self.time5m = 5 self.time10m = 10 self.time20m = 20 self.time_m_intervals = [2, 3, 5, 10, 20] self.time_m_symbols = ['2m', '3m', '5m', '10m', '20m'] self.time30m = 30 self.time45m = 45 self.time_m15_intervals = [30, 45] self.time_m15_symbols = ['30m', '45m'] 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.time_d_intervals = [2*24*60, 3*24*60] self.time_d_symbols = ['2d', '3d'] self.time1w = 7*24*60 self.time2w = 14*24*60 self.time_w_intervals = [14*24*60] self.time_w_symbols = ['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', '2m', '3m', '5m', '10m', '15m', '20m', '30m', '45m','1h', '2h', '4h', '6h', '8h', '12h', '16h', '1d', '2d', '3d', '1w', '2w', '1M', '3M', '6M', '1y'] self.tf_df_dict = {} self.ema_symbols = ['1m', '2m', '3m', '5m', '10m', '15m', '20m', '30m', '45m', '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=None, dataframe_15m=None, dataframe_h=None, dataframe_d=None, dataframe_w=None, dataframe_M=None): if dataframe_m is not None: self.tf_df_dict['1m'] = TF_DF(dataframe_m, 1, '1m') self.init_data(dataframe_m, self.time_m_intervals, self.time_m_symbols) if dataframe_15m is not None: self.tf_df_dict['15m'] = TF_DF(dataframe_15m, 1, '15m') self.init_data(dataframe_15m, self.time_m15_intervals, self.time_m15_symbols) if dataframe_h is not None: self.tf_df_dict['1h'] = TF_DF(dataframe_h, 1, '1h') self.init_data(dataframe_h, self.time_h_intervals, self.time_h_symbols) if dataframe_d is not None: self.tf_df_dict['1d'] = TF_DF(dataframe_d, 1, '1d') self.init_data(dataframe_d, self.time_d_intervals, self.time_d_symbols) if dataframe_w is not None: self.tf_df_dict['1w'] = TF_DF(dataframe_w, 1, '1w') self.init_data(dataframe_w, self.time_w_intervals, self.time_w_symbols) if dataframe_M is not None: 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 def get_tf_df_by_timeframe(self, timeframe): if timeframe in self.tf_df_dict: return self.tf_df_dict[timeframe] return None def check_price_ema52(self, price): key_list = [] if len(self.tf_df_dict) > 0: ema52_dict = self.get_ema52_dict() for key in self.ema_symbols: if ema52_dict[key] is not None: if abs(price - ema52_dict[key]) < 100: key_list.append(key) return key_list # 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))