Browse Source

feat: test warn

pull/14/head
QuentinN42 10 months ago
parent
commit
4fbcd2a2b9
Signed by: number42 GPG Key ID: 2CD7D563712B3A50
  1. 46
      auto_trading/interfaces.py
  2. 7
      tests/interfaces/test_ptf.py

46
auto_trading/interfaces.py

@ -14,7 +14,8 @@ from .errors import OrderException, UnknowOrder, PTFException
@dataclass
class Order:
""" An order to execute on a market. """
"""An order to execute on a market."""
successfull: Optional[bool] = field(default=None, init=False)
creation_date: datetime
@ -25,30 +26,30 @@ class CandlesProperties:
class DataBroker(ABC):
""" Somethink that give you data. """
"""Somethink that give you data."""
def __init__(self):
""" Init the class. """
"""Init the class."""
self.logger = logging.getLogger(self.__class__.__name__)
@abstractproperty
def properties(self) -> CandlesProperties:
""" Return the properties of the candles for this broker. """
"""Return the properties of the candles for this broker."""
@abstractproperty
def current_change(self) -> DataFrame:
""" Return the current change for each money. """
"""Return the current change for each money."""
def __iter__(self) -> "DataBroker":
""" Initialise the iterator. """
"""Initialise the iterator."""
return self
@abstractmethod
def __next__(self) -> DataFrame:
"""Next values.
Return the dataframe of all stock history for the strategy / indicators.
Returns:
DataFrame: Time-Stock valuated candlestick data.
For each time and each stock give (high, low, open, close).
@ -56,10 +57,10 @@ class DataBroker(ABC):
class Indicator(ABC):
""" Somethink that give you an insight of the market. """
"""Somethink that give you an insight of the market."""
def __init__(self):
""" Init the class. """
"""Init the class."""
self.logger = logging.getLogger(self.__class__.__name__)
@abstractmethod
@ -77,11 +78,12 @@ class Indicator(ABC):
class Strategy(ABC):
""" What order should you take on the market. """
"""What order should you take on the market."""
indicators: Dict[str, Indicator]
def __init__(self, indicators: Dict[str, Indicator]):
""" Init the class with some inticators. """
"""Init the class with some inticators."""
self.logger = logging.getLogger(self.__class__.__name__)
self.indicators = indicators
@ -99,14 +101,13 @@ class Strategy(ABC):
return self.execute(data, indicators_results)
@abstractmethod
def execute(self, data: DataFrame,
indicators_results: DataFrame) -> List[Order]:
def execute(self, data: DataFrame, indicators_results: DataFrame) -> List[Order]:
"""Execute the strategy with the indicators insights.
Args:
data (DataFrame): The Data broker output.
For each time and each stock give (high, low, open, close).
indicators_results (DataFrame): Indicator-Stock valuated float.
For each indicator and each stock give -1 if realy bad and +1 if realy good.
@ -116,13 +117,13 @@ class Strategy(ABC):
class PTF(ABC):
""" Somethink that buy or sell stocks."""
"""Somethink that buy or sell stocks."""
executors: Dict[Any, Callable[[Any, Any], None]] = {}
history: List[Order]
def __init__(self, skip_errors: bool = True, save_errors: bool = True):
""" Init the class.
"""Init the class.
Args:
skip_errors (bool, optional): Do we skip orders in failure ? Defaults to True.
@ -135,7 +136,7 @@ class PTF(ABC):
@abstractproperty
def balance(self) -> float:
""" Return the current total balance. """
"""Return the current total balance."""
def execute_multiples(self, orders: List[Order]) -> None:
"""Execute all orders
@ -150,18 +151,17 @@ class PTF(ABC):
order.successfull = True
except OrderException as e:
if not self.skip_errors:
raise PTFException(
f"Got and order exception : {e.message}") from e
self.logger.warn("Got an order exception : %s", e.message)
raise PTFException(f"Got and order exception : {e.message}") from e
self.logger.warning("Got an order exception : %s", e.message)
order.successfull = False
if self.save_errors or order.successfull:
self.history.append(order)
def _execute(self, order: Order) -> None:
"""Execute one order.
Try to execute the order on the market.
Raises:
OrderError: if the execution was unsuccessfull.

7
tests/interfaces/test_ptf.py

@ -75,3 +75,10 @@ def test_ptf_executions(test_class):
inst = test_class(skip_errors=False)
with pytest.raises(PTFException):
inst.execute_multiples([short_order, long_order])
@pytest.mark.parametrize("test_class", [_TestPTFLong, _TestPTFShort, _TestPTFNone])
def test_ptf_warn_only(test_class):
inst = test_class()
# Do not raise, only print a warning
inst.execute_multiples([short_order, long_order])

Loading…
Cancel
Save