Make Feeds objects instead of tuples. (#464)

* Make Feeds objects instead of tuples.
* Add namings for different feed objects.
* Move signal at the end.

pdr_backend/aimodel/aimodel_data_factory.py

@@ -1,14 +1,14 @@
 import sys
 from typing import Tuple
 
-from enforce_typing import enforce_types
 import numpy as np
 import pandas as pd
 import polars as pl
+from enforce_typing import enforce_types
 
 from pdr_backend.ppss.data_pp import DataPP
 from pdr_backend.ppss.data_ss import DataSS
-from pdr_backend.util.mathutil import has_nan, fill_nans
+from pdr_backend.util.mathutil import fill_nans, has_nan
 
 
 @enforce_types
@@ -79,8 +79,7 @@ def create_xy(
         x_df = pd.DataFrame()  # build this up
 
         target_hist_cols = [
-            f"{exch_str}:{pair_str}:{signal_str}"
-            for exch_str, signal_str, pair_str in ss.input_feed_tups
+            f"{feed.exchange}:{feed.pair}:{feed.signal}" for feed in ss.input_feeds
         ]
 
         for hist_col in target_hist_cols:

pdr_backend/aimodel/test/test_aimodel_data_factory.py

@@ -28,13 +28,13 @@ def test_create_xy__0(tmpdir):
     data_pp = DataPP(
         {
             "timeframe": "5m",
-            "predict_feeds": ["binanceus c ETH/USDT"],
+            "predict_feeds": ["binanceus ETH/USDT c"],
             "sim_only": {"test_n": 2},
         }
     )
     data_ss = DataSS(
         {
-            "input_feeds": ["binanceus oc ETH/USDT"],
+            "input_feeds": ["binanceus ETH/USDT oc"],
             "parquet_dir": str(tmpdir),
             "st_timestr": "2023-06-18",  # not used by AimodelDataFactory
             "fin_timestr": "2023-06-21",  # ""
@@ -82,7 +82,7 @@ def test_create_xy__0(tmpdir):
 
 @enforce_types
 def test_create_xy__1exchange_1coin_1signal(tmpdir):
-    _, ss, _, aimodel_data_factory = _data_pp_ss_1feed(tmpdir, "binanceus h ETH/USDT")
+    _, ss, _, aimodel_data_factory = _data_pp_ss_1feed(tmpdir, "binanceus ETH/USDT h")
     mergedohlcv_df = merge_rawohlcv_dfs(ETHUSDT_RAWOHLCV_DFS)
 
     # =========== have testshift = 0
@@ -212,10 +212,10 @@ def test_create_xy__2exchanges_2coins_2signals(tmpdir):
         },
     }
 
-    pp = _data_pp(["binanceus h ETH/USDT"])
+    pp = _data_pp(["binanceus ETH/USDT h"])
     ss = _data_ss(
         parquet_dir,
-        ["binanceus hl BTC/USDT,ETH/USDT", "kraken hl BTC/USDT,ETH/USDT"],
+        ["binanceus BTC/USDT,ETH/USDT hl", "kraken BTC/USDT,ETH/USDT hl"],
     )
     assert ss.autoregressive_n == 3
     assert ss.n == (4 + 4) * 3
@@ -319,7 +319,7 @@ def test_create_xy__input_type(tmpdir):
 @enforce_types
 def test_create_xy__handle_nan(tmpdir):
     # create mergedohlcv_df
-    _, _, _, aimodel_data_factory = _data_pp_ss_1feed(tmpdir, "binanceus h ETH/USDT")
+    _, _, _, aimodel_data_factory = _data_pp_ss_1feed(tmpdir, "binanceus ETH/USDT h")
     mergedohlcv_df = merge_rawohlcv_dfs(ETHUSDT_RAWOHLCV_DFS)
 
     # initial mergedohlcv_df should be ok

pdr_backend/analytics/test/test_check_network.py

@@ -98,7 +98,7 @@ def test_check_network_main(  # pylint: disable=unused-argument
     monkeypatch,
 ):
     del_network_override(monkeypatch)
-    ppss = mock_ppss("5m", ["binance c BTC/USDT"], "sapphire-mainnet", str(tmpdir))
+    ppss = mock_ppss("5m", ["binance BTC/USDT c"], "sapphire-mainnet", str(tmpdir))
 
     mock_get_opf_addresses.return_value = {
         "dfbuyer": "0xdfBuyerAddress",

pdr_backend/analytics/test/test_get_predictions_info.py

@@ -15,7 +15,7 @@ def test_get_predictions_info_main_mainnet(
     monkeypatch,
 ):
     del_network_override(monkeypatch)
-    ppss = mock_ppss("5m", ["binance c BTC/USDT"], "sapphire-mainnet", str(tmpdir))
+    ppss = mock_ppss("5m", ["binance BTC/USDT c"], "sapphire-mainnet", str(tmpdir))
 
     mock_getids = Mock(return_value=["0x123", "0x234"])
     mock_fetch = Mock(return_value=_sample_first_predictions)

pdr_backend/analytics/test/test_get_predictoors_info.py

@@ -11,7 +11,7 @@
 @enforce_types
 def test_get_predictoors_info_main_mainnet(tmpdir, monkeypatch):
     del_network_override(monkeypatch)
-    ppss = mock_ppss("5m", ["binance c BTC/USDT"], "sapphire-mainnet", str(tmpdir))
+    ppss = mock_ppss("5m", ["binance BTC/USDT c"], "sapphire-mainnet", str(tmpdir))
 
     mock_fetch = Mock(return_value=[])
     mock_save = Mock()

pdr_backend/analytics/test/test_get_traction_info.py

@@ -17,7 +17,7 @@ def test_get_traction_info_main_mainnet(
     monkeypatch,
 ):
     del_network_override(monkeypatch)
-    ppss = mock_ppss("5m", ["binance c BTC/USDT"], "sapphire-mainnet", str(tmpdir))
+    ppss = mock_ppss("5m", ["binance BTC/USDT c"], "sapphire-mainnet", str(tmpdir))
 
     mock_traction_stat = Mock()
     mock_plot_cumsum = Mock()

pdr_backend/lake/ohlcv_data_factory.py

@@ -1,29 +1,25 @@
 import os
 from typing import Dict
 
-from enforce_typing import enforce_types
 import numpy as np
 import polars as pl
+from enforce_typing import enforce_types
 
-from pdr_backend.lake.constants import (
-    OHLCV_MULT_MIN,
-    OHLCV_MULT_MAX,
-    TOHLCV_SCHEMA_PL,
-)
+from pdr_backend.lake.constants import OHLCV_MULT_MAX, OHLCV_MULT_MIN, TOHLCV_SCHEMA_PL
 from pdr_backend.lake.fetch_ohlcv import safe_fetch_ohlcv
 from pdr_backend.lake.merge_df import merge_rawohlcv_dfs
 from pdr_backend.lake.plutil import (
-    initialize_rawohlcv_df,
     concat_next_df,
-    load_rawohlcv_file,
-    save_rawohlcv_file,
     has_data,
-    oldest_ut,
+    initialize_rawohlcv_df,
+    load_rawohlcv_file,
     newest_ut,
+    oldest_ut,
+    save_rawohlcv_file,
 )
 from pdr_backend.ppss.data_pp import DataPP
 from pdr_backend.ppss.data_ss import DataSS
-from pdr_backend.util.timeutil import pretty_timestr, current_ut
+from pdr_backend.util.timeutil import current_ut, pretty_timestr
 
 
 @enforce_types
@@ -217,9 +213,9 @@ def _load_rawohlcv_files(self, fin_ut: int) -> Dict[str, Dict[str, pl.DataFrame]
             assert "/" in pair_str, f"pair_str={pair_str} needs '/'"
             filename = self._rawohlcv_filename(exch_str, pair_str)
             cols = [
-                signal_str  # cols is a subset of TOHLCV_COLS
-                for e, signal_str, p in self.ss.input_feed_tups
-                if e == exch_str and p == pair_str
+                feed.signal  # cols is a subset of TOHLCV_COLS
+                for feed in self.ss.input_feeds
+                if feed.exchange == exch_str and feed.pair == pair_str
             ]
             rawohlcv_df = load_rawohlcv_file(filename, cols, st_ut, fin_ut)
 

pdr_backend/lake/test/resources.py

@@ -21,7 +21,7 @@
 
 @enforce_types
 def _mergedohlcv_df_ETHUSDT(tmpdir):
-    _, _, _, aimodel_data_factory = _data_pp_ss_1feed(tmpdir, "binanceus h ETH/USDT")
+    _, _, _, aimodel_data_factory = _data_pp_ss_1feed(tmpdir, "binanceus ETH/USDT h")
     mergedohlcv_df = merge_rawohlcv_dfs(ETHUSDT_RAWOHLCV_DFS)
     return mergedohlcv_df, aimodel_data_factory
 

pdr_backend/lake/test/test_gql_data_factory.py

@@ -123,7 +123,7 @@ def _test_update_gql(
 
     _, gql_data_factory = _gql_data_factory(
         tmpdir,
-        "binanceus h ETH/USDT",
+        "binanceus ETH/USDT h",
         st_timestr,
         fin_timestr,
     )
@@ -209,7 +209,7 @@ def test_load_and_verify_schema(
 
     _, gql_data_factory = _gql_data_factory(
         tmpdir,
-        "binanceus h ETH/USDT",
+        "binanceus ETH/USDT h",
         st_timestr,
         fin_timestr,
     )
@@ -248,7 +248,7 @@ def test_get_gql_dfs_calls(
 
     _, gql_data_factory = _gql_data_factory(
         tmpdir,
-        "binanceus h ETH/USDT",
+        "binanceus ETH/USDT h",
         st_timestr,
         fin_timestr,
     )

pdr_backend/lake/test/test_ohlcv_data_factory.py

@@ -93,7 +93,7 @@ def fetch_ohlcv(self, since, limit, *args, **kwargs) -> list:
 
     _, ss, factory, _ = _data_pp_ss_1feed(
         tmpdir,
-        "binanceus h ETH/USDT",
+        "binanceus ETH/USDT h",
         st_timestr,
         fin_timestr,
     )
@@ -180,10 +180,10 @@ def test_get_mergedohlcv_df_happypath(tmpdir):
 def _test_get_mergedohlcv_df_happypath(tmpdir):
     parquet_dir = str(tmpdir)
 
-    pp = _data_pp(["binanceus h BTC/USDT"])
+    pp = _data_pp(["binanceus BTC/USDT h"])
     ss = _data_ss(
         parquet_dir,
-        ["binanceus h BTC-USDT,ETH/USDT", "kraken h BTC/USDT"],
+        ["binanceus BTC-USDT,ETH/USDT h", "kraken BTC/USDT h"],
         st_timestr="2023-06-18",
         fin_timestr="2023-06-19",
     )
@@ -225,7 +225,7 @@ def _test_mergedohlcv_df__low_vs_high_level(tmpdir, ohlcv_val):
     """
 
     # setup
-    _, _, factory, _ = _data_pp_ss_1feed(tmpdir, "binanceus h BTC/USDT")
+    _, _, factory, _ = _data_pp_ss_1feed(tmpdir, "binanceus BTC/USDT h")
     filename = factory._rawohlcv_filename("binanceus", "BTC/USDT")
     st_ut = factory.ss.st_timestamp
     fin_ut = factory.ss.fin_timestamp
@@ -284,7 +284,7 @@ def test_exchange_hist_overlap(tmpdir):
     """DataFactory get_mergedohlcv_df() and concat is executing e2e correctly"""
     _, _, factory, _ = _data_pp_ss_1feed(
         tmpdir,
-        "binanceus h ETH/USDT",
+        "binanceus ETH/USDT h",
         st_timestr="2023-06-18",
         fin_timestr="2023-06-19",
     )
@@ -303,7 +303,7 @@ def test_exchange_hist_overlap(tmpdir):
     # let's get more data from exchange with overlap
     _, _, factory2, _ = _data_pp_ss_1feed(
         tmpdir,
-        "binanceus h ETH/USDT",
+        "binanceus ETH/USDT h",
         st_timestr="2023-06-18",  # same
         fin_timestr="2023-06-20",  # different
     )
@@ -326,7 +326,7 @@ def test_get_mergedohlcv_df_calls(
     tmpdir,
 ):
     mock_merge_rawohlcv_dfs.return_value = Mock(spec=pl.DataFrame)
-    _, _, factory, _ = _data_pp_ss_1feed(tmpdir, "binanceus h ETH/USDT")
+    _, _, factory, _ = _data_pp_ss_1feed(tmpdir, "binanceus ETH/USDT h")
 
     factory._update_rawohlcv_files = Mock(return_value=None)
     factory._load_rawohlcv_files = Mock(return_value=None)

pdr_backend/models/feed.py

@@ -8,7 +8,7 @@
 from pdr_backend.util.timeframestr import Timeframe
 
 
-class Feed(StrMixin):  # pylint: disable=too-many-instance-attributes
+class SubgraphFeed(StrMixin):  # pylint: disable=too-many-instance-attributes
     @enforce_types
     def __init__(
         self,
@@ -54,7 +54,7 @@ def __str__(self) -> str:
 
 
 @enforce_types
-def print_feeds(feeds: Dict[str, Feed], label: Optional[str] = None):
+def print_feeds(feeds: Dict[str, SubgraphFeed], label: Optional[str] = None):
     label = label or "feeds"
     print(f"{len(feeds)} {label}:")
     if not feeds:
@@ -76,10 +76,10 @@ def _rnd_eth_addr() -> str:
 
 
 @enforce_types
-def mock_feed(timeframe_str: str, exchange_str: str, pair_str: str) -> Feed:
+def mock_feed(timeframe_str: str, exchange_str: str, pair_str: str) -> SubgraphFeed:
     addr = _rnd_eth_addr()
     name = f"Feed {addr} {pair_str}|{exchange_str}|{timeframe_str}"
-    feed = Feed(
+    feed = SubgraphFeed(
         name=name,
         address=addr,
         symbol=f"SYM: {addr}",

pdr_backend/models/slot.py

@@ -1,7 +1,7 @@
-from pdr_backend.models.feed import Feed
+from pdr_backend.models.feed import SubgraphFeed
 
 
 class Slot:
-    def __init__(self, slot_number: int, feed: Feed):
+    def __init__(self, slot_number: int, feed: SubgraphFeed):
         self.slot_number = slot_number
         self.feed = feed

pdr_backend/models/test/test_feed.py

@@ -1,11 +1,11 @@
 from enforce_typing import enforce_types
 
-from pdr_backend.models.feed import Feed, mock_feed, print_feeds
+from pdr_backend.models.feed import SubgraphFeed, mock_feed, print_feeds
 
 
 @enforce_types
 def test_feed():
-    feed = Feed(
+    feed = SubgraphFeed(
         "Contract Name",
         "0x12345",
         "SYM:TEST",

pdr_backend/models/test/test_slot.py

@@ -1,9 +1,9 @@
 from pdr_backend.models.slot import Slot
-from pdr_backend.models.feed import Feed
+from pdr_backend.models.feed import SubgraphFeed
 
 
 def test_slot_initialization():
-    feed = Feed(
+    feed = SubgraphFeed(
         "Contract Name",
         "0x12345",
         "test",
@@ -20,4 +20,4 @@ def test_slot_initialization():
 
     assert slot.slot_number == slot_number
     assert slot.feed == feed
-    assert isinstance(slot.feed, Feed)
+    assert isinstance(slot.feed, SubgraphFeed)