You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

215 lines
6.6 KiB

"""
Define basic interfaces for trading strategies.
"""
import logging
from abc import ABC, abstractmethod, abstractproperty
from dataclasses import dataclass, field
from datetime import datetime
import pandas as pd
from pandas import DataFrame, Timedelta, Series # type: ignore
from typing import Dict, List, Optional, Any, Callable
from .errors import OrderException, UnknowOrder, PTFException
@dataclass
class PTFState:
"""A fixed state of a PTF (read only)."""
balance: float
stocks: Dict[str, float]
def total_balance(self, conversion_rate: Dict[str, float]) -> float:
return self.balance + sum(
conversion_rate[stock_name] * amount
for stock_name, amount in self.stocks.items()
)
def to_dict(self):
return {'USD': self.balance, **self.stocks}
@dataclass
class Order:
"""An order to execute on a market."""
successfull: Optional[bool] = field(default=None, init=False)
creation_date: datetime = field(default_factory=datetime.now, init=False)
@dataclass
class CandlesProperties:
period: Timedelta
class DataBroker(ABC):
"""Somethink that give you data."""
def __init__(self):
"""Init the class."""
self.logger = logging.getLogger(self.__class__.__name__)
@abstractproperty
def properties(self) -> CandlesProperties:
"""Return the properties of the candles for this broker."""
@abstractproperty
def current_change(self) -> DataFrame:
"""Return the current change for each money."""
@abstractmethod
def __iter__(self) -> "DataBroker":
"""Initialise the iterator."""
@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).
"""
@abstractmethod
def __len__(self) -> int:
"""Total number of values"""
class Indicator(ABC):
"""Somethink that give you an insight of the market."""
def __init__(self):
"""Init the class."""
self.logger = logging.getLogger(self.__class__.__name__)
@abstractmethod
def __call__(self, data: DataFrame) -> Series:
"""Return a dataframe of valuation of each stock from the input data.
Args:
data (DataFrame): Time-Stock valuated candlestick data.
For each time and each stock give (high, low, open, close).
Returns:
DataFrame: Stock valuated float.
For each stock give -1 if realy bad and +1 if realy good.
"""
class Strategy(ABC):
"""What order should you take on the market."""
indicators: Dict[str, Indicator]
def __init__(self, indicators: Dict[str, Indicator] = None):
"""Init the class with some inticators."""
self.logger = logging.getLogger(self.__class__.__name__)
self.indicators = indicators or {}
def run(self, data: DataFrame, state: PTFState) -> List[Order]:
"""Execute the strategy from the data.
Args:
data (DataFrame): The Data broker output.
For each time and each stock give (high, low, open, close).
Returns:
List[Order]: A list of orders to execute.
"""
indicators_results = DataFrame(
{k: v(data) for k, v in self.indicators.items()}
).T
return self.execute(data, indicators_results, state)
@abstractmethod
def execute(
self, data: DataFrame, indicators_results: DataFrame, ptf_state: PTFState
) -> 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.
Returns:
List[Order]: A list of orders to execute.
"""
class PTF(ABC):
"""Somethink that buy or sell stocks."""
executors: Dict[Any, Callable[[Any, Any], None]] = {}
orders_history: List[Order]
states_history: pd.DataFrame
def __init__(self, skip_errors: bool = True, save_errors: bool = True):
"""Init the class.
Args:
skip_errors (bool, optional): Do we skip orders in failure ? Defaults to True.
save_errors (bool, optional): Do we save orders in failure in the orders_history ? Defaults to True.
"""
self.logger = logging.getLogger(self.__class__.__name__)
self.orders_history = []
self.states_history = pd.DataFrame()
self.skip_errors = skip_errors
self.save_errors = save_errors
@abstractproperty
def state(self) -> PTFState:
"""Return the current state."""
@property
def balance(self) -> float:
"""Return the current total balance."""
return self.state.balance
def total_balance(self, conversion_rate: Dict[str, float]) -> float:
"""Return the current total balance."""
return self.state.total_balance(conversion_rate)
def execute_multiples(self, orders: List[Order]) -> None:
"""Execute all orders
Args:
orders (List[Order]): The list of all orders to execute.
"""
for order in orders:
self.logger.debug("Applying order %s...", order)
try:
self._execute(order)
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.warning("Got an order exception : %s", e.message)
order.successfull = False
if self.save_errors or order.successfull:
self.orders_history.append(order)
# self.states_history.append(self.state.to_dict(), ignore_index=True)
self.states_history = self.states_history.append(self.state.to_dict(), ignore_index=True)
def _execute(self, order: Order) -> None:
"""Execute one order.
Try to execute the order on the market.
Raises:
OrderError: if the execution was unsuccessfull.
Args:
order (Order): One order to execute.
"""
for order_type, executor in self.executors.items():
if isinstance(order, order_type):
executor(self, order)
return
raise UnknowOrder(f"Can not process order of type {type(order)}")