Dynamic Subgraphs with Adapters

[Roh-codeur]

Hi, can you please help with the below please, ta!

Describe the bug
I recall that AdapterManager are not yet supported in DynamicGraphManager.

To Reproduce
I am trying to combine the examples: https://github.com/Point72/csp/blob/main/examples/06_advanced/e1_dynamic.py and https://github.com/Point72/csp/blob/main/examples/04_writing_adapters/e5_adaptermanager_pushinput.py as below:

1. Create adapter manager
2. Inside a sub-graph, I am hoping I can use the updates from the adapter

from datetime import datetime, timedelta

import csp
from csp import ts


class Order(csp.Struct):
    symbol: str
    size: int
    price: float


@csp.graph
def process_symbol(symbol: str, order: ts[Order], initial_order: Order, timer: ts[MyData], scalar: str) -> ts[int]:
    print("Starting sub-graph to process symbol ", symbol, " with initial order: ", initial_order, " scalar: ", scalar)

    csp.print(symbol + " orders", order)
    csp.print(symbol + " timer", timer)

    cum_size = csp.accum(timer.value)
    return cum_size


@csp.node
def process_results(x: {ts[str]: ts[int]}):
    if csp.ticked(x):
        print(csp.now(), "cum_sizes:", dict(x.tickeditems()))


@csp.graph
def main_graph():
    # We have a stream of incoming orders to deal with, we dont know the symbols up front
    adapter_manager = MyAdapterManager(timedelta(seconds=0.75))
    symbols = ["AAPL", "IBM", "TSLA", "GS", "JPM"]

    orders = csp.curve(
        Order,
        [
            (timedelta(seconds=0), Order(symbol="AAPL", price=135, size=100)),
            (timedelta(seconds=1), Order(symbol="FB", price=350, size=-200)),
            (timedelta(seconds=2), Order(symbol="GME", price=210, size=1000)),
            (timedelta(seconds=3), Order(symbol="AAPL", price=138, size=-100)),
            (timedelta(seconds=4), Order(symbol="FB", price=330, size=100)),
            (timedelta(seconds=5), Order(symbol="AMC", price=57, size=400)),
            (timedelta(seconds=6), Order(symbol="GME", price=200, size=800)),
        ],
    )

    # Get a dynamic basket keys by symbol
    trigger = csp.dynamic_demultiplex(orders, orders.symbol)

    some_ts = csp.count(csp.timer(timedelta(seconds=1)))
    some_scalar = "howdy"

    # dynamic graphs
    cum_sizes = csp.dynamic(
        trigger,
        process_symbol,
        csp.snapkey(),  # csp.snapkey() provides the key that triggers a new dynamic graph as a scalar argument
        csp.attach(),  # csp.attach() will pass the corresponding timeseries of the key for the graph instance
        csp.snap(
            orders
        ),  # csp.snap will provide the current value of the given timeseries at the time of dynamic graph instantiation
        adapter_manager.subscribe("GS"),  # regular time series can be passed along, which will be shared across all instances
        some_scalar,  # regular scalar values can be passed as arguments to the sub-graph as well
    )

    # cum_sizes is a dynamic basket of results, keyed by the trigger keys
    process_results(cum_sizes)


def main():
    csp.run(main_graph, starttime=datetime.utcnow().replace(microsecond=0), endtime=timedelta(seconds=10))


if __name__ == "__main__":
    main()

Expected behavior
I was hoping I can use the adapter this way with dynamic graph manager. in the future, the adapter would subscribe to all symbols and I will be able to filter for the pertinent symbol

Error Message
No error message, just no output from

csp.print(symbol + " timer", timer) 

Runtime Environment

>>> import sys, csp; print(csp.__version__); print(sys.version); print(sys.platform)
0.0.5
3.10.14 | packaged by Anaconda, Inc. | (main, May  6 2024, 19:44:50) [MSC v.1916 64 bit (AMD64)]

[timkpaine]

Adapters are not supported with dynamic, this is part of the motivation for the new pattern in #277

[Roh-codeur]

thanks @timkpaine. can you please suggest a pattern I could follow for the below please

1. on startup, I have a list of tickers
2. I create a sub-graph for each of these tickers
3. I would like to subscribe to market data for these tickers and feed to sub-graph

[timkpaine]

In contrast with some other csp developers, I tend to be a "fat pipe" kind of guy. Here is how I would do it (and how I do do it in other places):

• have an adapter following Expand websocket adapter, ensure connections/disconnections can be done dynamically, establish better pattern for dynamic subscriptions for use in other adapters #277 that accepts "new subscription" requests dynamically
• as new tickers become known, dynamically subscribe and return all data for all tickers in a fat pipe (e.g. one edge, not N edges)
• mux data as needed per symbol downstream

I will note that if you know all tickers on startup, you dont need to use dynamic, you could e.g.:

@csp.graph
def graph(tickers: List[str]):
    subscriptions = {}
    for ticker in tickers:
        subscriptions[ticker] = MarketData.subscribe(ticker)
    ...

[Roh-codeur]

thanks mate. :

1. Looks like the PR for that Issue is being worked on. thanks for that. Add dynamic endpoints to WebsocketAdapter #378. do you think I can do something before its merged?
2. I am working with Futures, so, I do indeed know the list of contracts. however, at a given point in time, only a set of contracts are active, in backtesting, I am only trying to create sub-graphs for a list of contracts active at graph time. hence my approach towards dynamic graphs. is there another way to achieve this please?

ta

[timkpaine]

1. Looks like the PR for that Issue is being worked on. thanks for that. [WIP] Add dynamic endpoints to WebsocketAdapter #378. do you think I can do something before its merged?

We are starting with websockets and http adapter as a POC of the new structure, to be used as a pattern for the forthcoming redis adapter (and then will likely go back and do for other types of adapters like kafka).

2. I am working with Futures, so, I do indeed know the list of contracts. however, at a given point in time, only a set of contracts are active, in backtesting, I am only trying to create sub-graphs for a list of contracts active at graph time. hence my approach towards dynamic graphs. is there another way to achieve this please?

I don't think there is a timeline for dynamic adapters right now

[Roh-codeur]

got it, thanks mate. I am thinking will start off with static sub-graphs for now, for all given contracts. to minimize the graph size(num of sub-graphs), will run shorter time periods. shouldnt be an issue when realtime=True.

Meanwhile, look forward to dynamic adapter changes. thanks again for looking into that. if you do think of a potential solution, please do let me know

ta

[timkpaine]

I'm going to keep open for now as I realize we don't have an external ticket for dynamic adapters

[robambalu]

This particular example looks like it should work though, since the adapter is created and subscribed to outside of the dynamic graph. The limitation at the moment ( perhaps forever ) is that you cannot instantiate adapter managers in a dynamic graph, but I dont see that happening in this example