Dynamic Integration of Daily Fama-French Factors for Live and Backtest Use

Here’s a robust, production-grade implementation you can use to dynamically source and integrate the daily Fama-French 3-Factor data (Mkt-RF, SMB, HML, RF) from Kenneth French’s actual published files into QuantConnect, suitable for both backtest and live deployments:

from AlgorithmImports import *
import zipfile
import io
import csv

class FamaFrenchThreeFactorDaily(PythonData):
    """
    Custom data type for the Fama-French 3-Factor daily dataset.
    It retrieves Mkt-RF, SMB, HML, and RF factors from Kenneth French's website.

    Data is published as a CSV inside a ZIP, with text lines before/after the data rows.
    We override the required methods to handle this format robustly.

    Key columns in each valid row:
        Date(YYYYMMDD), Mkt-RF, SMB, HML, RF
        e.g. 19260701, -0.10, -0.20, 0.30, 0.01
    """

    def __init__(self) -> None:
        super().__init__()
        # These private fields store the factor values
        self._mkt_rf: float = 0.0
        self._smb: float = 0.0
        self._hml: float = 0.0
        self._rf: float = 0.0

    # Expose public read-only properties
    @property
    def mkt_rf(self) -> float:
        return self._mkt_rf

    @property
    def SMB(self) -> float:
        return self._smb

    @property
    def HML(self) -> float:
        return self._hml

    @property
    def RF(self) -> float:
        return self._rf

    def get_source(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
        """
        Provide the URL of the zip file for both backtesting and live.
        The file is updated periodically by Kenneth French, so re-fetch at each deployment.
        """
        url = "https://mba.tuck.dartmouth.edu/pages/faculty/ken.french/ftp/F-F_Research_Data_Factors_daily_CSV.zip"
        # We'll specify RemoteFile with the default CSV reading. We'll do the unzip ourselves in Reader().
        return SubscriptionDataSource(url, SubscriptionTransportMedium.REMOTE_FILE)

    def reader(self,
               config: SubscriptionDataConfig,
               stream: Any,   # We receive the entire downloaded stream
               date: datetime,
               isLiveMode: bool) -> List[BaseData]:
        """
        Custom reader that:
        1) Unzips the file
        2) Skips the textual fluff lines before/after data
        3) Parses rows of the daily factor data
        4) Returns a list of FamaFrenchThreeFactorDaily objects
        """
        data_list: List[BaseData] = []

        # Ensure we can seek in the stream
        stream.seek(0)

        try:
            with zipfile.zip_file(stream) as z:
                # Typically there's a single CSV file inside. We'll parse the first one we find.
                for name in z.namelist():
                    with z.open(name) as f:
                        text = f.read().decode('utf-8', errors='ignore')
                        # Split file contents by newlines
                        lines = text.split('\n')

                        for line in lines:
                            line = line.strip()
                            if not line:
                                continue

                            parts = [x.strip() for x in line.split(',')]
                            # Expect at least 5 columns: date, Mkt-RF, SMB, HML, RF
                            if len(parts) < 5:
                                continue

                            try:
                                raw_date = parts[0]
                                # Date must be 8 digits, e.g. 19260701
                                if len(raw_date) != 8:
                                    continue

                                year = int(raw_date[0:4])
                                month = int(raw_date[4:6])
                                day = int(raw_date[6:8])
                                current_date = datetime(year, month, day)

                                mkt_rf = float(parts[1])
                                smb = float(parts[2])
                                hml = float(parts[3])
                                rf = float(parts[4])
                            except ValueError:
                                # Any parse error, skip
                                continue

                            # Construct data point
                            obj = FamaFrenchThreeFactorDaily()
                            # The data is typically available at T+1 midnight UTC for a daily timestamp
                            obj.end_time = current_date + timedelta(days=1)
                            obj.symbol = config.symbol
                            obj._mkt_rf = mkt_rf
                            obj._smb = smb
                            obj._hml = hml
                            obj._rf = rf
                            # For the 'Value' property that PythonData requires, we store Mkt-RF
                            obj.value = mkt_rf

                            data_list.append(obj)
                return data_list

        except Exception as e:
            # If anything unexpected happens, return an empty list (no data)
            # You could also log a warning
            return data_list

        return data_list

class FamaFrenchAlgorithm(QCAlgorithm):
    """
    Example algorithm demonstrating how to use FamaFrenchThreeFactorDaily in LEAN.
    """

    def initialize(self) -> None:
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100000)

        # Register the custom data with daily resolution, UTC time zone
        self._ff_symbol = self.add_data(
            FamaFrenchThreeFactorDaily,
            "FamaFrench3FactorsDaily",
            Resolution.DAILY,
            TimeZones.UTC
        ).symbol

        # Demonstrate using History() to retrieve factor data
        history = self.history(
            FamaFrenchThreeFactorDaily,
            self._ff_symbol,
            5,
            Resolution.DAILY
        )

        if not history.empty:
            # history is a pandas DataFrame with multi-index: [time, symbol]
            # Each row has columns named after the property names. E.g. 'MktRf', 'SMB', 'HML', 'RF'.
            for dt, row in history.loc[self._ff_symbol].iterrows():
                self.debug(f"History {dt} -> Mkt-RF: {row['MktRf']} SMB: {row['SMB']} "
                           f"HML: {row['HML']} RF: {row['RF']}")

    def on_data(self, data: Slice) -> None:
        # Fetch the newest data point for the day
        ff_data = data.get(FamaFrenchThreeFactorDaily, self._ff_symbol)
        if ff_data is None:
            return

        # Access factor values
        mkt_rf = ff_data.mkt_rf
        smb = ff_data.smb
        hml = ff_data.hml
        rf = ff_data.rf

        # Just print them for demonstration
        self.debug(f"Fama-French Factors {self.time} -> "
                   f"Mkt-RF={mkt_rf}% SMB={smb}% HML={hml}% RF={rf}%")

How it works:

• On deployment (backtest or live), downloads and parses the latest Fama-French daily ZIP from the official site.
• Handles edge-case text headers/footers and robustly parses valid numeric rows.
• Provides factor values as properties and aligns LEAN timestamps for daily workflow.
• Register it just once with add_data(), then access it in OnData, via data.get(), or with self.history() (for research/historical).
• No static file hosting or GitHub mirroring needed; works seamlessly with the QuantConnect API/Cloud ecosystem.

Feel free to copy, use, and extend as needed for research or production!

Sources: Expected Idiosyncratic Skewness, Residual Momentum, Composite Factor Bundle - QuantConnect.com, US Federal Reserve (FRED) - QuantConnect.com), US Fundamental Data - QuantConnect.com