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.

223 lines
6.8 KiB

1 year ago
"""
Define basic interfaces for trading strategies.
"""
import logging
1 year ago
from abc import ABC, abstractmethod, abstractproperty
from dataclasses import dataclass, field
from datetime import datetime
10 months ago
from typing import Dict, List, Optional, Any, Callable
1 year ago
from pandas import DataFrame, Timedelta, Series # type: ignore
1 year ago
from .errors import OrderException, UnknowOrder, PTFException
1 year ago
10 months ago
@dataclass
class PTFState:
"""A fixed state of a PTF (read only)."""
balance: float
stocks: Dict[str, float]
# TODO: stocker le conversion rate à l'instant t pour facilement calculer le total balance
# conversion_rate: Dict[str, float]
def total_balance(self, conversion_rate: Dict[str, float]) -> float:
10 months ago
return self.balance + sum(
conversion_rate[stock_name] * amount
for stock_name, amount in self.stocks.items()
10 months ago
)
9 months ago
def to_dict(self):
return {"USD": self.balance, **self.stocks}
9 months ago
10 months ago
@dataclass
class Order:
10 months ago
"""An order to execute on a market."""
successfull: Optional[bool] = field(default=None, init=False)
10 months ago
creation_date: datetime = field(default_factory=datetime.now, init=False)
1 year ago
@dataclass
class CandlesProperties:
10 months ago
period: Timedelta
1 year ago
class DataBroker(ABC):
10 months ago
"""Somethink that give you data."""
10 months ago
def __init__(self):
10 months ago
"""Init the class."""
self.logger = logging.getLogger(self.__class__.__name__)
10 months ago
@abstractproperty
def properties(self) -> CandlesProperties:
10 months ago
"""Return the properties of the candles for this broker."""
10 months ago
@abstractproperty
def current_change(self) -> DataFrame:
"""Return the current change for each money."""
10 months ago
@abstractmethod
def __iter__(self) -> "DataBroker":
10 months ago
"""Initialise the iterator."""
1 year ago
@abstractmethod
def __next__(self) -> DataFrame:
"""Next values.
10 months ago
Return the dataframe of all stock history for the strategy / indicators.
10 months ago
Returns:
DataFrame: Time-Stock valuated candlestick data.
For each time and each stock give (high, low, open, close).
"""
10 months ago
@abstractmethod
def __len__(self) -> int:
"""Total number of values"""
class Indicator(ABC):
10 months ago
"""Somethink that give you an insight of the market."""
10 months ago
def __init__(self):
10 months ago
"""Init the class."""
self.logger = logging.getLogger(self.__class__.__name__)
10 months ago
@abstractmethod
def __call__(self, data: DataFrame) -> Series:
"""Return a dataframe of valuation of each stock from the input data.
1 year ago
Args:
data (DataFrame): Time-Stock valuated candlestick data.
For each time and each stock give (high, low, open, close).
1 year ago
Returns:
DataFrame: Stock valuated float.
For each stock give -1 if realy bad and +1 if realy good.
"""
1 year ago
class Strategy(ABC):
10 months ago
"""What order should you take on the market."""
indicators: Dict[str, Indicator]
10 months ago
def __init__(self, indicators: Dict[str, Indicator] = None):
10 months ago
"""Init the class with some inticators."""
self.logger = logging.getLogger(self.__class__.__name__)
10 months ago
self.indicators = indicators or {}
10 months ago
def run(self, data: DataFrame, state: PTFState) -> List[Order]:
"""Execute the strategy from the data.
1 year ago
Args:
data (DataFrame): The Data broker output.
For each time and each stock give (high, low, open, close).
1 year ago
Returns:
List[Order]: A list of orders to execute.
"""
indicators_results = DataFrame(
{k: v(data) for k, v in self.indicators.items()}
).T
10 months ago
return self.execute(data, indicators_results, state)
1 year ago
@abstractmethod
10 months ago
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).
10 months ago
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):
10 months ago
"""Somethink that buy or sell stocks."""
10 months ago
executors: Dict[Any, Callable[[Any, Any], None]] = {}
orders_history: List[Order]
states_history: DataFrame
10 months ago
def __init__(self, skip_errors: bool = True, save_errors: bool = True):
10 months ago
"""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 = DataFrame()
self.skip_errors = skip_errors
self.save_errors = save_errors
10 months ago
@abstractproperty
10 months ago
def state(self) -> PTFState:
"""Return the current state."""
@property
def balance(self) -> float:
10 months ago
"""Return the current total balance."""
10 months ago
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:
10 months ago
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)
9 months ago
# 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.
10 months ago
Try to execute the order on the market.
10 months ago
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)}")