Refactor to new data providers

Author: MDUYN

examples/bitvavo_trading_bot.py

@@ -2,8 +2,7 @@
 import logging.config
 
 from investing_algorithm_framework import TimeUnit, TradingStrategy, \
-    CCXTOHLCVMarketDataSource, CCXTTickerMarketDataSource, \
-    create_app, DEFAULT_LOGGING_CONFIG, Context
+    create_app, DEFAULT_LOGGING_CONFIG, Context, DataSource
 
 """
 Bitvavo trading bot example with market data sources of bitvavo.
@@ -19,24 +18,14 @@
 # Load the environment variables from the .env file
 load_dotenv()
 
-# Define your market data sources for coinbase
-bitvavo_btc_eur_ohlcv_2h = CCXTOHLCVMarketDataSource(
-    identifier="BTC/EUR-ohlcv",
-    market="bitvavo",
-    symbol="BTC/EUR",
-    time_frame="2h",
-    window_size=200
-)
-bitvavo_btc_eur_ticker = CCXTTickerMarketDataSource(
-    identifier="BTC/EUR-ticker",
-    market="bitvavo",
-    symbol="BTC/EUR",
-)
-
+# Define your bitvavo trading strategy and register the data sources
 class BitvavoTradingStrategy(TradingStrategy):
     time_unit = TimeUnit.SECOND
     interval = 10
-    market_data_sources = [bitvavo_btc_eur_ohlcv_2h, bitvavo_btc_eur_ticker]
+    data_sources = [
+        DataSource(data_type="OHLCV", market="bitvavo", symbol="BTC/EUR", window_size=200, time_frame="2h", indentifier="BTC/EUR-ohlcv"),
+        DataSource(data_type="Ticker", market="bitvavo", symbol="BTC/EUR", indentifier="BTC/EUR-ticker")
+    ]
 
     def run_strategy(self, context: Context, market_data):
         print(market_data["BTC/EUR-ohlcv"])
@@ -45,6 +34,9 @@ def run_strategy(self, context: Context, market_data):
 # Create an app and add the market data sources and market credentials to it
 app = create_app()
 app.add_strategy(BitvavoTradingStrategy)
+
+# Market credentials for bitvavo 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", initial_balance=400)
 
 if __name__ == "__main__":

examples/coinbase_trading_bot.py

@@ -1,42 +1,26 @@
 from dotenv import load_dotenv
 import logging.config
 from investing_algorithm_framework import TimeUnit, \
-    CCXTOHLCVMarketDataSource, CCXTTickerMarketDataSource, TradingStrategy, \
-    create_app, DEFAULT_LOGGING_CONFIG, Context
+    DataSource, TradingStrategy, create_app, DEFAULT_LOGGING_CONFIG, Context
 """
-Coinbase market data sources example. Coinbase requires you to have an API key
+Coinbase trading bot example. Coinbase requires you to have an API key
 and secret key to access their market data. You can create them here:
 https://www.coinbase.com/settings/api
-
-You need to add a market credential to the app, and then add market
-data sources to the app. You can then use the market data
-sources in your trading strategy.
 """
 
 logging.config.dictConfig(DEFAULT_LOGGING_CONFIG)
 
 # Load the environment variables from the .env file
 load_dotenv()
 
-# Define your market data sources for coinbase
-coinbase_btc_eur_ohlcv_2h = CCXTOHLCVMarketDataSource(
-    identifier="BTC/EUR-ohlcv",
-    market="coinbase",
-    symbol="BTC/EUR",
-    time_frame="2h",
-    window_size=200
-)
-coinbase_btc_eur_ticker = CCXTTickerMarketDataSource(
-    identifier="BTC/EUR-ticker",
-    market="coinbase",
-    symbol="BTC/EUR",
-)
-
-
+# Define your coinbase trading strategy and register the data sources
 class CoinbaseTradingStrategy(TradingStrategy):
     time_unit = TimeUnit.SECOND
     interval = 10
-    market_data_sources = [coinbase_btc_eur_ohlcv_2h, coinbase_btc_eur_ticker]
+    data_sources = [
+        DataSource(data_type="OHLCV", market="coinbase", symbol="BTC/EUR", window_size=200, time_frame="2h", indentifier="BTC/EUR-ohlcv"),
+        DataSource(data_type="Ticker", market="coinbase", symbol="BTC/EUR", indentifier="BTC/EUR-ticker")
+    ]
 
     def apply_strategy(self, context: Context, market_data):
         print(market_data["BTC/EUR-ohlcv"])
@@ -45,6 +29,9 @@ def apply_strategy(self, context: Context, market_data):
 # Create an app and configure it with coinbase
 app = create_app()
 app.add_strategy(CoinbaseTradingStrategy)
+
+# 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="coinbase", trading_symbol="EUR", initial_balance=400)
 
 if __name__ == "__main__":

investing_algorithm_framework/__init__.py

@@ -21,7 +21,7 @@
     CCXTTickerMarketDataSource, CSVOHLCVMarketDataSource, \
     CSVTickerMarketDataSource, AzureBlobStorageStateHandler, \
     PandasOHLCVBacktestMarketDataSource, PandasOHLCVMarketDataSource, \
-    AWSS3StorageStateHandler
+    AWSS3StorageStateHandler, CSVOHLCVDataProvider, CCXTOHLCVDataProvider
 from .create_app import create_app
 from .download_data import download
 
@@ -92,5 +92,7 @@
     "AWS_S3_STATE_BUCKET_NAME",
     "AWS_LAMBDA_LOGGING_CONFIG",
     'select_backtest_date_ranges',
-    'DataType'
+    'DataType',
+    'CSVOHLCVDataProvider',
+    "CCXTOHLCVDataProvider"
 ]

investing_algorithm_framework/app/strategy.py

@@ -5,7 +5,7 @@
 from investing_algorithm_framework.domain import OperationalException, Position
 from investing_algorithm_framework.domain import \
     TimeUnit, StrategyProfile, Trade, ENVIRONMENT, Environment, \
-    BACKTESTING_INDEX_DATETIME
+    BACKTESTING_INDEX_DATETIME, DataSource
 from .context import Context
 
 
@@ -23,17 +23,17 @@ class TradingStrategy:
         worker_id (optional): str - the id of the worker
         strategy_id (optional): str - the id of the strategy
         decorated (optional): function - the decorated function
-        data_sources (List[DataSource] optional):the list of data
+        data_sources (List[DataSource] optional): the list of data
             sources to use for the strategy. The data sources will be used
-            to indetify data providers that will be called to gather data
+            to indentify data providers that will be called to gather data
             and pass to the strategy before its run.
     """
     time_unit: str = None
     interval: int = None
     worker_id: str = None
     strategy_id: str = None
     decorated = None
-    data_sources = None
+    data_sources: List[DataSource] = None
     traces = None
     context: Context = None