heavy reorganization

This commit is contained in:
Giulio De Pasquale 2020-11-30 14:38:28 +00:00
parent 7521756115
commit c7a582e6c5
9 changed files with 817 additions and 648 deletions

485
bfxbot.py
View File

@ -1,485 +0,0 @@
#!/usr/bin/env python
import asyncio
import inspect
import shutil
import time
from enum import Enum
from time import sleep
from typing import Dict, List
import dotenv
import termplotlib
from asciimatics.screen import Screen
from bfxapi import Client, Order
from bfxapi.models.position import Position
from playsound import playsound
from termcolor import colored
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,
CLOSE_POSITION = 8,
TRAILING_STOP_SET = 9,
TRAILING_STOP_MOVED = 10,
ORDER_SUBMITTED = 11,
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 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:
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, printer):
self.ticker: Ticker = Ticker(tick_duration)
self.events: List[Event] = []
self.symbol = symbol
self.ticks: Dict[int, (float, Position)] = {}
self.current_state: State = State.LOSS
self.printer: Printer = printer
self.stop_percentage: float = None
async def update(self, position: Position):
self.ticks[self.get_current_tick()] = (await get_current_price(self.symbol), position)
def wait(self):
sleep(self.ticker.seconds)
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 last_position(self) -> Position:
return self.ticks[self.ticker.current_tick][1]
async def add_event(self, event: Event):
self.events.append(event)
await eh.call_event(event, self)
async def last_price(self) -> float:
return await get_current_price(self.symbol)
async def set_state(self, state: State):
if self.current_state != state:
event: Event = 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)
await eh.call_event(event, self)
self.current_state = state
await eh.call_state(self.current_state, self)
def get_current_state(self) -> State:
return self.current_state
class EventHandler:
def __init__(self):
self.event_handlers = {}
self.state_handlers = {}
async def call_event(self, event: Event, status: Status):
value = event.kind.value
if value in self.event_handlers:
for h in self.event_handlers[value]:
if inspect.iscoroutinefunction(h):
await h(event, status)
else:
h(event, status)
async def call_state(self, state: State, status: Status):
if state in self.state_handlers:
for h in self.state_handlers[state]:
if inspect.iscoroutinefunction(h):
await h(status)
else:
h(status)
def on_event(self, kind: EventKind):
value = kind.value
def registerhandler(handler):
if value in self.event_handlers:
self.event_handlers[value].append(handler)
else:
self.event_handlers[value] = [handler]
return handler
return registerhandler
def on_state(self, state: State):
def registerhandler(handler):
if state in self.state_handlers:
self.state_handlers[state].append(handler)
else:
self.state_handlers[state] = [handler]
return handler
return registerhandler
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 = -3.75
OFFER_PERC = 0.01
TRAIL_STOP_PERCENTAGES = {
State.MINIMUM_PROFIT: 0.2,
State.PROFIT: 0.1
}
@eh.on_event(EventKind.REACHED_GOOD_PROFIT)
def on_good_profit(event: Event, status: Status):
playsound("sounds/coin.mp3")
@eh.on_event(EventKind.REACHED_MIN_PROFIT)
def on_min_profit(event: Event, status: Status):
playsound("sounds/1up.mp3")
@eh.on_event(EventKind.REACHED_MAX_LOSS)
def on_critical(event: Event, status: Status):
playsound("sounds/gameover.mp3")
@eh.on_state(State.MINIMUM_PROFIT)
def on_state_min_profit(status: Status):
update_stop_percentage(State.MINIMUM_PROFIT, status)
current_pl_perc = net_pl_percentage(
status.last_position().profit_loss_percentage, TAKER_FEE)
if current_pl_perc < status.stop_percentage:
status.add_event(Event(EventKind.CLOSE_POSITION,
status.get_current_tick()))
@eh.on_state(State.CRITICAL)
async def on_state_critical(status: Status):
await status.add_event(Event(EventKind.CLOSE_POSITION,
status.get_current_tick()))
@eh.on_state(State.PROFIT)
def on_state_min_profit(status: Status):
update_stop_percentage(State.PROFIT, status)
current_pl_perc = net_pl_percentage(
status.last_position().profit_loss_percentage, TAKER_FEE)
if current_pl_perc < status.stop_percentage:
status.add_event(Event(EventKind.CLOSE_POSITION,
status.get_current_tick()))
def update_stop_percentage(state: State, status: Status):
last_position = status.last_position()
last_pl_net_perc = net_pl_percentage(
last_position.profit_loss_percentage, TAKER_FEE)
# set stop percentage for first time
if not status.stop_percentage:
status.add_event(Event(EventKind.TRAILING_STOP_SET,
status.get_current_tick()))
status.stop_percentage = last_pl_net_perc - \
TRAIL_STOP_PERCENTAGES[state]
return
# moving trailing stop
if last_pl_net_perc - TRAIL_STOP_PERCENTAGES[state] > status.stop_percentage:
status.add_event(Event(EventKind.TRAILING_STOP_MOVED,
status.get_current_tick()))
status.stop_percentage = last_pl_net_perc - \
TRAIL_STOP_PERCENTAGES[state]
return
@eh.on_event(EventKind.CLOSE_POSITION)
async def on_close_position(event: Event, status: Status):
closing_price = await calculate_best_closing_price(status)
amount = status.last_position().amount * -1
await bfx.submit_order(status.symbol, closing_price, amount, Order.Type.LIMIT)
await status.add_event(Event(EventKind.ORDER_SUBMITTED, status.get_current_tick()))
@eh.on_event(EventKind.ORDER_SUBMITTED)
def on_order_submitted(event: Event, status: Status):
status.printer.print_next_line("ORDER SUBMITTED!")
return
async def calculate_best_closing_price(status: Status):
p: Position = status.last_position()
is_long_pos = p.amount < 0
pub_tick = await bfx.get_public_ticker(status.symbol)
bid_price = pub_tick[0]
ask_price = pub_tick[2]
if is_long_pos:
closing_price = bid_price * (1 - OFFER_PERC / 100)
else:
closing_price = ask_price * (1 + OFFER_PERC / 100)
return closing_price
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, printer)
balance = await get_usd_balance()
while True:
positions = [p for p in await bfx.get_active_position() if p.symbol == status.symbol]
orders = await bfx.get_active_orders(symbol)
current_price = await status.last_price()
screen.clear()
printer.print_next_line(
"Balance: ${} | Current {} price: {} | Current tick ({} sec): {}".format(colored_float(balance, "white"),
symbol,
colored_float(
current_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]:
await status.update(p)
plfees_percentage = net_pl_percentage(
p.profit_loss_percentage, TAKER_FEE)
if plfees_percentage > GOOD_PROFIT_PERC:
await status.set_state(State.PROFIT)
elif MIN_PROFIT_PERC <= plfees_percentage < GOOD_PROFIT_PERC:
await status.set_state(State.MINIMUM_PROFIT)
elif 0.0 <= plfees_percentage < MIN_PROFIT_PERC:
await status.set_state(State.BREAK_EVEN)
elif MAX_LOSS_PERC < plfees_percentage < 0.0:
await status.set_state(State.LOSS)
else:
await status.set_state(State.CRITICAL)
status_color = status.get_current_state().color()
#
# min / max calculations
#
if plfees_percentage > max_perc:
max_perc = plfees_percentage
await status.add_event(Event(EventKind.NEW_MAXIMUM,
status.get_current_tick()))
if plfees_percentage < min_perc:
min_perc = plfees_percentage
await 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()
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[0] for x in status.ticks.values()]
figure.plot(x, y, width=printer.screen.width,
height=printer.screen.height - printer.get_current_line())
printer.print_next_line(figure.get_string())
async def get_current_price(symbol):
tickers = await bfx.get_public_ticker(symbol)
return tickers[6]
async def get_usd_balance():
balance = 0.0
wallets = await bfx.get_wallets()
for w in wallets:
if w.currency == "USD":
balance += w.balance
else:
current_price = await get_current_price(f"t{w.currency}USD")
balance += current_price * w.balance
return balance
if __name__ == "__main__":
asyncio.run(Screen.wrapper(main))

View File

@ -0,0 +1 @@
from .bfxbot import BfxBot

View File

@ -1,87 +1,59 @@
import inspect
from time import sleep
from typing import Dict
from bfxapi import Position
from bfxbot.bfxwrapper import BfxWrapper
from bfxbot.event import Event, EventKind
from bfxbot.status import Status, PositionState, Ticker
from bfxbot.currency import Symbol
from bfxbot.models import SymbolStatus, Ticker, EventHandler, Strategy
class BfxBot:
class EventHandler:
def __init__(self):
self.event_handlers = {}
self.state_handlers = {}
async def call_event(self, event: Event, status: Status):
value = event.kind.value
if value in self.event_handlers:
for h in self.event_handlers[value]:
if inspect.iscoroutinefunction(h):
await h(event, status)
else:
h(event, status)
async def call_state(self, state: PositionState, status: Status):
if state in self.state_handlers:
for h in self.state_handlers[state]:
if inspect.iscoroutinefunction(h):
await h(status)
else:
h(status)
def on_event(self, kind: EventKind):
value = kind.value
def registerhandler(handler):
if value in self.event_handlers:
self.event_handlers[value].append(handler)
else:
self.event_handlers[value] = [handler]
return handler
return registerhandler
def on_state(self, state: PositionState):
def registerhandler(handler):
if state in self.state_handlers:
self.state_handlers[state].append(handler)
else:
self.state_handlers[state] = [handler]
return handler
return registerhandler
def __init__(self, api_key: str, api_secret: str, tick_duration: int = 60):
def __init__(self, api_key: str, api_secret: str, tick_duration: int = 1):
self.bfx: BfxWrapper = BfxWrapper(api_key, api_secret)
self.status: Dict[str, Status] = {}
self.eh: BfxBot.EventHandler = BfxBot.EventHandler()
self.status: Dict[Symbol, SymbolStatus] = {}
self.ticker: Ticker = Ticker(tick_duration)
await self.__update_status__()
async def __update_status__(self):
active_positions = await self.bfx.get_active_position()
for p in active_positions:
if p.symbol not in self.status:
self.status[p.symbol] = Status(p.symbol)
symbol = Symbol.from_str(p.symbol)
self.status[p.symbol].positions[self.ticker.current_tick].append(p)
if symbol not in self.status:
self.status[symbol] = SymbolStatus(symbol)
await self.status[symbol].add_position(p)
for symbol in self.status.keys():
active_orders = await self.bfx.get_active_orders(symbol)
for o in active_orders:
if symbol not in self.status:
self.status[symbol] = Status(symbol)
self.status[symbol].orders[self.ticker.current_tick].append(o)
self.status[symbol] = SymbolStatus(symbol)
def __trigger__events(self):
return
self.status[symbol].add_order(o)
def event_handler(self, symbol) -> EventHandler:
if symbol not in self.status:
return None
return self.status[symbol].eh
def status(self, symbol: Symbol) -> SymbolStatus:
if symbol not in self.status:
return None
return self.status[symbol]
async def start(self):
await self.__update_status__()
async def update(self):
self.ticker.inc()
sleep(self.ticker.seconds)
await self.__update_status__()
def set_strategy(self, symbol, strategy: Strategy):
if symbol in self.status:
self.status[symbol].strategy = strategy
else:
self.status[symbol] = SymbolStatus(symbol, strategy)

32
bfxbot/currency.py Normal file
View File

@ -0,0 +1,32 @@
import re
from enum import Enum
class Symbol(Enum):
XMR = "XMR"
BTC = "BTC"
ETH = "ETH"
def __repr__(self):
return f"t{self.value}USD"
def __str__(self):
return self.__repr__()
@staticmethod
def from_str(str: str):
match = re.compile("t([a-zA-Z]+)USD").match(str)
if not match:
raise ValueError
currency = match.group(1).lower()
if currency in ("xmr"):
return Symbol.XMR
elif currency in ("btc"):
return Symbol.BTC
elif currency in ("eth"):
return Symbol.ETH
else:
return NotImplementedError

View File

@ -1,24 +0,0 @@
from enum import Enum
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,
CLOSE_POSITION = 8,
TRAILING_STOP_SET = 9,
TRAILING_STOP_MOVED = 10,
ORDER_SUBMITTED = 11,
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}"

