Module pytrading212.trading212

API for Trading212 Platform

Expand source code
"""API for Trading212 Platform"""

import json
import logging
import re
import time
from datetime import datetime
from time import strftime
from urllib.parse import urlencode

import requests
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.ui import WebDriverWait

from pytrading212 import constants, console
from pytrading212.instrument import CFDInstrument
from pytrading212.order import CFDOrder, EquityOrder
from pytrading212.position import Position


class Trading212:
    def __init__(
            self,
            email: str,
            password: str,
            driver: webdriver,
            mode: constants.Mode = constants.Mode.DEMO,
            trading: constants.Trading = constants.Trading.EQUITY,
    ):
        self.session = f"TRADING212_SESSION_{mode.name}"
        self.base_url = f"https://{mode.name.lower()}.trading212.com"

        console.log(f"Starting PyTrading212 in [green]{mode.name}[/green] Mode")

        self.driver = driver
        self.driver.get(constants.URL_LOGIN)

        # Click Accept all cookies if it appears
        try:
            self.driver.find_element(By.CLASS_NAME, constants.CLASS_COOKIES_NOTICE_BUTTON).click()
        except NoSuchElementException:
            pass  # ignore

        console.log("Authenticating")

        # Authenticate
        self.driver.find_element(By.NAME, "email").send_keys(email)
        self.driver.find_element(By.NAME, "password").send_keys(password)

        # Click login button
        self.driver.find_element(By.CLASS_NAME, constants.CLASS_LOGIN_BUTTON).click()

        # Wait until the site is fully loaded, 120 seconds is a lot, but the site sometimes is very slow
        WebDriverWait(self.driver, 120).until(expected_conditions.
                                              visibility_of_element_located((By.CLASS_NAME, "company-logo")))

        self.user_agent = self.driver.execute_script("return navigator.userAgent;")

        # Redirect to correct mode, DEMO or LIVE
        if mode.name not in self.driver.current_url:
            self.driver.get(self.base_url)
            WebDriverWait(self.driver, 120).until(expected_conditions.
                                                  visibility_of_element_located((By.CLASS_NAME, "company-logo")))

        # Switch to right trading session: CFD or EQUITY
        self.switch_to(trading=trading)

        # Get session cookie
        cookies = self.driver.get_cookies()
        if cookies is not None:
            for cookie in cookies:
                # Get appropriate cookie for this session, live or demo
                if self.session in cookie['name']:
                    self.cookie = f"{self.session}={cookie['value']};"
        else:
            raise Exception("Unable to get cookies, aborting.")

        # Necessary headers for requests
        self.headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "User-Agent": self.user_agent,
            "Cookie": self.cookie,
        }

        self.companies = self.get_companies()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.finish()

    def finish(self):
        console.log("Closing session.")
        self.driver.close()

    def switch_to(self, trading: constants.Trading):
        self.driver.find_element(By.CLASS_NAME, "account-menu-info").click()
        WebDriverWait(self.driver, 10).until(expected_conditions.
                                             visibility_of_element_located((By.CLASS_NAME, "account-types")))
        element_account_types = self.driver.find_element(By.CLASS_NAME, "account-types")
        if trading == constants.Trading.CFD:
            element_account_types.find_element(By.CLASS_NAME, "cfd").click()
            WebDriverWait(self.driver, 60).until(expected_conditions.
                                                 visibility_of_element_located((By.CLASS_NAME, "cfd-icon")))
        elif trading == constants.Trading.EQUITY:
            element_account_types.find_element(By.CLASS_NAME, "equity").click()
            WebDriverWait(self.driver, 60).until(expected_conditions.
                                                 visibility_of_element_located((By.CLASS_NAME, "equity-icon")))

    def get_funds(self):
        """Get your funds, free, available."""
        response = requests.get(
            f"{self.base_url}/rest/v2/customer/accounts/funds", headers=self.headers
        )
        return json.loads(response.content.decode("utf-8"))

    def last_hour_hotlist(self):
        """Trading 212 last hour hotlist"""
        response = requests.get(
            f"{self.base_url}/trading212.com/rest/positions-tracker/deltas/hourly/1"
        )
        return json.loads(response.content.decode("utf-8"))

    def get_orders(self, older_than: datetime, newer_than: datetime):
        """Get orders within a range of dates"""
        params = {
            'olderThan': strftime(older_than.isoformat()),
            'newerThan': strftime(newer_than.isoformat())
        }

        response = requests.get(
            f"{self.base_url}/rest/history/orders", headers=self.headers, params=urlencode(params)
        )
        return json.loads(response.content.decode("utf-8"))

    def get_transactions(self, older_than: datetime, newer_than: datetime):
        """Get transactions within a range of dates"""
        params = {
            'olderThan': strftime(older_than.isoformat()),
            'newerThan': strftime(newer_than.isoformat())
        }
        response = requests.get(
            f"{self.base_url}/rest/history/transactions", headers=self.headers, params=urlencode(params)
        )
        return json.loads(response.content.decode("utf-8"))

    def get_order_details(self, details_path):
        """Get Order Details"""
        response = requests.get(f"{self.base_url}/rest/history{details_path}", headers=self.headers)
        return json.loads(response.content.decode("utf-8"))

    def get_dividends(self, older_than: datetime, newer_than: datetime):
        """Get dividends within a range of dates"""
        params = {'olderThan': strftime(older_than.isoformat()),
                  'newerThan': strftime(newer_than.isoformat())
                  }

        response = requests.get(
            f"{self.base_url}/rest/history/dividends", headers=self.headers, params=urlencode(params)
        )
        return json.loads(response.content.decode("utf-8"))

    def get_fundamentals(self, ticker, language_code: str = "en"):
        """Get fundamentals of a company by its isin"""
        params = {'ticker': ticker,
                  'languageCode': language_code
                  }

        response = requests.get(f"{self.base_url}/rest/companies/v2/fundamentals",
                                params=params)
        return json.loads(response.content.decode("utf-8"))

    def get_portfolio_performance(self, time_period: constants.Period):
        """Get Portfolio Performance"""
        response = requests.get(
            url=f"{self.base_url}/rest/v2/portfolio?period={time_period}",
            headers=self.headers,
        )
        return json.loads(response.content.decode("utf-8"))

    def get_portfolio_composition(self):
        """Get Portfolio Composition"""
        # click portfolio section on right-sidepanel
        right_sidepanel_portfolio_class = 'portfolio-icon'
        condition = expected_conditions.visibility_of_element_located(
            (By.CLASS_NAME, right_sidepanel_portfolio_class)
        )
        WebDriverWait(self.driver, 30).until(condition)
        self.driver.find_element(By.CLASS_NAME, right_sidepanel_portfolio_class).click()

        positions = []
        try:
            # click on investments
            self.driver.find_element(By.CLASS_NAME, 'investment-tab').click()
            for item in self.driver.find_elements(By.CLASS_NAME, "investment-item"):
                ticker = item.get_attribute("data-qa-item")
                value = item.find_element(By.CLASS_NAME, "total-value").text
                quantity = item.find_element(By.CLASS_NAME, "quantity").text
                total_return = item.find_element(By.CLASS_NAME, "return").text
                position = Position(ticker, value, quantity, total_return)
                positions.append(position.__dict__)
        except Exception as e:
            logging.error(e)  # portfolio is empty
        return positions

    def get_companies(self):
        """Get Ticker of all Trading212 tradable companies"""
        response = requests.get(
            url=f"{self.base_url}/rest/companies",
            headers=self.headers
        )
        return json.loads(response.content.decode("utf-8"))

    def search(self, query):
        """Search a company"""
        found = []
        companies = self.get_companies()
        for company in companies:
            if query in company["ticker"]:
                found.append(company["ticker"])

        response = requests.post(
            f"{self.base_url}/charting/prices?withFakes=false",
            headers=self.headers,
            data=found.__str__().replace("'", '"'),
        )  # ' with " for Trading212 compatibility
        return json.loads(response.content.decode("utf-8"))


