Python run code after particular interval of time (Mimic candles)
def round_down_to_interval(dt, interval_minutes):
remainder = dt.minute % interval_minutes
minutes_to_current_boundary = remainder
rounded_dt = dt - timedelta(minutes=minutes_to_current_boundary)
rounded_dt = rounded_dt.replace(second=0, microsecond=0)
return rounded_dt
def determine_min(minstr):
min = 0
if minstr == "1":
min = 1
if minstr == "3":
min = 3
if minstr == "5":
min = 5
if minstr == "15":
min = 15
if minstr == "30":
min = 30
return min
Function calling
if current_time>EntryTime and current_time < ExitTime and datetime.now() >= params["runtime"]:
try:
time.sleep(int(1))
presentDay = FyresIntegration.fetchOHLC(symbol=formatedsymbol, resolution=params['TimeFrame'],atrperiod=params['Atr_Period'])
previousBar = presentDay.iloc[-1]
currentBar = presentDay.iloc[-1]
currentBar_close = currentBar[4]
# print("previousBar: ",previousBar)
previousBar_close=previousBar[4]
previousBar_low = previousBar[3]
previousBar_high = previousBar[2]
if params['Trade']=="BUY":
atr = previousBar[6]
params["SecondarySl"] = previousBar_low-(atr*float(params['Atr_Multiplier']))
if params['Trade']=="SHORT":
atr = previousBar[6]
params["SecondarySl"] = previousBar_high + (atr*float(params['Atr_Multiplier']))
next_specific_part_time = datetime.now() + timedelta(
seconds=determine_min(params["TimeFrame"]) * 60)
next_specific_part_time = round_down_to_interval(next_specific_part_time,
determine_min(str(params["TimeFrame"])))
print("Next datafetch time = ", next_specific_part_time)
params['runtime'] = next_specific_part_time
except Exception as e:
print("Error happened in fetching data : ", str(e))
traceback.print_exc()
Comments
Post a Comment