6. Dynamic Graphs

csp graphs are somewhat limiting in that they cannot change shape once the process starts up. csp dynamic graphs addresses this issue by introducing a construct to allow applications to dynamically add / remove sub-graphs from a running graph.

csp.DynamicBasket

csp dynamic baskets are a pre-requisite construct needed for dynamic graphs. csp.DynamicBaskets work just like regular static csp baskets, however dynamic baskets can change their shape over time. csp.DynamicBaskts can only be created from either csp nodes or from csp.dynamic calls, as described below. A node can take a csp.DynamicBasket as an input or generate a dynamic basket as an output. Dynamic baskets are always dictionary-style baskets, where time series can be added by key. Note that timeseries can also be removed from dynamic baskets.

Syntax

Dynamic baskets are denoted by the type csp.DynamicBasket[key_type, ts_type], so for example csp.DynamicBasket[str,int] would be a dynamic basket that will have keys of type str, and timeseries of type int. One can also use the non-python shorthand { ts[str] : ts[int] } to signify the same.

Generating dynamic basket output

For nodes that generate dynamic basket output, they would use the same interface as regular basket outputs. The difference being that if you output a key that hasn't been seen before, it will automatically be added to the dynamic basket. In order to remove a key from a dynamic basket output, you would use the csp.remove_dynamic_key method. NOTE that it is illegal to add and remove a key in the same cycle:

@csp.node
def dynamic_demultiplex_example(data : ts[ 'T' ], key : ts['K']) -> csp.DynamicBasket['T', 'K']:
    if csp.ticked(data) and csp.valid(key):
        csp.output({ key : data })


        ## To remove a key, which wouldn't be done in this example node:
        ## csp.remove_dynamic_key(key)

To remove a key one would use csp.remove_dynamic_key. For a single unnamed output, the method expects the key. For named outputs, the arguments would be csp.remove_dynamic_key(output_name, key)

Consuming dynamic basket input

Taking dynamic baskets as input is exactly the same as static baskets. There is one additional bit of information available on dynamic basket inputs though, which is the .shape property. As keys are added or removed, the basket.shape property will tick the the change events. The .shape property behaves effectively as a ts[csp.DynamicBasketEvents]:

@csp.node
def consume_dynamic_basket(data : csp.DynamicBasket[str,int]):
     if csp.ticked(data.shape):
         for key in data.shape.added:
             print(f'key {key} was added')
         for key in data.shape.removed:
             print(f'key {key} was removed')


      if csp.ticked(data):
          for key,value in data.tickeditems():
             #...regular basket access here

csp.dynamic

• csp.dynamic(trigger, sub_graph, graph_args...) → csp.DynamicBasket[ ... ]
  • trigger: a csp.DynamicBasket input. As new keys are added to the basket, they will trigger sub_graph instances to be created. As keys are removed, they will shutdown their respective sub-graph
  • sub_graph - a regular csp.graph method that will be wired as new keys are added on trigger
  • graph_args: these are the args passed to the sub_graph at the time of creation. Note the special semantics of argument passing to dynamic sub-graphs:
    • scalars: can be passed as is, assuming they are known at main graph build time
    • timeseries - can be passed as is, assuming they are known at main graph build time
    • csp.snap(ts): this will convert a timeseries input to a scalar at the time of graph creation, allowing you to get a "dynamic" scalar value to use at sub_graph build time
    • csp.snapkey(): this will pass through the key that was added which triggered this dynamic sub-graph. One can use this to get the key triggering the sub-graph.
    • csp.attach(): this will pass through the timeseries of the input trigger for the key which triggered this dynamic sub-graph. For example, say we have a dynamic basket of { symbol : ts[orders ]} as our input trigger. As a new symbol is added, we will trigger a sub-graph to process this symbol. Say we also want to feed in the ts[orders] for the given symbol into our sub_graph, we would pass csp.attach() as the argument.
  • output: every output of sub_graph (if there are any) will be returned as a member of a csp.DynamicBasket output. As new keys are added to the trigger, which generates sub-graphs, keys will be added to the output dynamic basket (Note, output keys will only generate on first tick of some output data, not upon instantiation of the sub-graph, since csp.DynamicBasket requires all keys to have valid values)

@csp.graph
def my_sub_graph(symbol : str, orders : ts[ Orders ], portfolio_position : ts[int], some_scalar : int) -> ts[Fill]:
   ... regular csp.graph code ...


@csp.graph
def main():
    # position as ts[int]
    portfolio_position = get_portfolio_position()


    all_orders = get_orders()
    # demux fat-pipe of orders into a dynamic basket keyed by symbol
    demuxed_orders = csp.dynamic_demultiplex(all_orders, all_orders.symbol)


    result = csp.dynamic(demuxed_orders, my_sub_graph,
                          csp.snap(all_orders.symbol), # Grab scalar value of all_orders.symbol at time of instantiation
                          #csp.snapkey(),                # Alternative way to grab the key that instantiated the sub-graph
                          csp.attach(),                  # extract the demuxed_orders[symbol] time series of the symbol being created in the sub_graph
                          portfolio_position,            # pass in regular ts[]
                          123)                          # pass in some scalar


    # process result.fills which will be a csp.DynamicBasket of { symbol : ts[Fill] }