`turtle_quant_1.data_processing.base`¶

turtle_quant_1.data_processing.base ¶

Base classes for data processing components.

BaseDataFetcher ¶

BaseDataFetcher(symbols: List[str])

Bases: ABC

Base class for data fetchers.

Initialize the data fetcher.

Parameters:

Name	Type	Description	Default
`symbols`	`List[str]`	List of symbols to fetch data for.	required

Source code in src/turtle_quant_1/data_processing/base.py

def __init__(self, symbols: List[str]):
    """Initialize the data fetcher.

    Args:
        symbols: List of symbols to fetch data for.
    """
    self.symbols = symbols

fetch_ohlcv `abstractmethod` ¶

fetch_ohlcv(
    symbol: str,
    start_date: datetime,
    end_date: datetime,
    freq: CandleUnit = "5M",
) -> DataFrame

Fetch OHLCV data for a symbol at the requested candle frequency.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Symbol to fetch data for.	required
`start_date`	`datetime`	Start date for the data.	required
`end_date`	`datetime`	End date for the data.	required
`freq`	`CandleUnit`	Candle frequency for the returned data.	`'5M'`

Returns:

Type	Description
`DataFrame`	DataFrame with OHLCV data.

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def fetch_ohlcv(
    self,
    symbol: str,
    start_date: datetime,
    end_date: datetime,
    freq: CandleUnit = "5M",
) -> pd.DataFrame:
    """Fetch OHLCV data for a symbol at the requested candle frequency.

    Args:
        symbol: Symbol to fetch data for.
        start_date: Start date for the data.
        end_date: End date for the data.
        freq: Candle frequency for the returned data.

    Returns:
        DataFrame with OHLCV data.
    """
    pass

BaseDataMaintainer ¶

BaseDataMaintainer(
    symbols: Optional[List[str]] = None,
    freq: CandleUnit = "5M",
    fetcher: Optional[BaseDataFetcher] = None,
)

Bases: ABC

Base class for data maintainers.

Initialize the data maintainer.

Parameters:

Name	Type	Description	Default
`symbols`	`Optional[List[str]]`	List of symbols to maintain.	`None`
`freq`	`CandleUnit`	Candle frequency for the data being maintained.	`'5M'`
`fetcher`	`Optional[BaseDataFetcher]`	Data fetcher to use. If None, uses YFinanceDataFetcher.	`None`

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def __init__(
    self,
    symbols: Optional[List[str]] = None,
    freq: CandleUnit = "5M",
    fetcher: Optional[BaseDataFetcher] = None,
):
    """Initialize the data maintainer.

    Args:
        symbols: List of symbols to maintain.
        freq: Candle frequency for the data being maintained.
        fetcher: Data fetcher to use. If None, uses YFinanceDataFetcher.
    """
    pass

impute_data `abstractmethod` ¶

impute_data(
    symbol: str,
    data: DataFrame,
    end_date: Optional[datetime] = None,
) -> DataFrame

Ensure continuous data for a symbol.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Symbol to impute data for.	required
`data`	`DataFrame`	DataFrame with data.	required
`end_date`	`Optional[datetime]`	End date to ensure data up to. If None, uses current time.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with updated data.

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def impute_data(
    self,
    symbol: str,
    data: pd.DataFrame,
    end_date: Optional[datetime] = None,
) -> pd.DataFrame:
    """Ensure continuous data for a symbol.

    Args:
        symbol: Symbol to impute data for.
        data: DataFrame with data.
        end_date: End date to ensure data up to. If None, uses current time.

    Returns:
        DataFrame with updated data.
    """
    pass

BaseDataProcessor ¶

Bases: ABC

Base class for data processors.

load_data `abstractmethod` ¶

load_data(
    symbol: str, start_date: datetime, end_date: datetime
) -> DataFrame

Load data for a symbol.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Symbol to load data for.	required
`start_date`	`datetime`	Start date for the data.	required
`end_date`	`datetime`	End date for the data.	required

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def load_data(
    self,
    symbol: str,
    start_date: datetime,
    end_date: datetime,
) -> pd.DataFrame:
    """Load data for a symbol.

    Args:
        symbol: Symbol to load data for.
        start_date: Start date for the data.
        end_date: End date for the data.
    """
    pass

save_data `abstractmethod` ¶

save_data(symbol: str, data: DataFrame) -> None

Save data for a symbol.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Symbol to save data for.	required
`data`	`DataFrame`	DataFrame with data.	required

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def save_data(self, symbol: str, data: pd.DataFrame) -> None:
    """Save data for a symbol.

    Args:
        symbol: Symbol to save data for.
        data: DataFrame with data.
    """
    pass

BaseDataStorageAdapter ¶

Bases: ABC

Base class for data storage.

delete_ohlcv_data `abstractmethod` ¶

delete_ohlcv_data(
    symbol: str, freq: CandleUnit = "5M"
) -> None

Delete data for a symbol from storage.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Symbol to delete data for.	required
`freq`	`CandleUnit`	Candle frequency of the data to delete.	`'5M'`

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def delete_ohlcv_data(self, symbol: str, freq: CandleUnit = "5M") -> None:
    """Delete data for a symbol from storage.

    Args:
        symbol: Symbol to delete data for.
        freq: Candle frequency of the data to delete.
    """
    pass

load_ohlcv_data `abstractmethod` ¶

load_ohlcv_data(
    symbol: str,
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None,
    freq: CandleUnit = "5M",
) -> DataFrame

Load OHLCV data for a symbol.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Symbol to load data for.	required
`start_date`	`Optional[datetime]`	Start date for the data.	`None`
`end_date`	`Optional[datetime]`	End date for the data.	`None`
`freq`	`CandleUnit`	Candle frequency of the data to load.	`'5M'`

Returns:

Type	Description
`DataFrame`	DataFrame with OHLCV data.

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def load_ohlcv_data(
    self,
    symbol: str,
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None,
    freq: CandleUnit = "5M",
) -> pd.DataFrame:
    """Load OHLCV data for a symbol.

    Args:
        symbol: Symbol to load data for.
        start_date: Start date for the data.
        end_date: End date for the data.
        freq: Candle frequency of the data to load.

    Returns:
        DataFrame with OHLCV data.
    """
    pass

save_ohlcv_data `abstractmethod` ¶

save_ohlcv_data(
    symbol: str, data: DataFrame, freq: CandleUnit = "5M"
) -> None

Save OHLCV data for a symbol.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Symbol to save data for.	required
`data`	`DataFrame`	DataFrame with OHLCV data.	required
`freq`	`CandleUnit`	Candle frequency of the data being saved.	`'5M'`

Source code in src/turtle_quant_1/data_processing/base.py

@abstractmethod
def save_ohlcv_data(
    self,
    symbol: str,
    data: pd.DataFrame,
    freq: CandleUnit = "5M",
) -> None:
    """Save OHLCV data for a symbol.

    Args:
        symbol: Symbol to save data for.
        data: DataFrame with OHLCV data.
        freq: Candle frequency of the data being saved.
    """
    pass

turtle_quant_1.data_processing.base¶