[BUG-REPORT] Selection-based aggregate functions for each column row

[ravigoku]

• Currently using the latest Vaex 3.0.0
• I have the following tables:
  Table 1

     date        ticker x
0    2020-11-03  AAA    1
1    2020-11-04  AAA    1
2    2020-11-05  ZZZ    1

Table 2

            datetime          ticker   volume    
0           2020-11-03 12:48  AAA      100      
1           2020-11-04 11:17  AAA      100       
2           2020-11-04 14:35  AAA      100       
35,228,709  2020-12-18 15:57  ZZZ      300
35,228,710  2020-12-18 15:57  ZZZ      300

• I want to create a new column in Table 1:
  Table1[sum vol] = sum of all volume data in Table 2 matching the date and ticker of each corresponding row in Table 1
• After many variations, I believe this is the closest, but it seems to not evaluate per row

@vaex.register_function()
def sum_volume(x, date, ticker):
    sel = table2.datetime == date)
    sel = sel & table2.ticker == ticker & sel
    d = table2.sum(table2.volume, selection=sel)
    return d*x

To execute:
table1['sum vol'] = table1.func.sum_volume(table1.x, table1.date, table1.ticker)

Note: x is a column full of 1s because I wasn't able to return just an integer without multiplying it against a column

• Let me know if missed something, but I don't think this is currently supported. Any work arounds would be appreciated as well. :)

[kmcentush]

If you want to use register_function appropriately, I believe you actually need to call it off of the table's func attribute as follows: table1['sum_vol'] = table1.func.sum_volume(table1.x, table1.date, table1.ticker). Right now I think you're calling it as if it's a normal function, so you're passing the actual column objects instead of row values.

Give that a shot. In either case, a more efficient solution might be to create a hybrid column that you can groupby in Table2 (using the "sum" aggregator), then join on to Table1. Pseudo-code as follows:

# Prep columns using a "|" delimeter
table1['concat_col'] = table1['date'].astype('string') + '|' + table1['ticker'].astype('string')
table2['date'] = table2['datetime'].round_to_the_day() # <-- not sure what the actual call would be, but your goal is to match the date format of table1
table2['concat_col'] = table2['date'].astype('string') + '|' + table2['ticker'].astype('string')

# Group and join
grouped = table2.groupby(by='concat_col').agg({'volume': 'sum'})
result = table1.join(grouped , left_on='concat_col', right_on='concat_col')

# Output the result
result

[ravigoku]

Thanks! Great catch on the typo table1.func.sum_volume(table1.x, table1.date, table1.ticker). Unfortunatley the selections still seem to not be updating with each pass.

Your pseudocode is much appreciated for the groupby. My use case involves playing around with different time variations and selection types in general in table 2, and am searching for a more modular solution to manipulating selections or filters withing groups

Do you think the delayed and execute decorators can play a role in solving this? - I haven't played around with it, but I assumed they only work for vaex-standard functions.

[kmcentush]

In that sense, could you not write a modular function that converts the datetime column into a specified time bin (i.e. minute, hour, day, etc.?), then do the join approach? Given that you're working with financial tick(?) data, I'm assuming you're looking to ultimately optimize time bins in some way to generate a signal of interest. In essence, if you can modularly generate a bin ID, then you can still use the join approach.

Would you be able to provide a working example with your starting tables and an example of some of the parameters you would iterate over, along with the expected result for one or two cases? I'm happy to try to generate some working code given the groundwork. It will also be useful to understand to what extent this is a bug with Vaex.

[ravigoku]

That sparked an idea to look at binby the underlying code is written and I'm thinking to look further into (mapping, delayed, execution). The surface level functionality is identical, but in my use case, I'd like to perform unique selections dependent on the value of each row, which is why the bin function hasn't sufficed. In the sample below and the initial post. I dwindle the concept down to just summing the values for simplicity. But I appreciate your investigating this with me.

• Attached are three files. Convert them to csv as github does not support csvs.
  table1.xlsx
  table1-ideal-results.xlsx
  table2.xlsx

import vaex
import numpy as np
import pandas as pd

• Import the tables:

table2 = vaex.open('table2.csv')
table2['datetime'] = table2.datetime.astype('datetime64[m]')
table2['date'] = table2.datetime.astype('datetime64[D]')

table1 = vaex.open('table1.csv')
table1['date'] = table1.date.astype('datetime64[D]')

• The functionality I'm trying to achieve works with the following segment.. play around with the date time, and ticker variables and it works:

date = np.datetime64('2020-11-04','D') # type datetime64
ticker = 'A' # type string

sel = table2.ticker.str.equals(ticker)
sel = (table2.date == date) & sel
table2.sum(table2.volume, selection=sel)