class Equity(Trading212):
    """Trading 212 Equity"""

    def __init__(self, email: str, password: str, driver: webdriver, mode: constants.Mode = constants.Mode.DEMO):
        super().__init__(email, password, driver, mode, constants.Trading.EQUITY)

    def review_order(self, order: EquityOrder):
        """Preview of the order, with added costs and other useful data"""
        # Check if it is a 'value' order or a 'quantity' order
        if hasattr(order, 'value'):
            url = f"{self.base_url}/rest/v1/equity/value-order/review"
        else:
            url = f"{self.base_url}/rest/public/added-costs"
        response = requests.post(
            url=url,
            headers=self.headers,
            data=order.to_json(),
        )
        return json.loads(response.content.decode("utf-8"))

    def execute_order(self, order: EquityOrder):
        """Execute equity order"""
        # Check if it is a 'value' order or a 'quantity' order
        if hasattr(order, 'value'):
            url = f"{self.base_url}/rest/v1/equity/value-order"
        else:
            url = f"{self.base_url}/rest/public/v2/equity/order"
        response = requests.post(
            url=url,
            headers=self.headers,
            data=order.to_json(),
        )
        return json.loads(response.content.decode("utf-8"))

    def cancel_order(self, order_id):
        """Cancel a pending order"""
        response = requests.delete(
            url=f"{self.base_url}/rest/public/v2/equity/order/{order_id}",
            headers=self.headers,
        )
        return json.loads(response.content.decode("utf-8"))

    def check_order(self, equity_order: EquityOrder) -> [bool, str]:
        """Check if Order is valid."""
        is_valid_ticker = False, f"Instrument Code {equity_order.instrument_code} is not a valid Trading212 Ticker"
        for company in self.get_companies():
            if company['ticker'] == equity_order.instrument_code:
                is_valid_ticker = True, f"Instrument Code {equity_order.instrument_code} is valid Trading212 Ticker"

        return is_valid_ticker

    def min_max_sell_buy(self, instrument_code: str):
        params = {'instrumentCode': instrument_code}
        response = requests.get(
            f"{self.base_url}/rest/v1/equity/value-order/min-max",
            headers=self.headers,
            params=params
        )
        return json.loads(response.content.decode("utf-8"))


