Looking for experiences using QuantConnect/LEAN with B3 (Brazil) data

Here’s a template to help you (Ari) get started using QuantConnect/LEAN with B3 (Brazil) custom data for research/backtesting. This solution assumes you have access to historical B3 data files (like CSVs) because B3 assets are not natively supported in QuantConnect. The code shows how to import, parse, and analyze custom equity data—just adapt the file paths and column structure to your needs!

from AlgorithmImports import *

class B3EquityCustomData(PythonData):
    """
    Template for Brazilian equity daily bars.

    Expected CSV columns (no header):
      Date,Open,High,Low,Close,Volume
      20210104,94.41,96.28,93.51,95.98,30665000

    Date format  : YYYYMMDD
    File location: ./data/b3/<TICKER>.csv  (e.g. ./data/b3/VALE3.csv)

    Adjust `GetSource()` and `Reader()` to match your data-file structure.
    """

    # ----------------------------------------------------------------------
    # Tell LEAN where (and how) to fetch the data
    # ----------------------------------------------------------------------
    def get_source(
        self,
        config: SubscriptionDataConfig,
        date: datetime,
        isLiveMode: bool
    ) -> SubscriptionDataSource:

        if isLiveMode:
            # Replace the empty string with a websocket/API endpoint for live data
            return SubscriptionDataSource("", SubscriptionTransportMedium.REST)

        # Backtesting / research ─ read local CSV
        file_path = f"../data/b3/{config.symbol.value}.csv"    # <── tweak path as required
        return SubscriptionDataSource(file_path, SubscriptionTransportMedium.LOCAL_FILE)

    # ----------------------------------------------------------------------
    # Parse each line of the CSV file into a PythonData object
    # ----------------------------------------------------------------------
    def reader(
        self,
        config: SubscriptionDataConfig,
        line: str,
        date: datetime,
        isLiveMode: bool
    ):
        if not line.strip():
            return None

        parts = line.split(",")
        if len(parts) < 6:
            return None   # malformed row

        try:
            trade_date  = datetime.strptime(parts[0], "%Y%m%d")
            open_price  = float(parts[1])
            high_price  = float(parts[2])
            low_price   = float(parts[3])
            close_price = float(parts[4])
            volume      = float(parts[5])
        except (ValueError, IndexError):
            return None   # skip bad rows

        bar          = B3EquityCustomData()
        bar.symbol   = config.symbol
        bar.time     = trade_date
        bar.end_time  = trade_date + timedelta(days=1)

        # Store OHLCV so they can be accessed in OnData
        bar.open     = open_price
        bar.high     = high_price
        bar.low      = low_price
        bar.close    = close_price
        bar.volume   = volume

        bar.value    = close_price      # default data point LEAN uses
        return bar

class B3ResearchTemplate(QCAlgorithm):
    def initialize(self):
        # 1. Dates & capital
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100_000)

        # 2. Add custom B3 equity (ensure the CSV exists in ./data/b3/)
        self._b_3_symbol = self.add_data(B3EquityCustomData, "VALE3", Resolution.DAILY).symbol

        # Optional chart for quick visual checks
        self.add_chart(Chart("B3 Price"))
        self.debug(f"Added custom B3 symbol: {self._b_3_symbol}")

    # 3. Inspect the data as it streams in
    def on_data(self, data: Slice):
        if self._b_3_symbol not in data:
            return

        bar: B3EquityCustomData = data[self._b_3_symbol]
        if bar is None:
            return

        # Console output for research
        self.debug(
            f"{self.time:%Y-%m-%d} | {self._b_3_symbol.value} | "
            f"O:{bar.open:.2f} H:{bar.high:.2f} L:{bar.low:.2f} C:{bar.close:.2f} "
            f"Vol:{bar.volume:,.0f}"
        )

        # Plot Close price
        self.plot("B3 Price", "Close", bar.close)

        # --------------------------------------------------
        # Replace this section with your strategy/research:
        #   - indicators
        #   - alpha rules
        #   - execution logic
        # --------------------------------------------------

How to use:

1. Save your B3 historical data file as ./data/b3/VALE3.csv (or your preferred ticker).
2. Adjust the CSV columns and parsing logic if your data format differs.
3. Add new symbols or customize your strategy inside on_data() as needed!

If you have other B3 asset types (futures, etc.), you would just adapt the custom data class accordingly. Let me know if you need help parsing a specific data layout!

Sources: CSV Format Example - QuantConnect.com, CSV Format Example - QuantConnect.com, docs writing algorithms importing data streaming data key concepts, docs writing algorithms securities requesting data, Custom Data - QuantConnect.com

In the custom data template there’s this snippet:

if isLiveMode:
# Replace the empty string with a websocket/API endpoint for live data
return SubscriptionDataSource("", SubscriptionTransportMedium.REST)

If I route my real-time B3 (Brazil) data through that SubscriptionDataSource in live mode, can I run my algorithm/strategy and see charts updating in Live (not backtest)? Is my understanding correct?

