Browse Source

feat: base struct done

pull/14/head
QuentinN42 8 months ago
parent
commit
37957f56c2
Signed by: number42 GPG Key ID: 2CD7D563712B3A50
  1. 10
      .vscode/settings.json
  2. 9
      README.md
  3. 21
      auto_trading/errors.py
  4. 183
      auto_trading/interfaces.py
  5. 47
      auto_trading/main.py
  6. 30
      auto_trading/orders.py
  7. 2
      requirements.txt
  8. 5
      scripts/install.sh
  9. 5
      scripts/test.sh
  10. 68
      tests/test_interfaces.py

10
.vscode/settings.json

@ -0,0 +1,10 @@
{
"python.testing.pytestPath": "./scripts/test.sh",
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true,
"testOnSave.enabled": true,
"testOnSave.testCommand": "./scripts/test.sh",
"python.testing.pytestArgs": [
"tests"
]
}

9
README.md

@ -1,3 +1,12 @@
# AutoTrading
Used [this](https://www.kaggle.com/sudalairajkumar/cryptocurrencypricehistory) dataset but for now just making a mean of the values.
## Use it
# install
./scripts/install.sh
# run
python main.py
# test
./scripts/test.sh

21
auto_trading/errors.py

@ -0,0 +1,21 @@
"""Some errors to catch."""
class TradingException(Exception):
""" An error connected to the trading logic. """
message: str
def __init__(self, message):
super().__init__(message)
self.message = message
class OrderException(TradingException):
""" An order execution was unsuccessfull. """
class UnknowOrder(OrderException):
""" The exectuor can't process this order. """
class OrderFails(OrderException):
""" The order placement failed. """
class PTFException(TradingException):
""" An error occur inside the PTF. """

183
auto_trading/interfaces.py

@ -1,54 +1,173 @@
"""
Define basic interfaces for trading strategies.
"""
from abc import ABC, abstractmethod
import logging
import pandas as pd # type: ignore
from abc import ABC, abstractmethod, abstractproperty
from dataclasses import dataclass, field
from datetime import timedelta, datetime
from pandas import DataFrame
from typing import Dict, List, Optional, Type, Callable
from .errors import OrderException, UnknowOrder, PTFException
class Portfolio(ABC):
"""How may of each money do I have ?"""
@abstractmethod
def content(self) -> dict:
"""return the content in each currency"""
@dataclass
class Order:
""" An order to execute on a market. """
successfull: Optional[bool] = field(default=None, init=False)
creation_date: datetime
@abstractmethod
def withdraw(self, amount: int, currency: str) -> None:
"""Withdraw money from the portfolio"""
@abstractmethod
def deposit(self, amount: int, currency: str) -> None:
"""Deposit money into the portfolio"""
@dataclass
class CandlesProperties:
period: timedelta
@abstractmethod
def convert(self, amount: int, to_amount: int, currency: str, to_currency: str) -> None:
"""Convert money from one currency to another"""
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. """
class Strategy(ABC):
"""When do I buy and how many ?"""
def __iter__(self) -> "DataBroker":
""" Initialise the iterator. """
return self
@abstractmethod
def run(self, ptf: Portfolio, result: dict, current_conversion_rate: dict) -> None:
"""Run the strategy"""
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).
"""
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) -> DataFrame:
"""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).
class Broker(ABC):
"""Return the data"""
Returns:
DataFrame: Stock valuated float.
For each stock give -1 if realy bad and +1 if realy good.
"""
@abstractmethod
def __bool__(self):
"""Return True if the broker has data to retrive"""
@abstractmethod
def next(self) -> pd.DataFrame:
"""Return the next data to process"""
class Strategy(ABC):
""" 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. """
self.logger = logging.getLogger(self.__class__.__name__)
self.indicators = indicators
def run(self, data: DataFrame) -> 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).
class Predictor(ABC):
"""What is the future ?"""
Returns:
List[Order]: A list of orders to execute.
"""
indicators_results: DataFrame = ...
return self.execute(data, indicators_results)
@abstractmethod
def predict(self, data: pd.DataFrame) -> dict:
"""Return the prediction"""
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.
Returns:
List[Order]: A list of orders to execute.
"""
class PTF(ABC):
""" Somethink that buy or sell stocks."""
executors: Dict[object, Callable[["PTF", Order], None]] = {}
history: List[Order]
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 history ? Defaults to True.
"""
self.logger = logging.getLogger(self.__class__.__name__)
self.history = []
self.skip_errors = skip_errors
self.save_errors = save_errors
@abstractproperty
def balance(self) -> float:
""" Return the current total balance. """
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.warn("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.
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)}")

47
auto_trading/main.py

@ -2,39 +2,44 @@
One script to rule them all, One script to find them,
One script to bring them all and in the darkness bind them
"""
from .interfaces import Portfolio, Strategy, Broker, Predictor
import logging
from tqdm import tqdm
from typing import List
from pandas import DataFrame
from .interfaces import Order, CandlesProperties, DataBroker, Indicator, Strategy, PTF
from .errors import OrderException
class Bot:
"""the main class"""
"""The class that wrap all the classes together.
For more information, read /documentation/Diagramme.svg
"""
def __init__(self, ptf: Portfolio, strategy: Strategy, broker: Broker, predictor: Predictor):
"""initialize the bot"""
def __init__(self, ptf: PTF, strategy: Strategy, broker: DataBroker):
"""Initialize the bot."""
self.logger = logging.getLogger(self.__class__.__name__)
self.ptf = ptf
self.strategy = strategy
self.broker = broker
self.predictor = predictor
self.current_conversion_rate = None
self._balance = 0 # USD$
def run(self):
"""run the bot"""
while self.broker:
self.run_once()
iterator = tqdm(self.broker) if self.logger.level >= logging.DEBUG else self.broker
for data in iterator:
self.run_once(data)
def run_once(self):
def run_once(self, data: DataFrame):
"""run the bot once"""
data = self.broker.next()
self.current_conversion_rate = data.iloc[-1].fillna(0).to_dict()
self.strategy.run(self.ptf, self.predictor.predict(data), self.current_conversion_rate)
orders = self.strategy.run(data)
self.logger.debug("Get %d orders to execute", len(orders))
self.ptf.execute_multiples(orders)
def print_results(self):
"""print the results"""
for k, v in self.ptf.content().items():
print(f"{k}: {v:e}")
def _get_error_msg(self, exception: BaseException):
return getattr(exception, "message", getattr(exception, "msg", str(exception)))
@property
def balance(self):
money = 0
for k, v in self.ptf.content().items():
money += self.current_conversion_rate[k] * v
return money
return self.ptf.balance