class CFD(Trading212):
    """ Trading 212 CFD """

    def __init__(self, email: str, password: str, driver: webdriver, mode: constants.Mode = constants.Mode.DEMO):
        super().__init__(email, password, driver, mode, constants.Trading.CFD)

    def execute_order(self, order: CFDOrder):
        """Execute CFD order"""

        # Check if it is Limit Stop Order
        if hasattr(order, 'is_limit_stop') and order.is_limit_stop == True:
            url = f"{self.base_url}/rest/v2/pending-orders/entry-dep-limit-stop/{order.instrument_code}"
        # Check if it is OCO Order
        elif hasattr(order, 'is_oco') and order.is_oco == True:
            url = f"{self.base_url}/rest/v2/pending-orders/entry-oco/{order.instrument_code}"
        # Market Order
        else:
            url = f"{self.base_url}/rest/v2/trading/open-positions"

        response = requests.post(
            url=url,
            headers=self.headers,
            data=order.to_json(),
        )
        return json.loads(response.content.decode("utf-8"))

    def trading_additional_info(self, order: CFDOrder):
        """Get additional trading info before executing order."""
        params = {'instrumentCode': order.instrument_code,
                  'quantity': order.quantity,
                  'positionId': 'null'}
        response = requests.get(
            f"{self.base_url}/rest/v1/tradingAdditionalInfo",
            headers=self.headers,
            params=params
        )
        return json.loads(response.content.decode("utf-8"))

    def close_position(self, position_id):
        """Close an open position."""
        response = requests.delete(
            url=f"{self.base_url}/rest/v2/trading/open-positions/close/{position_id}",
            headers=self.headers,
        )
        return json.loads(response.content.decode("utf-8"))

    def get_current_price(self, instrument_code):
        """Workaround to get the current price of a CFD."""
        # Simulate an order with target price 0, T212 will respond with a business exception so we can get the
        # current price
        cfd_order = CFDOrder(instrument_code=instrument_code,
                             target_price=0.0,
                             quantity=0.1)
        # Return only the current price
        return float(self.execute_order(cfd_order)['context']['current'])

Classes

class CFD (email: str, password: str, driver: , mode: Mode = Mode.DEMO)

Trading 212 CFD

