KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _order_dict method in localstack
Best Python code snippet using localstack_python
current_orders.py
Source:current_orders.py 
1from sourse.ui.modules.base_qdockwidget_module import BaseUIModule2from datetime import datetime3from PyQt5 import QtWidgets, QtCore, QtGui4from crypto_futures_py import AbstractExchangeHandler5import dataclasses6import typing7import numpy as np8class CurrentOrdersModule(BaseUIModule):9 order_canceled = QtCore.pyqtSignal(str)10 all_orders_canceled = QtCore.pyqtSignal()11 rebuild_grid = QtCore.pyqtSignal()12 def _create_widgets(self):13 self.layout = QtWidgets.QHBoxLayout(self.base_widget)14 self.parent_widget.setWindowTitle("Orders")15 self.base_widget.setLayout(self.layout)16 self._order_dict: typing.Dict[17 str, typing.Tuple[int, AbstractExchangeHandler.OrderUpdate]18 ] = {}19 self._historical_order_dict: typing.Dict[20 str, typing.Tuple[int, AbstractExchangeHandler.OrderUpdate]21 ] = {}22 self.counter = 023 self.historical_counter = 024 self.horizontalHeaderLabelsList = [25 "Order id", # 026 "Client order id", # 127 "Status", # 228 "Symbol", # 329 "Price", # 430 "Average price", # 531 "Fee", # 632 "Fee asset", # 733 "Volume", # 834 "Volume realized", # 935 "Time", # 1036 "id", # 1137 ]38 self.colorfull_dictionary = {2: 2, 4: 8, 5: 8, 6: 6, 8: 8, 9: 9}39 self.table = QtWidgets.QTableWidget(40 len(self._order_dict), len(self.horizontalHeaderLabelsList)41 )42 self.table_historical = QtWidgets.QTableWidget(43 len(self._order_dict), len(self.horizontalHeaderLabelsList)44 )45 self.table.setSortingEnabled(True)46 self.table_historical.setSortingEnabled(True)47 self.table.setHorizontalHeaderLabels(self.horizontalHeaderLabelsList)48 self.table_historical.setHorizontalHeaderLabels(self.horizontalHeaderLabelsList)49 self.table.verticalHeader().hide()50 self.table_historical.verticalHeader().hide()51 header = self.table.horizontalHeader()52 header_historical = self.table_historical.horizontalHeader()53 for i in range(len(self.horizontalHeaderLabelsList)):54 header.setSectionResizeMode(i, QtWidgets.QHeaderView.Stretch)55 header_historical.setSectionResizeMode(i, QtWidgets.QHeaderView.Stretch)56 header_historical.setSectionResizeMode(57 len(self.horizontalHeaderLabelsList) - 1,58 QtWidgets.QHeaderView.ResizeToContents,59 )60 self.menu = QtWidgets.QMenu()61 self.tabwidget = QtWidgets.QTabWidget()62 self.tabwidget.addTab(self.table, "Current orders")63 self.tabwidget.addTab(self.table_historical, "Historical orders")64 self.layout.addWidget(self.tabwidget)65 self.table.sortItems(66 len(self.horizontalHeaderLabelsList) - 1, QtCore.Qt.AscendingOrder67 )68 self.table_historical.sortItems(69 len(self.horizontalHeaderLabelsList) - 1, QtCore.Qt.AscendingOrder70 )71 self.table.setColumnHidden(len(self.horizontalHeaderLabelsList) - 1, True)72 self.table_historical.setColumnHidden(73 len(self.horizontalHeaderLabelsList) - 1, True74 )75 self.table.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)76 self.table.customContextMenuRequested.connect(self.generateMenu)77 self.table.viewport().installEventFilter(self)78 self._stashed_orders: typing.List[79 typing.Tuple[AbstractExchangeHandler.OrderUpdate, bool]80 ] = []81 self._orders_to_delete: typing.Dict[str, datetime] = {}82 self.order_update_timer = QtCore.QTimer(self)83 self.order_update_timer.setInterval(200)84 self.order_update_timer.timeout.connect(self._update_orders_threadsafe)85 self.order_update_timer.start()86 def get_current_orders(self) -> typing.List[AbstractExchangeHandler.OrderUpdate]:87 return [v[1] for v in self._order_dict.values()]88 def get_historical_orders(self) -> typing.List[AbstractExchangeHandler.OrderUpdate]:89 return [v[1] for v in self._historical_order_dict.values()]90 def eventFilter(self, source, event):91 if (92 event.type() == QtCore.QEvent.MouseButtonPress93 and event.buttons() == QtCore.Qt.RightButton94 ):95 self._on_right_button_clicked(event)96 return True97 else:98 return False99 def _on_right_button_clicked(self, event):100 self.menu = QtWidgets.QMenu()101 item = self.table.itemAt(event.pos())102 if item is not None:103 if self.table.item(item.row(), 2).text() == "NEW":104 client_order_ID = self.table.item(item.row(), 1).text()105 self.menu.addAction(106 "Cancel order", lambda: self.order_canceled.emit(client_order_ID),107 )108 self.menu.addAction(109 "Cancel all orders", lambda: self.all_orders_canceled.emit()110 )111 self.menu.addAction("Rebuild grid", lambda: self.rebuild_grid.emit())112 def generateMenu(self, pos):113 try:114 self.menu.exec_(self.table.mapToGlobal(pos))115 except:116 pass117 @QtCore.pyqtSlot()118 def _update_orders_threadsafe(self):119 for inp in self._stashed_orders.copy():120 self.add_order(*inp)121 self._stashed_orders.remove(inp)122 for order, time in self._orders_to_delete.copy().items():123 if order not in self._order_dict:124 del self._orders_to_delete[order]125 elif (datetime.now() - time).total_seconds() > 3:126 self.remove_order(order)127 del self._orders_to_delete[order]128 def add_order_threadsafe(129 self, order: AbstractExchangeHandler.OrderUpdate, historical_table: bool = False130 ):131 self._stashed_orders.append((order, historical_table))132 def add_order(133 self, order: AbstractExchangeHandler.OrderUpdate, historical_table: bool = False134 ) -> str:135 order_id = order.client_orderID if order.client_orderID != "" else order.orderID136 if order.status in ["FILLED", "CANCELED", "FAILED", "EXPIRED"]:137 self._orders_to_delete[order_id] = datetime.now()138 current_sorted_index = self.table.horizontalHeader().sortIndicatorSection()139 current_sorted_type = self.table.horizontalHeader().sortIndicatorOrder()140 self.table.sortItems(141 len(self.horizontalHeaderLabelsList) - 1, QtCore.Qt.AscendingOrder142 )143 self.table.setSortingEnabled(False)144 current_sorted_historical_index = (145 self.table_historical.horizontalHeader().sortIndicatorSection()146 )147 current_sorted_historical_type = (148 self.table_historical.horizontalHeader().sortIndicatorOrder()149 )150 self.table_historical.sortItems(10, QtCore.Qt.AscendingOrder)151 self.table_historical.setSortingEnabled(False)152 if not historical_table:153 if order_id in self._order_dict.keys():154 res = self._edit_order(order)155 elif order_id in self._historical_order_dict.keys():156 res = self.add_order(order, True)157 else:158 self._order_dict[order_id] = (len(self._order_dict), order)159 self.table.setRowCount(len(self._order_dict))160 for i, value in enumerate(dataclasses.asdict(order).values()):161 color, value = self.highlight(i, order, value)162 self.table.setItem(163 len(self._order_dict) - 1, i, self.createItem(str(value), color)164 )165 self.table.setItem(166 len(self._order_dict) - 1,167 len(self.horizontalHeaderLabelsList) - 1,168 self.QTableWidgetIntegerItem(str(self.counter)),169 )170 self.counter += 1171 res = order_id172 else:173 if order_id in self._historical_order_dict.keys():174 res = self._edit_order(order, True)175 else:176 self._historical_order_dict[order_id] = (177 len(self._historical_order_dict),178 order,179 )180 self.table_historical.setRowCount(len(self._historical_order_dict))181 for i, value in enumerate(dataclasses.asdict(order).values()):182 color, value = self.highlight(i, order, value)183 self.table_historical.setItem(184 len(self._historical_order_dict) - 1,185 i,186 self.createItem(str(value), color),187 )188 self.table_historical.setItem(189 len(self._historical_order_dict) - 1,190 len(self.horizontalHeaderLabelsList) - 1,191 self.QTableWidgetIntegerItem(str(self.historical_counter)),192 )193 self.historical_counter += 1194 res = order_id195 self.table.setSortingEnabled(True)196 self.table.sortItems(current_sorted_index, current_sorted_type)197 self.table_historical.setSortingEnabled(True)198 self.table_historical.sortItems(199 current_sorted_historical_index, current_sorted_historical_type200 )201 return res202 def _edit_order(203 self, order: AbstractExchangeHandler.OrderUpdate, historical_table: bool = False204 ) -> str:205 order_id = order.client_orderID if order.client_orderID != "" else order.orderID206 if historical_table:207 for i in range(self.table_historical.rowCount()):208 if str(order_id) == str(209 self.table_historical.item(210 i, 1 if order.client_orderID != "" else 0211 ).text()212 ):213 row = i214 break215 else:216 raise ValueError(f"{order_id} not historical table")217 order_index = self._historical_order_dict[order_id][0]218 self._historical_order_dict[order_id] = order_index, order219 else:220 for i in range(self.table.rowCount()):221 if str(order_id) == str(222 self.table.item(i, 1 if order.client_orderID != "" else 0).text()223 ):224 row = i225 break226 else:227 raise ValueError(f"{order_id} not historical table")228 order_index = self._order_dict[order_id][0]229 self._order_dict[order_id] = order_index, order230 for i, (key, value) in enumerate(dataclasses.asdict(order).items()):231 color, value = self.highlight(i, order, value)232 if key == "message":233 continue234 elif not historical_table:235 self.table.item(row, i).setText(str(value))236 self.table.item(row, i).setForeground(237 QtGui.QBrush(QtGui.QColor(*color))238 )239 else:240 self.table_historical.item(row, i).setText(str(value))241 self.table_historical.item(row, i).setForeground(242 QtGui.QBrush(QtGui.QColor(*color))243 )244 return order_id245 def highlight(246 self,247 i: int,248 order: AbstractExchangeHandler.OrderUpdate,249 value: typing.Union[int, float, str],250 ) -> typing.Tuple[typing.Tuple[int, int, int], typing.Union[int, float, str]]:251 if i in self.colorfull_dictionary.keys():252 j = list(dataclasses.asdict(order).values())[self.colorfull_dictionary[i]]253 try:254 if i == 5:255 if float(j) < 0:256 color = Colors.limegreen257 elif float(j) > 0:258 color = Colors.orangered259 else:260 color = Colors.yellow261 value = np.format_float_positional(value)262 elif float(j) > 0:263 color = Colors.limegreen264 elif float(j) < 0:265 color = Colors.orangered266 else:267 color = Colors.yellow268 except:269 color = (270 Colors.limegreen271 if str(j) == "FILLED"272 else Colors.orangered273 if str(j) == "CANCELED"274 else Colors.yellow275 if str(j) == "NEW"276 else Colors.darkorange277 if str(j) == "PENDING"278 else Colors.white279 )280 else:281 color = Colors.white282 return color, value283 def remove_order(self, order_id) -> None:284 for i in range(self.table.rowCount()):285 if str(order_id) == str(self.table.item(i, 1).text()):286 row = i287 break288 else:289 raise ValueError(f"{order_id} not in 'Client order id' column")290 current_sorted_index = self.table.horizontalHeader().sortIndicatorSection()291 current_sorted_type = self.table.horizontalHeader().sortIndicatorOrder()292 self.table.sortItems(293 len(self.horizontalHeaderLabelsList) - 1, QtCore.Qt.AscendingOrder294 )295 self.table.setSortingEnabled(False)296 self.add_order(self._order_dict[order_id][1], True)297 self.table.removeRow(row)298 del self._order_dict[order_id]299 self.table.setSortingEnabled(True)300 self.table.sortItems(current_sorted_index, current_sorted_type)301 def remove_all_orders(self) -> None:302 self.table.setSortingEnabled(False)303 self._transfer_table()304 self._order_dict = {}305 self.table.setRowCount(0)306 self.table.setSortingEnabled(True)307 def _transfer_table(self) -> None:308 for i in self._order_dict:309 self.add_order(self._order_dict[i][1], True)310 class QTableWidgetIntegerItem(QtWidgets.QTableWidgetItem):311 def __lt__(self, other):312 try:313 return int(self.text()) < int(other.text())314 except ValueError:315 return self.text() < other.text()316 def createItem(317 self, text: str, color: typing.Tuple[int, int, int]318 ) -> QtWidgets.QTableWidgetItem:319 tableWidgetItem = QtWidgets.QTableWidgetItem(text)320 tableWidgetItem.setForeground(QtGui.QBrush(QtGui.QColor(*color)))321 tableWidgetItem.setFlags(QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled)322 return tableWidgetItem323class Colors:324 red = (255, 0, 0)325 green = (0, 255, 0)326 blue = (0, 0, 255)327 white = (255, 255, 255)328 black = (0, 0, 0)329 yellow = (255, 255, 0)330 darkorange = (255, 140, 0)331 orangered = (255, 69, 0)332 limegreen = (50, 205, 50)333 neutralred = (239, 83, 80)334 neutralgreen = (38, 166, 154)335 def __init__(self):...
agent.py
Source:agent.py
1import asyncio2import typing as t3from .entities import SymbolPair4from .entities.account import Account5from .entities.order import Order6from .errors import (7 AgentError,8 OrderCancellationError,9 OrderCreationError,10)11from .exchange import Exchange12class CreateMarketOrder(t.NamedTuple):13 pair: SymbolPair14 side: Order.Side15 amount: float16 acc_name: str17 order_id: int18class CreateLimitOrder(t.NamedTuple):19 pair: SymbolPair20 price: float21 side: Order.Side22 amount: float23 acc_name: str24 order_id: int25class CancelOrder(t.NamedTuple):26 pair: SymbolPair27 order_id: int28class Sleep(t.NamedTuple):29 time: int30class Agent:31 _account: Account32 _instruction_list: t.List[33 t.Union[Sleep, CreateMarketOrder, CreateLimitOrder, CancelOrder]34 ]35 _order_book: t.Dict[int, Order]36 _working_event: asyncio.Event37 _task: t.Any38 _order_dict: t.Dict[int, int]39 _exchange = Exchange()40 is_run: bool41 def __init__(42 self,43 instructions: t.List[44 t.Union[Sleep, CreateMarketOrder, CreateLimitOrder, CancelOrder]45 ],46 account: Account,47 ):48 self._account = account49 self._instruction_list = instructions50 self._working_event = asyncio.Event()51 self._order_dict = {}52 self.is_run = False53 async def _run(self) -> None:54 for instruction in self._instruction_list:55 await self._working_event.wait()56 if isinstance(instruction, CreateLimitOrder):57 if instruction.order_id in self._order_dict:58 raise OrderCreationError("Order is already exist")59 order = await self._exchange.create_limit(60 instruction.pair,61 instruction.price,62 instruction.side,63 instruction.amount,64 instruction.acc_name,65 )66 self._order_dict[instruction.order_id] = order.order_id67 elif isinstance(instruction, CreateMarketOrder):68 if instruction.order_id in self._order_dict:69 raise OrderCreationError("Order is already exist")70 order = await self._exchange.create_market(71 instruction.pair,72 instruction.side,73 instruction.amount,74 instruction.acc_name,75 )76 self._order_dict[instruction.order_id] = order.order_id77 elif isinstance(instruction, CancelOrder):78 try:79 order_id = self._order_dict[instruction.order_id]80 except KeyError:81 raise OrderCancellationError82 await self._exchange.cancel_order(instruction.pair, order_id)83 elif isinstance(instruction, Sleep):84 await asyncio.sleep(instruction.time)85 self.is_run = False86 async def run(self) -> None:87 if self.is_run:88 raise AgentError("Agent is already running")89 task = asyncio.create_task(self._run())90 self._task = task91 self._working_event.set()92 self.is_run = True93 await task94 def pause(self) -> None:95 self._working_event.clear()96 def stop(self) -> None:97 if not self.is_run:98 raise AgentError("Agent is already closed")99 self.pause()100 self._task.cancel()...
lob.py
Source:lob.py
1from limit_price import LimitPriceNode, AVL_Tree2from order import order, order_wrapper3class LOB:4 5 def __init__(self) -> None:6 self.ask_tree = None7 self.bid_tree = None8 self.best_ask = None9 self.best_bid = None10 self.ask_tree_self = AVL_Tree()11 self.bid_tree_self = AVL_Tree()12 self._order_dict = {}13 self._price_dict = {}14 15 def process(self, new_order: order):16 ## Need also to update _order_dict and _price_dict17 if new_order.id in self._order_dict:18 # Order currently in the record19 if new_order.quantity == 0:20 # Remove order21 if not self._order_dict[new_order.id].prev_order and not self._order_dict[new_order.id].next_order:22 # Need also to delete the limit price node23 if new_order.is_sell:24 # Update the best ask25 if self.best_ask.price == new_order.price:26 self.best_ask = self._price_dict[new_order.price].parent if self._price_dict[new_order.price].parent else None27 self.ask_tree = self.ask_tree_self.delete(self.ask_tree, new_order.price)28 else:29 # Update the best bid30 if self.best_bid.price == new_order.price:31 self.best_bid = self._price_dict[new_order.price].parent if self._price_dict[new_order.price].parent else None32 self.bid_tree = self.bid_tree_self.delete(self.bid_tree, new_order.price)33 34 del self._price_dict[new_order.price]35 else: 36 self._order_dict[new_order.id].pop()37 38 del self._order_dict[new_order.id]39 else:40 # Update order41 self._order_dict[new_order.id].update(new_order)42 else:43 # New order44 if new_order.price in self._price_dict:45 # The LimitPriceNode exists46 self._price_dict[new_order.price].wrapper.append(new_order)47 self._order_dict[new_order.id] = new_order48 else:49 # The LimitPriceNode doesn't exist50 new_node = LimitPriceNode(new_order)51 if not new_order.is_sell:52 # A new sell order53 self.best_bid = new_node if (self.best_bid is None or self.best_bid.price<new_node.price) else self.best_bid54 self.bid_tree = self.bid_tree_self.insert(self.bid_tree, new_node)55 else:56 # A new buy order57 self.best_ask = new_node if (self.best_ask is None or self.best_ask.price>new_node.price) else self.best_ask58 self.ask_tree = self.ask_tree_self.insert(self.ask_tree, new_node)59 60 self._price_dict[new_order.price] = new_node...
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?