• Now I attempt to wrap it in a function

# Attempt 1
# Wrapping it in a function
@vaex.register_function()
def sum_day(date, ticker):
    sel = table2.ticker.str.equals(ticker)
    sel = (table2.date == date) & sel
    return table2.sum(table2.volume, selection=sel)

• Executing Function: table1['sum day'] = table1.func.sum_day(table1.date, table1.ticker) results in the error:

# ERROR:MainThread:vaex:error evaluating: sum_day at rows 0-4
# Traceback (most recent call last):
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 3575, in table_part
#     values = dict(zip(column_names, df.evaluate(column_names)))
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 2616, in evaluate
#     return self._evaluate_implementation(expression, i1=i1, i2=i2, out=out, selection=selection, filtered=filtered, internal=internal, parallel=parallel, chunk_size=chunk_size)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 5352, in _evaluate_implementation
#     dtypes[expression] = df.data_type(expression, internal=False)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 1999, in data_type
#     dtype = data.dtype
# AttributeError: 'function' object has no attribute 'dtype'
#
# During handling of the above exception, another exception occurred:
#
# Traceback (most recent call last):
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 3580, in table_part
#     values[name] = df.evaluate(name)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 2616, in evaluate
#     return self._evaluate_implementation(expression, i1=i1, i2=i2, out=out, selection=selection, filtered=filtered, internal=internal, parallel=parallel, chunk_size=chunk_size)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 5352, in _evaluate_implementation
#     dtypes[expression] = df.data_type(expression, internal=False)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 1999, in data_type
#     dtype = data.dtype
# AttributeError: 'function' object has no attribute 'dtype'
#   #  date        ticker    sum day
#   0  2020-11-03  A         error
#   1  2020-11-03  B         error
#   2  2020-11-04  A         error
#   3  2020-11-04  B         error

• Attempt 2: I added a dummy 'x' column to mimic an arithmetic expression table1['x'] = np.ones(table1.length(), dtype=int) and translate the dtype, but that patch still did not work.

@vaex.register_function()
def sum_day(x, date, ticker):
    sel = table2.ticker.str.equals(ticker)
    sel = (table2.date == date) & sel
    return x*table2.sum(table2.volume, selection=sel)

• Executing table1['sum day'] = table1.func.sum_day(table1.x, table1.date, table1.ticker) results in:

#ERROR:MainThread:vaex:error evaluating: sum_day at rows 0-4
# Traceback (most recent call last):
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 3575, in table_part
#     values = dict(zip(column_names, df.evaluate(column_names)))
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 2616, in evaluate
#     return self._evaluate_implementation(expression, i1=i1, i2=i2, out=out, selection=selection, filtered=filtered, internal=internal, parallel=parallel, chunk_size=chunk_size)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 5352, in _evaluate_implementation
#     dtypes[expression] = df.data_type(expression, internal=False)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 1999, in data_type
#     dtype = data.dtype
# AttributeError: 'function' object has no attribute 'dtype'
#
# During handling of the above exception, another exception occurred:
#
# Traceback (most recent call last):
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 3580, in table_part
#     values[name] = df.evaluate(name)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 2616, in evaluate
#     return self._evaluate_implementation(expression, i1=i1, i2=i2, out=out, selection=selection, filtered=filtered, internal=internal, parallel=parallel, chunk_size=chunk_size)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 5352, in _evaluate_implementation
#     dtypes[expression] = df.data_type(expression, internal=False)
#   File "/home/ec2-user/stonkswork/venv/lib64/python3.7/site-packages/vaex/dataframe.py", line 1999, in data_type
#     dtype = data.dtype
# AttributeError: 'function' object has no attribute 'dtype'
#   #  date        ticker      x  sum day
#   0  2020-11-03  A           1  error
#   1  2020-11-03  B           1  error
#   2  2020-11-04  A           1  error
#   3  2020-11-04  B           1  error

[kmcentush]

I dug around a bit. I think the issue with using register_function here is that it passes vectors as the function arguments. Your code is then trying to see if a single value of table2 (ex: ticker) equals a list of values, which will never be true. I was not able to get anything to work using register function thus far. But here was my general idea:

@vaex.register_function()
def sum_day(date, ticker):
    if len(date) > 1:  # ticker will be the same length
        print('hi_if', date, ticker)
        vals = []
        for i in range(len(date)):
            sel = table2['ticker'] == str(ticker[i])
            sel = sel & (table2['date'] == date[i])
            # sum_vol = table2.sum(table2.volume, selection=sel)  # This started throwing some async eval errors for me that I don't have the time to dig into right now, but you get the idea
            # vals.append(sum_vol)
            vals.append(i**2)  # I put this here so that the function doesn't break
        return vals
    else:
        print('hi_else')
        sel = table2['ticker'] == str(ticker[0])
        sel = sel & (table2['date'] == date[0])
        sum_vol = table2.sum(table2.volume, selection=sel)
        print('hi_else', date, ticker, sum_vol)
        return sum_vol

