How to calculate rolling conditional mean for values above the window’s median?

I’m trying to compute a rolling conditional mean in DolphinDB where, for each window of 5 elements, I calculate the mean of values that are greater than or equal to the window’s median. My current approach returns unexpected results, and I need help debugging or finding the correct implementation.

My current code:

x = [3, 2, 1, 6, 5, 4, 7, 9, 9, 8, 5, 8, 6, 9, 10]
y = mavg(iif(x >= mmed(x,5), x, NULL), 5)
//Output:  [NULL, NULL, NULL, NULL, 3.4, 3.6, 4.6, 6.2, 6.8, 7.4, 10, 8.25, 8.5, 8.3333, 8.3333, 9]

But this seems incorrect compared to manual calculations (see expected output below).

For each window of 5 values:

• Compute the median (mmed).
• Filter values ≥ median, then take their mean (mavg).

For example:

• Window [3,2,1,6,5]:
  • Median = 3
  • Values ≥ median = [3,6,5] → Mean = (3+6+5)/3 = 4.666...
• Window [2,1,6,5,4]:
  • Median = 4
  • Values ≥ median = [6,5,4] → Mean = 5

So expected output is：

[NULL, NULL, NULL, NULL, 4.666, 5, 6, 7.333, 8.333, 8.666, 8.666,8.5，8.333,8.333,9]

I tried using rolling with a custom function but struggled to align windows properly.I also Checked DolphinDB documentation for window alignment parameters (adjust), but unsure how to apply them here.

How can I correctly implement this rolling conditional mean calculation in DolphinDB? A reproducible solution using vectorized operations or rolling functions would be appreciated.

Answer

The core issue with your original code is that mmed and mavg operate on globally aligned windows, not enforcing per-window median conditions. To correctly calculate the rolling mean of values ≥ the window’s median, use DolphinDB’s moving function with a custom aggregation. Here’s the solution:

x = [3,2,1,6,5,4,7,9,9,8,5,8,6,9,10]

// Define a custom aggregation function for per-window logic
defg averageif(x) {
    median_val = x.med()        // Step 1: Compute median of the current window
    filtered = x[x >= median_val] // Step 2: Filter values ≥ median
    return filtered.avg()       // Step 3: Return mean of filtered values
}

windows = 5
y = moving(averageif, x, windows)  // Apply rolling window computation