Skip to content

candle

candle

Classes:

Name Description
Candle

Candle

Candle(
    open: float,
    high: float,
    low: float,
    close: float,
    volume: int,
    timestamp: Optional[datetime | str] = None,
    timeframe: Optional[str | TimeFrame | timedelta | int] = None,
    indicators: Optional[
        Dict[str, float | Dict[str, float | None] | None]
    ] = None,
    sub_indicators: Optional[
        Dict[str, float | Dict[str, float | None] | None]
    ] = None,
)

Methods:

Name Description
from_dict

Create a Candle object from a dictionary representation.

from_dicts

Create's a list of Candle object's from a list of dictionary representation.

from_list

Create a Candle object from a list representation.

from_lists

Create a list of Candle object's from a list of list representation.

merge

Merge another Candle object into the current candle.

Source code in hexital/core/candle.py
def __init__(
    self,
    open: float,
    high: float,
    low: float,
    close: float,
    volume: int,
    timestamp: Optional[datetime | str] = None,
    timeframe: Optional[str | TimeFrame | timedelta | int] = None,
    indicators: Optional[Dict[str, float | Dict[str, float | None] | None]] = None,
    sub_indicators: Optional[Dict[str, float | Dict[str, float | None] | None]] = None,
):
    self.open = open
    self.high = high
    self.low = low
    self.close = close
    self.volume = volume
    self.timeframe = convert_timeframe_to_timedelta(timeframe) if timeframe else None

    if isinstance(timestamp, datetime):
        self.timestamp = timestamp
    elif isinstance(timestamp, str):
        self.timestamp = datetime.fromisoformat(timestamp)
    else:
        if UTC:
            self.timestamp = datetime.now(UTC)
        else:
            self.timestamp = datetime.utcnow()

    self._clean_values = {}
    self.indicators = indicators if indicators else {}
    self.sub_indicators = sub_indicators if sub_indicators else {}

from_dict classmethod

from_dict(candle: Dict[str, Any]) -> Candle

Create a Candle object from a dictionary representation.

The dictionary is expected to have the following keys: - Required: 'open', 'high', 'low', 'close', 'volume' - Optional: 'timestamp', 'time', 'date' - Optional: 'timeframe'

The method extracts the values for these keys. If the optional keys for time ('timestamp', 'time', etc.) are present, the first match is used as the timestamp.

Parameters:

Name Type Description Default
candle Dict[str, Any]

A dictionary containing the candle data.

required

Returns:

Name Type Description
Candle Candle

A Candle object initialized with the provided dictionary data.

Source code in hexital/core/candle.py
@classmethod
def from_dict(cls, candle: Dict[str, Any]) -> Candle:
    """
    Create a `Candle` object from a dictionary representation.

    The dictionary is expected to have the following keys:
    - Required: 'open', 'high', 'low', 'close', 'volume'
    - Optional: 'timestamp', 'time', 'date'
    - Optional: 'timeframe'

    The method extracts the values for these keys. If the optional keys for time ('timestamp',
    'time', etc.) are present, the first match is used as the timestamp.

    Args:
        candle (Dict[str, Any]): A dictionary containing the candle data.

    Returns:
        Candle: A `Candle` object initialized with the provided dictionary data.
    """
    time = [
        v
        for k, v in candle.items()
        if k in ["timestamp", "Timestamp", "time", "Time", "date", "Date"]
    ]
    return cls(
        candle.get("open", candle.get("Open", 0.0)),
        candle.get("high", candle.get("High", 0.0)),
        candle.get("low", candle.get("Low", 0.0)),
        candle.get("close", candle.get("Close", 0.0)),
        candle.get("volume", candle.get("Volume", 0)),
        timestamp=time[0] if time else None,
        timeframe=candle.get("timeframe", candle.get("Timeframe")),
    )

from_dicts staticmethod

from_dicts(candles: List[Dict[str, float]]) -> List[Candle]