Opa Eduardo Angelim. Que bom que tem pelo menos mais um brazuka na área. Estou fazendo outros testes usando a versão online e carregando csv para backtest. Assim que concluir vou dar uma conferida nesta sugestão da Mia Alissi.
 

Mia Alissi  Thanks for the suggestion. I'll try it out as soon as I have time and will post the results here.

Here’s a friendly, step-by-step demo for you, Ari, showing exactly how to load your B3 CSV into the QuantConnect Object Store (online IDE) and run research/backtests over it as custom data. This uses the PETR4.csv example, but you can change the symbol/file as needed.

# =======================================================
#  B3 CSV → Object Store → Custom Data (PETR4) Demo
# =======================================================
#
# 0)  CSV FORMAT EXAMPLE (daily OHLCV; header row optional)
#     Date,Open,High,Low,Close,Volume
#     2021-01-04,29.50,29.90,28.98,29.05,12345678
#
# 1A) MANUAL UPLOAD (online IDE)
#     • Click the                 icon in the left toolbar.
#     • Press “Upload”, select PETR4.csv, then “Save”.
#
# 1B) ONE-TIME PROGRAMMATIC UPLOAD (Research Notebook)
#     from AlgorithmImports import *
#     qb = QuantBook()
#     with open('/local/path/PETR4.csv', 'r', encoding='utf-8') as f:
#         qb.object_store.save('PETR4.csv', f.read())
#
# 2)  CUSTOM DATA CLASS (reads from Object Store)
# 3)  ALGORITHM: adds the data, logs, and plots the close
# =======================================================

from AlgorithmImports import *
class B3Daily(PythonData):
    """
    Custom data type for B3 daily bars stored as a CSV inside
    the QuantConnect Object Store.
    Columns: Date,Open,High,Low,Close,Volume
    """

    # ------------------------------------------------------------------
    #  Tell Lean where the file lives (Object Store) and its name/key
    # ------------------------------------------------------------------
    def get_source(
        self,
        config: SubscriptionDataConfig,
        date: datetime,
        is_live: bool
    ) -> SubscriptionDataSource:  # noqa: N802
        file_name: str = "PETR4.csv"  # must match the uploaded key
        return SubscriptionDataSource(
            file_name, SubscriptionTransportMedium.OBJECT_STORE, FileFormat.CSV
        )

    # ------------------------------------------------------------------
    #  Convert one line of the CSV into a tradable bar
    # ------------------------------------------------------------------
    def reader(
        self,
        config: SubscriptionDataConfig,
        line: str,
        date: datetime,
        is_live: bool
    ) -> "B3Daily":  # noqa: N802
        if not line or line.startswith("Date"):
            return None

        parts: List[str] = line.split(",")
        if len(parts) < 6:
            return None

        try:
            bar_date: datetime = datetime.strptime(parts[0], "%Y-%m-%d")
            open_: float = float(parts[1])
            high_: float = float(parts[2])
            low_: float = float(parts[3])
            close_: float = float(parts[4])
            volume_: float = float(parts[5])
        except (ValueError, IndexError):
            return None  # skip malformed rows safely

        bar: B3Daily = B3Daily()
        bar.symbol = config.symbol
        bar.time = bar_date
        bar.end_time = bar_date + timedelta(days=1)
        bar.open = open_
        bar.high = high_
        bar.low = low_
        bar.close = close_
        bar.volume = volume_
        bar.value = close_
        return bar

class B3CsvAlgorithm(QCAlgorithm):
    """
    Template algorithm that consumes the PETR4.csv file from the
    Object Store, prints each close, and plots it to the chart.
    """

    def initialize(self) -> None:
        # Basic backtest parameters
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100_000)

        # Add the PETR4 custom data stream
        self._petr_4_symbol: Symbol = self.add_data(B3Daily, "PETR4").symbol

        # Create a chart for visual inspection
        chart = Chart("PETR4")
        chart.add_series(Series("Close", SeriesType.LINE, index=0))
        self.add_chart(chart)

    def on_data(self, data: Slice) -> None:
        if data.contains_key(self._petr_4_symbol):
            bar: B3Daily = data[self._petr_4_symbol]

            # Log the close price
            self.debug(f"{bar.time:%Y-%m-%d}  PETR4 Close = {bar.close:.2f}")

            # Plot for quick research visuals
            self.plot("PETR4", "Close", bar.close)

Resumo dos passos:

1. Suba seu CSV para o Object Store via IDE online (ou pelo notebook, ver bloco 1B).
2. Ajuste o nome do arquivo em get_source se precisar.
3. Use/adapte a classe de dados para seus outros arquivos/símbolos da B3 facilmente.

Assim, você pode explorar qualquer ativo da B3 em poucos minutos na plataforma QuantConnect! Qualquer dúvida, só chamar.

Sources: Object Store - QuantConnect.com, CSV Format Example - QuantConnect.com, Key Concepts - QuantConnect.com, docs writing algorithms securities requesting data