Plugin

Seasonality

How the instrument did in past months, weekdays, turns of the month or hours like the current one: mean return, hit rate, count and t-statistic.

IndicatorNo networkDesktop onlyv1.0.0

Get early access Plugin API documentation Built in indicators

About

Some markets are said to do better in some months, on some weekdays, around the turn of the month or at some hours. This indicator measures that on the run's own data as it goes: for the bucket the current bar is in (this calendar month, this weekday, turn of the month or not, this hour), the number of earlier periods in it, their mean return, the share that went up, and a t-statistic.

Only finished periods count. One observation is the close to close return of one finished hour, day or month; the period the current bar is in has not finished and is left out, so the bar never feeds the number it reads and a backtest cannot see ahead. Each figure therefore grows with the history: early in a run there is little to go on.

Be careful with what it says. A pattern found by looking at twelve months or seven weekdays is often luck, mean, hit_rate and t_stat stay None until a bucket has min_count observations, and the t-statistic assumes independent returns, which market returns rarely are. A t-statistic around 2 from a dozen observations is weak evidence.

Parameters

Name	Type	Default	Range or options	Description
`by`	choice	`"weekday"`	month, weekday, turn_of_month, hour	What makes periods alike. The turn of the month is the last weekday of a month and the first three weekdays of the next, by the calendar (holidays ignored).
`tz`	text	`"UTC"`		The clock days, hours and months are counted on. A time zone name (IANA), like Europe/Berlin or America/New_York.
`min_count`	int	`10`	min 2, max 1000	Observations a bucket needs before mean, hit_rate and t_stat appear.

Use it in a strategy

pythonsea = ctx.plugin_indicator("seasonality", "seasonality", {"by": "weekday", "min_count": 20})
if sea is not None and sea.t_stat is not None:
    if sea.t_stat > 2.0 and sea.hit_rate > 0.55:
        ctx.signal("LONG")
    elif sea.t_stat < 0:
        ctx.signal("EXIT")

The indicator library has this as seasonality: ctx.ind("seasonality", by="weekday", zone="utc", min_count=10) reads it on the clock of UTC, London, New York, Tokyo or Sydney, with the turn of the month as bucket 1 and the rest as 0, and the chart, Blocks, Rust and the Rust core have it too. This template stays as an example of a plugin indicator; only it takes any time zone.

What it provides

`value, mean`Value	The mean close to close return of the finished periods in the current bucket, as a fraction (0.001 is 0.1 %); None below min_count.
`count`Value	How many finished periods of the current bucket came before this bar; 0 when none.
`hit_rate`Value	The share of those periods with a return above zero, 0 to 1; None below min_count.
`t_stat`Value	mean / (sample standard deviation / sqrt(count)); None below min_count or when every return was the same.
`bucket`Value	The current bucket: month 1 to 12, weekday 0 (Monday) to 6, "turn" or "rest", hour 0 to 23.

Limits

It needs history: a month bucket gains one observation a year. Load enough data before the part of the run you judge, or results early in the run rest on almost nothing.
Returns are close to close between periods that had bars, so the first hour or day after a weekend includes the weekend gap. With forex data on UTC days the few Sunday evening hours form a day of their own.
The hour grouping needs bars of an hour or shorter. Nothing here says a pattern will continue; it describes the past of this data only.
A plugin indicator is not drawn on the chart and is not available in Blocks, Pine, Rust or C#; it runs on the run's own instrument only.

Preview

json{
  "bucket": 4,
  "count": 52,
  "hit_rate": 0.54,
  "mean": 0.00031,
  "t_stat": 0.81,
  "value": 0.00031
}

A made-up sample that shows the shape of the output. Not vendor data and not a result.

The code

main.py, the file the app installs.

python"""Seasonality: how the instrument did, in the past, in periods like the current one.

Choose what "like" means with `by`:

    month           the calendar month (March bars read every finished March)
    weekday         the day of the week
    turn_of_month   the turn of the month (the last weekday of a month and the
                    first three of the next) against all other days
    hour            the hour of the day

After every bar a strategy reads, for the bucket the bar is in:

    count      how many finished periods of this bucket came before
    mean       their mean return (also `value`)
    hit_rate   the share of them that went up
    t_stat     mean / (sample standard deviation / sqrt(count))
    bucket     the bucket: 1 to 12, 0 (Monday) to 6, "turn" or "rest", 0 to 23

One observation is the return of one finished period, close to close: the last
close of an hour, day or month against the last close of the one before it
that had bars. The period the current bar is in has not finished, so it never
counts: the bar never feeds the number it reads, and nothing from a later bar
can reach an earlier one.

Small samples are noisy, and seasonal effects found by looking at many buckets
are often luck. mean, hit_rate and t_stat stay None until a bucket has
`min_count` observations; count is always there. A t_stat around 2 from a
dozen observations is weak evidence, and it assumes the returns are
independent, which they rarely are.

Days, hours and months are on the clock of `tz` (UTC by default).
"""