Expand source code
class CFD(Trading212):
    """ Trading 212 CFD """

    def __init__(self, email: str, password: str, driver: webdriver, mode: constants.Mode = constants.Mode.DEMO):
        super().__init__(email, password, driver, mode, constants.Trading.CFD)

    def execute_order(self, order: CFDOrder):
        """Execute CFD order"""

        # Check if it is Limit Stop Order
        if hasattr(order, 'is_limit_stop') and order.is_limit_stop == True:
            url = f"{self.base_url}/rest/v2/pending-orders/entry-dep-limit-stop/{order.instrument_code}"
        # Check if it is OCO Order
        elif hasattr(order, 'is_oco') and order.is_oco == True:
            url = f"{self.base_url}/rest/v2/pending-orders/entry-oco/{order.instrument_code}"
        # Market Order
        else:
            url = f"{self.base_url}/rest/v2/trading/open-positions"

        response = requests.post(
            url=url,
            headers=self.headers,
            data=order.to_json(),
        )
        return json.loads(response.content.decode("utf-8"))

    def trading_additional_info(self, order: CFDOrder):
        """Get additional trading info before executing order."""
        params = {'instrumentCode': order.instrument_code,
                  'quantity': order.quantity,
                  'positionId': 'null'}
        response = requests.get(
            f"{self.base_url}/rest/v1/tradingAdditionalInfo",
            headers=self.headers,
            params=params
        )
        return json.loads(response.content.decode("utf-8"))

    def close_position(self, position_id):
        """Close an open position."""
        response = requests.delete(
            url=f"{self.base_url}/rest/v2/trading/open-positions/close/{position_id}",
            headers=self.headers,
        )
        return json.loads(response.content.decode("utf-8"))

    def get_current_price(self, instrument_code):
        """Workaround to get the current price of a CFD."""
        # Simulate an order with target price 0, T212 will respond with a business exception so we can get the
        # current price
        cfd_order = CFDOrder(instrument_code=instrument_code,
                             target_price=0.0,
                             quantity=0.1)
        # Return only the current price
        return float(self.execute_order(cfd_order)['context']['current'])

Ancestors

Methods

def close_position(self, position_id)

Close an open position.

Expand source code
def close_position(self, position_id):
    """Close an open position."""
    response = requests.delete(
        url=f"{self.base_url}/rest/v2/trading/open-positions/close/{position_id}",
        headers=self.headers,
    )
    return json.loads(response.content.decode("utf-8"))
def execute_order(self, order: CFDOrder)

Execute CFD order

Expand source code
def execute_order(self, order: CFDOrder):
    """Execute CFD order"""

    # Check if it is Limit Stop Order
    if hasattr(order, 'is_limit_stop') and order.is_limit_stop == True:
        url = f"{self.base_url}/rest/v2/pending-orders/entry-dep-limit-stop/{order.instrument_code}"
    # Check if it is OCO Order
    elif hasattr(order, 'is_oco') and order.is_oco == True:
        url = f"{self.base_url}/rest/v2/pending-orders/entry-oco/{order.instrument_code}"
    # Market Order
    else:
        url = f"{self.base_url}/rest/v2/trading/open-positions"

    response = requests.post(
        url=url,
        headers=self.headers,
        data=order.to_json(),
    )
    return json.loads(response.content.decode("utf-8"))
def get_current_price(self, instrument_code)

Workaround to get the current price of a CFD.

Expand source code
def get_current_price(self, instrument_code):
    """Workaround to get the current price of a CFD."""
    # Simulate an order with target price 0, T212 will respond with a business exception so we can get the
    # current price
    cfd_order = CFDOrder(instrument_code=instrument_code,
                         target_price=0.0,
                         quantity=0.1)
    # Return only the current price
    return float(self.execute_order(cfd_order)['context']['current'])
def trading_additional_info(self, order: CFDOrder)

Get additional trading info before executing order.

Expand source code
def trading_additional_info(self, order: CFDOrder):
    """Get additional trading info before executing order."""
    params = {'instrumentCode': order.instrument_code,
              'quantity': order.quantity,
              'positionId': 'null'}
    response = requests.get(
        f"{self.base_url}/rest/v1/tradingAdditionalInfo",
        headers=self.headers,
        params=params
    )
    return json.loads(response.content.decode("utf-8"))

Inherited members

class Equity (email: str, password: str, driver: , mode: Mode = Mode.DEMO)

Trading 212 Equity

