Signal#

Signal generation from spread residuals.

Direction#

class spreadpy.signal.signal.Direction(*values)[source]#

Spread position direction.

LONG (+1): buy y, sell x — spread expected to rise (revert upward). SHORT (−1): sell y, buy x — spread expected to fall (revert downward). FLAT (0): no position / exit signal.

Signal#

class spreadpy.signal.signal.Signal(direction: Direction, zscore: float, timestamp: Timestamp, prob: float = nan, is_entry: bool = False)[source]#

Output of a SignalGenerator at a single bar.

Parameters:

  • direction (Direction) – Desired position direction (LONG, SHORT, or FLAT).

  • zscore (float) – Z-score of the spread at signal time, used for position sizing.

  • timestamp (pd.Timestamp) – Bar timestamp.

  • prob (float) – Conditional mean-reversion probability from a copula signal, or nan for z-score signals.

  • is_entry (bool) – True for new position entries; False for holds or exits.

SignalGenerator#

Abstract base class for all signal generators.

class spreadpy.signal.signal.SignalGenerator[source]#

Abstract base class for signal generators.

Enforces the fit / generate discipline required for walk-forward backtesting: fit() is called on in-sample data to calibrate any parameters, then generate() is called on out-of-sample data to produce signals without lookahead.

abstractmethod fit(spread: SpreadSeries) → SignalGenerator[source]#

Calibrate parameters on in-sample spread data.

Must be called before generate(). Implementations should compute any statistics (thresholds, copula parameters, etc.) from spread and store them as instance attributes.

Parameters:

spread (SpreadSeries) – In-sample spread series used for fitting.

Returns:

self (for method chaining).

Return type:

SignalGenerator

abstractmethod generate(spread: SpreadSeries) → Series[source]#

Generate a Signal for each bar of spread.

Should be called after fit(). spread may be the same in-sample series or a disjoint out-of-sample window; in either case no lookahead is permitted — signal at bar t may only depend on observations up to and including t.

Parameters:

spread (SpreadSeries) – Spread series to generate signals for.

Returns:

Series of Signal objects indexed by spread.index.

Return type:

pd.Series

ZScoreSignal#

class spreadpy.signal.zScoreSignal.ZScoreSignal(window: int = 60, entry_threshold: float = 1.0, revert_threshold: float = 0.0)[source]#

Classic z-score entry / exit signal generator.

At each bar, computes a rolling z-score of the spread residuals:

\[z_t = \frac{s_t - \hat{\mu}_t}{\hat{\sigma}_t}\]

where \(\hat{\mu}_t\) and \(\hat{\sigma}_t\) are the rolling mean and standard deviation over the last window bars (no lookahead).

Entry / exit rules:

LONG   if z_t < -entry_threshold
SHORT  if z_t > +entry_threshold
FLAT   if LONG  and z_t > -revert_threshold  (z reverted back up)
FLAT   if SHORT and z_t < +revert_threshold  (z reverted back down)

Setting revert_threshold=0 exits at the mean crossing (\(z\) crosses 0). Setting it to a positive value exits before the mean is fully reached.

Parameters:

  • window (int) – Number of bars for the rolling z-score computation.

  • entry_threshold (float) – \(|z|\) level above which a position is opened.

  • revert_threshold (float) – \(z\) level at which mean reversion is considered complete and the position is closed. Must be \(\leq\) entry_threshold. Use 0.0 to exit at the mean (default).

fit(spread: SpreadSeries) → ZScoreSignal[source]#

Compute in-sample mean and standard deviation of the spread residuals.

These statistics are stored but not used by generate(), which relies on the rolling estimators instead. Calling fit is required by the SignalGenerator interface.

Parameters:

spread (SpreadSeries) – In-sample spread series.

Returns:

self.

Return type:

ZScoreSignal

generate(spread: SpreadSeries) → Series[source]#

Compute the rolling z-score and map each bar to a Signal.

At each bar \(t\) the z-score is:

\[z_t = \frac{s_t - \hat{\mu}_{t,w}}{\hat{\sigma}_{t,w}}\]

where \(\hat{\mu}_{t,w}\) and \(\hat{\sigma}_{t,w}\) are the rolling mean and standard deviation over the previous window bars (no lookahead). Bars with fewer than window predecessors yield Direction.FLAT with zscore=NaN.

Parameters:

spread (SpreadSeries) – Spread series to generate signals for (may be out-of-sample).

Returns:

Series of Signal objects indexed by spread.index.

Return type:

pd.Series

RollingADFFilter#

class spreadpy.signal.rollingADFFilter.RollingADFFilter(base: SignalGenerator, adf_window: int = 120, p_threshold: float = 0.05, min_confirm: int = 1)[source]#

Wraps any SignalGenerator and suppresses new entries when the spread fails a rolling ADF stationarity test.

At each bar t, the ADF test is evaluated on the preceding adf_window bars of the spread residuals. A new position is only opened if the p-value is below p_threshold (i.e. the spread is stationary).

Exits and holds are always passed through unchanged. If an entry is blocked and the spread becomes stationary while still in the entry zone, the position opens at the first bar where the confirmation condition is met — with is_entry=True so sizers compute a fresh size.

To reduce false signals from a noisy p-value, set min_confirm > 1: entry is only allowed after min_confirm consecutive bars where the p-value stayed below p_threshold.

Usage:

signal_gen = RollingADFFilter(
    base=ZScoreSignal(window=60, entry_threshold=1.5),
    adf_window=120,
    p_threshold=0.05,
    min_confirm=3,
)
Parameters:

  • base (SignalGenerator) – Underlying signal generator to wrap.

  • adf_window (int) – Number of bars for the rolling ADF window.

  • p_threshold (float) – Maximum p-value to allow an entry (default 0.05).

  • min_confirm (int) – Number of consecutive bars the p-value must stay above p_threshold to block an entry (default 1). With the default of 1, any bar above the threshold blocks entry. Increase to require sustained non-stationarity before blocking — a single noisy bar above the threshold will not prevent entry.

fit(spread: SpreadSeries) → RollingADFFilter[source]#

Calibrate parameters on in-sample spread data.

Must be called before generate(). Implementations should compute any statistics (thresholds, copula parameters, etc.) from spread and store them as instance attributes.

Parameters:

spread (SpreadSeries) – In-sample spread series used for fitting.

Returns:

self (for method chaining).

Return type:

SignalGenerator

generate(spread: SpreadSeries) → Series[source]#

Generate a Signal for each bar of spread.

Should be called after fit(). spread may be the same in-sample series or a disjoint out-of-sample window; in either case no lookahead is permitted — signal at bar t may only depend on observations up to and including t.

Parameters:

spread (SpreadSeries) – Spread series to generate signals for.

Returns:

Series of Signal objects indexed by spread.index.

Return type:

pd.Series