import math
from datetime import timedelta

from orionquant_engine.plugins.kit import dates

GROUPINGS = ("month", "weekday", "turn_of_month", "hour")


def _field(bar, name):
    return bar[name] if isinstance(bar, dict) else getattr(bar, name)


def is_turn_of_month(day):
    """The last weekday of a month or one of the first three weekdays of a month.

    A calendar rule, so it is known in advance; exchange holidays are ignored.
    """
    if day.weekday() >= dates.SAT:
        return False
    last = dates.month_end(day.year, day.month)
    while last.weekday() >= dates.SAT:
        last -= timedelta(days=1)
    if day == last:
        return True
    first = day.replace(day=1)
    weekdays_so_far = sum(1 for d in dates.days(first, day) if d.weekday() < dates.SAT)
    return weekdays_so_far <= 3


class _Bucket:
    """Running count, mean, variance (Welford) and ups of one bucket's returns."""

    def __init__(self):
        self.n = 0
        self.mean = 0.0
        self.m2 = 0.0
        self.ups = 0

    def add(self, r):
        self.n += 1
        delta = r - self.mean
        self.mean += delta / self.n
        self.m2 += delta * (r - self.mean)
        if r > 0:
            self.ups += 1


class Seasonality:
    """update(bar), then read the properties listed in the module docstring."""

    def __init__(self, by="weekday", tz="UTC", min_count=10):
        if by not in GROUPINGS:
            raise ValueError(f"by must be one of {', '.join(GROUPINGS)}")
        self.by = by
        self.tz = tz
        dates.zone(tz)  # an unknown zone fails here, not on the first bar
        self.min_count = max(2, int(min_count))
        self._buckets = {}
        self._period = None
        self._last_close = None
        self._prev_period_close = None
        self.bucket = None

    def _period_of(self, ts):
        local = dates.to_local(ts, self.tz)
        if self.by == "hour":
            return (local.date(), local.hour)
        if self.by == "month":
            return (local.year, local.month)
        return local.date()

    def _bucket_of(self, period):
        if self.by in ("hour", "month"):
            return period[1]
        if self.by == "weekday":
            return period.weekday()
        return "turn" if is_turn_of_month(period) else "rest"

    def update(self, bar):
        close = float(_field(bar, "close"))
        period = self._period_of(_field(bar, "ts"))
        if period != self._period:
            if self._period is not None:
                self._finish(self._period)
            self._period = period
            self.bucket = self._bucket_of(period)
        self._last_close = close

    def _finish(self, period):
        """The period before this bar's has ended: its return joins its bucket."""
        last, prev = self._last_close, self._prev_period_close
        if prev is not None and prev > 0:
            key = self._bucket_of(period)
            self._buckets.setdefault(key, _Bucket()).add(last / prev - 1.0)
        self._prev_period_close = last

    def _current(self):
        b = self._buckets.get(self.bucket)
        return b if b is not None and b.n >= self.min_count else None

    @property
    def count(self):
        b = self._buckets.get(self.bucket)
        return b.n if b is not None else 0

    @property
    def mean(self):
        b = self._current()
        return b.mean if b else None

    @property
    def value(self):
        return self.mean

    @property
    def hit_rate(self):
        b = self._current()
        return b.ups / b.n if b else None

    @property
    def t_stat(self):
        b = self._current()
        if b is None:
            return None
        variance = b.m2 / (b.n - 1)
        if variance <= 0:
            return None
        return b.mean / math.sqrt(variance / b.n)


def make(params):
    params = params or {}
    return Seasonality(
        by=params.get("by", "weekday"),
        tz=params.get("tz", "UTC"),
        min_count=params.get("min_count", 10),
    )


def register(api):
    api.register_indicator(
        "seasonality",
        "Seasonality",
        make,
        [
            {"key": "by", "label": "Group by", "type": "choice", "default": "weekday", "options": list(GROUPINGS)},
            {"key": "tz", "label": "Time zone", "type": "text", "default": "UTC"},
            {"key": "min_count", "label": "Minimum observations", "type": "int", "default": 10, "min": 2,
             "max": 1000},
        ],
    )