Earning calendar

Here’s how you can construct the earnings calendar for each symbol in your universe and block trading around earnings in QuantConnect, as requested:

from AlgorithmImports import *
from collections import defaultdict
import datetime

class EarningsAwareUniverse(QCAlgorithm):
    """
    Keeps an earnings calendar (self._calendar) built from the EODHDUpcomingEarnings
    universe.  Symbols with earnings in the next *N* days (default N = 2 ⇒
    today, tomorrow, and the next day) are blocked from trading.
    """

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

        # 2. Earnings calendar → { Symbol: [report_date, …] }
        self._calendar: dict[Symbol, list[datetime.date]] = defaultdict(list)

        # 3. Universe settings (daily US equities)
        self.universe_settings.resolution = Resolution.DAILY
        self.universe_settings.leverage   = 1
        self.universe_settings.fill_forward = True

        # 4. Universe selection using the EODHDUpcomingEarnings provider
        self.add_universe(EODHDUpcomingEarnings, self.selection_function)

    # ----------------------------------------------------------------------
    # Universe Selection
    # ----------------------------------------------------------------------
    def selection_function(
        self, upcoming_earnings: list[EODHDUpcomingEarnings]
    ) -> list[Symbol]:
        """
        1. Refresh the earnings calendar with the latest EODHDUpcomingEarnings data.
        2. Return the list of symbols that do **not** have earnings in the next two days.
        """
        today: datetime.date = self.time.date()

        # 1) Remove past-dated entries so the calendar stays current
        for calendar_symbol, date_list in list(self._calendar.items()):
            self._calendar[calendar_symbol] = [d for d in date_list if d >= today]
            if not self._calendar[calendar_symbol]:
                del self._calendar[calendar_symbol]

        # 2) Add/refresh upcoming earnings from the provider
        incoming_symbols: set[Symbol] = set()
        for earning_item in upcoming_earnings:
            symbol_incoming: Symbol = earning_item.symbol
            report_date:    datetime.date = earning_item.report_date.date()

            incoming_symbols.add(symbol_incoming)

            if report_date >= today and report_date not in self._calendar[symbol_incoming]:
                self._calendar[symbol_incoming].append(report_date)

        # 3) Keep only symbols whose earnings are NOT within the block window
        return [s for s in incoming_symbols
                if not self.block_today_or_tomorrow(s, self.time)]

    # ----------------------------------------------------------------------
    # Blocking Logic
    # ----------------------------------------------------------------------
    @staticmethod
    def _days_between(start: datetime.date, end: datetime.date) -> int:
        """Inclusive, integer day difference (end − start)."""
        return (end - start).days

    def block_today_or_tomorrow(
        self, symbol: Symbol, current: datetime.datetime, window: int = 2
    ) -> bool:
        """
        Returns True when *symbol* has an earnings date
        between *current.date()* and *(current + window days)* (inclusive).
        """
        if symbol not in self._calendar:
            return False

        today: datetime.date = current.date()
        for rpt_date in self._calendar[symbol]:
            if 0 <= self._days_between(today, rpt_date) <= window:
                return True
        return False

    # ----------------------------------------------------------------------
    # Trading Logic (example EQ-weight buy/hold)
    # ----------------------------------------------------------------------
    def on_data(self, data: Slice) -> None:
        # 1) Symbols we *can* trade today (have data & no imminent earnings)
        tradable = [
            s for s in self.active_securities.keys
            if s in data.bars and not self.block_today_or_tomorrow(s, self.time)
        ]

        if not tradable:
            return

        # 2) Allocate equal weight across the tradable universe
        weight: float = 1.0 / len(tradable)
        for sym in tradable:
            if not self.portfolio[sym].invested:
                self.set_holdings(sym, weight)

        # 3) Liquidate positions that became blocked
        for sym, holdings in self.portfolio.items():
            if holdings.invested and self.block_today_or_tomorrow(sym, self.time):
                self.liquidate(sym)

Key takeaways:

• The calendar is kept up to date with each universe refresh.
• block_today_or_tomorrow(symbol, time) checks if a symbol has an earnings date within your blocking window—today, tomorrow or the day after.
• You can use this blocking utility to prevent buying (or trigger liquidation) around earnings events, as in your example.

Let me know if you'd like to see further integration or adaptations!

Sources: Upcoming Earnings - QuantConnect.com, docs writing algorithms datasets eod historical data upcoming dividends, docs writing algorithms datasets eod historical data upcoming ipos, Dropbox Base Data Universe Selection Algorithm, Custom Data Universe Algorithm