3

Apologies if the question is poorly phrased, I'll do my best.

If I have a sequence of values with times as an Observable[(U,T)] where U is a value and T is a time-like type (or anything difference-able I suppose), how could I write an operator which is an auto-reset one-touch barrier, which is silent when abs(u_n - u_reset) < barrier, but spits out t_n - t_reset if the barrier is touched, at which point it also resets u_reset = u_n.

That is to say, the first value this operator receives becomes the baseline, and it emits nothing. Henceforth it monitors the values of the stream, and as soon as one of them is beyond the baseline value (above or below), it emits the elapsed time (measured by the timestamps of the events), and resets the baseline. These times then will be processed to form a high-frequency estimate of the volatility.

For reference, I am trying to write a volatility estimator outlined in http://www.amazon.com/Volatility-Trading-CD-ROM-Wiley/dp/0470181990 , where rather than measuring the standard deviation (deviations at regular homogeneous times), you repeatedly measure the time taken to breach a barrier for some fixed barrier amount.

Specifically, could this be written using existing operators? I'm a bit stuck on how the state would be reset, though maybe I need to make two nested operators, one which is one-shot and another which keeps creating that one-shot... I know it could be done by writing one by hand, but then I need to write my own publisher etc etc.

Thanks!

experquisite
  • 879
  • 5
  • 14

1 Answers1

5

I don't fully understand the algorithm and your variables in the example, but you can use flatMap with some heap-state and return empty() or just() as needed:

int[] var1 = { 0 };
source.flatMap(v -> {
    var1[0] += v;
    if ((var1[0] & 1) == 0) {
       return Observable.just(v);
    }
    return Observable.empty();
});

If you need a per-sequence state because of multiple consumers, you can defer the whole thing:

Observable.defer(() -> {
    int[] var1 = { 0 };
    return source.flatMap(v -> {
        var1[0] += v;
        if ((var1[0] & 1) == 0) {
           return Observable.just(v);
        }
        return Observable.empty();
    });
}).subscribe(...);
akarnokd
  • 69,132
  • 14
  • 157
  • 192
  • Thanks, that's a big hint. In RxScala, say, what's the usual idiom for creating one's own reusable stream filters like this? (ie: without just closing over source) For more generic ones, I've caked them directly onto Observable, but this is a pretty specific one. – experquisite Sep 01 '15 at 21:02
  • useful trick using defer for per subscriber state, thanks @akarnokd – Dave Moten Sep 02 '15 at 02:33