candle_manager

candle_manager

Classes:

Name	Description
CandleManager

CandleManager

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

Methods:

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

Source code in hexital/core/candle_manager.py

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

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

    if self._candles:
        self._candle_tasks()

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: Optional[int] = None)

resamples the given list of candles into specific timeframe candles. This can re-ran with same list to resample latest candles. This method is destructive, generating a new list for the resampled candles

Source code in hexital/core/candle_manager.py

def resample_candles(
    self,
    mode: CalcMode,
    index: Optional[int] = None,
):
    """resamples the given list of candles into specific timeframe candles.
    This can re-ran with same list to resample latest candles.
    This method is destructive, generating a new list for the resampled candles"""
    if mode == CalcMode.INSERT:
        start_index = 0
    elif index is not None:
        start_index = index
    else:
        start_index = self._find_resample_index()

    if len(self._candles) <= start_index + 1 or not self.timeframe:
        return

    end_index = len(self._candles)

    candles_ = [self._candles.pop(start_index)]

    init_candle = candles_[0]
    init_candle.timeframe = self.timeframe
    if not init_candle.timestamp:
        return

    start_time = round_down_timestamp(init_candle.timestamp, self.timeframe)
    end_time = start_time + self.timeframe

    if not on_timeframe(init_candle.timestamp, self.timeframe):
        init_candle.set_resampled_timestamp(end_time)

    for _ in range(abs(end_index - start_index)):
        if len(self._candles) <= start_index:
            break

        candle = self._candles.pop(start_index)
        prev_candle = candles_[-1]

        if not candle.timestamp:
            return

        if (
            mode != CalcMode.INSERT
            and candle.timeframe == self.timeframe
            and prev_candle.timeframe == self.timeframe
        ):
            candles_.append(candle)
            break

        next_end_time = end_time + self.timeframe
        candle.timestamp = trim_timestamp(candle.timestamp)
        candle.timeframe = self.timeframe

        if start_time < candle.timestamp <= end_time and prev_candle.timestamp == end_time:
            prev_candle.merge(candle)
        elif (
            start_time - self.timeframe < candle.timestamp <= start_time
            and prev_candle.timestamp == start_time
        ):
            prev_candle.merge(candle)
        elif start_time < candle.timestamp <= end_time:
            candle.set_resampled_timestamp(end_time)
            candles_.append(candle)
        elif end_time < candle.timestamp <= next_end_time:
            candle.set_resampled_timestamp(next_end_time)
            candles_.append(candle)
            start_time = end_time
            end_time = next_end_time
        elif start_time < candle.timestamp and on_timeframe(candle.timestamp, self.timeframe):
            start_time = round_down_timestamp(candle.timestamp, self.timeframe)
            end_time = start_time + self.timeframe
            candle.set_resampled_timestamp(start_time)
            candles_.append(candle)
        elif next_end_time < candle.timestamp:
            start_time = round_down_timestamp(candle.timestamp, self.timeframe)
            end_time = start_time + self.timeframe
            candle.set_resampled_timestamp(end_time)
            candles_.append(candle)
        else:
            # Shit's fucked yo
            raise InvalidCandleOrder(
                f"Failed to resample_candles due to invalid candle order prev: [{prev_candle}] - current: [{candle}]",
            )

    if self.timeframe_fill:
        candles_ = self._fill_timeframe_candles(
            candles_, self.timeframe, start_index, end_index
        )

    self._candles[start_index:start_index] = candles_

sort_candles

sort_candles(candles: Optional[List[Candle]] = None)

Sorts Candles in order of timestamp, accounts for collapsing

Source code in hexital/core/candle_manager.py

def sort_candles(self, candles: Optional[List[Candle]] = None):
    """Sorts Candles in order of timestamp, accounts for collapsing"""
    if candles:
        candles.sort(key=cmp_to_key(self._sort_comparison))
    else:
        self._candles.sort(key=cmp_to_key(self._sort_comparison))