Afternoon, I want to be able to get

raw_data

and split it using

groupby(key="exchange")

. Then have a module that can be applied for each

exchange

. I'd like to be able to

dr.execute()

each exchange independently or all at once. Is this reasonable?

from hamilton.function_modifiers import subdag, parameterized_subdag

def raw_data() -> pd.DataFrame:
    return ...

@extract_fields({
    f"{xch}_data": pd.DataFrame for xch in ["xch1","xch2"]
})
def _split_exchanges(raw_data: pd.DataFrame) -> pd.DataFrame:
    return {
        xch: df for xch, df in raw_data.groupby(key="exchange")
    }

# how can i apply these modules to these extracted fields
# calculate_rates_xch1.py
# calculate_rates_xch2.py
calculate_rates_xch1 = ...
calculate_rates_xch2 = ...

@subdag(calculate_rates_xch1, external_inputs=[f"{xch1}_data"])
def rates_xch1(calculated_rates: pd.DataFrame) -> pd.DataFrame:
    return calculated_rates

@subdag(calculate_rates_xch2, external_inputs=[f"{xch2}_data"])
def rates_xch2(calculated_rates: pd.DataFrame) -> pd.DataFrame:
    return calculated_rates

def rates_all(rates_xch1: pd.DataFrame, rates_xch2: pd.DataFrame) -> pd.DataFrame:
    return pd.concat([*locals().values()])

yes I think that’s do-able. But are the values in the groupby known ahead of time? are they dynamic? or?

Dynamic would be cool, but they could be known.

so it looks like your current code kind of does what you want though? You can ask just for

rates_xch2

or you can ask for

rates_all

(though I think the input parameters should be

rates_xch1

&

rates_xch2

?)

yes correct

For more dynamism: • there’s @resolve_with_config that could be used here with parameterized_subdag? you’d pass in the groupby values? Other options / ideas • fitting this into parallelizable & collect could be done - if somehow the module code could be parameterized? • hamilton within hamilton