isaacharrisholt

# Creates copy slice of main DataFrame to pass to parser
transaction = transaction_df.iloc[i].copy()

# Checks if transaction IDs are being used
if use_transaction_id:
    transaction_id = transaction["Transaction IDs"]

    # Checks that current transaction has not already been processed to avoid incorrect output to final spreadsheet
    if transaction_id in transaction_history:
        pprint("\n{BLUE}Transaction processed previously, skipping...{RESET}")

isaacharrisholt

# Main function
def get_data(queue, sport, markets=None):
    if markets is None:
        markets = []

    # Initialise the webdriver
    driver = initialise_webdriver()

    # Open page and accept cookies
    driver.get(SITE_LINK)

isaacharrisholt

# Formula to find surebets in dataframes
def find_surebets(surebet_df, market):
    # Separate odds into separate columns and clean
    # x column
    surebet_df[[f'{market}_x_1',
                f'{market}_x_2']] = surebet_df[f'{market}_x'].apply(utils.replace_comma).str.split('\n', expand=True) \
                                                             .iloc[:, 0:2].apply(pd.Series)
    surebet_df[f'{market}_x_1'] = surebet_df[f'{market}_x_1'].apply(utils.convert_odds).astype(float)
    surebet_df[f'{market}_x_2'] = surebet_df[f'{market}_x_2'].apply(utils.convert_odds).astype(float)
    # y column

isaacharrisholt

# Unrounded calculations for two way bets
def two_way_unrounded_calculations(odds1, odds2, total_stake):
    x, y = symbols('x y')

    # Equation for total stake
    total_stake_eq = Eq(x + y - total_stake, 0)

    # Odds multiplied by their stake must be equal
    individual_stakes_eq = Eq((odds2 * y) - (odds1 * x), 0)

isaacharrisholt

from quiffen import Qif
qif = Qif.parse('test.qif')
print(qif.accounts)
# {'Quiffen Default Account': Account(name='Quiffen Default Account', desc='The default account created by Quiffen when no
# other accounts were present')}
acc = qif.accounts['Quiffen Default Account']
print(acc.transactions)
# {'Bank': TransactionList(Transaction(date=datetime.datetime(2021, 2, 14, 0 , 0), amount=150.0, ...), ...),
# 'Invst': TransactionList(...)}
tr = acc.transactions['Bank'][0]

isaacharrisholt

import quiffen
from datetime import datetime
qif = quiffen.Qif()
acc = quiffen.Account('Personal Bank Account', desc='My personal bank account with Barclays.')
qif.add_account(acc)
groceries = quiffen.Category('Groceries')
essentials = quiffen.Category('Essentials')
groceries.add_child(essentials)
print(groceries.render_tree())
# Groceries (root)

isaacharrisholt

from typing import List

import requests

API_ROOT = 'http://localhost:8000/api/v1'  # This is the default for Airbyte


def get_workspaces() -> List[str]:
    response = requests.post(f'{API_ROOT}/workspaces/list')
    response.raise_for_status()  # Either handle this yourself, or use a tool like Sentry for logging

isaacharrisholt

def get_connections_for_workspace(workspace_id: str) -> List[str]:
    response = requests.post(
        f'{API_ROOT}/connections/list',
        json={'workspaceId': workspace_id},
    )
    response.raise_for_status()
    return [
        connection['connectionId']
        for connection in response.json()['connections']
        if connection['status'] == 'active'  # So we can still disable connections in the UI

isaacharrisholt

from typing import List

import requests

API_ROOT = 'http://localhost:8000/api/v1'  # This is the default for Airbyte


def get_workspaces() -> List[str]:
    response = requests.post(f'{API_ROOT}/workspaces/list')
    response.raise_for_status()  # Either handle this yourself, or use a tool like Sentry for logging

isaacharrisholt

import zipfile

from datetime import datetime

from dagster import op, job


@op
def get_todays_date() -> str:
    return datetime.today().strftime()