79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
|
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
|