How to make multiple PushAdapters tick simultaneously in one graph?

[mizuse-emiya]

In our setup, a graph usually has three real-time data sources. We use three PushAdapters, and each data source publishes data once every 10 minutes.

The issue we see is that the graph downstream does not receive all three updates at the same time. Instead, each PushAdapter seems to tick the entire graph separately.

Question:
Is there a way to implement the graph or adapters so that multiple PushAdapters can tick simultaneously?

[AdamGlustein]

Considering all the sources are real-time and distinct, your best bet would be to just synchronize the 3 sources within the graph using https://github.com/Point72/csp/wiki/Basket-Nodes-API#sync. There's no easy way to say "these 3 push adapters should always be on the same cycle" - each will call push_tick individually on a separate thread asynchronously from the others.

[AdamGlustein]

So you should be able to sync with a 10 minute threshold to cover this - it will only release all 3 ticks once they have provided a value within the current 10 minute interval.

[mizuse-emiya]

import threading
import time
from datetime import datetime, timedelta

import csp
from csp import ts, basketlib
from csp.impl.pushpulladapter import PushPullInputAdapter
from csp.impl.wiring import py_pushpull_adapter_def

class MyPushPullAdapter(PushPullInputAdapter):
    def __init__(self, history_start: datetime, history_end: datetime, catchup_time: datetime, symbol: str):
        self._history_start = history_start
        self._history_end = history_end
        self._catchup_time = catchup_time
        self.symbol = symbol

        self._running = False
        self._thread = None
        super().__init__()

    def start(self, starttime, endtime):
        print("MyPushPullAdapter::start")
        self._running = True
        self._thread = threading.Thread(target=self._run)
        self._thread.start()

    def stop(self):
        print("MyPushPullAdapter::stop")
        if self._running:
            self._running = False
            if self._thread is not None:
                self._thread.join()

    def _run(self):

        current_ts = self._history_start
        counter = 0
        while self._running and current_ts <= self._history_end:

            self.push_tick(False, current_ts, counter)
            print(f"{self.symbol} pull 30m @ {current_ts} -> {counter}")
            counter += 1
            current_ts += timedelta(minutes=30)

        self.flag_replay_complete()

        current_ts = self._catchup_time
        while self._running:
            self.push_tick(True, current_ts, counter)
            print(f"{self.symbol} push 5s @ {current_ts} -> {counter}")
            counter += 1
            current_ts += timedelta(seconds=5)
            time.sleep(5)


MyPushPull = py_pushpull_adapter_def(
    "MyPushPull",
    MyPushPullAdapter,
    ts[int],
    history_start=datetime,
    history_end=datetime,
    catchup_time=datetime,
    symbol=str
)

@csp.node
def print_node(a: ts[int], b: ts[int], c: ts[int]):
    if csp.ticked(a) and csp.ticked(b) and csp.ticked(c):
        print(f"all ticker a: {a} b: {b} c: {c}")


@csp.graph
def demo_graph():
    now = datetime.now().replace(minute=0, second=0, microsecond=0)
    data1 = MyPushPull(
        history_start=now - timedelta(hours=200),
        history_end=now,
        catchup_time=now + timedelta(seconds=10),
        symbol="s1"
    )
    data2 = MyPushPull(
        history_start=now - timedelta(hours=200),
        history_end=now,
        catchup_time=now + timedelta(seconds=10),
        symbol="s2"
    )
    data3 = MyPushPull(
        history_start=now - timedelta(hours=200),
        history_end=now,
        catchup_time=now + timedelta(seconds=10),
        symbol="s3"
    )

    print_node(data1, data2, data3)


def main():
    csp.run(demo_graph, starttime=datetime(2020, 12, 1), realtime=True)


if __name__ == "__main__":
    main()

Very sry to bother you again.This is my code for testing the pullpushingadapter.
Why i can't trigger print_node whenever pull or push stage?

[robambalu]

I just ran this ( adding csp.now() to your print ) and as expected I do see it printing during the pull phase
2025-09-26 08:00:00 all ticker a: 2 b: 2 c: 2
2025-09-26 08:30:00 all ticker a: 3 b: 3 c: 3
2025-09-26 09:00:00 all ticker a: 4 b: 4 c: 4

[mizuse-emiya]

Thanks, and sorry for the noise earlier. I realized there were some mistakes in my code, which I’ve now fixed, and it’s running as expected. That said, I’m still a bit unclear on the difference between the pull and push stages—how is the graph actually driven? Also, I noticed that unless I use sync_list before print_node, the print_node function doesn’t get triggered during push stage.

[robambalu]

It should be explained in the docs but simply put, in "pull" mode the adapter is running on historical / playback data, and will preschedule data at the exact time that it occured. If multiple pull adapters schedule the same time, they will execute at the exact same time.
Once you are in "push" mode there is no telling when data will arrive from some arbitrary async thread, so the time of the tick is just grabbed as wallclock of the tick and its extremely unlikely multiple adapters will tick with the same arbitrary wallclock time

If the source of your multiple adapters is the same thread / driver, then you can ensure they tick together by using PushGroups. However if they are coming from multiple sources there is no way to ensure this so you need to rely on sync_list to ensure they all tick at the same time ( if needed )

[robambalu]

Agreed with Adam. In the slight chance that all 3 of these push adapters are actually driven by the same backend, then you can also create a single AdapterManager to service them all, and have it use PushGroups ( concept is explained here but doesnt look like there is currently a doc explaining how to use it in a push adapter manager yet