Expand source code
class Equity(Trading212):
    """Trading 212 Equity"""

    def __init__(self, email: str, password: str, driver: webdriver, mode: constants.Mode = constants.Mode.DEMO):
        super().__init__(email, password, driver, mode, constants.Trading.EQUITY)

    def review_order(self, order: EquityOrder):
        """Preview of the order, with added costs and other useful data"""
        # Check if it is a 'value' order or a 'quantity' order
        if hasattr(order, 'value'):
            url = f"{self.base_url}/rest/v1/equity/value-order/review"
        else:
            url = f"{self.base_url}/rest/public/added-costs"
        response = requests.post(
            url=url,
            headers=self.headers,
            data=order.to_json(),
        )
        return json.loads(response.content.decode("utf-8"))

    def execute_order(self, order: EquityOrder):
        """Execute equity order"""
        # Check if it is a 'value' order or a 'quantity' order
        if hasattr(order, 'value'):
            url = f"{self.base_url}/rest/v1/equity/value-order"
        else:
            url = f"{self.base_url}/rest/public/v2/equity/order"
        response = requests.post(
            url=url,
            headers=self.headers,
            data=order.to_json(),
        )
        return json.loads(response.content.decode("utf-8"))

    def cancel_order(self, order_id):
        """Cancel a pending order"""
        response = requests.delete(
            url=f"{self.base_url}/rest/public/v2/equity/order/{order_id}",
            headers=self.headers,
        )
        return json.loads(response.content.decode("utf-8"))

    def check_order(self, equity_order: EquityOrder) -> [bool, str]:
        """Check if Order is valid."""
        is_valid_ticker = False, f"Instrument Code {equity_order.instrument_code} is not a valid Trading212 Ticker"
        for company in self.get_companies():
            if company['ticker'] == equity_order.instrument_code:
                is_valid_ticker = True, f"Instrument Code {equity_order.instrument_code} is valid Trading212 Ticker"

        return is_valid_ticker

    def min_max_sell_buy(self, instrument_code: str):
        params = {'instrumentCode': instrument_code}
        response = requests.get(
            f"{self.base_url}/rest/v1/equity/value-order/min-max",
            headers=self.headers,
            params=params
        )
        return json.loads(response.content.decode("utf-8"))

Ancestors

Methods

def cancel_order(self, order_id)

Cancel a pending order

Expand source code
def cancel_order(self, order_id):
    """Cancel a pending order"""
    response = requests.delete(
        url=f"{self.base_url}/rest/public/v2/equity/order/{order_id}",
        headers=self.headers,
    )
    return json.loads(response.content.decode("utf-8"))
def check_order(self, equity_order: EquityOrder) ‑> []

Check if Order is valid.

Expand source code
def check_order(self, equity_order: EquityOrder) -> [bool, str]:
    """Check if Order is valid."""
    is_valid_ticker = False, f"Instrument Code {equity_order.instrument_code} is not a valid Trading212 Ticker"
    for company in self.get_companies():
        if company['ticker'] == equity_order.instrument_code:
            is_valid_ticker = True, f"Instrument Code {equity_order.instrument_code} is valid Trading212 Ticker"

    return is_valid_ticker
def execute_order(self, order: EquityOrder)

Execute equity order

Expand source code
def execute_order(self, order: EquityOrder):
    """Execute equity order"""
    # Check if it is a 'value' order or a 'quantity' order
    if hasattr(order, 'value'):
        url = f"{self.base_url}/rest/v1/equity/value-order"
    else:
        url = f"{self.base_url}/rest/public/v2/equity/order"
    response = requests.post(
        url=url,
        headers=self.headers,
        data=order.to_json(),
    )
    return json.loads(response.content.decode("utf-8"))
def min_max_sell_buy(self, instrument_code: str)
Expand source code
def min_max_sell_buy(self, instrument_code: str):
    params = {'instrumentCode': instrument_code}
    response = requests.get(
        f"{self.base_url}/rest/v1/equity/value-order/min-max",
        headers=self.headers,
        params=params
    )
    return json.loads(response.content.decode("utf-8"))
def review_order(self, order: EquityOrder)

Preview of the order, with added costs and other useful data

Expand source code
def review_order(self, order: EquityOrder):
    """Preview of the order, with added costs and other useful data"""
    # Check if it is a 'value' order or a 'quantity' order
    if hasattr(order, 'value'):
        url = f"{self.base_url}/rest/v1/equity/value-order/review"
    else:
        url = f"{self.base_url}/rest/public/added-costs"
    response = requests.post(
        url=url,
        headers=self.headers,
        data=order.to_json(),
    )
    return json.loads(response.content.decode("utf-8"))

Inherited members

