turtle_quant_1.data_processing.adapters.yfinance_fetcher

turtle_quant_1.data_processing.adapters.yfinance_fetcher

YFinance data fetcher implementation.

YFinanceDataFetcher

YFinanceDataFetcher(symbols: List[str])

Bases: BaseDataFetcher

YFinance implementation of data fetcher.

Initialize the YFinance 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/adapters/yfinance_fetcher.py

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

    Args:
        symbols: List of symbols to fetch data for.
    """
    super().__init__(symbols)
    self.tickers = {symbol: yf.Ticker(symbol) for symbol in symbols}

fetch_ohlcv

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

Fetch OHLCV data from YFinance at the requested candle frequency.

2H and 4H candles are not supported natively by yfinance; they are fetched at 1H resolution and resampled before being returned.

Parameters:

Name	Type	Description	Default
symbol	str	The symbol to fetch 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
freq	CandleUnit	Candle frequency for the returned data.	'5M'

Returns:

Type	Description
DataFrame	DataFrame with columns: datetime, Open, High, Low, Close, Volume.

Source code in src/turtle_quant_1/data_processing/adapters/yfinance_fetcher.py

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

    2H and 4H candles are not supported natively by yfinance; they are
    fetched at 1H resolution and resampled before being returned.

    Args:
        symbol: The symbol to fetch data for.
        start_date: Timezone-aware start date to fetch data from.
        end_date: Timezone-aware end date to fetch data up to.
        freq: Candle frequency for the returned data.

    Returns:
        DataFrame with columns: datetime, Open, High, Low, Close, Volume.
    """
    logger.info(
        f"Fetching {freq} data for {symbol} from {start_date} to {end_date}"
    )

    if symbol not in self.tickers:
        raise ValueError(f"Symbol {symbol} not initialized")

    yf_interval = _CANDLE_UNIT_TO_YF_INTERVAL[freq]

    try:
        df = self.tickers[symbol].history(
            start=start_date,
            end=end_date,
            interval=yf_interval,
        )

        if df.empty:
            logger.warning(
                f"No data returned for {symbol} from {start_date} to {end_date}"
            )
            return _EMPTY_DF.copy()

        # Reset index and normalise the datetime column name
        df = df.reset_index()
        if "Date" in df.columns:
            df = df.rename(columns={"Date": "datetime"})
        elif "Datetime" in df.columns:
            df = df.rename(columns={"Datetime": "datetime"})
        elif "index" in df.columns:
            df = df.rename(columns={"index": "datetime"})
        else:
            df.columns = ["datetime"] + list(df.columns[1:])

        df["datetime"] = pd.to_datetime(df["datetime"])

        result_df = (
            df[["datetime", "Open", "High", "Low", "Close", "Volume"]]
            .copy()
            .sort_values("datetime")
            .reset_index(drop=True)
        )

        # Resample 1H base data up to 2H / 4H where yfinance has no native interval
        resample_rule = _CANDLE_UNIT_TO_RESAMPLE_RULE.get(freq)
        if resample_rule is not None:
            result_df = (
                result_df.set_index("datetime")
                .resample(resample_rule)
                .agg(
                    Open=("Open", "first"),
                    High=("High", "max"),
                    Low=("Low", "min"),
                    Close=("Close", "last"),
                    Volume=("Volume", "sum"),
                )
                .dropna(subset=["Open"])
                .reset_index()
            )

        logger.info(f"Successfully fetched {len(result_df)} records for {symbol}")
        return result_df

    except Exception as e:
        logger.error(
            f"Error fetching {freq} data for {symbol} "
            f"from {start_date} to {end_date}: {str(e)}"
        )
        return _EMPTY_DF.copy()