Create's a list of Candle object's from a list of dictionary representation.

Each dictionary is expected to have the following keys: - Required: 'open', 'high', 'low', 'close', 'volume' - Optional: 'timestamp', 'time', 'date' - Optional: 'timeframe'

The method extracts the values for these keys. If the optional keys for time ('timestamp', 'time', etc.) are present, the first match is used as the timestamp. Returning a list of Candle objects initialized with the provided dictionary data.

Parameters:

Name Type Description Default
candles List[Dict[str, Any]]

A dictionary containing the candle data.

required

Returns:

Type Description
List[Candle]

List[Candle]: A list of Candle object's.

Source code in hexital/core/candle.py
@staticmethod
def from_dicts(candles: List[Dict[str, float]]) -> List[Candle]:
    """
    Create's a list of `Candle` object's from a list of dictionary representation.

    Each dictionary is expected to have the following keys:
    - Required: 'open', 'high', 'low', 'close', 'volume'
    - Optional: 'timestamp', 'time', 'date'
    - Optional: 'timeframe'

    The method extracts the values for these keys. If the optional keys for time ('timestamp',
    'time', etc.) are present, the first match is used as the timestamp.
    Returning a list of `Candle` objects initialized with the provided dictionary data.

    Args:
        candles (List[Dict[str, Any]]): A dictionary containing the candle data.

    Returns:
        List[Candle]: A list of `Candle` object's.
    """
    return [Candle.from_dict(candle) for candle in candles]

from_list classmethod

from_list(candle: list) -> Candle

Create a Candle object from a list representation.

The list is expected to contain the following elements: - Required: [open, high, low, close, volume] in that order. - Optional: A timestamp at the beginning of the list. - Optional: A timeframe at the end of the list.

If the first element is a str or datetime, it is treated as the timestamp. If the last element is a str, int, TimeFrame, or timedelta, it is treated as the timeframe.

Parameters:

Name Type Description Default
candle list

A list containing the candle data.

required

Returns:

Name Type Description
Candle Candle

A Candle object initialized with the data from the list.

Source code in hexital/core/candle.py
@classmethod
def from_list(cls, candle: list) -> Candle:
    """
    Create a `Candle` object from a list representation.

    The list is expected to contain the following elements:
    - Required: `[open, high, low, close, volume]` in that order.
    - Optional: A `timestamp` at the beginning of the list.
    - Optional: A `timeframe` at the end of the list.

    If the first element is a `str` or `datetime`, it is treated as the `timestamp`.
    If the last element is a `str`, `int`, `TimeFrame`, or `timedelta`, it is treated as the `timeframe`.

    Args:
        candle (list): A list containing the candle data.

    Returns:
        Candle: A `Candle` object initialized with the data from the list.
    """
    timestamp = None
    timeframe = None

    if len(candle) > 5 and isinstance(candle[0], (str, datetime)):
        timestamp = candle.pop(0)
    if len(candle) > 5 and isinstance(candle[-1], (str, int, TimeFrame, timedelta)):
        timeframe = candle.pop(-1)

    return cls(
        open=candle[0],
        high=candle[1],
        low=candle[2],
        close=candle[3],
        volume=candle[4],
        timestamp=timestamp,
        timeframe=timeframe,
    )

from_lists staticmethod

from_lists(candles: List[list]) -> List[Candle]

Create a list of Candle object's from a list of list representation.

Each list is expected to contain the following elements: - Required: [open, high, low, close, volume] in that order. - Optional: A timestamp at the beginning of the list. - Optional: A timeframe at the end of the list.

If the first element is a str or datetime, it is treated as the timestamp. If the last element is a str, int, TimeFrame, or timedelta, it is treated as the timeframe.

Parameters:

Name Type Description Default
candles List[list]

A list of list's containing the candle data.

required

Returns:

Type Description
List[Candle]