class Trading212 (email: str, password: str, driver: , mode: Mode = Mode.DEMO, trading: Trading = Trading.EQUITY)
Expand source code
class Trading212:
    def __init__(
            self,
            email: str,
            password: str,
            driver: webdriver,
            mode: constants.Mode = constants.Mode.DEMO,
            trading: constants.Trading = constants.Trading.EQUITY,
    ):
        self.session = f"TRADING212_SESSION_{mode.name}"
        self.base_url = f"https://{mode.name.lower()}.trading212.com"

        console.log(f"Starting PyTrading212 in [green]{mode.name}[/green] Mode")

        self.driver = driver
        self.driver.get(constants.URL_LOGIN)

        # Click Accept all cookies if it appears
        try:
            self.driver.find_element(By.CLASS_NAME, constants.CLASS_COOKIES_NOTICE_BUTTON).click()
        except NoSuchElementException:
            pass  # ignore

        console.log("Authenticating")

        # Authenticate
        self.driver.find_element(By.NAME, "email").send_keys(email)
        self.driver.find_element(By.NAME, "password").send_keys(password)

        # Click login button
        self.driver.find_element(By.CLASS_NAME, constants.CLASS_LOGIN_BUTTON).click()

        # Wait until the site is fully loaded, 120 seconds is a lot, but the site sometimes is very slow
        WebDriverWait(self.driver, 120).until(expected_conditions.
                                              visibility_of_element_located((By.CLASS_NAME, "company-logo")))

        self.user_agent = self.driver.execute_script("return navigator.userAgent;")

        # Redirect to correct mode, DEMO or LIVE
        if mode.name not in self.driver.current_url:
            self.driver.get(self.base_url)
            WebDriverWait(self.driver, 120).until(expected_conditions.
                                                  visibility_of_element_located((By.CLASS_NAME, "company-logo")))

        # Switch to right trading session: CFD or EQUITY
        self.switch_to(trading=trading)

        # Get session cookie
        cookies = self.driver.get_cookies()
        if cookies is not None:
            for cookie in cookies:
                # Get appropriate cookie for this session, live or demo
                if self.session in cookie['name']:
                    self.cookie = f"{self.session}={cookie['value']};"
        else:
            raise Exception("Unable to get cookies, aborting.")

        # Necessary headers for requests
        self.headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "User-Agent": self.user_agent,
            "Cookie": self.cookie,
        }

        self.companies = self.get_companies()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.finish()

    def finish(self):
        console.log("Closing session.")
        self.driver.close()

    def switch_to(self, trading: constants.Trading):
        self.driver.find_element(By.CLASS_NAME, "account-menu-info").click()
        WebDriverWait(self.driver, 10).until(expected_conditions.
                                             visibility_of_element_located((By.CLASS_NAME, "account-types")))
        element_account_types = self.driver.find_element(By.CLASS_NAME, "account-types")
        if trading == constants.Trading.CFD:
            element_account_types.find_element(By.CLASS_NAME, "cfd").click()
            WebDriverWait(self.driver, 60).until(expected_conditions.
                                                 visibility_of_element_located((By.CLASS_NAME, "cfd-icon")))
        elif trading == constants.Trading.EQUITY:
            element_account_types.find_element(By.CLASS_NAME, "equity").click()
            WebDriverWait(self.driver, 60).until(expected_conditions.
                                                 visibility_of_element_located((By.CLASS_NAME, "equity-icon")))

    def get_funds(self):
        """Get your funds, free, available."""
        response = requests.get(
            f"{self.base_url}/rest/v2/customer/accounts/funds", headers=self.headers
        )
        return json.loads(response.content.decode("utf-8"))

    def last_hour_hotlist(self):
        """Trading 212 last hour hotlist"""
        response = requests.get(
            f"{self.base_url}/trading212.com/rest/positions-tracker/deltas/hourly/1"
        )
        return json.loads(response.content.decode("utf-8"))

    def get_orders(self, older_than: datetime, newer_than: datetime):
        """Get orders within a range of dates"""
        params = {
            'olderThan': strftime(older_than.isoformat()),
            'newerThan': strftime(newer_than.isoformat())
        }

        response = requests.get(
            f"{self.base_url}/rest/history/orders", headers=self.headers, params=urlencode(params)
        )
        return json.loads(response.content.decode("utf-8"))

    def get_transactions(self, older_than: datetime, newer_than: datetime):
        """Get transactions within a range of dates"""
        params = {
            'olderThan': strftime(older_than.isoformat()),
            'newerThan': strftime(newer_than.isoformat())
        }
        response = requests.get(
            f"{self.base_url}/rest/history/transactions", headers=self.headers, params=urlencode(params)
        )
        return json.loads(response.content.decode("utf-8"))

    def get_order_details(self, details_path):
        """Get Order Details"""
        response = requests.get(f"{self.base_url}/rest/history{details_path}", headers=self.headers)
        return json.loads(response.content.decode("utf-8"))

    def get_dividends(self, older_than: datetime, newer_than: datetime):
        """Get dividends within a range of dates"""
        params = {'olderThan': strftime(older_than.isoformat()),
                  'newerThan': strftime(newer_than.isoformat())
                  }

        response = requests.get(
            f"{self.base_url}/rest/history/dividends", headers=self.headers, params=urlencode(params)
        )
        return json.loads(response.content.decode("utf-8"))

    def get_fundamentals(self, ticker, language_code: str = "en"):
        """Get fundamentals of a company by its isin"""
        params = {'ticker': ticker,
                  'languageCode': language_code
                  }

        response = requests.get(f"{self.base_url}/rest/companies/v2/fundamentals",
                                params=params)
        return json.loads(response.content.decode("utf-8"))

    def get_portfolio_performance(self, time_period: constants.Period):
        """Get Portfolio Performance"""
        response = requests.get(
            url=f"{self.base_url}/rest/v2/portfolio?period={time_period}",
            headers=self.headers,
        )
        return json.loads(response.content.decode("utf-8"))

    def get_portfolio_composition(self):
        """Get Portfolio Composition"""
        # click portfolio section on right-sidepanel
        right_sidepanel_portfolio_class = 'portfolio-icon'
        condition = expected_conditions.visibility_of_element_located(
            (By.CLASS_NAME, right_sidepanel_portfolio_class)
        )
        WebDriverWait(self.driver, 30).until(condition)
        self.driver.find_element(By.CLASS_NAME, right_sidepanel_portfolio_class).click()

        positions = []
        try:
            # click on investments
            self.driver.find_element(By.CLASS_NAME, 'investment-tab').click()
            for item in self.driver.find_elements(By.CLASS_NAME, "investment-item"):
                ticker = item.get_attribute("data-qa-item")
                value = item.find_element(By.CLASS_NAME, "total-value").text
                quantity = item.find_element(By.CLASS_NAME, "quantity").text
                total_return = item.find_element(By.CLASS_NAME, "return").text
                position = Position(ticker, value, quantity, total_return)
                positions.append(position.__dict__)
        except Exception as e:
            logging.error(e)  # portfolio is empty
        return positions

    def get_companies(self):
        """Get Ticker of all Trading212 tradable companies"""
        response = requests.get(
            url=f"{self.base_url}/rest/companies",
            headers=self.headers
        )
        return json.loads(response.content.decode("utf-8"))

    def search(self, query):
        """Search a company"""
        found = []
        companies = self.get_companies()
        for company in companies:
            if query in company["ticker"]:
                found.append(company["ticker"])

        response = requests.post(
            f"{self.base_url}/charting/prices?withFakes=false",
            headers=self.headers,
            data=found.__str__().replace("'", '"'),
        )  # ' with " for Trading212 compatibility
        return json.loads(response.content.decode("utf-8"))

