Exploring the concept of value fluctuation using RXJS

Question

Exploring the concept of value fluctuation using RXJS

My current dilemma involves an Observable that continuously generates a rapid sequence of positive-integer values.

I am seeking a way to transform this sequence into value-change percentages, particularly when those values exceed a certain threshold within a specified time interval (in milliseconds).

Snippet:

function getPressurePercents(input: Observable, minChange: number, iw: number) {
   // Create an observable that emits value-change percentages outside
   // the designated minimum change percentage during each interval window
}

const obs = getPressurePercents(input, 0.5, 3000).subscribe(val => {
    // Notify when input values have changed by at least 0.5% in the last 3 seconds
    // The emitted value represents the percent change, where Math.abs(value) >= 0.5
    // Positive values indicate percent increases, while negative values signify decreases
});

In the scenario outlined above, I aim to receive an observable that notifies me of any value-change percentages exceeding 0.5% within a 3-second timeframe.

The objective is to identify sudden spikes in value changes and react to all percentages outside the defined threshold, such as abrupt fluctuations in equity prices (price spikes).

CRUCIAL NOTE: The solution should account for both upward and downward changes, meaning a decrease in percentage within the window should emit a corresponding negative percent value.

RESOLUTION

After exploring various options, I devised my own solution that operates independently from the RXJS pipeline but can be seamlessly integrated into it. This approach is more straightforward and compact compared to previously suggested pure RXJS methods. Check out my answer below for details.

typescript rxjs

Answer 1

Answer №1

In my opinion, a potential solution could resemble the following approach:

The system promptly sends spike notifications as soon as a spike is detected, without any waiting interval, while taking into account values from the previous spikeIntervalMs. This enables an instant response to spikes.
If there is a spike wave, where the percent change gradually increases or decreases, it only sends spike notifications for values higher than the maximum in that particular spike wave. This continues until the spike wave subsides (no spikes detected during a certain period).

export function getSpikedPercentChangeAbs$(
  value$: Observable<number>,
  spikePercentage: number,
  spikeIntervalMs: number,
): Observable<number> {
  // Implementation logic here
}

function getPercentChangeAbs(
  previousValue: number,
  currentValue: number,
): number {
  // Definition of percent change calculation
}

function removeOldValues(
  oldTime: Date,
  valueAndTimeList: IValueAndTime[],
): void {
  // Logic to remove outdated values from the list
}

export interface IValueAndTime {
  time: Date;
  value: number;
}

Answer 2

In my opinion, a potential solution could resemble the following approach:

The system promptly sends spike notifications as soon as a spike is detected, without any waiting interval, while taking into account values from the previous spikeIntervalMs. This enables an instant response to spikes.
If there is a spike wave, where the percent change gradually increases or decreases, it only sends spike notifications for values higher than the maximum in that particular spike wave. This continues until the spike wave subsides (no spikes detected during a certain period).

export function getSpikedPercentChangeAbs$(
  value$: Observable<number>,
  spikePercentage: number,
  spikeIntervalMs: number,
): Observable<number> {
  // Implementation logic here
}

function getPercentChangeAbs(
  previousValue: number,
  currentValue: number,
): number {
  // Definition of percent change calculation
}

function removeOldValues(
  oldTime: Date,
  valueAndTimeList: IValueAndTime[],
): void {
  // Logic to remove outdated values from the list
}

export interface IValueAndTime {
  time: Date;
  value: number;
}

Answer 3

Answer №2

While I initially searched for a RXJS solution that was fully functional, I couldn't find one that suited my needs. So here's my take on it - a solution that doesn't involve RXJS but can easily be integrated with it:

class SpikeDetector {

    private buffer: Array<{ ts: number, value: number }> = [];

    constructor(private change: number, private interval: number) {
    }

    next(value: number, spike: (change: number) => void): void {
        const ts = Date.now(); // timestamp
        const b = this.buffer; // abbreviation        
        while (b.length && ts - b[0].ts > this.interval) {
            b.shift();
        }
        b.push({ts, value});
        if (b.length >= 2) {
            const change = 100 * value / b[0].value - 100;
            if (Math.abs(change) >= this.change) {
                b.length = 0;
                spike(change);
            }
        }
    }
}

Using with RXJS:

const sd = new SpikeDetector(0.5, 3000);

input.subscribe(value => {
    sd.next(value, change => {
        // value spike by >=0.5% within 3 seconds
    });
});

Alternatively, you can utilize an RXJS helper as shown below:

function changeSpike(input: Observable<number>, change: number, i: number) {
    const sd = new SpikeDetector(change, i);
    return input.pipe(switchMap(value =>
        new Observable<number>(obs => {
            sd.next(value, change => obs.next(change));
        })
    ));
}

UPDATE:

Here's an enhanced and comprehensive solution that I've implemented in my financial project for detecting equity price spikes.

Answer 4

While I initially searched for a RXJS solution that was fully functional, I couldn't find one that suited my needs. So here's my take on it - a solution that doesn't involve RXJS but can easily be integrated with it:

class SpikeDetector {

    private buffer: Array<{ ts: number, value: number }> = [];

    constructor(private change: number, private interval: number) {
    }

    next(value: number, spike: (change: number) => void): void {
        const ts = Date.now(); // timestamp
        const b = this.buffer; // abbreviation        
        while (b.length && ts - b[0].ts > this.interval) {
            b.shift();
        }
        b.push({ts, value});
        if (b.length >= 2) {
            const change = 100 * value / b[0].value - 100;
            if (Math.abs(change) >= this.change) {
                b.length = 0;
                spike(change);
            }
        }
    }
}

Using with RXJS:

const sd = new SpikeDetector(0.5, 3000);

input.subscribe(value => {
    sd.next(value, change => {
        // value spike by >=0.5% within 3 seconds
    });
});

Alternatively, you can utilize an RXJS helper as shown below:

function changeSpike(input: Observable<number>, change: number, i: number) {
    const sd = new SpikeDetector(change, i);
    return input.pipe(switchMap(value =>
        new Observable<number>(obs => {
            sd.next(value, change => obs.next(change));
        })
    ));
}

UPDATE:

Here's an enhanced and comprehensive solution that I've implemented in my financial project for detecting equity price spikes.

Exploring the concept of value fluctuation using RXJS

Answer №1

Answer №2

Similar questions

My troubleshooting journey: Why isn't Angular's HostListener preventDefault function

Dealing with delays in rendering certain values with Angular interpolation & ways to eliminate an element from an array of TypeScript objects

Enhance Axios to support receiving a URL in the form of a URL address

React typescript wormhole is struggling to display content due to createPortal and useContext combination

Exploring the differences between two Observables by utilizing a combination of .forEach and .some on an array

Assign a variable to set the property of a class

Display a complete inventory of installed typings using the tsd command

What could be causing my AJAX request to send a null object to the server?

Receiving "this" as an undefined value within the TypeScript class

Error encountered while executing vitest within DevContainer

Angular: Incorporating a custom validation function into the controller - Techniques for accessing the 'this' keyword

One of the modules ionic declaration consists of

I am interested in creating an input range slider using React and TypeScript

Utilizing Node.JS Tools for Visual Studio to share classes across numerous TypeScript files

the data chart is malfunctioning

Exploring the World of TypeScript Decorators

The continuous invocation of the ngStyle background-image in Angular 4 is causing a decrease in the loading speed of the page

Is there a way to automatically populate the result input field with the dynamic calculation results from a dynamic calculator in Angular6?

Tips on how to access the previous and current state within a selector in ngrx selectors

Add a unique class to an element within a main component