Improved naming

roboquant/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.7.0"
+__version__ = "0.7.1"
 
 import logging
 

roboquant/feeds/__init__.py

@@ -1,13 +1,13 @@
-from .aggregate import AggregatorFeed
-from .collect import CollectorFeed
+from .util import AggregatorFeed, CollectorFeed
 from .csvfeed import CSVFeed
 from .eventchannel import EventChannel
 from .feed import Feed
 from .historic import HistoricFeed
 from .randomwalk import RandomWalk
-from .sqllitefeed import SQLFeed
-from .parquetfeed import ParquetFeed
-from .avrofeed import AvroFeed
+from .sql import SQLFeed
+
+# from .parquetfeed import ParquetFeed
+# from .avrofeed import AvroFeed
 
 try:
     from .yahoo import YahooFeed

roboquant/feeds/aggregate.py → roboquant/feeds/util.py

@@ -86,3 +86,32 @@ def play(self, channel: EventChannel):
             items = list(bars.values())
             evt = Event(next_time, items)
             channel.put(evt)
+
+
+class CollectorFeed(Feed):
+    """Collect events that occur close after ech other into a single new event.
+
+    Close to eachother is defined by the timeout in seconds. If there is no new
+    event in the specified timeout, all the previously collected event items will
+    be bundled together and put on the channel as a single event.
+    """
+
+    def __init__(
+        self,
+        feed: Feed,
+        timeout=5.0,
+    ):
+        super().__init__()
+        self.feed = feed
+        self.timeout = timeout
+
+    def play(self, channel: EventChannel):
+        src_channel = self.feed.play_background(channel.timeframe, channel.maxsize)
+        items = []
+        while event := src_channel.get(self.timeout):
+            if event.is_empty() and items:
+                new_event = Event(event.time, items)
+                channel.put(new_event)
+                items = []
+
+            items.extend(event.items)

samples/alpaca_record_avro.py

@@ -1,6 +1,6 @@
 # %%
 from roboquant.alpaca import AlpacaHistoricStockFeed
-from roboquant.feeds.avrofeed import AvroFeed
+from roboquant.feeds.avro import AvroFeed
 
 # %%
 print("The retrieval of historical data will take some time....")

samples/parquet_recording.py

@@ -1,11 +1,12 @@
 # %%
 import roboquant as rq
+from roboquant.feeds.parquet import ParquetFeed
 
 # %%
 yahoo_feed = rq.feeds.YahooFeed("JPM", "IBM", "F", "MSFT", "TSLA", start_date="2000-01-01")
 
 # %%
-feed = rq.feeds.ParquetFeed("/tmp/stocks.parquet")
+feed = ParquetFeed("/tmp/stocks.parquet")
 if not feed.exists():
     feed.record(yahoo_feed)
 

samples/sb3_strategy_quotes.py

@@ -5,7 +5,7 @@
 from roboquant.asset import Stock
 from roboquant.ml.features import EquityFeature, QuoteFeature
 from roboquant.ml.rl import TradingEnv, SB3PolicyStrategy
-from roboquant.feeds.parquetfeed import ParquetFeed
+from roboquant.feeds.parquet import ParquetFeed
 from roboquant.timeframe import Timeframe
 
 # %%

tests/unit/test_aggregator.py

@@ -1,7 +1,7 @@
 import unittest
 from datetime import timedelta
 
-from roboquant.feeds.aggregate import AggregatorFeed
+from roboquant.feeds.util import AggregatorFeed
 from roboquant.feeds.randomwalk import RandomWalk
 from tests.common import run_price_item_feed
 

tests/unit/test_avrofeed.py

@@ -2,7 +2,7 @@
 import unittest
 from pathlib import Path
 
-from roboquant.feeds.avrofeed import AvroFeed
+from roboquant.feeds.avro import AvroFeed
 from tests.common import get_feed, run_price_item_feed
 
 

tests/unit/test_parquetfeed.py

@@ -2,7 +2,7 @@
 import unittest
 from pathlib import Path
 
-from roboquant.feeds.parquetfeed import ParquetFeed
+from roboquant.feeds.parquet import ParquetFeed
 from tests.common import get_feed, run_price_item_feed