Helpers

Time-based utilities for periodic operations, particularly useful for trading automation and scheduled tasks.

Tip

The TimeTrigger class is designed for modulo-based scheduling (e.g., “every 4 hours aligned to midnight”) rather than interval-based scheduling (e.g., “every 4 hours from now”).

Quick overview

• TimeTrigger: Detect time boundaries for periodic operations (candle closes, scheduled restarts)

• Supports human-readable intervals: "4h", "15m", "1d"

• Configurable margin for early triggering

• Statistics tracking for trigger events

Usage patterns

Basic boundary detection

from kstlib.helpers import TimeTrigger

# Trigger at 4-hour boundaries (00:00, 04:00, 08:00, ...)
trigger = TimeTrigger("4h")

if trigger.is_at_boundary():
    print("We are at a 4-hour mark!")

# Check with margin (trigger 60s before boundary)
if trigger.should_trigger(margin=60):
    restart_connection()

Time calculations

from kstlib.helpers import TimeTrigger

trigger = TimeTrigger("15m")

# Seconds until next boundary
wait_time = trigger.time_until_next()
print(f"Next 15-minute mark in {wait_time:.0f} seconds")

# Get the next boundary as datetime
next_boundary = trigger.next_boundary_time()
print(f"Next boundary at {next_boundary}")

WebSocket restart scheduling

from kstlib.helpers import TimeTrigger
from kstlib.websocket import WebSocketManager

trigger = TimeTrigger("8h")

async def maybe_restart(ws: WebSocketManager) -> None:
    """Restart WebSocket at 8-hour boundaries for stability."""
    if trigger.should_trigger(margin=30):
        await ws.shutdown()
        await ws.connect()
        trigger.record_trigger()  # Track for statistics

Context manager for timed operations

from kstlib.helpers import TimeTrigger

with TimeTrigger("1h") as trigger:
    while True:
        if trigger.should_trigger():
            perform_hourly_task()
            trigger.record_trigger()
        await asyncio.sleep(1)

Interval formats

Format	Meaning
"15m"	15 minutes
"4h"	4 hours
"1d"	1 day (24 hours)
"30s"	30 seconds
3600	3600 seconds (int)

Module reference

Helper utilities for time-based operations.

This module provides utilities for time-based triggering and scheduling:

• TimeTrigger: Detect time boundaries (modulo-based) for periodic operations

Examples

Check if current time is at a 4-hour boundary:

>>> from kstlib.helpers import TimeTrigger
>>> trigger = TimeTrigger("4h")
>>> trigger.is_at_boundary()  
True  # If current time is 00:00, 04:00, 08:00, etc.

Get time until next boundary:

>>> trigger.time_until_next()  
3542.5  # Seconds until next 4-hour mark

Use with WebSocket for periodic restart:

>>> trigger = TimeTrigger("8h")
>>> if trigger.should_trigger(margin=60):  
...     await ws.shutdown()
...     await ws.connect()

exception kstlib.helpers.InvalidModuloError

Bases: TimeTriggerError

Raised when modulo string is invalid or out of bounds.

class kstlib.helpers.TimeTrigger(modulo, *, timezone='UTC')

Bases: object

Time-based trigger for detecting modulo boundaries.

Detects when current time aligns with periodic intervals (boundaries). Useful for coordinating operations with market candle closes or scheduling periodic restarts.

modulo

Original modulo string (e.g., “4h”).

modulo_seconds

Modulo duration in seconds.

stats

Trigger statistics.

Parameters:

• modulo (str) – Duration string for the interval (e.g., “30m”, “4h”, “8h”, “1d”).

• timezone (str) – Timezone for calculations (default: UTC).

Raises:

InvalidModuloError – If modulo format is invalid or out of bounds.

Examples

Create a 4-hour trigger:

>>> trigger = TimeTrigger("4h")
>>> trigger.modulo_seconds
14400

Check boundary status:

>>> trigger.is_at_boundary()  
False
>>> trigger.time_until_next()  
1234.5

Create with different timezone:

>>> trigger = TimeTrigger("1d", timezone="Europe/Paris")
>>> trigger.timezone
'Europe/Paris'

__init__(self, modulo: 'str', *, timezone: 'str' = 'UTC') → 'None' -> None

Initialize TimeTrigger.

Parameters:

• modulo (str) – Duration string (e.g., “30m”, “4h”, “1d”).

