rustico/core
2
0
Fork 0
Code Issues 9 Pull Requests Projects Releases Wiki Activity

refactored names in symbolstatus. now a strategy has to assemble a positionwrapper. raising exceptions if types are not as expected.

Browse Source
This commit is contained in:
Giulio De Pasquale 2020-12-15 20:10:51 +00:00
parent 3534aef408
commit 2f5259ebd2
1 changed files with 35 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
bfxbot/models.py
View File

@ -1,7 +1,7 @@
import inspect
import time
from enum import Enum
from typing import List, Dict
from typing import List, Dict, Tuple
from bfxapi import Order, Position
@ -77,18 +77,23 @@ class Event:
class PositionWrapper:
def __init__(self, position: Position):
def __init__(self, position: Position, net_profit_loss: float = None, net_profit_loss_percentage: float = None):
self.position: Position = position
self.state: PositionState = PositionState.LOSS
self.__net_profit_loss: float = net_profit_loss
self.__net_profit_loss_percentage: float = net_profit_loss_percentage
self.__state: PositionState = PositionState.LOSS
def get_state(self) -> PositionState:
return self.state
def net_profit_loss(self) -> float:
return self.__net_profit_loss
def net_profit_loss_percentage(self) -> float:
return self.__net_profit_loss_percentage
def set_state(self, state: PositionState):
if not isinstance(state, PositionState):
return
self.__state = state
self.state = state
def state(self) -> PositionState:
return self.__state
class SymbolStatus:
@ -109,10 +114,12 @@ class SymbolStatus:
# Applies strategy and adds position to list
async def add_position(self, position: Position):
pw = PositionWrapper(position)
# if a strategy is defined then the strategy takes care of creating a PW for us
if self.strategy:
await self.__apply_strategy_to_position__(pw)
if not self.strategy:
pw = PositionWrapper(position)
else:
pw = await self.__apply_strategy_to_position__(position)
__add_to_dict_list__(self.positions, self.current_tick, pw)
def all_prices(self) -> List[float]:
@ -140,38 +147,36 @@ class SymbolStatus:
self.events.append(event)
await self.eh.call_event(self, event)
async def __apply_strategy_to_position__(self, pw: PositionWrapper):
async def __apply_strategy_to_position__(self, position: Position):
if not self.strategy:
return
new_state, events = self.strategy.position_on_new_tick(pw.position, self)
pw, events = self.strategy.position_on_new_tick(position, self)
if isinstance(new_state, PositionState):
await self.__update_position_state__(pw, new_state)
if not isinstance(pw, PositionWrapper):
raise ValueError
if isinstance(events, List):
for e in events:
if isinstance(e, Event):
await self.__add_event__(e)
if not isinstance(events, list):
raise ValueError
async def __update_position_state__(self, pw: PositionWrapper, state: PositionState):
pw.set_state(state)
# triggering state callbacks
await self.__trigger_position_state_callbacks__(pw)
for e in events:
if not isinstance(e, Event):
raise ValueError
await self.__add_event__(e)
async def __trigger_position_state_callbacks__(self, pw: PositionWrapper):
await self.eh.call_position_state(self, pw)
class Strategy:
"""
Defines a list of events on a new tick
"""
def on_new_tick(self, ss: SymbolStatus) -> List[Event]:
pass
"""
Defines new position state and events after tick.
"""
def position_on_new_tick(self, position: Position, ss: SymbolStatus) -> (PositionState, List[Event]):
def position_on_new_tick(self, position: Position, ss: SymbolStatus) -> Tuple[PositionWrapper, List[Event]]:
pass
"""
@ -207,7 +212,7 @@ class EventHandler:
h(event, status)
async def call_position_state(self, status: SymbolStatus, pw: PositionWrapper):
state = pw.state
state = pw.state()
if state in self.state_handlers:
for h in self.state_handlers[state]: