Standardize signal generation

Author: MDUYN

examples/simple_trading_bot_example.py

@@ -0,0 +1,253 @@
+from typing import Dict, Any
+from datetime import datetime, timezone
+
+import pandas as pd
+from pyindicators import ema, rsi, crossover, crossunder
+
+from investing_algorithm_framework import TradingStrategy, DataSource, \
+    TimeUnit, DataType, PositionSize, create_app, RESOURCE_DIRECTORY, \
+    BacktestDateRange, BacktestReport
+
+
+class RSIEMACrossoverStrategy(TradingStrategy):
+    time_unit = TimeUnit.HOUR
+    interval = 2
+    symbols = ["BTC"]
+    position_sizes = [
+        PositionSize(
+            symbol="BTC", percentage_of_portfolio=20.0
+        ),
+        PositionSize(
+            symbol="ETH", percentage_of_portfolio=20.0
+        )
+    ]
+
+    def __init__(
+        self,
+        time_unit: TimeUnit,
+        interval: int,
+        market: str,
+        rsi_time_frame: str,
+        rsi_period: int,
+        rsi_overbought_threshold,
+        rsi_oversold_threshold,
+        ema_time_frame,
+        ema_short_period,
+        ema_long_period,
+        ema_cross_lookback_window: int = 10
+    ):
+        self.rsi_time_frame = rsi_time_frame
+        self.rsi_period = rsi_period
+        self.rsi_result_column = f"rsi_{self.rsi_period}"
+        self.rsi_overbought_threshold = rsi_overbought_threshold
+        self.rsi_oversold_threshold = rsi_oversold_threshold
+        self.ema_time_frame = ema_time_frame
+        self.ema_short_result_column = f"ema_{ema_short_period}"
+        self.ema_long_result_column = f"ema_{ema_long_period}"
+        self.ema_crossunder_result_column = "ema_crossunder"
+        self.ema_crossover_result_column = "ema_crossover"
+        self.ema_short_period = ema_short_period
+        self.ema_long_period = ema_long_period
+        self.ema_cross_lookback_window = ema_cross_lookback_window
+        data_sources = []
+
+        for symbol in self.symbols:
+            full_symbol = f"{symbol}/EUR"
+            data_sources.append(
+                DataSource(
+                    identifier=f"{symbol}_rsi_data",
+                    data_type=DataType.OHLCV,
+                    time_frame=self.rsi_time_frame,
+                    market=market,
+                    symbol=full_symbol,
+                    pandas=True,
+                    window_size=800
+                )
+            )
+            data_sources.append(
+                DataSource(
+                    identifier=f"{symbol}_ema_data",
+                    data_type=DataType.OHLCV,
+                    time_frame=self.ema_time_frame,
+                    market=market,
+                    symbol=full_symbol,
+                    pandas=True,
+                    window_size=800
+                )
+            )
+
+        super().__init__(
+            data_sources=data_sources, time_unit=time_unit, interval=interval
+        )
+
+        self.buy_signal_dates = {}
+        self.sell_signal_dates = {}
+
+        for symbol in self.symbols:
+            self.buy_signal_dates[symbol] = []
+            self.sell_signal_dates[symbol] = []
+
+    def _prepare_indicators(
+        self,
+        rsi_data,
+        ema_data
+    ):
+        ema_data = ema(
+            ema_data,
+            period=self.ema_short_period,
+            source_column="Close",
+            result_column=self.ema_short_result_column
+        )
+        ema_data = ema(
+            ema_data,
+            period=self.ema_long_period,
+            source_column="Close",
+            result_column=self.ema_long_result_column
+        )
+        # Detect crossover (short EMA crosses above long EMA)
+        ema_data = crossover(
+            ema_data,
+            first_column=self.ema_short_result_column,
+            second_column=self.ema_long_result_column,
+            result_column=self.ema_crossover_result_column
+        )
+        # Detect crossunder (short EMA crosses below long EMA)
+        ema_data = crossunder(
+            ema_data,
+            first_column=self.ema_short_result_column,
+            second_column=self.ema_long_result_column,
+            result_column=self.ema_crossunder_result_column
+        )
+        rsi_data = rsi(
+            rsi_data,
+            period=self.rsi_period,
+            source_column="Close",
+            result_column=self.rsi_result_column
+        )
+
+        return ema_data, rsi_data
+
+    def generate_buy_signals(self, data: Dict[str, Any]) -> Dict[str, pd.Series]:
+        """
+        Generate buy signals based on the moving average crossover.
+
+        data (Dict[str, Any]): Dictionary containing all the data for
+            the strategy data sources.
+
+        Returns:
+            Dict[str, pd.Series]: A dictionary where keys are symbols and values
+                are pandas Series indicating buy signals (True/False).
+        """
+
+        signals = {}
+
+        for symbol in self.symbols:
+            ema_data_identifier = f"{symbol}_ema_data"
+            rsi_data_identifier = f"{symbol}_rsi_data"
+            ema_data, rsi_data = self._prepare_indicators(
+                data[ema_data_identifier].copy(),
+                data[rsi_data_identifier].copy()
+            )
+
+            # crossover confirmed
+            ema_crossover_lookback = ema_data[
+                self.ema_crossover_result_column].rolling(
+                window=self.ema_cross_lookback_window
+            ).max().astype(bool)
+
+            # use only RSI column
+            rsi_oversold = rsi_data[self.rsi_result_column] \
+                < self.rsi_oversold_threshold
+
+            buy_signal = rsi_oversold & ema_crossover_lookback
+            buy_signals = buy_signal.fillna(False).astype(bool)
+            signals[symbol] = buy_signals
+
+            # Get all dates where there is a sell signal
+            buy_signal_dates = buy_signals[buy_signals].index.tolist()
+
+            if buy_signal_dates:
+                self.buy_signal_dates[symbol] += buy_signal_dates
+
+        return signals
+
+    def generate_sell_signals(self, data: Dict[str, Any]) -> Dict[str, pd.Series]:
+        """
+        Generate sell signals based on the moving average crossover.
+
+        Args:
+            data (Dict[str, Any]): Dictionary containing all the data for
+                the strategy data sources.
+
+        Returns:
+            Dict[str, pd.Series]: A dictionary where keys are symbols and values
+                are pandas Series indicating sell signals (True/False).
+        """
+
+        signals = {}
+        for symbol in self.symbols:
+            ema_data_identifier = f"{symbol}_ema_data"
+            rsi_data_identifier = f"{symbol}_rsi_data"
+            ema_data, rsi_data = self._prepare_indicators(
+                data[ema_data_identifier].copy(),
+                data[rsi_data_identifier].copy()
+            )
+
+            # Confirmed by crossover between short-term EMA and long-term EMA
+            # within a given lookback window
+            ema_crossunder_lookback = ema_data[
+                self.ema_crossunder_result_column].rolling(
+                window=self.ema_cross_lookback_window
+            ).max().astype(bool)
+
+            # use only RSI column
+            rsi_overbought = rsi_data[self.rsi_result_column] \
+               >= self.rsi_overbought_threshold
+
+            # Combine both conditions
+            sell_signal = rsi_overbought & ema_crossunder_lookback
+            sell_signal = sell_signal.fillna(False).astype(bool)
+            signals[symbol] = sell_signal
+
+            # Get all dates where there is a sell signal
+            sell_signal_dates = sell_signal[sell_signal].index.tolist()
+
+            if sell_signal_dates:
+                self.sell_signal_dates[symbol] += sell_signal_dates
+
+        return signals
+
+
+if __name__ == "__main__":
+    app = create_app()
+    app.add_strategy(
+        RSIEMACrossoverStrategy(
+            time_unit=TimeUnit.HOUR,
+            interval=2,
+            market="bitvavo",
+            rsi_time_frame="2h",
+            rsi_period=14,
+            rsi_overbought_threshold=70,
+            rsi_oversold_threshold=30,
+            ema_time_frame="2h",
+            ema_short_period=12,
+            ema_long_period=26,
+            ema_cross_lookback_window=10
+        )
+    )
+
+    # Market credentials for coinbase for both the portfolio connection and data sources will
+    # be read from .env file, when not registering a market credential object in the app.
+    app.add_market(
+        market="bitvavo",
+        trading_symbol="EUR",
+    )
+    backtest_range = BacktestDateRange(
+        start_date=datetime(2023, 1, 1, tzinfo=timezone.utc),
+        end_date=datetime(2024, 6, 1, tzinfo=timezone.utc)
+    )
+    backtest = app.run_backtest(
+        backtest_date_range=backtest_range, initial_amount=1000
+    )
+    report = BacktestReport(backtest)
+    report.show(backtest_date_range=backtest_range, browser=True)

investing_algorithm_framework/__init__.py

@@ -8,7 +8,7 @@
     get_monthly_returns_heatmap_chart, create_weights, \
     get_yearly_returns_bar_chart, get_entry_and_exit_signals, \
     get_ohlcv_data_completeness_chart
-from .domain import ApiException, combine_backtests, \
+from .domain import ApiException, combine_backtests, PositionSize, \
     OrderType, OperationalException, OrderStatus, OrderSide, \
     TimeUnit, TimeInterval, Order, Portfolio, Backtest, \
     Position, TimeFrame, INDEX_DATETIME, MarketCredential, \
@@ -165,4 +165,5 @@
     "get_growth_percentage",
     "BacktestEvaluationFocus",
     "combine_backtests",
+    "PositionSize"
 ]

investing_algorithm_framework/app/app.py

@@ -20,7 +20,7 @@
     PortfolioConfiguration, SnapshotInterval, DataType, \
     PortfolioProvider, OrderExecutor, ImproperlyConfigured, \
     DataProvider, INDEX_DATETIME, tqdm, BacktestPermutationTest, \
-    LAST_SNAPSHOT_DATETIME
+    LAST_SNAPSHOT_DATETIME, BACKTESTING_FLAG
 from investing_algorithm_framework.infrastructure import setup_sqlalchemy, \
     create_all_tables, CCXTOrderExecutor, CCXTPortfolioProvider, \
     BacktestOrderExecutor, CCXTOHLCVDataProvider, clear_db, \
@@ -312,7 +312,8 @@ def initialize_backtest_config(
             ),
             BACKTESTING_INITIAL_AMOUNT: initial_amount,
             INDEX_DATETIME: backtest_date_range.start_date,
-            LAST_SNAPSHOT_DATETIME: None
+            LAST_SNAPSHOT_DATETIME: None,
+            BACKTESTING_FLAG: True
         }
         configuration_service = self.container.configuration_service()
         configuration_service.initialize_from_dict(data)
@@ -451,12 +452,15 @@ def initialize_data_sources_backtest(
         if not show_progress:
             for _, data_provider in data_providers:
                 data_provider.prepare_backtest_data(
-                    backtest_date_range.start_date,
-                    backtest_date_range.end_date
+                    backtest_start_date=backtest_date_range.start_date,
+                    backtest_end_date=backtest_date_range.end_date
                 )
         else:
             for _, data_provider in \
-                    tqdm(data_providers, desc=description, colour="green"):
+                    tqdm(
+                        data_providers, desc=description, colour="green"
+                    ):
+
                 data_provider.prepare_backtest_data(
                     backtest_start_date=backtest_date_range.start_date,
                     backtest_end_date=backtest_date_range.end_date

investing_algorithm_framework/app/context.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import datetime, timezone
 from typing import List
 
 from investing_algorithm_framework.services import ConfigurationService, \
@@ -420,9 +421,52 @@ def get_portfolio(self, market=None) -> Portfolio:
         """
 
         if market is None:
-            return self.portfolio_service.get_all()[0]
+            portfolio = self.portfolio_service.get_all()[0]
+        else:
+            portfolio = self.portfolio_service.find({"market": market})
+
+        # Retrieve positions
+        positions = self.position_service.get_all(
+            {"portfolio": portfolio.id}
+        )
+
+        if BACKTESTING_FLAG in self.configuration_service.config \
+                and self.configuration_service.config[BACKTESTING_FLAG]:
+            date = self.configuration_service.config[INDEX_DATETIME]
+        else:
+            date = datetime.now(tz=timezone.utc)
+
+        allocated = 0.0
+
+        for position in positions:
+
+            if position.symbol != portfolio.trading_symbol:
+                ticker = self.data_provider_service.get_ticker_data(
+                    symbol=f"{position.symbol}/{portfolio.trading_symbol}",
+                    market=portfolio.market,
+                    date=date
+                )
+                if ticker is not None and "bid" in ticker:
+                    allocated += position.get_amount() * ticker["bid"]
+
+        portfolio.allocated = allocated
+        return portfolio
+
+    def get_latest_price(self, symbol, market=None):
+
+        if BACKTESTING_FLAG in self.configuration_service.config \
+                and self.configuration_service.config[BACKTESTING_FLAG]:
+            date = self.configuration_service.config[INDEX_DATETIME]
+        else:
+            date = datetime.now(tz=timezone.utc)
+
+        ticker = self.data_provider_service.get_ticker_data(
+            symbol=symbol,
+            market=market,
+            date=date
+        )
 
-        return self.portfolio_service.find({"market": market})
+        return ticker['bid'] if ticker and 'bid' in ticker else None
 
     def get_portfolios(self):
         """
@@ -1597,3 +1641,27 @@ def get_market_credentials(self) -> List[MarketCredential]:
             List[MarketCredential]: A list of all market credentials
         """
         return self.market_credential_service.get_all()
+
+    def get_trading_symbol(self, portfolio_id=None):
+        """
+        Function to get the trading symbol of a portfolio. If the
+        portfolio_id parameter is specified, the function will return
+        the trading symbol of the portfolio with the specified id.
+
+        Args:
+            portfolio_id: The id of the portfolio to get the trading symbol for
+
+        Returns:
+            str: The trading symbol of the portfolio
+        """
+        if portfolio_id is None:
+            if self.portfolio_service.count() > 1:
+                raise OperationalException(
+                    "Multiple portfolios found. Please specify a "
+                    "portfolio identifier."
+                )
+            portfolio = self.portfolio_service.get_all()[0]
+        else:
+            portfolio = self.portfolio_service.get(portfolio_id)
+
+        return portfolio.trading_symbol