• timezone (str) – Timezone for boundary calculations.

property modulo: str

Return the original modulo string.

property modulo_seconds: int

Return the modulo duration in seconds.

property timezone: str

Return the timezone used for calculations.

property stats: TimeTriggerStats

Return trigger statistics.

time_until_next(self) → 'float' -> float

Calculate seconds until next boundary.

Returns:

Seconds remaining until the next modulo boundary.

Return type:

float

Examples

>>> trigger = TimeTrigger("4h")
>>> remaining = trigger.time_until_next()  
>>> 0 <= remaining <= 14400  
True

is_at_boundary(self, margin: 'float' = 1.0) → 'bool' -> bool

Check if current time is at a boundary.

A boundary is when the timestamp is divisible by the modulo. The margin allows for slight timing imprecision.

Parameters:

margin (float) – Tolerance in seconds around the boundary (default: 1.0).

Returns:

True if within margin seconds of a boundary.

Return type:

bool

Examples

>>> trigger = TimeTrigger("4h")
>>> trigger.is_at_boundary()  
True  # If time is 00:00:00, 04:00:00, etc.
>>> trigger.is_at_boundary(margin=5.0)  
True  # If time is within 5 seconds of boundary

should_trigger(self, margin: 'float' = 30.0) → 'bool' -> bool

Check if trigger should fire (boundary approaching).

Use this to prepare for an upcoming boundary (e.g., start shutdown sequence before the boundary hits).

Parameters:

margin (float) – Seconds before boundary to trigger (default: 30.0).

Returns:

True if boundary is within margin seconds.

Return type:

bool

Examples

>>> trigger = TimeTrigger("4h")
>>> if trigger.should_trigger(margin=60):  
...     print("Boundary in less than 60 seconds!")

next_boundary(self) → 'pendulum.DateTime' -> pendulum.DateTime

Get the datetime of the next boundary.

Returns:

Pendulum DateTime of the next boundary.

Return type:

DateTime

Examples

>>> trigger = TimeTrigger("4h")
>>> next_time = trigger.next_boundary()  
>>> print(next_time.to_iso8601_string())  
'2024-01-15T08:00:00+00:00'

previous_boundary(self) → 'pendulum.DateTime' -> pendulum.DateTime

Get the datetime of the previous boundary.

Returns:

Pendulum DateTime of the previous boundary.

Return type:

DateTime

Examples

>>> trigger = TimeTrigger("4h")
>>> prev_time = trigger.previous_boundary()  
>>> print(prev_time.to_iso8601_string())  
'2024-01-15T04:00:00+00:00'

async wait_for_boundary(self, margin: 'float' = 0.0) → 'None' -> None

Wait until the next boundary (async).

Sleeps until the next boundary minus the margin.

Parameters:

margin (float) – Seconds before boundary to wake up (default: 0.0).

Examples

>>> import asyncio
>>> trigger = TimeTrigger("30m")
>>> await trigger.wait_for_boundary()  
>>> print("Boundary reached!")  

async run_on_boundary(self, callback: 'Callable[[], None] | Callable[[], Awaitable[None]]', *, margin: 'float' = 0.0, run_immediately: 'bool' = False) → 'None' -> None

Run callback at each boundary (async loop).

Continuously waits for boundaries and invokes the callback. Call stop() to terminate the loop.

Parameters:

• callback (Callable[[], None] | Callable[[], Awaitable[None]]) – Function to call at each boundary (sync or async).

• margin (float) – Seconds before boundary to invoke callback.

• run_immediately (bool) – If True, run callback immediately before first wait.

Examples

>>> import asyncio
>>> async def restart():  
...     print("Restarting...")
>>> trigger = TimeTrigger("4h")
>>> task = asyncio.create_task(  
...     trigger.run_on_boundary(restart, margin=30)
... )
>>> # Later: trigger.stop()

stop(self) → 'None' -> None

Stop the boundary loop.

__repr__(self) → 'str' -> str

Return string representation.

async __aenter__(self) → 'Self' -> Self

Async context manager entry.

async __aexit__(self, exc_type: 'type[BaseException] | None', exc_val: 'BaseException | None', exc_tb: 'object') → 'None' -> None

Async context manager exit.

exception kstlib.helpers.TimeTriggerError

Bases: KstlibError

Base exception for TimeTrigger errors.