This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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}") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import zipfile | |
from datetime import datetime | |
from dagster import op, job | |
@op | |
def get_todays_date() -> str: | |
return datetime.today().strftime() |
OlderNewer