turtle_quant_1.data_processing.processor

turtle_quant_1.data_processing.processor

Data processor for orchestrating data fetching and storage.

DataProcessor

DataProcessor(
    symbols: Optional[List[str]] = None,
    freq: CandleUnit = CANDLE_UNIT,
    storage: Optional[BaseDataStorageAdapter] = None,
    fetcher: Optional[BaseDataFetcher] = None,
    maintainer: Optional[BaseDataMaintainer] = None,
)

Bases: BaseDataProcessor

Data processor for orchestrating data fetching and storage.

Initialize the data processor.

Parameters:

Name	Type	Description	Default
symbols	Optional[List[str]]	List of symbols to process. If None, uses SYMBOLS from config.	None
freq	CandleUnit	Candle frequency to use for fetching and processing data.	CANDLE_UNIT
storage	Optional[BaseDataStorageAdapter]	Data storage to use. If None, uses GCSDataStorage.	None
fetcher	Optional[BaseDataFetcher]	Data fetcher to use. If None, uses YFinanceDataFetcher.	None
maintainer	Optional[BaseDataMaintainer]	Data maintainer to use. If None, uses DataMaintainer.	None

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

def __init__(
    self,
    symbols: Optional[List[str]] = None,
    freq: CandleUnit = CANDLE_UNIT,
    storage: Optional[BaseDataStorageAdapter] = None,
    fetcher: Optional[BaseDataFetcher] = None,
    maintainer: Optional[BaseDataMaintainer] = None,
):
    """Initialize the data processor.

    Args:
        symbols: List of symbols to process. If None, uses SYMBOLS from config.
        freq: Candle frequency to use for fetching and processing data.
        storage: Data storage to use. If None, uses GCSDataStorage.
        fetcher: Data fetcher to use. If None, uses YFinanceDataFetcher.
        maintainer: Data maintainer to use. If None, uses DataMaintainer.
    """
    self.symbols = symbols or LIVE_SYMBOLS
    self.freq = freq

    self.storage = storage or GCSDataStorageAdapter()
    self.fetcher = fetcher or YFinanceDataFetcher(symbols=self.symbols)
    self.maintainer = maintainer or DataMaintainer(
        symbols=self.symbols, freq=self.freq, fetcher=self.fetcher
    )

    self.data_cache: dict[str, pd.DataFrame] = {}

load_data

load_data(
    symbol: str,
    start_date: datetime,
    end_date: datetime,
    impute_data: bool = True,
) -> DataFrame

Load data for a symbol.

Auto-corrects start and end dates to be at the bounds of market hours for those days.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol to load data for.	required
start_date	datetime	Timezone-aware start date to fetch data from.	required
end_date	datetime	Timezone-aware end date to fetch data up to.	required
impute_data	bool	Whether to impute data.	True

Returns:

Type	Description
DataFrame	DataFrame with OHLCV data.

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

def load_data(
    self,
    symbol: str,
    start_date: datetime,
    end_date: datetime,
    impute_data: bool = True,
) -> pd.DataFrame:
    """Load data for a symbol.

    Auto-corrects start and end dates to be at the bounds of market hours for those days.

    Args:
        symbol: Symbol to load data for.
        start_date: Timezone-aware start date to fetch data from.
        end_date: Timezone-aware end date to fetch data up to.
        impute_data: Whether to impute data.

    Returns:
        DataFrame with OHLCV data.
    """
    if DANGER_CLEAR_CACHE:
        self.data_cache.pop(symbol, None)
    if DANGER_CLEAR_STORAGE:
        self.storage.delete_ohlcv_data(symbol, freq=self.freq)
        self.data_cache.pop(symbol, None)

    is_data_updated = False

    _start_date, _end_date = get_expected_market_hours_bounds(
        symbol, start_date, end_date
    )

    if symbol in self.data_cache:
        df = self.data_cache[symbol]
    else:
        logger.warning(
            f"No data found for {symbol} in cache. Fetching from {self.storage}..."
        )
        df = self.storage.load_ohlcv_data(
            symbol=symbol,
            start_date=None,  # TODO: Work out why.
            end_date=None,  # TODO: Work out why.
            freq=self.freq,
        )

    if df.empty:
        logger.warning(
            f"No data found for {symbol}. Fetching from {self.fetcher}..."
        )
        df = self.fetcher.fetch_ohlcv(
            symbol=symbol,
            start_date=_start_date,
            end_date=_end_date,
            freq=self.freq,
        )
        is_data_updated = True

    if impute_data:
        _df = self.maintainer.impute_data(
            symbol=symbol, data=df, end_date=_end_date
        )
        if not _df.index.equals(df.index):
            is_data_updated = True
        df = _df

    # Save updated data in relevant locations
    self.data_cache[symbol] = df
    if is_data_updated:
        logger.info(
            f"Data has been changed. Updating data for {symbol} in cache and storage..."
        )
        self.save_data(symbol, df)

    return df

save_data

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

Save data to storage.

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/processor.py

def save_data(self, symbol: str, data: pd.DataFrame) -> None:
    """Save data to storage.

    Args:
        symbol: Symbol to save data for.
        data: DataFrame with data.
    """
    self.storage.save_ohlcv_data(symbol=symbol, data=data, freq=self.freq)