initial commit

.gitignore

@@ -0,0 +1,156 @@
+
+# Created by https://www.toptal.com/developers/gitignore/api/python,visualstudiocode
+# Edit at https://www.toptal.com/developers/gitignore?templates=python,visualstudiocode
+
+### Python ###
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+pytestdebug.log
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+doc/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+pythonenv*
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# profiling data
+.prof
+
+### VisualStudioCode ###
+.vscode/*
+!.vscode/tasks.json
+!.vscode/launch.json
+*.code-workspace
+
+### VisualStudioCode Patch ###
+# Ignore all local history of files
+.history
+.ionide
+
+# End of https://www.toptal.com/developers/gitignore/api/python,visualstudiocode

main.py

@@ -0,0 +1,315 @@
+#!/usr/bin/env python
+
+import asyncio
+from asyncio import events
+import time
+from enum import Enum
+from os import system, terminal_size
+from time import sleep
+from bfxapi import Client
+from bfxapi.models.order import Order
+from bfxapi.models.position import Position
+from termcolor import colored
+import shutil
+import termplotlib
+import numpy
+from playsound import playsound
+from asciimatics.screen import ManagedScreen, Screen
+import dotenv
+
+dotenv.load()
+API_KEY = dotenv.get('API_KEY', default='')
+API_SECRET = dotenv.get('API_SECRET', default='')
+
+bfx = Client(
+    API_KEY=API_KEY,
+    API_SECRET=API_SECRET
+).rest
+
+TAKER_FEE = 0.2
+MAKER_FEE = 0.1
+
+BREAK_EVEN_PERC = TAKER_FEE
+MIN_PROFIT_PERC = 0.65
+GOOD_PROFIT_PERC = MIN_PROFIT_PERC * 2.1
+MAX_LOSS_PERC = -4.0
+
+TRAILING_GOOD_PROFIT_PERC = 0.2
+
+
+class Ticker():
+    def __init__(self, sec) -> None:
+        self.seconds: int = sec
+        self.start_time = time.time()
+        self.current_tick: int = 1
+
+
+class EventKind(Enum):
+    NEW_MINIMUM = 1,
+    NEW_MAXIMUM = 2,
+    REACHED_LOSS = 3,
+    REACHED_BREAK_EVEN = 4,
+    REACHED_MIN_PROFIT = 5,
+    REACHED_GOOD_PROFIT = 6,
+    REACHED_MAX_LOSS = 7
+
+
+class Event():
+    def __init__(self, kind: EventKind, tick: int) -> None:
+        self.kind: EventKind = kind
+        self.tick: int = tick
+
+    def __repr__(self) -> str:
+        return f"{self.kind.name} @ Tick {self.tick}"
+
+
+class PositionStatus():
+    def __init__(self, position: Position) -> None:
+        self.position: Position = position
+
+
+class State(Enum):
+    CRITICAL = -1,
+    LOSS = 0,
+    BREAK_EVEN = 1,
+    MINIMUM_PROFIT = 2,
+    PROFIT = 3
+
+    def color(self) -> str:
+        if self == self.LOSS or self == self.CRITICAL:
+            return "red"
+        elif self == self.BREAK_EVEN:
+            return "yellow"
+        else:
+            return "green"
+
+
+class Printer():
+    def __init__(self, screen: Screen):
+        self.screen: Screen = screen
+        self.current_line: int = 0
+        (self.current_width, self.current_height) = shutil.get_terminal_size()
+
+    def get_current_line(self) -> int:
+        return self.current_line
+
+    def next(self) -> int:
+        # print("Current line: {}".format(self.current_line))
+        line = self.current_line
+        self.current_line += 1
+        return line
+
+    def print_next_line(self, text):
+        for line in text.split("\n"):
+            self.screen.print_at(line, 0, self.next(), 1)
+        self.screen.refresh()
+
+    def reset_current_line(self):
+        self.current_line = 0
+
+    def set_screen(self, screen: Screen):
+        self.screen = screen
+    
+    def has_screen_resized(self):
+        return (self.current_width, self.current_height) != shutil.get_terminal_size()
+
+    def to_current_screen_size(self):
+        (self.current_width, self.current_height) = shutil.get_terminal_size()
+
+class Status():
+    def __init__(self, tick_duration, symbol):
+        self.ticker: Ticker = Ticker(tick_duration)
+        self.events: list[Event] = []
+        self.symbol = symbol
+        self.ticks: dict[int, float] = {}
+        self.current_state: State = State.LOSS
+
+    async def wait(self) -> None:
+        sleep(self.ticker.seconds)
+
+        self.ticks[self.ticker.current_tick] = await get_current_price(self.symbol)
+        self.ticker.current_tick += 1
+
+    def get_current_tick(self) -> int:
+        return self.ticker.current_tick
+
+    def last_events(self, n):
+        return self.events[-n:]
+
+    def add_event(self, event: Event):
+        self.events.append(event)
+
+    async def last_price(self) -> float:
+        if self.get_current_tick() not in self.ticks.keys():
+            return await get_current_price(self.symbol)
+        self.ticks[self.get_current_tick()]
+
+    def set_state(self, state: State):
+        if self.current_state != state:
+            if state == State.CRITICAL:
+                self.add_event(
+                    Event(EventKind.REACHED_MAX_LOSS, self.get_current_tick()))
+            elif state == State.LOSS:
+                self.add_event(
+                    Event(EventKind.REACHED_LOSS, self.get_current_tick()))
+            elif state == State.BREAK_EVEN:
+                self.add_event(
+                    Event(EventKind.REACHED_BREAK_EVEN, self.get_current_tick()))
+            elif state == State.MINIMUM_PROFIT:
+                self.add_event(
+                    Event(EventKind.REACHED_MIN_PROFIT, self.get_current_tick()))
+            elif state == State.PROFIT:
+                self.add_event(
+                    Event(EventKind.REACHED_GOOD_PROFIT, self.get_current_tick()))
+
+        self.current_state = state
+
+    def get_current_state(self) -> State:
+        return self.current_state
+
+
+def net_pl_percentage(perc: float, reference_fee_perc: float):
+    return perc - reference_fee_perc
+
+
+async def main(screen: Screen):
+    min_perc = 999.0
+    max_perc = -999.0
+    symbol = "tBTCUSD"
+
+    printer = Printer(screen)
+    status = Status(20, symbol)
+
+    while True:
+        positions = [p for p in await bfx.get_active_position() if p.symbol == status.symbol]
+        orders = await bfx.get_active_orders(symbol)
+
+        last_price = await status.last_price()
+
+        screen.clear()
+        printer.print_next_line("Current {} price: {} | Current tick ({} sec): {}".format(symbol,
+                                                                                          colored_float(
+                                                                                              last_price, "white", attrs=["bold"]),
+                                                                                          status.ticker.seconds,
+                                                                                          status.get_current_tick(),
+                                                                                          ))
+
+        if positions:
+            printer.print_next_line("")
+            printer.print_next_line("Open {}:".format(
+                colored("POSITIONS", attrs=["underline"])))
+
+            for p in [p for p in positions if p.symbol == status.symbol]:
+                plfees_percentage = net_pl_percentage(
+                    p.profit_loss_percentage, TAKER_FEE)
+
+                if plfees_percentage > GOOD_PROFIT_PERC:
+                    status.set_state(State.PROFIT)
+                elif plfees_percentage >= MIN_PROFIT_PERC and plfees_percentage < GOOD_PROFIT_PERC:
+                    status.set_state(State.MINIMUM_PROFIT)
+                elif plfees_percentage >= 0.0 and plfees_percentage < MIN_PROFIT_PERC:
+                    status.set_state(State.BREAK_EVEN)
+                elif plfees_percentage < 0.0 and plfees_percentage > MAX_LOSS_PERC:
+                    status.set_state(State.LOSS)
+                else:
+                    status.set_state(State.CRITICAL)
+
+                status_color = status.get_current_state().color()
+
+                #
+                # min / max calculations
+                #
+                if plfees_percentage > max_perc:
+                    max_perc = plfees_percentage
+                    status.add_event(Event(EventKind.NEW_MAXIMUM,
+                                           status.get_current_tick()))
+                if plfees_percentage < min_perc:
+                    min_perc = plfees_percentage
+                    status.add_event(Event(EventKind.NEW_MINIMUM,
+                                           status.get_current_tick()))
+
+                min_perc_colored = colored_percentage(
+                    min_perc, "red") if min_perc < 0.0 else colored_percentage(min_perc, "green")
+                max_perc_colored = colored_percentage(
+                    max_perc, "red") if max_perc < 0.0 else colored_percentage(max_perc, "green")
+
+                #
+                # current status calculations
+                #
+                current_colored_format = "{} ({})".format(colored_percentage(plfees_percentage, status_color),
+                                                          colored_float(p.profit_loss, status_color))
+
+                #
+                # Status bar
+                #
+                printer.print_next_line("{:1.5f} {} @ {} | {} | min: {}, MAX: {}".format(
+                    p.amount,
+                    p.symbol,
+                    colored_float(p.base_price, "white", attrs=["underline"]),
+                    current_colored_format,
+                    min_perc_colored,
+                    max_perc_colored))
+
+            # Separator
+            printer.print_next_line("")
+
+        if orders:
+            printer.print_next_line("Open {}:".format(
+                colored("ORDERS", attrs=["underline"])))
+
+        print_last_events(status, 10, printer)
+        plot(status, printer)
+
+        if status.get_current_state() == State.PROFIT:
+            playsound("sounds/coin.mp3")
+
+        printer.reset_current_line()
+        await status.wait()
+
+    return
+
+
+def colored_percentage(perc, color, **kwargs):
+    return "{}".format(colored("{:1.2f}%".format(perc), color=color, **kwargs))
+
+
+def colored_float(num, color, **kwargs):
+    return "{}".format(colored("{:1.2f}".format(num), color=color, **kwargs))
+
+
+def print_last_events(status: Status, n: int, printer: Printer):
+    printer.print_next_line(colored(f"Last {n} events:", attrs=["bold"]))
+
+    for e in status.last_events(n):
+        printer.print_next_line(f"- {e}")
+
+
+def plot(status: Status, printer: Printer):
+    if status.ticks:
+        figure = termplotlib.figure()
+        x = range(1, status.get_current_tick() + 1)
+        y = [x for x in status.ticks.values()]
+
+        figure.plot(x, y, width=printer.screen.width,
+                    height=printer.screen.height - printer.get_current_line() - 1)
+
+        printer.print_next_line(figure.get_string())
+
+
+async def get_current_price(symbol):
+    tickers = await bfx.get_public_ticker(symbol)
+    return tickers[6]
+
+
+def clear():
+    system("clear")
+
+
+def screen_main():
+    return
+
+
+if __name__ == "__main__":
+    asyncio.run(Screen.wrapper(main))
+    # asyncio.run(main())
+    # screen_main()