turtle_quant_1.strategies.helpers.data_units

turtle_quant_1.strategies.helpers.data_units

DataUnitConverter

convert_to_15m_data classmethod

convert_to_15m_data(
    symbol: str, data: DataFrame
) -> DataFrame

Convert 5M OHLCV data to 15-minute bars.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def convert_to_15m_data(cls, symbol: str, data: pd.DataFrame) -> pd.DataFrame:
    """Convert 5M OHLCV data to 15-minute bars."""
    return cls._resample_with_datetime_grouper(symbol, data, "15M", "15min")

convert_to_1d_data classmethod

convert_to_1d_data(
    symbol: str, data: DataFrame
) -> DataFrame

Convert 5M OHLCV data to daily bars.

The indices of the original DataFrame will be retained, i.e. the row number of a day will be the row number of the last timestamp of the day, so the row numbers will not be contiguous.

Parameters:

Name	Type	Description	Default
symbol	str	The symbol of the data.	required
data	DataFrame	The data to convert.	required

Returns:

Type	Description
DataFrame	The converted data.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def convert_to_1d_data(cls, symbol: str, data: pd.DataFrame) -> pd.DataFrame:
    """Convert 5M OHLCV data to daily bars.

    The indices of the original DataFrame will be retained,
    i.e. the row number of a day will be the row number of the last
    timestamp of the day, so the row numbers will not be contiguous.

    Args:
        symbol: The symbol of the data.
        data: The data to convert.

    Returns:
        The converted data.
    """
    if data.empty:
        return pd.DataFrame(
            columns=["datetime", "Open", "High", "Low", "Close", "Volume"]
        )

    cache_key = cls._get_cache_key(symbol, "1D")
    cached_data = get_global_cache().get(cache_key)
    if cached_data is not None:
        logger.debug(f"{cache_key} already exists. Using cached data...")
        return cached_data.loc[cached_data.index.intersection(data.index)]

    data = data.copy()
    data["datetime"] = pd.to_datetime(data["datetime"])
    data["date"] = data["datetime"].dt.date

    daily_groups = data.groupby(pd.Grouper(key="date"))

    daily_df = daily_groups.agg(
        {
            "datetime": "last",
            "Open": "first",
            "High": "max",
            "Low": "min",
            "Close": "last",
            "Volume": "sum",
        }
    )

    # pyrefly: ignore
    last_indices = daily_groups.apply(lambda x: x.index[-1], include_groups=False)
    daily_df.index = last_indices.values
    daily_df = daily_df[["datetime", "Open", "High", "Low", "Close", "Volume"]]

    return daily_df

convert_to_1h_data classmethod

convert_to_1h_data(
    symbol: str, data: DataFrame
) -> DataFrame

Convert 5M OHLCV data to 1-hour bars.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def convert_to_1h_data(cls, symbol: str, data: pd.DataFrame) -> pd.DataFrame:
    """Convert 5M OHLCV data to 1-hour bars."""
    return cls._resample_with_datetime_grouper(symbol, data, "1H", "1h")

convert_to_1w_data classmethod

convert_to_1w_data(
    symbol: str, data: DataFrame
) -> DataFrame

Convert 5M OHLCV data to weekly bars (week ending on Friday).

The indices of the original DataFrame will be retained, i.e. the row number of a week will be the row number of the last timestamp of the week, so the row numbers will not be contiguous.

Parameters:

Name	Type	Description	Default
symbol	str	The symbol of the data.	required
data	DataFrame	The data to convert.	required

Returns:

Type	Description
DataFrame	The converted data.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def convert_to_1w_data(cls, symbol: str, data: pd.DataFrame) -> pd.DataFrame:
    """Convert 5M OHLCV data to weekly bars (week ending on Friday).

    The indices of the original DataFrame will be retained,
    i.e. the row number of a week will be the row number of the last
    timestamp of the week, so the row numbers will not be contiguous.

    Args:
        symbol: The symbol of the data.
        data: The data to convert.

    Returns:
        The converted data.
    """
    if data.empty:
        return pd.DataFrame(
            columns=["datetime", "Open", "High", "Low", "Close", "Volume"]
        )

    cache_key = cls._get_cache_key(symbol, "1W")
    cached_data = get_global_cache().get(cache_key)
    if cached_data is not None:
        logger.debug(f"{cache_key} already exists. Using cached data...")
        return cached_data.loc[cached_data.index.intersection(data.index)]

    data = data.copy()
    data["datetime"] = pd.to_datetime(data["datetime"])

    weekly_groups = data.groupby(pd.Grouper(key="datetime", freq="W-FRI"))

    weekly_df = weekly_groups.agg(
        {
            "datetime": "last",
            "Open": "first",
            "High": "max",
            "Low": "min",
            "Close": "last",
            "Volume": "sum",
        }
    )

    # pyrefly: ignore
    last_indices = weekly_groups.apply(lambda x: x.index[-1], include_groups=False)
    weekly_df.index = last_indices.values
    weekly_df = weekly_df[["datetime", "Open", "High", "Low", "Close", "Volume"]]

    return weekly_df

convert_to_2h_data classmethod

convert_to_2h_data(
    symbol: str, data: DataFrame
) -> DataFrame

Convert 5M OHLCV data to 2-hour bars.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def convert_to_2h_data(cls, symbol: str, data: pd.DataFrame) -> pd.DataFrame:
    """Convert 5M OHLCV data to 2-hour bars."""
    return cls._resample_with_datetime_grouper(symbol, data, "2H", "2h")

convert_to_30m_data classmethod

convert_to_30m_data(
    symbol: str, data: DataFrame
) -> DataFrame

Convert 5M OHLCV data to 30-minute bars.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def convert_to_30m_data(cls, symbol: str, data: pd.DataFrame) -> pd.DataFrame:
    """Convert 5M OHLCV data to 30-minute bars."""
    return cls._resample_with_datetime_grouper(symbol, data, "30M", "30min")

convert_to_4h_data classmethod

convert_to_4h_data(
    symbol: str, data: DataFrame
) -> DataFrame

Convert 5M OHLCV data to 4-hour bars.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def convert_to_4h_data(cls, symbol: str, data: pd.DataFrame) -> pd.DataFrame:
    """Convert 5M OHLCV data to 4-hour bars."""
    return cls._resample_with_datetime_grouper(symbol, data, "4H", "4h")

preload_global_instance_cache classmethod

preload_global_instance_cache(
    symbol: str, data: DataFrame, freq: CandleUnit
) -> None

NOTE: Used for backtesting only.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

@classmethod
def preload_global_instance_cache(
    cls, symbol: str, data: pd.DataFrame, freq: CandleUnit
) -> None:
    """NOTE: Used for backtesting only."""
    cache_key = cls._get_cache_key(symbol, freq)
    conversion_func = {
        "15M": cls.convert_to_15m_data,
        "30M": cls.convert_to_30m_data,
        "1H": cls.convert_to_1h_data,
        "2H": cls.convert_to_2h_data,
        "4H": cls.convert_to_4h_data,
        "1D": cls.convert_to_1d_data,
        "1W": cls.convert_to_1w_data,
    }.get(freq)

    if conversion_func is None:
        raise ValueError(f"Invalid frequency provided: {freq}")

    agg_data = conversion_func(symbol, data)
    get_global_cache().set(cache_key, agg_data)

    logger.info(f"Preloaded {freq} OHLCV data for {symbol} into cache")

get_global_cache

get_global_cache()

Get the global cache instance.

Source code in src/turtle_quant_1/strategies/helpers/data_units.py

def get_global_cache():
    """Get the global cache instance."""
    return _global_cache