Merge branch 'feature/dispatch-loop-started-event' into develop

Author: gbeced

.github/copilot-instructions.md

@@ -13,7 +13,8 @@ Key data flows: Bar events → Strategies → Trading signals → Position manag
 ## Development Workflow
 - **Dependencies**: Use Poetry (`poetry install --all-extras`). Python 3.10+ required.
 - **Linting**: `invoke lint` runs mypy and ruff (line length 120, excludes `pocs/`)
-- **Testing**: `invoke test` runs pytest with 100% coverage requirement. Use `--html_report` for coverage HTML.
+- **Testing**: `invoke test` runs lint + pytest with 100% line coverage required.
+- **Single test**: `poetry run pytest tests/test_file.py::test_name -vv --no-cov`
 - **Cleaning**: `invoke clean` removes caches and build artifacts
 - **Documentation**: `invoke build_docs` generates Sphinx docs in `docs/_build/html/`
 
@@ -43,13 +44,17 @@ Key data flows: Bar events → Strategies → Trading signals → Position manag
 - **Live Trading**: Requires API keys. Use `realtime_dispatcher()` instead of `backtesting_dispatcher()`.
 - **Precision Handling**: Use `round_decimal()`/`truncate_decimal()` for display/calculation precision.
 
+## Testing Conventions
+- `asyncio_mode = auto` (setup.cfg) — async test functions run without `@pytest.mark.asyncio`.
+- Pytest fixtures `backtesting_dispatcher` and `realtime_dispatcher` are available globally via `tests/conftest.py`.
+- Shared test pairs/pair-infos live in `tests/common.py` (e.g. `btc_pair`, `btc_pair_info`).
+- 100% line coverage is required; use `# pragma: no cover` only for unreachable/platform-specific branches.
+
 ## Common Pitfalls
 - Naive datetimes cause issues - always use timezone-aware (`dt.utc_now()`).
 - Concurrent position modifications require locks (`asyncio.Lock` per pair).
 - WebSocket reconnections handled automatically, but monitor for gaps.
 - Borrowing disabled by default in samples - set `borrowing_disabled=False` to enable shorts.</content>
 
 ## Permissions
-- Do not run git commit, git add, are create any branches. Let me stage and/or commit manually.
-- Always ask before installing new packages and never do it unless its inside a virtual environment.
-
+- Always ask before installing new packages and do it inside a virtual environment.

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features
 
 * `fees.Percentage` now supports charging fees in base currency.
+* Both `BacktestingDispatcher` and `RealtimeDispatcher` now support `subscribe_event_loop_started` to register handlers that are called once when the dispatch loop starts.
 
 ## 1.10.2
 

basana/core/dispatcher/__init__.py

@@ -19,6 +19,7 @@
 from .base import (
     EventDispatcher,
     EventHandler,
+    LoopStartedHandler,
     SchedulerJob,
 )
 

basana/core/dispatcher/backtesting.py

@@ -53,6 +53,9 @@ def _set_now(self, now: datetime.datetime):
         self._last_dt = now
 
     async def _dispatch_loop(self):
+        self._last_dt = self._event_mux.peek_next_event_dt()
+        await self._notify_loop_started()
+
         while not self.stopped:
             # Time is driven by event timestamps, so "now" advances only when events are processed.
             next_dt = self._event_mux.peek_next_event_dt()

basana/core/dispatcher/base.py

@@ -33,6 +33,7 @@
 
 logger = logging.getLogger(__name__)
 EventHandler = Callable[[core_event.Event], Awaitable[Any]]
+LoopStartedHandler = Callable[[], Awaitable[Any]]
 
 
 # Indices for prefetched event tuples stored in the EventMultiplexer heap.
@@ -114,6 +115,7 @@ def __init__(self, max_concurrent: int):
         self._event_handlers: Dict[core_event.EventSource, List[EventHandler]] = defaultdict(list)
         self._sniffers_pre: List[EventHandler] = []
         self._sniffers_post: List[EventHandler] = []