30
auto_trading/orders.py

@ -0,0 +1,30 @@
"""Some orders on the market."""
from dataclasses import dataclass
from .interfaces import Order
@dataclass
class Long(Order):
"""Buy stock."""
stock: str
amount: float
price: float
@property
def amount_usd(self) -> float:
"""The amount in $"""
return self.amount * self.price
@dataclass
class Short(Order):
"""Buy stock."""
stock: str
amount: float
price: float
@property
def amount_usd(self) -> float:
"""The amount in $"""
return self.amount * self.price

2
requirements.txt

@ -3,3 +3,5 @@ numpy
pandas
tqdm
plotly
pytest
pytest-cov

5
scripts/install.sh

@ -0,0 +1,5 @@
#!/bin/bash
python3.10 -m venv venv
source venv/bin/activate
python -m pip install -r requirements.txt

5
scripts/test.sh

@ -0,0 +1,5 @@
#!/bin/bash
source venv/bin/activate
python -m pytest $@
coverage xml

68
tests/test_interfaces.py

@ -0,0 +1,68 @@
import pytest
from datetime import datetime
from auto_trading.errors import UnknowOrder
from auto_trading.interfaces import PTF
from auto_trading.orders import Short, Long
long_order = Long(datetime.now(), "BTC", 1, 1)
short_order = Short(datetime.now(), "GOLD", 1, 1)
class _TestPTFLong(PTF):
@property
def balance(self):
return 0
def execute_long(self, order: Long):
assert order == long_order
executors = {Long: execute_long}
class _TestPTFLongShort(PTF):
@property
def balance(self):
return 0
def execute_long(self, order: Long):
assert order == long_order
def execute_short(self, order: Long):
assert order == short_order
executors = {Long: execute_long, Short: execute_short}
class _TestPTFShort(PTF):
@property
def balance(self):
return 0
def execute_short(self, order: Long):
assert order == short_order
executors = {Short: execute_short}
class _TestPTFNone(PTF):
@property
def balance(self):
return 0
@pytest.mark.parametrize("test_class, must_match", [
(_TestPTFLong, [long_order]),
(_TestPTFShort, [short_order]),
(_TestPTFLongShort, [long_order, short_order]),
(_TestPTFNone, [])
])
def test_ptf_execution(test_class, must_match):
inst = test_class()
for o in [short_order, long_order]:
if o in must_match:
inst._execute(o)
else:
with pytest.raises(UnknowOrder):
inst._execute(o)
Loading…
Cancel
Save