tradezero_api.main

  1from __future__ import annotations
  2
  3import time
  4import os
  5import warnings
  6from collections import namedtuple
  7
  8from selenium.webdriver.chrome.service import Service as ChromeService
  9from webdriver_manager.chrome import ChromeDriverManager
 10from selenium import webdriver
 11from selenium.webdriver.common.by import By
 12from selenium.webdriver.common.keys import Keys
 13from selenium.webdriver.support.ui import Select
 14from selenium.common.exceptions import NoSuchElementException, WebDriverException, StaleElementReferenceException
 15from termcolor import colored
 16
 17from .time_helpers import Time, Timer, time_it
 18from .watchlist import Watchlist
 19from .portfolio import Portfolio
 20from .notification import Notification
 21from .account import Account
 22from .enums import Order, TIF
 23
 24os.system('color')
 25
 26TZ_HOME_URL = 'https://standard.tradezeroweb.us/'
 27
 28
 29class TradeZero(Time):
 30    def __init__(self, user_name: str, password: str, headless: bool = False,
 31                 hide_attributes: bool = False):
 32        """
 33        :param user_name: TradeZero user_name
 34        :param password: TradeZero password
 35        :param headless: default: False, True will run the browser in headless mode, which means it won't be visible
 36        :param hide_attributes: bool, if True: Hide account attributes (acc username, equity, total exposure...)
 37        """
 38        super().__init__()
 39        self.user_name = user_name
 40        self.password = password
 41        self.hide_attributes = hide_attributes
 42
 43        service = ChromeService(ChromeDriverManager().install())
 44        options = webdriver.ChromeOptions()
 45        options.add_experimental_option('excludeSwitches', ['enable-logging'])
 46        if headless is True:
 47            options.headless = headless
 48
 49        self.driver = webdriver.Chrome(service=service, options=options)
 50        self.driver.get(TZ_HOME_URL)
 51
 52        self.Watchlist = Watchlist(self.driver)
 53        self.Portfolio = Portfolio(self.driver)
 54        self.Notification = Notification(self.driver)
 55        self.Account = Account(self.driver)
 56
 57        # to instantiate the time, pytz, and datetime modules:
 58        Timer()
 59        self.time_between(time1=(9, 30), time2=(10, 30))
 60
 61    def _dom_fully_loaded(self, iter_amount: int = 1):
 62        """
 63        check that webpage elements are fully loaded/visible.
 64        there is no need to call this method, but instead call tz_conn() and that will take care of all the rest.
 65
 66        :param iter_amount: int, default: 1, number of times it will iterate.
 67        :return: if the elements are fully loaded: return True, else: return False.
 68        """
 69        container_xpath = "//*[contains(@id,'portfolio-container')]//div//div//h2"
 70        for i in range(iter_amount):
 71            elements = self.driver.find_elements(By.XPATH, container_xpath)
 72            text_elements = [x.text for x in elements]
 73            if 'Portfolio' in text_elements:
 74                return True
 75            time.sleep(0.5)
 76        return False
 77
 78    @time_it
 79    def login(self, log_time_elapsed: bool = False):
 80        """
 81        log-in TradeZero's website
 82
 83        :param log_time_elapsed: bool, if True it will print time elapsed for login
 84        """
 85        login_form = self.driver.find_element(By.ID, "login")
 86        login_form.send_keys(self.user_name)
 87
 88        password_form = self.driver.find_element(By.ID, "password")
 89        password_form.send_keys(self.password, Keys.RETURN)
 90
 91        self._dom_fully_loaded(150)
 92        if self.hide_attributes:
 93            self.Account.hide_attributes()
 94
 95        Select(self.driver.find_element(By.ID, "trading-order-select-type")).select_by_index(1)
 96
 97    def conn(self, log_tz_conn: bool = False):
 98        """
 99        make sure that the website stays connected and is fully loaded.
100        TradeZero will ask for a Login twice a day, and sometimes it will require the page to be reloaded,
101        so this will make sure that its fully loaded, by reloading or doing the login.
102
103        :param log_tz_conn: bool, default: False. if True it will print if it reconnects through the login or refresh.
104        :return: True if connected
105        :raises Exception: if it fails to reconnect after a while
106        """
107        if self._dom_fully_loaded(1):
108            return True
109
110        try:
111            self.driver.find_element(By.ID, "login")
112            self.login()
113
114            self.Watchlist.restore()
115
116            if log_tz_conn is True:
117                print(colored('tz_conn(): Login worked', 'cyan'))
118            return True
119
120        except NoSuchElementException:
121            self.driver.get("https://standard.tradezeroweb.us/")
122            if self._dom_fully_loaded(150):
123
124                if self.hide_attributes:
125                    self.Account.hide_attributes()
126
127                self.Watchlist.restore()
128
129                if log_tz_conn is True:
130                    print(colored('tz_conn(): Refresh worked', 'cyan'))
131                return True
132
133        raise Exception('@ tz_conn(): Error: not able to reconnect, max retries exceeded')
134
135    def exit(self):
136        """close Selenium window and driver"""
137        try:
138            self.driver.close()
139        except WebDriverException:
140            pass
141
142        self.driver.quit()
143
144    def load_symbol(self, symbol: str):
145        """
146        make sure the data for the symbol is fully loaded and that the symbol itself is valid
147
148        :param symbol: str
149        :return: True if symbol data loaded, False if prices == 0.00 (mkt closed), Error if symbol not found
150        :raises Exception: if symbol not found
151        """
152        if symbol.upper() == self.current_symbol():
153            price = self.driver.find_element(By.ID, "trading-order-ask").text.replace('.', '').replace(',', '')
154            if price.isdigit() and float(price) > 0:
155                return True
156
157        input_symbol = self.driver.find_element(By.ID, "trading-order-input-symbol")
158        input_symbol.send_keys(symbol.lower(), Keys.RETURN)
159        time.sleep(0.04)
160
161        for i in range(300):
162            price = self.driver.find_element(By.ID, "trading-order-ask").text.replace('.', '').replace(',', '')
163            if price == '':
164                time.sleep(0.01)
165
166            elif price.isdigit() and float(price) == 0:
167                warnings.warn(f"Market Closed, ask/bid = {price}")
168                return False
169
170            elif price.isdigit():
171                return True
172
173            elif i == 15 or i == 299:
174                last_notif = self.Notification.get_last_notification_message()
175                message = f'Symbol not found: {symbol.upper()}'
176                if message == last_notif:
177                    raise Exception(f"ERROR: {symbol=} Not found")
178
179    def current_symbol(self):
180        """get current symbol"""
181        return self.driver.find_element(By.ID, 'trading-order-symbol').text.replace('(USD)', '')
182
183    @property
184    def bid(self):
185        """get bid price"""
186        return float(self.driver.find_element(By.ID, 'trading-order-bid').text.replace(',', ''))
187
188    @property
189    def ask(self):
190        """get ask price"""
191        return float(self.driver.find_element(By.ID, 'trading-order-ask').text.replace(',', ''))
192
193    @property
194    def last(self):
195        """get last price"""
196        return float(self.driver.find_element(By.ID, 'trading-order-p').text.replace(',', ''))
197
198    def data(self, symbol: str):
199        """
200        return a namedtuple with data for the given symbol, the properties are:
201        'open', 'high', 'low', 'close', 'volume', 'last', 'ask', 'bid'.
202
203        :param symbol: str: ex: 'aapl', 'amd', 'NVDA', 'GM'
204        :return: namedtuple = (open, high, low, close, volume, last, ask, bid)
205        """
206        Data = namedtuple('Data', ['open', 'high', 'low', 'close', 'volume', 'last', 'ask', 'bid'])
207
208        if self.load_symbol(symbol) is False:
209            return Data(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0)
210
211        element_ids = [
212            'trading-order-open',
213            'trading-order-high',
214            'trading-order-low',
215            'trading-order-close',
216            'trading-order-vol',
217            'trading-order-p',
218            'trading-order-ask',
219            'trading-order-bid',
220        ]
221        lst = []
222        for id_ in element_ids:
223            val = self.driver.find_element(By.ID, id_).text
224            val = float(val.replace(',', ''))  # replace comma for volume, and when prices > 999
225            lst.append(val)
226
227        return Data._make(lst)
228
229    def calculate_order_quantity(self, symbol: str, buying_power: float, float_option: bool = False):
230        """
231        returns the amount of shares you can buy with the given buying_power as int(), but if float_option is True,
232        it will return the amount as a float.
233
234        :param symbol: str
235        :param buying_power: float,
236        :param float_option: bool, default: False, if True returns the original number as float
237        :return: int or float
238        """
239        if self.load_symbol(symbol) is False:
240            return
241        quantity = (buying_power / self.last)
242
243        if float_option is True:
244            return quantity
245        return int(quantity)
246
247    def locate_stock(self, symbol: str, share_amount: int, max_price: float = 0, debug_info: bool = False):
248        """
249        Locate a stock, requires: stock symbol, and share_amount. optional: max_price.
250        if the locate_price is less than max_price: it will accept, else: decline.
251
252        :param symbol: str, symbol to locate.
253        :param share_amount: int, must be a multiple of 100 (100, 200, 300...)
254        :param max_price: float, default: 0, total price you are willing to pay for locates
255        :param debug_info: bool, if True it will print info about the locates in the console
256        :return: named tuple with the following attributes: 'price_per_share' and 'total'
257        :raises Exception: if share_amount is not divisible by 100
258        """
259        Data = namedtuple('Data', ['price_per_share', 'total'])
260
261        if share_amount is not None and share_amount % 100 != 0:
262            raise Exception(f'ERROR: share_amount is not divisible by 100 ({share_amount=})')
263
264        if not self.load_symbol(symbol):
265            return
266
267        if self.last <= 1.00:
268            print(f'Error: Cannot locate stocks priced under $1.00 ({symbol=}, price={self.last})')
269
270        self.driver.find_element(By.ID, "locate-tab-1").click()
271        input_symbol = self.driver.find_element(By.ID, "short-list-input-symbol")
272        input_symbol.clear()
273        input_symbol.send_keys(symbol, Keys.RETURN)
274
275        input_shares = self.driver.find_element(By.ID, "short-list-input-shares")
276        input_shares.clear()
277        input_shares.send_keys(share_amount)
278
279        while self.driver.find_element(By.ID, "short-list-locate-status").text == '':
280            time.sleep(0.1)
281
282        if self.driver.find_element(By.ID, "short-list-locate-status").text == 'Easy to borrow':
283            locate_pps = 0.00
284            locate_total = 0.00
285            if debug_info:
286                print(colored(f'Stock ({symbol}) is "Easy to borrow"', 'green'))
287            return Data(locate_pps, locate_total)
288
289        self.driver.find_element(By.ID, "short-list-button-locate").click()
290
291        for i in range(300):
292            try:
293                locate_pps = float(self.driver.find_element(By.ID, f"oitem-l-{symbol.upper()}-cell-2").text)
294                locate_total = float(self.driver.find_element(By.ID, f"oitem-l-{symbol.upper()}-cell-6").text)
295                break
296
297            except (ValueError, NoSuchElementException, StaleElementReferenceException):
298                time.sleep(0.15)
299                if i == 15 or i == 299:
300                    insufficient_bp = 'Insufficient BP to short a position with requested quantity.'
301                    last_notif = self.Notification.get_last_notification_message()
302                    if insufficient_bp in last_notif:
303                        warnings.warn(f"ERROR! {insufficient_bp}")
304                        return
305        else:
306            raise Exception(f'Error: not able to locate symbol element ({symbol=})')
307
308        if locate_total <= max_price:
309            self.driver.find_element(By.XPATH, f'//*[@id="oitem-l-{symbol.upper()}-cell-8"]/span[1]').click()
310            if debug_info:
311                print(colored(f'HTB Locate accepted ({symbol}, $ {locate_total})', 'cyan'))
312        else:
313            self.driver.find_element(By.XPATH, f'//*[@id="oitem-l-{symbol.upper()}-cell-8"]/span[2]').click()
314
315        return Data(locate_pps, locate_total)
316        # TODO create a function to get the pps and another one that just locates the shares
317
318    def credit_locates(self, symbol: str, quantity=None):
319        """
320        sell/ credit stock locates, if no value is given in 'quantity', it will credit all the shares
321        available of the given symbol.
322
323        :param symbol: str
324        :param quantity: amount of shares to sell, must be a multiple of 100, ie: 100, 200, 300
325        :return:
326        :raises Exception: if given symbol in not already located
327        :raises ValueError: if quantity is not divisible by 100 or quantity > located shares
328        """
329        located_symbols = self.driver.find_elements(By.XPATH, '//*[@id="locate-inventory-table"]/tbody/tr/td[1]')
330        located_symbols = [x.text for x in located_symbols]
331
332        if symbol.upper() not in located_symbols:
333            raise Exception(f"ERROR! cannot find {symbol} in located symbols")
334
335        if quantity is not None:
336            if quantity % 100 != 0:
337                raise ValueError(f"ERROR! quantity is not divisible by 100 ({quantity=})")
338
339            located_shares = float(self.driver.find_element(By.ID, f"inv-{symbol.upper()}-cell-1").text)
340            if quantity > located_shares:
341                raise ValueError(f"ERROR! you cannot credit more shares than u already have "
342                                 f"({quantity} vs {located_shares}")
343
344            input_quantity = self.driver.find_element(By.ID, f"inv-{symbol.upper()}-sell-qty")
345            input_quantity.clear()
346            input_quantity.send_keys(quantity)
347
348        self.driver.find_element(By.XPATH, f'//*[@id="inv-{symbol.upper()}-sell"]/button').click()
349        return
350
351    @time_it
352    def limit_order(self, order_direction: Order, symbol: str, share_amount: int, limit_price: float,
353                    time_in_force: TIF = TIF.DAY, log_info: bool = False):
354        """
355        Place a Limit Order, the following params are required: order_direction, symbol, share_amount, and limit_price.
356
357        :param order_direction: str: 'buy', 'sell', 'short', 'cover'
358        :param symbol: str: e.g: 'aapl', 'amd', 'NVDA', 'GM'
359        :param limit_price: float
360        :param share_amount: int
361        :param time_in_force: str, default: 'DAY', must be one of the following: 'DAY', 'GTC', or 'GTX'
362        :param log_info: bool, if True it will print information about the order
363        :return: True if operation succeeded
364        :raises AttributeError: if time_in_force argument not one of the following: 'DAY', 'GTC', 'GTX'
365        """
366        symbol = symbol.lower()
367        order_direction = order_direction.value
368        time_in_force = time_in_force.value
369
370        if time_in_force not in ['DAY', 'GTC', 'GTX']:
371            raise AttributeError(f"Error: time_in_force argument must be one of the following: 'DAY', 'GTC', 'GTX'")
372
373        self.load_symbol(symbol)
374
375        order_menu = Select(self.driver.find_element(By.ID, "trading-order-select-type"))
376        order_menu.select_by_index(1)
377
378        tif_menu = Select(self.driver.find_element(By.ID, "trading-order-select-time"))
379        tif_menu.select_by_visible_text(time_in_force)
380
381        input_quantity = self.driver.find_element(By.ID, "trading-order-input-quantity")
382        input_quantity.clear()
383        input_quantity.send_keys(share_amount)
384
385        price_input = self.driver.find_element(By.ID, "trading-order-input-price")
386        price_input.clear()
387        price_input.send_keys(limit_price)
388
389        self.driver.find_element(By.ID, f"trading-order-button-{order_direction}").click()
390
391        if log_info is True:
392            print(f"Time: {self.time}, Order direction: {order_direction}, Symbol: {symbol}, "
393                  f"Limit Price: {limit_price}, Shares amount: {share_amount}")
394
395    @time_it
396    def market_order(self, order_direction: Order, symbol: str, share_amount: int,
397                     time_in_force: TIF = TIF.DAY, log_info: bool = False):
398        """
399        Place a Market Order, The following params are required: order_direction, symbol, and share_amount
400
401        :param order_direction: str: 'buy', 'sell', 'short', 'cover'
402        :param symbol: str: e.g: 'aapl', 'amd', 'NVDA', 'GM'
403        :param share_amount: int
404        :param time_in_force: str, default: 'DAY', must be one of the following: 'DAY', 'GTC', or 'GTX'
405        :param log_info: bool, if True it will print information about the order
406        :return:
407        :raises Exception: if time not during market hours (9:30 - 16:00)
408        :raises AttributeError: if time_in_force argument not one of the following: 'DAY', 'GTC', 'GTX'
409        """
410        symbol = symbol.lower()
411        order_direction = order_direction.value
412        time_in_force = time_in_force.value
413
414        if not self.time_between((9, 30), (16, 0)):
415            raise Exception(f'Error: Market orders are not allowed at this time ({self.time})')
416
417        if time_in_force not in ['DAY', 'GTC', 'GTX']:
418            raise AttributeError(f"Error: time_in_force argument must be one of the following: 'DAY', 'GTC', 'GTX'")
419
420        self.load_symbol(symbol)
421
422        order_menu = Select(self.driver.find_element(By.ID, "trading-order-select-type"))
423        order_menu.select_by_index(0)
424
425        tif_menu = Select(self.driver.find_element(By.ID, "trading-order-select-time"))
426        tif_menu.select_by_visible_text(time_in_force)
427
428        input_quantity = self.driver.find_element(By.ID, "trading-order-input-quantity")
429        input_quantity.clear()
430        input_quantity.send_keys(share_amount)
431
432        self.driver.find_element(By.ID, f"trading-order-button-{order_direction}").click()
433
434        if log_info is True:
435            print(f"Time: {self.time}, Order direction: {order_direction}, Symbol: {symbol}, "
436                  f"Price: {self.last}, Shares amount: {share_amount}")
437
438    @time_it
439    def stop_market_order(self, order_direction: Order, symbol: str, share_amount: int, stop_price: float,
440                          time_in_force: TIF = TIF.DAY, log_info: bool = False):
441        """
442        Place a Stop Market Order, the following params are required: order_direction, symbol,
443        share_amount, and stop_price.
444        note that a Stop Market Order can only be placed during market-hours (09:30:00 - 16:00:00), therefore if a
445        Stop Market Order is placed outside market hours it will raise an error.
446
447        :param order_direction: str: 'buy', 'sell', 'short', 'cover'
448        :param symbol: str: e.g: 'aapl', 'amd', 'NVDA', 'GM'
449        :param stop_price: float
450        :param share_amount: int
451        :param time_in_force: str, default: 'DAY', must be one of the following: 'DAY', 'GTC', or 'GTX'
452        :param log_info: bool, if True it will print information about the order
453        :return: True if operation succeeded
454        :raises Exception: if time not during market hours (9:30 - 16:00)
455        :raises AttributeError: if time_in_force argument not one of the following: 'DAY', 'GTC', 'GTX'
456        """
457        symbol = symbol.lower()
458        order_direction = order_direction.value
459        time_in_force = time_in_force.value
460
461        if not self.time_between((9, 30), (16, 0)):
462            raise Exception(f'Error: Stop Market orders are not allowed at this time ({self.time})')
463
464        if time_in_force not in ['DAY', 'GTC', 'GTX']:
465            raise AttributeError(f"Error: time_in_force argument must be one of the following: 'DAY', 'GTC', 'GTX'")
466
467        self.load_symbol(symbol)
468
469        order_menu = Select(self.driver.find_element(By.ID, "trading-order-select-type"))
470        order_menu.select_by_index(2)
471
472        tif_menu = Select(self.driver.find_element(By.ID, "trading-order-select-time"))
473        tif_menu.select_by_visible_text(time_in_force)
474
475        input_quantity = self.driver.find_element(By.ID, "trading-order-input-quantity")
476        input_quantity.clear()
477        input_quantity.send_keys(share_amount)
478
479        price_input = self.driver.find_element(By.ID, "trading-order-input-sprice")
480        price_input.clear()
481        price_input.send_keys(stop_price)
482
483        self.driver.find_element(By.ID, f"trading-order-button-{order_direction}").click()
484
485        if log_info is True:
486            print(f"Time: {self.time}, Order direction: {order_direction}, Symbol: {symbol}, "
487                  f"Stop Price: {stop_price}, Shares amount: {share_amount}")