table1.func.sum_day(table1.date, table1.ticker)

The actual printout of this attempt is interesting. On my computer it seems to evaluate things in no particular order (or even multiple times!), which I assume is related to the parallelization.

hi_else
hi_else ['2020-11-03'] [
  "A"
] 15600
hi_else
hi_else ['2020-11-03'] [
  "A"
] 15600
hi_else
hi_else ['2020-11-03'] [
  "A"
] 15600
hi_if ['2020-11-03' '2020-11-03' '2020-11-04' '2020-11-04'] [
  "A",
  "B",
  "A",
  "B"
]
hi_else
hi_else ['2020-11-03'] [
  "A"
] 15600

Expression = sum_day(date, ticker)
Length: 4 dtype: int64 (expression)
-----------------------------------
0  0
1  1
2  4
3  9

I think you could probably use table1.apply() in some way, but that solution is not vectorized nor parallelized, so it's extremely slow. At that point I would advocate for just looping over the unique (date, ticker) combinations, making a dataframe, then concat'ing them all together at the end. But this is basically brute forcing what you're after.

I know you explained how the "sum" case was a simplification of your ideal filtering, but here's how I achieved it using groupby():

import vaex
vaex.__version__


# # Load Tables

# Import table1
table1 = vaex.open('table1.csv')  # note: had to change date string format in Excel to be YYYY-MM-DD as opposed to MM/DD/YYYY; otherwise, the astype() below fails
table1['date'] = table1.date.astype('datetime64[D]')

table1.head()

# Import table2
table2 = vaex.open('table2.csv')
table2['datetime'] = table2.datetime.astype('datetime64[m]')
table2['date'] = table2.datetime.astype('datetime64[D]')

table2.head()


# # Grouping

# Groupby
table2['date_ticker'] = table2['date'].astype('string') + '_' + table2['ticker']
result = table2.groupby('date_ticker', agg={'volume': 'sum'})

# Because vaex string splitting is a little buggy right now, we'll just join back to table 1 to get the relevant info
table1['date_ticker'] = table1['date'].astype('string') + '_' + table1['ticker']
result = table1.join(result, left_on='date_ticker', right_on='date_ticker')
result.drop('date_ticker', inplace=True)

result

[ravigoku]

I'm so pumped !!! This works perfectly as of now in vaex 4.0.0:
result = table2.groupby(by=[vaex.BinnerTime(table2.datetime, resolution='D'),'ticker'], agg={'volume': vaex.agg.sum(table2.volume)})

[ravigoku]

Your method of creating the date_ticker as an index column is more efficient! Thanks!

[maartenbreddels]

Hi,

short reply (vacation mode now)

On my computer it seems to evaluate things in no particular order
It will evaluate the first element to determine the dtype, the rest is indeed in parallel chunks.

I think you could probably use table1.apply() in some way, but that solution is not vectorized nor parallelized, so it's extremely slow.

In the latest alpha it is, but I think this should be solvable without https://vaex.io/docs/tutorial.html#The-escape-hatch:-apply .

I haven't had the time to fully understand your problem, but it seems groupby and join should work right? If not, why not.

cheers,

Maarten

[ravigoku]

Thank you for the quick response and Happy Holidays!

• I'm looking into the group by and joins now in a different light and will confirm my findings.
• Can you elaborate on what you mean by "in the latest alpha it is?"

[maartenbreddels]

PS: I move this to discussions, we just opened that, and this is a good candidate for it.

[ravigoku]

I want to create a selection before performing the aggregate groupby. Is the most efficient way of doing that to use the filter function or is there another function with a parameter similar to select that will be better?
Say I want to select all the items with a time between 9:00 and 12:00:

• I create a filtered dataset df2 = df[between 9 and 11] first (are there more efficient ways/functions to do this?)
• And then perform the group by aggregate function.
• then join

[JovanVeljanoski]

Sorry for joining late to the party..

I don't know how known this feature is, but in vaex you can do selection per aggregation during groupby operations.

Consider this example:

df.groupby('some_column').agg({'agg_col_1': vaex.agg.mean('x')  # usual aggregation
                               'agg_col_2 vaex.agg.mean('y', selection='(z1>5) & (z2<10)')   # The aggregation will happen only for samples that obey the selection.
})

Does this help you in any way?