List[Candle]: A list of Candle object's.

Source code in hexital/core/candle.py
@staticmethod
def from_lists(candles: List[list]) -> List[Candle]:
    """
    Create a list of `Candle` object's from a list of list representation.

    Each list is expected to contain the following elements:
    - Required: `[open, high, low, close, volume]` in that order.
    - Optional: A `timestamp` at the beginning of the list.
    - Optional: A `timeframe` at the end of the list.

    If the first element is a `str` or `datetime`, it is treated as the `timestamp`.
    If the last element is a `str`, `int`, `TimeFrame`, or `timedelta`, it is treated as the `timeframe`.

    Args:
        candles (List[list]): A list of list's containing the candle data.

    Returns:
        List[Candle]: A list of `Candle` object's.
    """
    return [Candle.from_list(candle) for candle in candles]

merge

merge(candle: Candle)

Merge another Candle object into the current candle.

This method updates the current candle by integrating data from the provided candle. It ensures that the merged values respect the timeframe boundaries and adjusts attributes such as open, high, low, close, volume, and timestamps accordingly.

Note: - Any calculated indicators will be wiped, as merging modifies the core candle values. - Any conversion or derived values associated with the candle will also be removed.

Parameters:

Name Type Description Default
candle Candle

The Candle object to merge into the current candle.

required
Behaviour
  • Adjusts the open if the merged candle's timestamp is earlier than the start timestamp.
  • Updates the close if the merged candle's timestamp is more recent.
  • Updates high and low based on the maximum and minimum values of the two candles.
  • Increases the volume by the volume of the merged candle.
  • Increments the aggregation_factor to account for the merged data.
  • Resets calculated indicators and cleans any derived values.
Source code in hexital/core/candle.py
def merge(self, candle: Candle):
    """
    Merge another `Candle` object into the current candle.

    This method updates the current candle by integrating data from the provided `candle`.
    It ensures that the merged values respect the timeframe boundaries and adjusts
    attributes such as open, high, low, close, volume, and timestamps accordingly.

    **Note:**
    - Any calculated indicators will be wiped, as merging modifies the core candle values.
    - Any conversion or derived values associated with the candle will also be removed.

    Args:
        candle (Candle): The `Candle` object to merge into the current candle.

    Behaviour:
        - Adjusts the `open` if the merged candle's timestamp is earlier than the start timestamp.
        - Updates the `close` if the merged candle's timestamp is more recent.
        - Updates `high` and `low` based on the maximum and minimum values of the two candles.
        - Increases the `volume` by the volume of the merged candle.
        - Increments the `aggregation_factor` to account for the merged data.
        - Resets calculated indicators and cleans any derived values.
    """

    if self.timeframe:
        if (candle.timestamp + self.timeframe > self.timestamp + self.timeframe) or (
            candle.timestamp < self.timestamp - self.timeframe
        ):
            return

    self.recover_clean_values()

    if self._start_timestamp and candle.timestamp < self._start_timestamp:
        self.open = candle.open
        if not self._end_timestamp:
            self._end_timestamp = self._start_timestamp
        self._start_timestamp = candle.timestamp
    elif (
        self._start_timestamp
        and not self._end_timestamp
        and candle.timestamp > self._start_timestamp
    ):
        self.close = candle.close
        self._end_timestamp = candle.timestamp
    elif (
        self._start_timestamp
        and self._end_timestamp
        and candle.timestamp > self._end_timestamp
    ):
        self.close = candle.close
        self._end_timestamp = candle.timestamp
    elif (
        self._start_timestamp
        and self._end_timestamp
        and self._start_timestamp > candle.timestamp > self._end_timestamp
    ):
        pass
    else:
        self.close = candle.close

    self.high = max(self.high, candle.high)
    self.low = min(self.low, candle.low)
    self.volume += candle.volume
    self.aggregation_factor += candle.aggregation_factor

    self.reset_candle()