Skip to content

Events Module

Convert continuous signals to discrete entry/exit events.

events

Convert continuous signals into discrete entry events.

signal_to_events(df, sig_col='sig', mode='transitions')

Convert continuous signals into discrete entry events (Long/Short).

Parameters:

Name Type Description Default
df DataFrame

DataFrame containing signal column

 required
sig_col str

Name of signal column (expected values: +1 for long, -1 for short, 0 for neutral)

 'sig'
mode (transitions, levels)

'transitions': Fire events only on state changes (enter_long, enter_short) 'levels': Fire events on all long/short signals

 'transitions'

Returns:

Type Description
dict

Dictionary with keys: - 'enter_long': numpy array of integer indices where long entries occur - 'enter_short': numpy array of integer indices where short entries occur - 'exit_long': numpy array of integer indices where long exits occur (optional) - 'exit_short': numpy array of integer indices where short exits occur (optional)
Notes

Transitions are defined as: - enter_long: previous signal <= 0 and current signal > 0 - enter_short: previous signal >= 0 and current signal < 0 - exit_long: previous signal > 0 and current signal <= 0 - exit_short: previous signal < 0 and current signal >= 0

Source code in signal_analyzer/core/events.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def signal_to_events(
    df: pd.DataFrame,
    sig_col: str = "sig",
    mode: Literal["transitions", "levels"] = "transitions",
) -> dict[str, np.ndarray]:
    """
    Convert continuous signals into discrete entry events (Long/Short).

    Parameters
    ----------
    df : pd.DataFrame
        DataFrame containing signal column
    sig_col : str
        Name of signal column (expected values: +1 for long, -1 for short, 0 for neutral)
    mode : {'transitions', 'levels'}
        'transitions': Fire events only on state changes (enter_long, enter_short)
        'levels': Fire events on all long/short signals

    Returns
    -------
    dict
        Dictionary with keys:
        - 'enter_long': numpy array of integer indices where long entries occur
        - 'enter_short': numpy array of integer indices where short entries occur
        - 'exit_long': numpy array of integer indices where long exits occur (optional)
        - 'exit_short': numpy array of integer indices where short exits occur (optional)

    Notes
    -----
    Transitions are defined as:
    - enter_long: previous signal <= 0 and current signal > 0
    - enter_short: previous signal >= 0 and current signal < 0
    - exit_long: previous signal > 0 and current signal <= 0
    - exit_short: previous signal < 0 and current signal >= 0
    """
    sig = df[sig_col].to_numpy(dtype=float)
    n = len(sig)

    if mode == "transitions":
        # Compute state changes
        sig_prev = np.roll(sig, 1)
        sig_prev[0] = 0  # assume neutral start

        # Enter long: was not-long, now long
        enter_long = np.where((sig_prev <= 0) & (sig > 0))[0]

        # Enter short: was not-short, now short
        enter_short = np.where((sig_prev >= 0) & (sig < 0))[0]

        # Exit long: was long, now not-long
        exit_long = np.where((sig_prev > 0) & (sig <= 0))[0]

        # Exit short: was short, now not-short
        exit_short = np.where((sig_prev < 0) & (sig >= 0))[0]

        return {
            "enter_long": enter_long,
            "enter_short": enter_short,
            "exit_long": exit_long,
            "exit_short": exit_short,
        }

    elif mode == "levels":
        # All bars where signal is long/short
        enter_long = np.where(sig > 0)[0]
        enter_short = np.where(sig < 0)[0]

        return {
            "enter_long": enter_long,
            "enter_short": enter_short,
        }

    else:
        raise ValueError(f"Unknown mode: {mode}. Use 'transitions' or 'levels'.")