Subclasses

Methods

def finish(self)
Expand source code
def finish(self):
    console.log("Closing session.")
    self.driver.close()
def get_companies(self)

Get Ticker of all Trading212 tradable companies

Expand source code
def get_companies(self):
    """Get Ticker of all Trading212 tradable companies"""
    response = requests.get(
        url=f"{self.base_url}/rest/companies",
        headers=self.headers
    )
    return json.loads(response.content.decode("utf-8"))
def get_dividends(self, older_than: datetime.datetime, newer_than: datetime.datetime)

Get dividends within a range of dates

Expand source code
def get_dividends(self, older_than: datetime, newer_than: datetime):
    """Get dividends within a range of dates"""
    params = {'olderThan': strftime(older_than.isoformat()),
              'newerThan': strftime(newer_than.isoformat())
              }

    response = requests.get(
        f"{self.base_url}/rest/history/dividends", headers=self.headers, params=urlencode(params)
    )
    return json.loads(response.content.decode("utf-8"))
def get_fundamentals(self, ticker, language_code: str = 'en')

Get fundamentals of a company by its isin

Expand source code
def get_fundamentals(self, ticker, language_code: str = "en"):
    """Get fundamentals of a company by its isin"""
    params = {'ticker': ticker,
              'languageCode': language_code
              }

    response = requests.get(f"{self.base_url}/rest/companies/v2/fundamentals",
                            params=params)
    return json.loads(response.content.decode("utf-8"))
def get_funds(self)

Get your funds, free, available.

Expand source code
def get_funds(self):
    """Get your funds, free, available."""
    response = requests.get(
        f"{self.base_url}/rest/v2/customer/accounts/funds", headers=self.headers
    )
    return json.loads(response.content.decode("utf-8"))
def get_order_details(self, details_path)

Get Order Details

Expand source code
def get_order_details(self, details_path):
    """Get Order Details"""
    response = requests.get(f"{self.base_url}/rest/history{details_path}", headers=self.headers)
    return json.loads(response.content.decode("utf-8"))
