Monday, 24 May 2021

Numba: how to parse arbitrary logic string into sequence of jitclassed instances in a loop

Tl Dr. If I were to explain the problem in short:

  1. I have signals:
np.random.seed(42)
x = np.random.randn(1000)
y = np.random.randn(1000)
z = np.random.randn(1000)
  1. and human readable string tuple logic like :
entry_sig_ = ((x,y,'crossup',False),)
exit_sig_ = ((x,z,'crossup',False), 'or_',(x,y,'crossdown',False))

where:

  • 'entry_sig_' means the output will be 1 when the time series unfolds from left to right and 'entry_sig_' is hit. (x,y,'crossup',False) means: x crossed y up at a particular time i, and False means signal doesn't have "memory". Otherwise number of hits accumulates.
  • 'exit_sig_' means the output will again become '0' when the 'exit_sig_' is hit.
  1. The output is generated through:
@njit
def run(x, entry_sig, exit_sig):
    '''
    x: np.array
    entry_sig, exit_sig: homogeneous tuples of tuple signals
    Returns: sequence of 0 and 1 satisfying entry and exit sigs
    ''' 
    L = x.shape[0]
    out = np.empty(L)
    out[0] = 0.0
    out[-1] = 0.0
    i = 1
    trade = True
    while i < L-1:
        out[i] = 0.0
        if reduce_sig(entry_sig,i) and i<L-1:
            out[i] = 1.0
            trade = True
            while trade and i<L-2:
                i += 1
                out[i] = 1.0
                if reduce_sig(exit_sig,i):
                    trade = False
        i+= 1
    return out

reduce_sig(sig,i) is a function (see definition below) that parses the tuple and returns resulting output for a given point in time.

Question:

As of now, an object of SingleSig class is instantiated in the for loop from scratch for any given point in time; thus, not having "memory", which totally cancels the merits of having a class, a bare function will do. Does there exist a workaround (a different class template, a different approach, etc) so that:

  1. combined tuple signal can be queried for its value at a particular point in time i.
  2. "memory" can be reset; i.e. e.g. MultiSig(sig_tuple).memory_field can be set to 0 at a constituent signals levels.

Fully reproducible (yet simplified) example:

  1. Signals
from numba import njit, int64, float64
from numba.types import Array, string, boolean
from numba.experimental import jitclass

np.random.seed(42)
x = np.random.randn(1000)
y = np.random.randn(1000)
z = np.random.randn(1000)

# Example of "human-readable" signals
entry_sig_ = ((x,y,'crossup',False),)
exit_sig_ = ((x,z,'crossup',False), 'or_',(x,y,'crossdown',False))

# Turn signals into homogeneous tuple
#entry_sig_
entry_sig = (((x,y,'crossup',False),'NOP'),)
#exit_sig_
exit_sig = (((x,z,'crossup',False),'or_'),((x,y,'crossdown',False),'NOP'))
  1. Generate single signal value for a particular point in time (i) via sig(i) method of SingleSig class:
@njit
def crossup(x, y, i):
    '''
    x,y: np.array
    i: int - point in time
    Returns: 1 or 0 when condition is met
    '''
    if x[i - 1] < y[i - 1] and x[i] > y[i]:
        out = 1
    else:
        out = 0
    return out


@njit
def crossdown(x, y, i):
    if x[i - 1] > y[i - 1] and x[i] < y[i]:
        out = 1
    else:
        out = 0
    return out


spec = [
    ("x", Array(dtype=float64, ndim=1, layout="C")),
    ("y", Array(dtype=float64, ndim=1, layout="C")),
    ("how", string),
    ("acc", boolean),
    ("i", int64),
    ("out", int64),
]


@jitclass(spec)
class SingleSig:
    def __init__(self, x, y, how, acc):
        '''
        x,y: np.array
        how: string - determines type of signal
        acc: boolean - if the signal has 'memory'
        out: int - accumulator
        '''
        self.x = x
        self.y = y
        self.how = how
        self.acc = acc
        self.out = 0

    def sig(self, i):
        '''
        i: int - point in time
        Returns either signal or accumulator
        '''
        if self.how == "crossup":
            out = crossup(self.x, self.y, i)
        elif self.how == "crossdown":
            out = crossdown(self.x, self.y, i)
        if self.acc:
            self.out += out
        else:
            self.out = out
        return self.out
  1. Parse tuple of tuples into single value (this is where it becomes apparent things go wrong):
@njit
def reduce_sig(sig, i):
    '''
    Parses multisignal
    sig: homogeneous tuple of tuples ("human-readable" signal definition)
    i: int - point in time
    Returns: resulting value of multisignal
    '''
    L = len(sig)
    s, logic = sig[0]
    out = SingleSig(*s).sig(i)
    for cnt in range(1, L):
        s = SingleSig(*sig[cnt][0]).sig(i)
        out = out | s if logic == 'or_' else out & s
        logic = sig[cnt][1]
    return out
  1. Finally run the whole exercise altogether:
@njit
def run(x, entry_sig, exit_sig):
    '''
    x: np.array
    entry_sig, exit_sig: homogeneous tuples of tuples
    Returns: sequence of 0 and 1 satisfying entry and exit sigs
    '''
    L = x.shape[0]
    out = np.empty(L)
    out[0] = 0.0
    out[-1] = 0.0
    i = 1
    trade = True
    while i < L-1:
        out[i] = 0.0
        if reduce_sig(entry_sig,i) and i<L-1:
            out[i] = 1.0
            trade = True
            while trade and i<L-2:
                i += 1
                out[i] = 1.0
                if reduce_sig(exit_sig,i):
                    trade = False
        i+= 1
    return out
run(x, entry_sig, exit_sig)

So here, it would be better to instantiate an object of say MultiSig class and query it within loop, which is basically one of the possible solutions I seek.

Any hints, as well as a solution to this problem, are appreciated.



from Numba: how to parse arbitrary logic string into sequence of jitclassed instances in a loop

No comments:

Post a Comment