turtle_quant_1.data_processing.maintainer

turtle_quant_1.data_processing.maintainer

Data maintainer for ensuring continuous historical data availability.

DataMaintainer

DataMaintainer(
    symbols: Optional[List[str]] = None,
    freq: CandleUnit = CANDLE_UNIT,
    fetcher: Optional[BaseDataFetcher] = None,
)

Bases: BaseDataMaintainer

Maintains continuous historical data for all symbols.

Initialize the data maintainer.

Parameters:

Name	Type	Description	Default
symbols	Optional[List[str]]	List of symbols to maintain. If None, uses SYMBOLS from config.	None
freq	CandleUnit	Candle frequency for the data being maintained. The maintainer always stores raw data at the 5M atomic resolution; this value is retained for informational purposes and future use.	CANDLE_UNIT
fetcher	Optional[BaseDataFetcher]	Data fetcher to use. If None, uses YFinanceDataFetcher.	None

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

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

    Args:
        symbols: List of symbols to maintain. If None, uses SYMBOLS from config.
        freq: Candle frequency for the data being maintained. The maintainer
            always stores raw data at the 5M atomic resolution; this value
            is retained for informational purposes and future use.
        fetcher: Data fetcher to use. If None, uses YFinanceDataFetcher.
    """
    self.symbols = symbols or LIVE_SYMBOLS
    self.freq = freq
    self.fetcher = fetcher or YFinanceDataFetcher(symbols=self.symbols)

impute_data

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

Ensure continuous data exists for a symbol up to the specified end date.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol to ensure data for.	required
data	DataFrame	DataFrame with data.	required
end_date	Optional[datetime]	Timezone-aware 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/maintainer.py

def impute_data(
    self,
    symbol: str,
    data: pd.DataFrame,
    end_date: Optional[datetime] = None,
) -> pd.DataFrame:
    """Ensure continuous data exists for a symbol up to the specified end date.

    Args:
        symbol: Symbol to ensure data for.
        data: DataFrame with data.
        end_date: Timezone-aware end date to ensure data up to. If None, uses current time.

    Returns:
        DataFrame with updated data.
    """
    now_date = datetime.now().astimezone(pytz.timezone(HOST_TIMEZONE))
    end_date = end_date or now_date
    start_date = now_date - timedelta(days=MAX_HISTORY_DAYS)

    logger.info(
        f"Ensuring continuous data for {symbol} from {start_date} to {end_date}..."
    )

    # Find gaps in the data
    gaps = self._get_data_gaps(symbol, data, start_date, end_date)
    logger.info(f"Found {len(gaps)} gaps in total in data for {symbol}")
    # Filter any gaps that are less than 1 hour to avoid too many API calls
    gaps = [gap for gap in gaps if gap[1] - gap[0] >= timedelta(hours=1)]
    logger.info(f"Found {len(gaps)} significant gaps in data for {symbol}")

    # Fill any gaps found
    for i, gap in enumerate(gaps):
        logger.info(f"    - Gap {i}: {gap[0]} to {gap[1]}")
    data = self._fill_data_gaps(symbol, data, gaps)

    # Delete outdated data as a "cron job"
    if datetime.now().astimezone(pytz.timezone(HOST_TIMEZONE)).day == 28:
        data = self._delete_outdated_data(symbol, data)
        logger.info(
            f"Deleted outdated data for {symbol}. Remaining dates: {data['datetime'].min()} to {data['datetime'].max()}"
        )

    return data