KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use terminate_session method in localstack
Best Python code snippet using localstack_python
DNN_trader.py
Source:DNN_trader.py 
...64 if attempt >= max_attempts:65 print("max_attempts reached !")66 try: #try to terminate session67 time.sleep(wait)68 self.terminate_session(cause = "Unexpected Session Stop ( too many errors).")69 except Exception as e:70 print(e, end = " | ")71 print("Could not terminate_session properly!")72 finally:73 break74 else: # try again 75 time.sleep(wait)76 wait += wait_increase77 self.tick_data = pd.DataFrame()78 def on_success(self,time,bid,ask):79 print(self.ticks, end = " ")80 # collect and store tick data81 recent_tick = pd.to_datetime(time)82 #define stop because of low market violatility83 # if recent_tick.time() >= pd.to_datetime("17:30").time():84 # self.stop_stream = True85 ############## Define Stop ##############86 # if self.ticks >= 100:87 # self.terminate_session(cause = "Scheduled Session End.")88 # return 89 #########################################90 df = pd.DataFrame({self.instrument:(ask+bid)/2},index = [recent_tick])91 self.tick_data = self.tick_data.append(df)92 #if a time longer than the bar_length has elapsed between last full bar and the most recent time93 if recent_tick - self.last_bar > self.bar_length:94 self.resample_and_join()95 self.define_strategy()96 self.execute_trades()97 def resample_and_join(self):98 #append the most recent ticks(resample) to self.data 99 self.raw_data = self.raw_data.append(self.tick_data.resample(self.bar_length,100 label="right").last().ffill().iloc[:-1])101 self.tick_data = self.tick_data.iloc[-1:] #only keep the latest tick(next bar)102 self.last_bar = self.raw_data.index[-1] #update time of last full bar 103 def define_strategy(self):104 df = self.raw_data.copy()105 #******************** define your strategy here ************************106 #create features107 df = df.append(self.tick_data) # append latest tick (== open price of current bar)108 df["returns"] = np.log(df[self.instrument] / df[self.instrument].shift())109 df["Dir"] = np.where(df["returns"] > 0, 1, 0)110 df["SMA"] = df[self.instrument].rolling(self.window).mean() - df[self.instrument].rolling(150).mean()111 df["Boll"] = (df[self.instrument] - df[self.instrument].rolling(self.window).mean()) / df[self.instrument].rolling(self.window).std()112 df["MACD"] = df[self.instrument].ewm(span = 12, adjust = False).mean() - df[self.instrument].ewm(span = 26, adjust = False).mean()113 df["Min"] = df[self.instrument].rolling(self.window).min() / df[self.instrument] - 1114 df["Max"] = df[self.instrument].rolling(self.window).max() / df[self.instrument] - 1115 df["Mom"] = df["returns"].rolling(3).mean()116 df["Vol"] = df["returns"].rolling(self.window).std()117 df.dropna(inplace = True)118 119 # create lags120 self.cols = []121 features = ["Dir", "SMA", "Boll", "Min", "Max", "Mom", "Vol"]122 for f in features:123 for lag in range(1, self.lags + 1):124 col = "{}_lag_{}".format(f, lag)125 df[col] = df[f].shift(lag)126 self.cols.append(col)127 df.dropna(inplace = True)128 129 # standardization130 df_s = (df - self.mu) / self.std131 # predict132 df["proba"] = self.model.predict(df_s[self.cols])133 134 #determine positions135 df = df.loc[self.start_time:].copy() # starting with first live_stream bar (removing historical bars)136 df["position"] = np.where(df.proba < 0.47, -1, np.nan)137 df["position"] = np.where(df.proba > 0.53, 1, df.position)138 df["position"] = df.position.ffill().fillna(0) # start with neutral position if no strong signal139 #***********************************************************************140 141 self.data = df.copy()142 def execute_trades(self):143 if self.data["position"].iloc[-1] == 1:144 if self.position == 0:145 order = self.create_order(self.instrument, self.units, suppress = True, ret = True)146 self.report_trade(order, "GOING LONG")147 elif self.position == -1:148 order = self.create_order(self.instrument, self.units * 2, suppress = True, ret = True) 149 self.report_trade(order, "GOING LONG")150 self.position = 1151 elif self.data["position"].iloc[-1] == -1: 152 if self.position == 0:153 order = self.create_order(self.instrument, -self.units, suppress = True, ret = True)154 self.report_trade(order, "GOING SHORT")155 elif self.position == 1:156 order = self.create_order(self.instrument, -self.units * 2, suppress = True, ret = True)157 self.report_trade(order, "GOING SHORT")158 self.position = -1159 elif self.data["position"].iloc[-1] == 0: 160 if self.position == -1:161 order = self.create_order(self.instrument, self.units, suppress = True, ret = True) 162 self.report_trade(order, "GOING NEUTRAL")163 elif self.position == 1:164 order = self.create_order(self.instrument, -self.units, suppress = True, ret = True)165 self.report_trade(order, "GOING NEUTRAL")166 self.position = 0167 def report_trade(self, order, going):168 time = order["time"]169 units = order["units"]170 price = order["price"]171 pl = float(order["pl"])172 self.profits.append(pl)173 cumpl = sum(self.profits)174 print("\n" + 100* "-")175 print("{} | {}".format(time, going))176 print("{} | units = {} | price = {} | P&L = {} | Cum P&L = {}".format(time, units, price, pl, cumpl))177 print(100 * "-" + "\n")178 def terminate_session(self, cause): #NEW179 self.stop_stream = True180 if self.position != 0:181 close_order = self.create_order(self.instrument, units = -self.position * self.units,182 suppress = True, ret = True)183 self.report_trade(close_order, "GOING NEUTRAL")184 self.position = 0185 print(cause)186#Only execute when directly called from terminal 187if __name__ == "__main__": 188 # Loading the model189 model = keras.models.load_model("DNN_model")190 # Loading mu and std191 params = pickle.load(open("params.pkl", "rb"))192 mu = params["mu"]...
run.py
Source:run.py
...66 continue67 except socket.error as e:68 print('Error occured waiting for R10 response:')69 sleep(2)70 garminConnect.terminate_session()71 check_gspro_status()72 garminConnect = GarminConnect(gsProConnect, _config["garmin"]["port"])73 continue74 except KeyboardInterrupt:75 break76 except Exception as e:77 _logger.exception("General error: ")78 _logger.exception(e)79 continue80 finally:81 print('closing sockets')82 if(gsProConnect):83 gsProConnect.terminate_session()84 if(garminConnect):85 garminConnect.terminate_session()86 sys.exit()87if __name__ == "__main__":...
group_api.py
Source:group_api.py
1#2# Author: Dlo Bagari3# created Date: 12-10-20194from flask import jsonify5class GroupApi:6 def __init__(self, config, logger):7 self._group = None8 self._config = config9 self._logger = logger10 def filter_group_by_name(self, group_name):11 error_code, error_message, group = self._group.filter_groups_by_name(group_name)12 if error_code != 0:13 self._logger.error(self, error_message)14 return jsonify({"error": error_message}), 40015 else:16 return jsonify(group), 20017 def set_group(self, group):18 self._group = group19 def change_group(self, user_id, current_group, new_group, terminate_session=True):20 error_code, error_message, group = self._group.change_group(user_id, current_group, new_group,21 terminate_session=terminate_session)22 if error_code != 0 or len(error_message) != 0:23 self._logger.error(self, error_message)24 return jsonify({"error": error_message, "status": "failed"}), 40025 else:...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
