-
Notifications
You must be signed in to change notification settings - Fork 180
Build out user dir #31
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from 16 commits
e0a6be0
1d1d20f
a0f146c
acb1bd9
eeba548
f115f41
b8027a1
ef4a981
bca643d
71fb806
a5a6542
e7af0ac
5393bca
ab8ae45
2c7edb2
e3710fb
73b605e
b69a08e
f2094be
9c6737f
463e9bf
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -16,6 +16,7 @@ | |
QMainWindow, QMessageBox, QTableWidgetItem) | ||
|
||
import algobot.assets | ||
from algobot import helpers | ||
from algobot.algodict import get_interface_dictionary | ||
from algobot.data import Data | ||
from algobot.enums import (AVG_GRAPH, BACKTEST, LIVE, LONG, NET_GRAPH, | ||
|
@@ -26,8 +27,8 @@ | |
setup_graph_plots, setup_graphs, | ||
update_backtest_graph_limits, | ||
update_main_graphs) | ||
from algobot.helpers import (ROOT_DIR, create_folder, create_folder_if_needed, | ||
get_caller_string, open_file_or_folder) | ||
from algobot.helpers import (PATHS, create_folder_if_needed, get_caller_string, | ||
open_file_or_folder) | ||
from algobot.interface.about import About | ||
from algobot.interface.config_utils.state_utils import load_state, save_state | ||
from algobot.interface.config_utils.strategy_utils import get_strategies | ||
|
@@ -47,7 +48,7 @@ | |
from algobot.traders.simulationtrader import SimulationTrader | ||
|
||
app = QApplication(sys.argv) | ||
mainUi = os.path.join(ROOT_DIR, 'UI', 'algobot.ui') | ||
mainUi = os.path.join(PATHS.get_ui_dir(), 'algobot.ui') | ||
|
||
|
||
class Interface(QMainWindow): | ||
|
@@ -252,21 +253,23 @@ def export_optimizer(self, file_type: str): | |
""" | ||
if self.optimizer: | ||
if len(self.optimizer.optimizerRows) > 0: | ||
optimizerFolderPath = create_folder('Optimizer Results') | ||
innerPath = os.path.join(optimizerFolderPath, self.optimizer.symbol) | ||
create_folder_if_needed(innerPath, optimizerFolderPath) | ||
defaultFileName = self.optimizer.get_default_result_file_name('optimizer', ext=file_type.lower()) | ||
defaultPath = os.path.join(innerPath, defaultFileName) | ||
filePath, _ = QFileDialog.getSaveFileName(self, 'Save Optimizer', defaultPath, | ||
f'{file_type} (*.{file_type.lower()})') | ||
if not filePath: | ||
optimizer_folder_path = helpers.PATHS.get_optimizer_results_dir() | ||
create_folder_if_needed(optimizer_folder_path) | ||
inner_path = os.path.join(optimizer_folder_path, self.optimizer.symbol) | ||
create_folder_if_needed(inner_path) | ||
default_file_name = self.optimizer.get_default_result_file_name(optimizer_folder_path, | ||
'optimizer', ext=file_type.lower()) | ||
default_path = os.path.join(inner_path, default_file_name) | ||
file_path, _ = QFileDialog.getSaveFileName(self, 'Save Optimizer', default_path, | ||
f'{file_type} (*.{file_type.lower()})') | ||
if not file_path: | ||
create_popup(self, "Export cancelled.") | ||
else: | ||
self.optimizer.export_optimizer_rows(filePath, file_type) | ||
create_popup(self, f'Exported successfully to {filePath}.') | ||
self.optimizer.export_optimizer_rows(file_path, file_type) | ||
create_popup(self, f'Exported successfully to {file_path}.') | ||
|
||
if open_from_msg_box(text='Do you want to open the optimization report?', title='Optimizer Report'): | ||
open_file_or_folder(filePath) | ||
open_file_or_folder(file_path) | ||
|
||
else: | ||
create_popup(self, "No table rows found.") | ||
|
@@ -407,18 +410,19 @@ def end_backtest(self): | |
""" | ||
Ends backtest and prompts user if they want to see the results. | ||
""" | ||
backtestFolderPath = create_folder('Backtest Results') | ||
innerPath = os.path.join(backtestFolderPath, self.backtester.symbol) | ||
create_folder_if_needed(innerPath, backtestFolderPath) | ||
defaultFile = os.path.join(innerPath, self.backtester.get_default_result_file_name()) | ||
fileName, _ = QFileDialog.getSaveFileName(self, 'Save Result', defaultFile, 'TXT (*.txt)') | ||
fileName = fileName.strip() | ||
fileName = fileName if fileName != '' else None | ||
backtest_folder_path = helpers.PATHS.get_backtest_results_dir() | ||
create_folder_if_needed(backtest_folder_path) | ||
inner_path = os.path.join(backtest_folder_path, self.backtester.symbol) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We should probably refactor this code to use makedirs() There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
create_folder_if_needed(inner_path) | ||
default_file = os.path.join(inner_path, self.backtester.get_default_result_file_name(backtest_folder_path)) | ||
file_name, _ = QFileDialog.getSaveFileName(self, 'Save Result', default_file, 'TXT (*.txt)') | ||
file_name = file_name.strip() | ||
file_name = file_name if file_name != '' else None | ||
|
||
if not fileName: | ||
if not file_name: | ||
self.add_to_backtest_monitor('Ended backtest.') | ||
else: | ||
path = self.backtester.write_results(resultFile=fileName) | ||
path = self.backtester.write_results(resultFile=file_name) | ||
self.add_to_backtest_monitor(f'Ended backtest and saved results to {path}.') | ||
|
||
if open_from_msg_box(text=f"Backtest results have been saved to {path}.", title="Backtest Results"): | ||
|
@@ -1293,12 +1297,13 @@ def export_trades(self, caller): | |
trade.append(item.text()) | ||
trades.append(trade) | ||
|
||
path = create_folder("Trade History") | ||
trade_history_dir = helpers.PATHS.get_trade_history_dir() | ||
create_folder_if_needed(trade_history_dir) | ||
|
||
if caller == LIVE: | ||
defaultFile = os.path.join(path, 'live_trades.csv') | ||
defaultFile = os.path.join(trade_history_dir, 'live_trades.csv') | ||
else: | ||
defaultFile = os.path.join(path, 'simulation_trades.csv') | ||
defaultFile = os.path.join(trade_history_dir, 'simulation_trades.csv') | ||
|
||
path, _ = QFileDialog.getSaveFileName(self, 'Export Trades', defaultFile, 'CSV (*.csv)') | ||
|
||
|
@@ -1317,8 +1322,9 @@ def import_trades(self, caller): | |
""" | ||
table = self.interfaceDictionary[caller]['mainInterface']['historyTable'] | ||
label = self.interfaceDictionary[caller]['mainInterface']['historyLabel'] | ||
path = create_folder("Trade History") | ||
path, _ = QFileDialog.getOpenFileName(self, 'Import Trades', path, "CSV (*.csv)") | ||
trade_history_dir = helpers.PATHS.get_trade_history_dir() | ||
create_folder_if_needed(trade_history_dir) | ||
path, _ = QFileDialog.getOpenFileName(self, 'Import Trades', trade_history_dir, "CSV (*.csv)") | ||
|
||
try: | ||
with open(path, 'r') as f: | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -9,7 +9,8 @@ | |
from binance.client import Client | ||
from binance.helpers import interval_to_milliseconds | ||
|
||
from algobot.helpers import (ROOT_DIR, get_logger, get_normalized_data, | ||
from algobot import helpers | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Here also, why not just have There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Updated all |
||
from algobot.helpers import (PATHS, get_logger, get_normalized_data, | ||
get_ups_and_downs) | ||
from algobot.typing_hints import DATA_TYPE | ||
|
||
|
@@ -142,12 +143,12 @@ def get_database_file(self) -> str: | |
Retrieves database file path. | ||
:return: Database file path. | ||
""" | ||
database_folder = os.path.join(ROOT_DIR, 'Databases') | ||
database_folder = PATHS.get_database_dir() | ||
if not os.path.exists(database_folder): | ||
os.mkdir(database_folder) | ||
os.makedirs(database_folder) | ||
|
||
filePath = os.path.join(database_folder, f'{self.symbol}.db') | ||
return filePath | ||
file_path = os.path.join(database_folder, f'{self.symbol}.db') | ||
return file_path | ||
|
||
def create_table(self): | ||
""" | ||
|
@@ -521,19 +522,20 @@ def get_interval_minutes(self) -> int: | |
else: | ||
raise ValueError("Invalid interval.", 4) | ||
|
||
def create_folders_and_change_path(self, folderName: str): | ||
def create_folders_and_change_path(self, folder_name: str): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Nice change! We should really convert camel case to snake case over time There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. snake case is the python way. If you're happy with something we could use black? https://black.readthedocs.io/en/stable/ We can configure it as a pre-commit hook even :) Although not sure how it handles variable naming come to think about it 🤔 There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. pylint will help enforce it ;) There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yeap. We should really start leveraging pylint and mypy |
||
""" | ||
Creates appropriate folders for data storage then changes current working directory to it. | ||
:param folderName: Folder to create. | ||
:param folder_name: Folder to create. | ||
""" | ||
os.chdir(ROOT_DIR) | ||
if not os.path.exists(folderName): # Create CSV folder if it doesn't exist | ||
os.mkdir(folderName) | ||
os.chdir(folderName) # Go inside the folder. | ||
if not os.path.exists(folder_name): | ||
helpers.create_folder_if_needed(folder_name) | ||
|
||
if not os.path.exists(self.symbol): # Create symbol folder inside CSV folder if it doesn't exist. | ||
os.chdir(folder_name) | ||
|
||
if not os.path.exists(self.symbol): | ||
os.mkdir(self.symbol) | ||
os.chdir(self.symbol) # Go inside the folder. | ||
|
||
os.chdir(self.symbol) | ||
|
||
def write_csv_data(self, totalData: list, fileName: str, armyTime: bool = True) -> str: | ||
""" | ||
|
@@ -544,7 +546,7 @@ def write_csv_data(self, totalData: list, fileName: str, armyTime: bool = True) | |
:return: Absolute path to CSV file. | ||
""" | ||
currentPath = os.getcwd() | ||
self.create_folders_and_change_path(folderName="CSV") | ||
self.create_folders_and_change_path(helpers.PATHS.get_csv_dir()) | ||
|
||
with open(fileName, 'w') as f: | ||
f.write("Date_UTC, Open, High, Low, Close, Volume, Quote_Asset_Volume, Number_of_Trades, " | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -6,11 +6,13 @@ | |
import random | ||
import re | ||
import subprocess | ||
import tempfile | ||
import time | ||
from datetime import datetime | ||
from typing import Dict, List, Tuple, Union | ||
|
||
import requests | ||
from appdirs import AppDirs | ||
from dateutil import parser | ||
|
||
import algobot | ||
|
@@ -19,7 +21,10 @@ | |
|
||
BASE_DIR = os.path.dirname(__file__) | ||
ROOT_DIR = os.path.dirname(BASE_DIR) | ||
LOG_FOLDER = 'Logs' | ||
|
||
APP_NAME = "algobot" | ||
APP_AUTHOR = "ZENALC" | ||
|
||
|
||
SHORT_INTERVAL_MAP = { | ||
'1m': '1 Minute', | ||
|
@@ -40,6 +45,69 @@ | |
LONG_INTERVAL_MAP = {v: k for k, v in SHORT_INTERVAL_MAP.items()} | ||
|
||
|
||
class AppDirTemp: | ||
def __init__(self): | ||
self.root = tempfile.mkdtemp() | ||
|
||
@property | ||
def user_data_dir(self): | ||
return os.path.join(self.root, "UserData") | ||
|
||
@property | ||
def user_log_dir(self): | ||
return os.path.join(self.root, "UserLog") | ||
|
||
|
||
class Paths: | ||
""" Encapsulates all the path information for the app to store its configuration. """ | ||
def __init__(self, root_dir: str, app_dirs): | ||
inverse marked this conversation as resolved.
Show resolved
Hide resolved
|
||
self.root_dir = root_dir | ||
self.app_dirs = app_dirs | ||
|
||
def get_ui_dir(self) -> str: | ||
return os.path.join(self.root_dir, 'UI') | ||
|
||
def get_log_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_log_dir, 'Logs') | ||
|
||
def get_database_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'Databases') | ||
|
||
def get_state_path(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'state.json') | ||
|
||
def get_optimizer_results_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'Optimizer Results') | ||
|
||
def get_backtest_results_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'Backtest Results') | ||
|
||
def get_trade_history_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'Trade History') | ||
|
||
def get_volatility_results_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'Volatility Results') | ||
|
||
def get_csv_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'CSV') | ||
|
||
def get_configuration_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'configuration') | ||
inverse marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
def get_credentials_dir(self) -> str: | ||
return os.path.join(self.app_dirs.user_data_dir, 'Credentials') | ||
|
||
|
||
def _get_app_dirs(): | ||
inverse marked this conversation as resolved.
Show resolved
Hide resolved
|
||
if os.getenv("ALGOBOT_TESTING"): | ||
return AppDirTemp() | ||
|
||
return AppDirs(APP_NAME, APP_AUTHOR) | ||
|
||
|
||
PATHS = Paths(ROOT_DIR, _get_app_dirs()) | ||
|
||
|
||
def get_latest_version() -> str: | ||
""" | ||
Gets the latest Algobot version from GitHub. | ||
|
@@ -90,65 +158,53 @@ def open_folder(folder: str): | |
""" | ||
This will open a folder even if it doesn't exist. It'll create one if it doesn't exist. | ||
""" | ||
targetPath = create_folder(folder) | ||
open_file_or_folder(targetPath) | ||
|
||
|
||
def create_folder(folder: str): | ||
""" | ||
This will create a folder if needed in the root directory. | ||
""" | ||
targetPath = os.path.join(ROOT_DIR, folder) | ||
create_folder_if_needed(targetPath) | ||
|
||
return targetPath | ||
create_folder_if_needed(folder) | ||
open_file_or_folder(folder) | ||
|
||
|
||
def create_folder_if_needed(targetPath: str, basePath: str = ROOT_DIR) -> bool: | ||
def create_folder_if_needed(target_path: str) -> bool: | ||
""" | ||
This function will create the appropriate folders in the root folder if needed. | ||
:param targetPath: Target path to have exist. | ||
:param basePath: Base path to start from. By default, it'll be the root directory. | ||
:param target_path: Target path to have exist. | ||
:return: Boolean whether folder was created or not. | ||
""" | ||
if not os.path.exists(targetPath): | ||
folder = os.path.basename(targetPath) | ||
os.mkdir(os.path.join(basePath, folder)) | ||
if not os.path.exists(target_path): | ||
os.makedirs(target_path, exist_ok=True) | ||
return True | ||
return False | ||
|
||
|
||
def open_file_or_folder(targetPath: str): | ||
def open_file_or_folder(target_path: str): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nice |
||
""" | ||
Opens a file or folder based on targetPath. | ||
:param targetPath: File or folder to open with system defaults. | ||
:param target_path: File or folder to open with system defaults. | ||
""" | ||
if platform.system() == "Windows": | ||
os.startfile(targetPath) | ||
os.startfile(target_path) | ||
elif platform.system() == "Darwin": | ||
subprocess.Popen(["open", targetPath]) | ||
subprocess.Popen(["open", target_path]) | ||
else: | ||
subprocess.Popen(["xdg-open", targetPath]) | ||
subprocess.Popen(["xdg-open", target_path]) | ||
|
||
|
||
def setup_and_return_log_path(fileName: str) -> str: | ||
def setup_and_return_log_path(filename: str) -> str: | ||
""" | ||
Creates folders (if needed) and returns default log path. | ||
:param fileName: Log filename to be created. | ||
:param filename: Log filename to be created. | ||
:return: Absolute path to log file. | ||
""" | ||
LOG_DIR = os.path.join(ROOT_DIR, LOG_FOLDER) | ||
if not os.path.exists(LOG_DIR): | ||
os.mkdir(LOG_DIR) | ||
log_dir = PATHS.get_log_dir() | ||
if not os.path.exists(log_dir): | ||
os.makedirs(log_dir) | ||
|
||
todayDate = datetime.today().strftime('%Y-%m-%d') | ||
LOG_DATE_FOLDER = os.path.join(LOG_DIR, todayDate) | ||
if not os.path.exists(LOG_DATE_FOLDER): | ||
os.mkdir(LOG_DATE_FOLDER) | ||
today_date = datetime.today().strftime('%Y-%m-%d') | ||
log_date_folder = os.path.join(log_dir, today_date) | ||
if not os.path.exists(log_date_folder): | ||
os.mkdir(log_date_folder) | ||
|
||
logFileName = f'{datetime.now().strftime("%H-%M-%S")}-{fileName}.log' | ||
fullPath = os.path.join(LOG_DATE_FOLDER, logFileName) | ||
return fullPath | ||
log_file_name = f'{datetime.now().strftime("%H-%M-%S")}-{filename}.log' | ||
full_path = os.path.join(log_date_folder, log_file_name) | ||
return full_path | ||
|
||
|
||
def get_logger(log_file: str, logger_name: str) -> logging.Logger: | ||
|
@@ -164,7 +220,7 @@ def get_logger(log_file: str, logger_name: str) -> logging.Logger: | |
log_level = logging.DEBUG | ||
logger.setLevel(log_level) | ||
formatter = logging.Formatter('%(message)s') | ||
handler = logging.FileHandler(filename=setup_and_return_log_path(fileName=log_file), delay=True) | ||
handler = logging.FileHandler(filename=setup_and_return_log_path(filename=log_file), delay=True) | ||
handler.setFormatter(formatter) | ||
logger.addHandler(handler) | ||
|
||
|
Uh oh!
There was an error while loading. Please reload this page.