def get_orders(self, older_than: datetime.datetime, newer_than: datetime.datetime)

Get orders within a range of dates

Expand source code
def get_orders(self, older_than: datetime, newer_than: datetime):
    """Get orders within a range of dates"""
    params = {
        'olderThan': strftime(older_than.isoformat()),
        'newerThan': strftime(newer_than.isoformat())
    }

    response = requests.get(
        f"{self.base_url}/rest/history/orders", headers=self.headers, params=urlencode(params)
    )
    return json.loads(response.content.decode("utf-8"))
def get_portfolio_composition(self)

Get Portfolio Composition

Expand source code
def get_portfolio_composition(self):
    """Get Portfolio Composition"""
    # click portfolio section on right-sidepanel
    right_sidepanel_portfolio_class = 'portfolio-icon'
    condition = expected_conditions.visibility_of_element_located(
        (By.CLASS_NAME, right_sidepanel_portfolio_class)
    )
    WebDriverWait(self.driver, 30).until(condition)
    self.driver.find_element(By.CLASS_NAME, right_sidepanel_portfolio_class).click()

    positions = []
    try:
        # click on investments
        self.driver.find_element(By.CLASS_NAME, 'investment-tab').click()
        for item in self.driver.find_elements(By.CLASS_NAME, "investment-item"):
            ticker = item.get_attribute("data-qa-item")
            value = item.find_element(By.CLASS_NAME, "total-value").text
            quantity = item.find_element(By.CLASS_NAME, "quantity").text
            total_return = item.find_element(By.CLASS_NAME, "return").text
            position = Position(ticker, value, quantity, total_return)
            positions.append(position.__dict__)
    except Exception as e:
        logging.error(e)  # portfolio is empty
    return positions
def get_portfolio_performance(self, time_period: Period)

Get Portfolio Performance

Expand source code
def get_portfolio_performance(self, time_period: constants.Period):
    """Get Portfolio Performance"""
    response = requests.get(
        url=f"{self.base_url}/rest/v2/portfolio?period={time_period}",
        headers=self.headers,
    )
    return json.loads(response.content.decode("utf-8"))
def get_transactions(self, older_than: datetime.datetime, newer_than: datetime.datetime)

Get transactions within a range of dates

Expand source code
def get_transactions(self, older_than: datetime, newer_than: datetime):
    """Get transactions within a range of dates"""
    params = {
        'olderThan': strftime(older_than.isoformat()),
        'newerThan': strftime(newer_than.isoformat())
    }
    response = requests.get(
        f"{self.base_url}/rest/history/transactions", headers=self.headers, params=urlencode(params)
    )
    return json.loads(response.content.decode("utf-8"))
def last_hour_hotlist(self)

Trading 212 last hour hotlist

Expand source code
def last_hour_hotlist(self):
    """Trading 212 last hour hotlist"""
    response = requests.get(
        f"{self.base_url}/trading212.com/rest/positions-tracker/deltas/hourly/1"
    )
    return json.loads(response.content.decode("utf-8"))
def search(self, query)

Search a company

Expand source code
def search(self, query):
    """Search a company"""
    found = []
    companies = self.get_companies()
    for company in companies:
        if query in company["ticker"]:
            found.append(company["ticker"])

    response = requests.post(
        f"{self.base_url}/charting/prices?withFakes=false",
        headers=self.headers,
        data=found.__str__().replace("'", '"'),
    )  # ' with " for Trading212 compatibility
    return json.loads(response.content.decode("utf-8"))
def switch_to(self, trading: Trading)
Expand source code
def switch_to(self, trading: constants.Trading):
    self.driver.find_element(By.CLASS_NAME, "account-menu-info").click()
    WebDriverWait(self.driver, 10).until(expected_conditions.
                                         visibility_of_element_located((By.CLASS_NAME, "account-types")))
    element_account_types = self.driver.find_element(By.CLASS_NAME, "account-types")
    if trading == constants.Trading.CFD:
        element_account_types.find_element(By.CLASS_NAME, "cfd").click()
        WebDriverWait(self.driver, 60).until(expected_conditions.
                                             visibility_of_element_located((By.CLASS_NAME, "cfd-icon")))
    elif trading == constants.Trading.EQUITY:
        element_account_types.find_element(By.CLASS_NAME, "equity").click()
        WebDriverWait(self.driver, 60).until(expected_conditions.
                                             visibility_of_element_located((By.CLASS_NAME, "equity-icon")))