+        self._loop_started_handlers: List[LoopStartedHandler] = []
         self._producers: Set[core_event.Producer] = set()
         # Task group for core tasks like producers and dispatch loop.
         self._core_tasks: Optional[helpers.TaskGroup] = None
@@ -179,6 +181,17 @@ def subscribe_all(self, event_handler: EventHandler, front_run: bool = False):
         if event_handler not in sniffers:
             sniffers.append(event_handler)
 
+    def subscribe_event_loop_started(self, handler: LoopStartedHandler):
+        """Registers an async callable that will be called once when the dispatch loop starts.
+
+        :param handler: An async callable that receives no arguments.
+        """
+
+        assert not self._running, "Subscribing once we're running is not currently supported."
+
+        if handler not in self._loop_started_handlers:
+            self._loop_started_handlers.append(handler)
+
     def schedule(self, when: datetime.datetime, job: SchedulerJob):
         """Schedules a function to be executed at a given time.
 
@@ -248,6 +261,10 @@ async def _core_task_group(self):
         finally:
             self._core_tasks = None
 
+    async def _notify_loop_started(self):
+        if self._loop_started_handlers:
+            await gather_no_raise(*[handler() for handler in self._loop_started_handlers])
+
     async def _dispatch_event(self, event: core_event.Event, handlers: List[EventHandler]):
         if self._debug_enabled:
             logger.debug(logs.StructuredMessage(

basana/core/dispatcher/realtime.py

@@ -55,6 +55,7 @@ def subscribe_idle(self, idle_handler: IdleHandler):
             self._idle_handlers.append(idle_handler)
 
     async def _dispatch_loop(self):
+        await self._notify_loop_started()
         while not self.stopped:
             now = dt.utc_now()
             # Feed the task pool with scheduled jobs and events that are ready for processing.

tests/test_backtesting_dispatcher.py

@@ -219,6 +219,45 @@ def test_now_fails_if_no_events_were_processed(backtesting_dispatcher):
         backtesting_dispatcher.now()
 
 
+async def test_event_loop_started(backtesting_dispatcher):
+    started = []
+
+    async def on_started():
+        assert backtesting_dispatcher.now_available
+        started.append(True)
+
+    src = event.FifoQueueEventSource(events=[event.Event(dt.utc_now())])
+
+    async def on_event(_event):
+        pass
+
+    backtesting_dispatcher.subscribe(src, on_event)
+    backtesting_dispatcher.subscribe_event_loop_started(on_started)
+    await backtesting_dispatcher.run()
+
+    assert len(started) == 1
+
+
+async def test_duplicate_loop_started_subscription_is_ignored(backtesting_dispatcher):
+    started = []
+
+    async def on_started():
+        assert backtesting_dispatcher.now_available
+        started.append(True)
+
+    src = event.FifoQueueEventSource(events=[event.Event(dt.utc_now())])
+
+    async def on_event(_event):
+        pass
+
+    backtesting_dispatcher.subscribe(src, on_event)
+    backtesting_dispatcher.subscribe_event_loop_started(on_started)
+    backtesting_dispatcher.subscribe_event_loop_started(on_started)
+    await backtesting_dispatcher.run()
+
+    assert len(started) == 1
+
+
 async def test_recursive_schedule_bug(backtesting_dispatcher):
     jobs_processed = 0
 

tests/test_realtime_dispatcher.py

@@ -218,6 +218,24 @@ async def on_idle():
     assert handler_calls == 2
 
 
+async def test_event_loop_started(realtime_dispatcher):
+    started = []
+
+    async def on_started():
+        started.append(True)
+
+    async def on_idle():
+        realtime_dispatcher.stop()
+
+    async def test_main():
+        realtime_dispatcher.subscribe_event_loop_started(on_started)
+        realtime_dispatcher.subscribe_idle(on_idle)
+        await realtime_dispatcher.run()
+        assert len(started) == 1
+
+    await asyncio.wait_for(test_main(), 2)
+
+
 async def test_cancelation_is_forwarded(realtime_dispatcher):
     async def test_main():
         with pytest.raises(asyncio.CancelledError):