core/main.py

#!/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


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 EventHandler:
    def __init__(self):
        self.handlers = {}

    def call(self, event: Event):
        value = event.kind.value
        if value in self.handlers:
            for h in self.handlers[value]:
                h(event)

    def on_event(self, kind: EventKind):
        value = kind.value

        def registerhandler(handler):
            if value in self.handlers:
                self.handlers[value].append(handler)
            else:
                self.handlers[value] = [handler]
            return handler
        return registerhandler


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:
            event: EventKind = None

            if state == State.CRITICAL:
                event = Event(EventKind.REACHED_MAX_LOSS,
                              self.get_current_tick())
            elif state == State.LOSS:
                event = Event(EventKind.REACHED_LOSS, self.get_current_tick())
            elif state == State.BREAK_EVEN:
                event = Event(EventKind.REACHED_BREAK_EVEN,
                              self.get_current_tick())
            elif state == State.MINIMUM_PROFIT:
                event = Event(EventKind.REACHED_MIN_PROFIT,
                              self.get_current_tick())
            elif state == State.PROFIT:
                event = Event(EventKind.REACHED_GOOD_PROFIT,
                              self.get_current_tick())

            self.events.append(event)
            eh.call(event)
            self.current_state = state

    def get_current_state(self) -> State:
        return self.current_state


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
eh = EventHandler()

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


@eh.on_event(EventKind.REACHED_GOOD_PROFIT)
def kaching(event):
    playsound("sounds/coin.mp3")


@eh.on_event(EventKind.REACHED_MIN_PROFIT)
def kaching(event):
    playsound("sounds/letsgo.mp3")


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)

        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()