205
bfxbot/models.py Normal file
View File

@ -0,0 +1,205 @@
import inspect
import time
from enum import Enum
from typing import List, Dict
from bfxapi import Order, Position
from bfxbot.currency import Symbol
def __add_to_dict_list__(dict: Dict[int, List], k, v):
if k not in dict:
dict[k] = [v]
return
dict[k].append(v)
class PositionState(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 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,
CLOSE_POSITION = 8,
TRAILING_STOP_SET = 9,
TRAILING_STOP_MOVED = 10,
ORDER_SUBMITTED = 11,
class Ticker:
def __init__(self, sec) -> None:
self.seconds: int = sec
self.start_time = time.time()
self.current_tick: int = 1
def inc(self):
self.current_tick += 1
class Event:
def __init__(self, kind: EventKind, pid: int, tick: int) -> None:
self.kind: EventKind = kind
self.tick: int = tick
# position ID
self.pid: int = pid
def __repr__(self) -> str:
return f"[{self.pid}]: {self.kind.name} @ Tick {self.tick}"
class PositionWrapper:
def __init__(self, position: Position):
self.position: Position = position
self.stop_percentage: float = None
self.state: PositionState = PositionState.LOSS
def get_stop_percentage(self) -> float:
return self.stop_percentage
def get_state(self) -> PositionState:
return self.state
def set_stop_percentage(self, perc: float):
self.stop_percentage = perc
def set_state(self, state: PositionState):
if not isinstance(state, PositionState):
return
self.state = state
class SymbolStatus:
def __init__(self, symbol: Symbol, strategy=None):
self.symbol = symbol
self.eh = EventHandler()
self.events: List[Event] = []
self.orders: Dict[int, List[Order]] = {}
self.positions: Dict[int, List[PositionWrapper]] = {}
self.current_tick: int = 1
self.strategy: Strategy = strategy
def last_events(self, n) -> List[Event]:
return self.events[-n:]
def last_positions(self) -> List[PositionWrapper]:
return self.positions[self.current_tick]
# Applies strategy and adds position to list
async def add_position(self, position: Position):
pw = PositionWrapper(position)
if self.strategy:
await self.__apply_strategy_to_position__(pw)
__add_to_dict_list__(self.positions, self.current_tick, pw)
def add_order(self, order: Order):
if self.strategy:
self.strategy.order_on_tick(order, self)
__add_to_dict_list__(self.orders, self.current_tick, order)
def previous_position_w(self, pid: int) -> PositionWrapper:
if self.current_tick == 1:
return None
return next(filter(lambda x: x.pid == pid, self.positions[self.current_tick - 1]))
async def __add_event__(self, event: Event):
self.events.append(event)
await self.eh.call_event(self, event)
async def __apply_strategy_to_position__(self, pw: PositionWrapper):
(new_state, events) = self.strategy.position_on_tick(pw.position, self)
if isinstance(new_state, PositionState):
await self.__update_position_state__(pw, new_state)
if isinstance(events, List):
for e in events:
if isinstance(e, Event):
await self.__add_event__(e)
async def __update_position_state__(self, pw: PositionWrapper, state: PositionState):
pw.set_state(state)
await self.eh.call_state(self, pw)
class Strategy:
"""
Defines new position state and events after tick.
"""
def position_on_tick(self, position: Position, ss: SymbolStatus) -> (PositionState, List[Event]):
pass
"""
Defines new order state and events after tick.
"""
def order_on_tick(self, order: Order, ss: SymbolStatus):
pass
class EventHandler:
def __init__(self):
self.event_handlers = {}
self.state_handlers = {}
async def call_event(self, status: SymbolStatus, event: Event):
value = event.kind.value
if value in self.event_handlers:
for h in self.event_handlers[value]:
if inspect.iscoroutinefunction(h):
await h(status, event)
else:
h(status, event)
async def call_state(self, status: SymbolStatus, pw: PositionWrapper):
if pw.state in self.state_handlers:
for h in self.state_handlers[pw.state]:
if inspect.iscoroutinefunction(h):
await h(status, pw)
else:
h(status, pw)
def on_event(self, kind: EventKind):
value = kind.value
def registerhandler(handler):
if value in self.event_handlers:
self.event_handlers[value].append(handler)
else:
self.event_handlers[value] = [handler]
return handler
return registerhandler
def on_state(self, state: PositionState):
def registerhandler(handler):
if state in self.state_handlers:
self.state_handlers[state].append(handler)
else:
self.state_handlers[state] = [handler]
return handler
return registerhandler

View File

@ -1,78 +0,0 @@
import time
from enum import Enum
from typing import List, Dict
from bfxapi import Position, Order
from bfxbot.bfxbot import BfxBot
from bfxbot.event import Event, EventKind
class Ticker:
def __init__(self, sec) -> None:
self.seconds: int = sec
self.start_time = time.time()
self.current_tick: int = 1
def inc(self):
self.current_tick += 1
class PositionState(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 Status:
def __init__(self, symbol):
self.events: List[Event] = []
self.symbol = symbol
self.current_state: PositionState = PositionState.LOSS
self.stop_percentage: float = None
self.orders: Dict[int, List[Order]] = {}
self.positions: Dict[int, List[Position]] = {}
def last_events(self, n) -> List[Event]:
return self.events[-n:]
def last_position(self) -> Position:
return [self.ticker.current_tick][1]
async def add_event(self, event: Event, event_handler: BfxBot.EventHandler):
self.events.append(event)
await event_handler.call_event(event, self)
async def set_state(self, state: PositionState, event_handler: BfxBot.EventHandler, tick: int):
if self.current_state != state:
event: Event = None
if state == PositionState.CRITICAL:
event = Event(EventKind.REACHED_MAX_LOSS, tick)
elif state == PositionState.LOSS:
event = Event(EventKind.REACHED_LOSS, tick)
elif state == PositionState.BREAK_EVEN:
event = Event(EventKind.REACHED_BREAK_EVEN, tick)
elif state == PositionState.MINIMUM_PROFIT:
event = Event(EventKind.REACHED_MIN_PROFIT, tick)
elif state == PositionState.PROFIT:
event = Event(EventKind.REACHED_GOOD_PROFIT, tick)
self.events.append(event)
await event_handler.call_event(event, self)
self.current_state = state
await event_handler.call_state(self.current_state, self)
def get_current_state(self) -> PositionState:
return self.current_state

6
bfxbot/utils.py Normal file
View File

@ -0,0 +1,6 @@
TAKER_FEE = 0.2
MAKER_FEE = 0.1
def net_pl_percentage(perc: float, reference_fee_perc: float):
return perc - reference_fee_perc

540
main.py Executable file
View File

@ -0,0 +1,540 @@
# #!/usr/bin/env python
#
# import asyncio
# import inspect
# import shutil
# import time
# from enum import Enum
# from time import sleep
# from typing import Dict, List
#
# import dotenv
# import termplotlib
# from asciimatics.screen import Screen
# from bfxapi import Client, Order
# from bfxapi.models.position import Position
# from playsound import playsound
# from termcolor import colored
#
#
# 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,
# CLOSE_POSITION = 8,
# TRAILING_STOP_SET = 9,
# TRAILING_STOP_MOVED = 10,
# ORDER_SUBMITTED = 11,
#
#
# 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 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:
# 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, printer):
# self.ticker: Ticker = Ticker(tick_duration)
# self.events: List[Event] = []
# self.symbol = symbol
# self.ticks: Dict[int, (float, Position)] = {}
# self.current_state: State = State.LOSS
# self.printer: Printer = printer
# self.stop_percentage: float = None
#
# async def update(self, position: Position):
# self.ticks[self.get_current_tick()] = (await get_current_price(self.symbol), position)
#
# def wait(self):
# sleep(self.ticker.seconds)
# 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 last_position(self) -> Position:
# return self.ticks[self.ticker.current_tick][1]
#
# async def add_event(self, event: Event):
# self.events.append(event)
# await eh.call_event(event, self)
#
# async def last_price(self) -> float:
# return await get_current_price(self.symbol)
#
# async def set_state(self, state: State):
# if self.current_state != state:
# event: Event = 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)
# await eh.call_event(event, self)
# self.current_state = state
#
# await eh.call_state(self.current_state, self)
#
# def get_current_state(self) -> State:
# return self.current_state
#
#
# class EventHandler:
# def __init__(self):
# self.event_handlers = {}
# self.state_handlers = {}
#
# async def call_event(self, event: Event, status: Status):
# value = event.kind.value
# if value in self.event_handlers:
# for h in self.event_handlers[value]:
# if inspect.iscoroutinefunction(h):
# await h(event, status)
# else:
# h(event, status)
#
# async def call_state(self, state: State, status: Status):
# if state in self.state_handlers:
# for h in self.state_handlers[state]:
# if inspect.iscoroutinefunction(h):
# await h(status)
# else:
# h(status)
#
# def on_event(self, kind: EventKind):
# value = kind.value
#
# def registerhandler(handler):
# if value in self.event_handlers:
# self.event_handlers[value].append(handler)
# else:
# self.event_handlers[value] = [handler]
# return handler
#
# return registerhandler
#
# def on_state(self, state: State):
# def registerhandler(handler):
# if state in self.state_handlers:
# self.state_handlers[state].append(handler)
# else:
# self.state_handlers[state] = [handler]
# return handler
#
# return registerhandler
#
#
# 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 = -3.75
# OFFER_PERC = 0.01
#
# TRAIL_STOP_PERCENTAGES = {
# State.MINIMUM_PROFIT: 0.2,
# State.PROFIT: 0.1
# }
#
#
# @eh.on_event(EventKind.REACHED_GOOD_PROFIT)
# def on_good_profit(event: Event, status: Status):
# playsound("sounds/coin.mp3")
#
#
# @eh.on_event(EventKind.REACHED_MIN_PROFIT)
# def on_min_profit(event: Event, status: Status):
# playsound("sounds/1up.mp3")
#
#
# @eh.on_event(EventKind.REACHED_MAX_LOSS)
# def on_critical(event: Event, status: Status):
# playsound("sounds/gameover.mp3")
#
#
# @eh.on_state(State.MINIMUM_PROFIT)
# def on_state_min_profit(status: Status):
# update_stop_percentage(State.MINIMUM_PROFIT, status)
#
# current_pl_perc = net_pl_percentage(
# status.last_position().profit_loss_percentage, TAKER_FEE)
#
# if current_pl_perc < status.stop_percentage:
# status.add_event(Event(EventKind.CLOSE_POSITION,
# status.get_current_tick()))
#
#
# @eh.on_state(State.CRITICAL)
# async def on_state_critical(status: Status):
# await status.add_event(Event(EventKind.CLOSE_POSITION,
# status.get_current_tick()))
#
#
# @eh.on_state(State.PROFIT)
# def on_state_min_profit(status: Status):
# update_stop_percentage(State.PROFIT, status)
#
# current_pl_perc = net_pl_percentage(
# status.last_position().profit_loss_percentage, TAKER_FEE)
#
# if current_pl_perc < status.stop_percentage:
# status.add_event(Event(EventKind.CLOSE_POSITION,
# status.get_current_tick()))
#
#
# def update_stop_percentage(state: State, status: Status):
# last_position = status.last_position()
# last_pl_net_perc = net_pl_percentage(
# last_position.profit_loss_percentage, TAKER_FEE)
#
# # set stop percentage for first time
# if not status.stop_percentage:
# status.add_event(Event(EventKind.TRAILING_STOP_SET,
# status.get_current_tick()))
# status.stop_percentage = last_pl_net_perc - \
# TRAIL_STOP_PERCENTAGES[state]
# return
#
# # moving trailing stop
# if last_pl_net_perc - TRAIL_STOP_PERCENTAGES[state] > status.stop_percentage:
# status.add_event(Event(EventKind.TRAILING_STOP_MOVED,
# status.get_current_tick()))
# status.stop_percentage = last_pl_net_perc - \
# TRAIL_STOP_PERCENTAGES[state]
#
# return
#
#
# @eh.on_event(EventKind.CLOSE_POSITION)
# async def on_close_position(event: Event, status: Status):
# closing_price = await calculate_best_closing_price(status)
# amount = status.last_position().amount * -1
#
# await bfx.submit_order(status.symbol, closing_price, amount, Order.Type.LIMIT)
# await status.add_event(Event(EventKind.ORDER_SUBMITTED, status.get_current_tick()))
#
#
# @eh.on_event(EventKind.ORDER_SUBMITTED)
# def on_order_submitted(event: Event, status: Status):
# status.printer.print_next_line("ORDER SUBMITTED!")
# return
#
#
# async def calculate_best_closing_price(status: Status):
# p: Position = status.last_position()
#
# is_long_pos = p.amount < 0
#
# pub_tick = await bfx.get_public_ticker(status.symbol)
#
# bid_price = pub_tick[0]
# ask_price = pub_tick[2]
#
# if is_long_pos:
# closing_price = bid_price * (1 - OFFER_PERC / 100)
# else:
# closing_price = ask_price * (1 + OFFER_PERC / 100)
#
# return closing_price
#
#
# 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, printer)
# balance = await get_usd_balance()
#
# while True:
# positions = [p for p in await bfx.get_active_position() if p.symbol == status.symbol]
# orders = await bfx.get_active_orders(symbol)
#
# current_price = await status.last_price()
#
# screen.clear()
# printer.print_next_line(
# "Balance: ${} | Current {} price: {} | Current tick ({} sec): {}".format(colored_float(balance, "white"),
# symbol,
# colored_float(
# current_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]:
# await status.update(p)
#
# plfees_percentage = net_pl_percentage(
# p.profit_loss_percentage, TAKER_FEE)
#
# if plfees_percentage > GOOD_PROFIT_PERC:
# await status.set_state(State.PROFIT)
# elif MIN_PROFIT_PERC <= plfees_percentage < GOOD_PROFIT_PERC:
# await status.set_state(State.MINIMUM_PROFIT)
# elif 0.0 <= plfees_percentage < MIN_PROFIT_PERC:
# await status.set_state(State.BREAK_EVEN)
# elif MAX_LOSS_PERC < plfees_percentage < 0.0:
# await status.set_state(State.LOSS)
# else:
# await status.set_state(State.CRITICAL)
#
# status_color = status.get_current_state().color()
#
# #
# # min / max calculations
# #
# if plfees_percentage > max_perc:
# max_perc = plfees_percentage
# await status.add_event(Event(EventKind.NEW_MAXIMUM,
# status.get_current_tick()))
# if plfees_percentage < min_perc:
# min_perc = plfees_percentage
# await 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()
# 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[0] for x in status.ticks.values()]
#
# figure.plot(x, y, width=printer.screen.width,
# height=printer.screen.height - printer.get_current_line())
#
# printer.print_next_line(figure.get_string())
#
#
# async def get_current_price(symbol):
# tickers = await bfx.get_public_ticker(symbol)
# return tickers[6]
#
#
# async def get_usd_balance():
# balance = 0.0
#
# wallets = await bfx.get_wallets()
#
# for w in wallets:
# if w.currency == "USD":
# balance += w.balance
# else:
# current_price = await get_current_price(f"t{w.currency}USD")
# balance += current_price * w.balance
#
# return balance
#
#
# if __name__ == "__main__":
# asyncio.run(Screen.wrapper(main))
import asyncio
from typing import List
from bfxapi import Position
from bfxbot import BfxBot
import dotenv
import os
import bfxapi
from bfxbot.symbolstatus import Strategy, SymbolStatus, PositionState, Event
from bfxbot.utils import TAKER_FEE
dotenv.load_dotenv()
class TrailingStopStrategy(Strategy):
BREAK_EVEN_PERC = TAKER_FEE
MIN_PROFIT_PERC = TAKER_FEE * 2.5
GOOD_PROFIT_PERC = MIN_PROFIT_PERC * 1.5
MAX_LOSS_PERC = -3.75
OFFER_PERC = 0.01
TRAIL_STOP_PERCENTAGES = {
PositionState.MINIMUM_PROFIT: 0.27,
PositionState.PROFIT: 0.14
}
def position_on_tick(self, position: Position, ss: SymbolStatus) -> (PositionState, List[Event]):
return
async def main():
API_KEY = os.getenv("API_KEY")
API_SECRET = os.getenv("API_SECRET")
if API_KEY == None:
print("API_KEY is not set! Set the var in the .env file.")
return
if API_SECRET == None:
print("API_SECRET is not set! Set the var in the .env file.")
return
bot = BfxBot(api_key=API_KEY, api_secret=API_SECRET)
strategy = TrailingStopStrategy()
bot.set_strategy("tBTCUSD", strategy)
await bot.update()
return
if __name__ == '__main__':
asyncio.run(main())