candle_manager

candle_manager

Classes:

Name	Description
CandleManager

CandleManager

CandleManager(
    candles: list[Candle] | None = None,
    candle_life: timedelta | None = None,
    timeframe: timedelta | None = None,
    timeframe_fill: bool = False,
    candlestick: CandlestickType | None = None,
)

Methods:

Name	Description
purge	Remove this indicator value from all Candles
resample_candles	Resample candles into the manager timeframe.
sort_candles	Sorts Candles in order of timestamp, accounts for collapsing

Attributes:

Name	Type	Description
candles	list[Candle]	Returns reference to the managed candles list.

Source code in hexital/core/candle_manager.py

def __init__(
    self,
    candles: list[Candle] | None = None,
    candle_life: timedelta | None = None,
    timeframe: timedelta | None = None,
    timeframe_fill: bool = False,
    candlestick: CandlestickType | None = None,
):
    self.candle_life = candle_life
    self.timeframe = timeframe
    self.timeframe_fill = timeframe_fill
    self._candles = []
    if candles:
        self._candles.extend(candles)

    if candlestick:
        self.candlestick = candlestick
        self.candlestick.set_candle_refs(self._candles)

    if self._candles:
        self._candle_tasks()

candles property writable

candles: list[Candle]

Returns reference to the managed candles list.

WARNING: This is a live reference, not a copy. Modifications affect all indicators sharing this manager. This is intentional to enable indicator chaining (e.g., one indicator reading another's results).

purge

purge(indicator: str | set[str])

Remove this indicator value from all Candles

Source code in hexital/core/candle_manager.py

def purge(self, indicator: str | set[str]):
    """Remove this indicator value from all Candles"""
    if isinstance(indicator, str):
        indicator = {indicator}

    for candle in self.candles:
        for name in indicator:
            candle.indicators.pop(name, None)
            candle.sub_indicators.pop(name, None)

resample_candles

resample_candles(
    mode: CalcMode, index: int | None = None, appended_count: int = 0
)

Resample candles into the manager timeframe.

Source code in hexital/core/candle_manager.py

def resample_candles(
    self,
    mode: CalcMode,
    index: int | None = None,
    appended_count: int = 0,
):
    """Resample candles into the manager timeframe."""
    tf = self.timeframe
    if not tf:
        return

    if not self.timeframe_fill:
        if mode == CalcMode.APPEND and appended_count >= 1:
            if appended_count == 1 and self._try_resample_append_inplace():
                return
            if appended_count > 1 and self._try_resample_multi_append(appended_count):
                return

    if mode == CalcMode.INSERT:
        start_index = 0
    elif index is not None:
        start_index = index
    else:
        start_index = self._find_resample_index()

    end_index = len(self._candles)
    if end_index <= start_index + 1:
        return

    if not self.timeframe_fill and self._range_already_resampled(
        start_index, end_index
    ):
        return

    self._resample_rebuild_range(start_index, end_index, mode)

sort_candles

sort_candles(candles: list[Candle] | None = None)

Sorts Candles in order of timestamp, accounts for collapsing

Source code in hexital/core/candle_manager.py

def sort_candles(self, candles: list[Candle] | None = None):
    """Sorts Candles in order of timestamp, accounts for collapsing"""
    if candles:
        target = candles
    else:
        target = self._candles

    if not self.timeframe:
        target.sort(
            key=lambda c: (
                c.timestamp is None,
                c.timestamp.timestamp() if c.timestamp else 0.0,
            )
        )
    else:
        target.sort(key=cmp_to_key(self._sort_comparison))