timeframe

TimeFrame

TimeFrame(tf: int | str | list | Timedelta | TimeFrame)

DataFeed TimeFrame

Parameters:

• tf (int | str | list | Timedelta | TimeFrame) –
  • int: TimeFrame in minutes. Example: TimeFrame(5) == TimeFrame("5m")
  • str: string format of TimeFrame in s, m, h, d, w.

Raises: RuntimeError: description

Source code in lettrade/data/timeframe.py

def __init__(self, tf: "int | str | list | pd.Timedelta | TimeFrame") -> None:
    """_summary_

    Args:
        tf (int | str | list | pd.Timedelta | TimeFrame):
            - `int`: TimeFrame in minutes. Example: TimeFrame(5) == TimeFrame("5m")
            - `str`: string format of TimeFrame in `s`, `m`, `h`, `d`, `w`.
    Raises:
        RuntimeError: _description_
    """
    if isinstance(tf, TimeFrame):
        self.value = tf.value
        self.unit = tf.unit
    elif isinstance(tf, str):
        match = _pattern_timeframe_str.search(tf)
        if not match:
            raise RuntimeError(f"TimeFrame value {tf} is invalid")

        self.value = int(match.group(1))
        self.unit = match.group(2)
    elif isinstance(tf, int):
        self.value = tf
        self.unit = "m"
    elif isinstance(tf, list):
        self.value = int(tf[0])
        self.unit = tf[1]
    elif isinstance(tf, pd.Timedelta):
        map = TIMEFRAME_DELTA_2_STR[tf]
        self.value = map[0]
        self.unit = map[1]
    else:
        raise RuntimeError(f"Timeframe {tf} is invalid format")

    # Validate
    self._validate()

    # Setup
    self.unit_pandas = TIMEFRAME_UNIT_LET_2_PANDAS[self.unit]
    self.delta = pd.Timedelta(self.value, self.unit_pandas)

    # Warning
    if self.delta not in TIMEFRAME_DELTA_2_STR:
        if self.unit not in ["s", "m"]:
            logger.warning(
                "Unsupport TimeFrame(%s), some function may not work floor()/ceil()...",
                self.delta,
            )

string property

string

TimeFrame as lettrade string

string_pandas property

string_pandas

TimeFrame as pandas string

ceil

ceil(
    at: datetime | timedelta | Timestamp | Timedelta,
) -> Timestamp | Timedelta

Get ceil of TimeFrame

Parameters:

• at (datetime | timedelta | Timestamp | Timedelta) –

  description

Raises:

• RuntimeError –

  description

Returns:

• Timestamp | Timedelta –

  pd.Timestamp: description

Source code in lettrade/data/timeframe.py

def ceil(
    self, at: datetime | timedelta | pd.Timestamp | pd.Timedelta
) -> pd.Timestamp | pd.Timedelta:
    """Get ceil of TimeFrame

    Args:
        at (datetime | timedelta | pd.Timestamp | pd.Timedelta): _description_

    Raises:
        RuntimeError: _description_

    Returns:
        pd.Timestamp: _description_
    """
    if isinstance(at, datetime):
        at = pd.Timestamp(at)
    elif isinstance(at, timedelta):
        at = pd.Timedelta(at)

    if self.unit in ["h", "m", "s"]:
        freq = self.string_pandas
        if self.unit == "m":
            freq += "in"
        return at.ceil(freq=freq)

    if isinstance(at, pd.Timestamp):
        if self.unit == "d":
            return pd.Timestamp(at.date() + pd.Timedelta(days=1))
        if self.unit == "w":
            return pd.Timestamp(at.date() + pd.Timedelta(days=7 - at.day_of_week))

    elif isinstance(at, pd.Timedelta):
        if self.unit == "d":
            return pd.Timedelta(days=at.days + (0 if at.seconds == 0 else 1))
        if self.unit == "w":
            weeks = math.ceil(at.days / 7)
            return pd.Timedelta(weeks=weeks)

    raise RuntimeError(f"Unit {self.unit} is not implement yet for {at}")

floor

floor(
    at: datetime | timedelta | Timestamp | Timedelta,
) -> Timestamp | Timedelta

Get floor of TimeFrame

Parameters:

• at (datetime | timedelta | Timestamp | Timedelta) –

  description

Raises:

• RuntimeError –

  description

Returns:

• Timestamp | Timedelta –

  pd.Timestamp | pd.Timedelta: description

Source code in lettrade/data/timeframe.py

def floor(
    self, at: datetime | timedelta | pd.Timestamp | pd.Timedelta
) -> pd.Timestamp | pd.Timedelta:
    """Get floor of TimeFrame

    Args:
        at (datetime | timedelta | pd.Timestamp | pd.Timedelta): _description_

    Raises:
        RuntimeError: _description_

    Returns:
        pd.Timestamp | pd.Timedelta: _description_
    """
    if isinstance(at, datetime):
        at = pd.Timestamp(at)
    elif isinstance(at, timedelta):
        at = pd.Timedelta(at)

    if self.unit in ["h", "m", "s"]:
        freq = self.string_pandas
        if self.unit == "m":
            freq += "in"
        return at.floor(freq=freq)

    if isinstance(at, pd.Timestamp):
        if self.unit == "d":
            return pd.Timestamp(at.date())
        if self.unit == "w":
            return pd.Timestamp(at.date() - pd.Timedelta(days=at.day_of_week))

    elif isinstance(at, pd.Timedelta):
        if self.unit == "d":
            return pd.Timedelta(days=at.days)
        if self.unit == "w":
            weeks = math.floor(at.days / 7)
            return pd.Timedelta(weeks=weeks)

    raise RuntimeError(f"Unit {self.unit} is not implement yet for {at}")