Prompt One Shot Failure: Alternative Data Universe

[AlexCatarino]

Prompt

Build an algorithm that combines two universes. The first selects the top 100 most liquid US Equities from fundamental data, caches the symbols, and leaves the active universe unchanged. The second pulls the Quiver CNBC-trader alternative dataset and keeps only entries whose traders field mentions "Cramer". The final universe is the intersection. When a security is added, subscribe to the underlying CNBC data feed for it, and log each incoming trader mention. Set $100,000 starting cash. Backtest from September 1, 2024 to December 31, 2024.

Error

The backtest ended with a runtime error.

[ERROR] FATAL UNHANDLED EXCEPTION: name 'QuiverCNBCsUniverseSelectionData' is not defined, at QuantConnect.AlgorithmFactory.Python.Wrappers.AlgorithmPythonWrapper..ctor(String moduleName) in AlgorithmFactory/Python/Wrappers/AlgorithmPythonWrapper.cs:line 188, at QuantConnect.AlgorithmFactory.Loader.TryCreatePythonAlgorithm(String assemblyPath, IAlgorithm& algorithmInstance, String& errorMessage) in AlgorithmFactory/Loader.cs:line 173,Engine.Run(): QuantConnect.Lean.Engine.Setup.AlgorithmSetupException: During the algorithm initialization, the following exception has occurred: Loader.TryCreatePythonAlgorithm(): Unable to import python module ./cache/algorithm/project/main.pyc. AlgorithmPythonWrapper(): name 'QuiverCNBCsUniverseSelectionData' is not defined, at CramerTrackerAlgorithm, def _select_cnbc(self, data: List[QuiverCNBCsUniverseSelectionData]) -> List[Symbol]:, ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^, in main.py: line 39, at , class CramerTrackerAlgorithm(QCAlgorithm):, in main.py: line 5, name 'QuiverCNBCsUniverseSelectionData' is not defined, at QuantConnect.Lean.Engine.Setup.BacktestingSetupHandler.CreateAlgorithmInstance(AlgorithmNodePacket algorithmNodePacket, String assemblyPath) in Engine/Setup/BacktestingSetupHandler.cs:line 113, at QuantConnect.Lean.Engine.Engine.Run(AlgorithmNodePacket job, AlgorithmManager manager, String assemblyPath, WorkerThread workerThread) in Engine/Engine.cs:line 119,ApiConnection.TryRequest(https://www.quantconnect.com/api/v2/cloud/backtest/status/update): Error: The operation was canceled., Response content: ,BacktestingResultHandler.SendFinalResult(): Error running backtest analysis System.NullReferenceException: Object reference not set to an instance of an object., at QuantConnect.Lean.Engine.Results.Analysis.Analyses.ParameterCountAnalysis.Run(QCAlgorithm algorithm, Language language) in Engine/Results/Analysis/Analyses/ParameterCountAnalysis.cs:line 61, at QuantConnect.Lean.Engine.Results.Analysis.Analyses.ParameterCountAnalysis.Run(ResultsAnalysisRunParameters parameters) in Engine/Results/Analysis/Analyses/ParameterCountAnalysis.cs:line 39, at QuantConnect.Lean.Engine.Results.Analysis.ResultsAnalyzer.Run(Int32 timeLimitSeconds, Int32 maxFailedAnalyses) in Engine/Results/Analysis/ResultsAnalyzer.cs:line 122, at QuantConnect.Lean.Engine.Results.BacktestingResultHandler.SendFinalResult() in Engine/Results/BacktestingResultHandler.cs:line 420

During the algorithm initialization, the following exception has occurred: Loader.TryCreatePythonAlgorithm(): Unable to import python module ./cache/algorithm/project/main.pyc. AlgorithmPythonWrapper(): name 'QuiverCNBCsUniverseSelectionData' is not defined
at CramerTrackerAlgorithm
def _select_cnbc(self, data: List[QuiverCNBCsUniverseSelectionData]) -> List[Symbol]:
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
in main.py: line 39
at
class CramerTrackerAlgorithm(QCAlgorithm):
in main.py: line 5
name 'QuiverCNBCsUniverseSelectionData' is not defined

Code

main.py

from AlgorithmImports import *
from QuantConnect.DataSource import *


class CramerTrackerAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(100_000)
        self.settings.seed_initial_prices = True

        # Cache of top-100 liquid symbols from fundamental universe
        self._liquid_symbols = set()

        # Fundamental universe: top 100 most liquid US equities
        self._fundamental_universe = self.add_universe(self._select_fundamentals)

        # Quiver CNBC-trader alternative data universe filtered to Cramer mentions
        self._cnbc_universe = self.add_universe(
            QuiverCNBCsUniverseSelectionData,
            "QuiverCNBCsUniverse",
            Resolution.DAILY,
            self._select_cnbc
        )

    def _select_fundamentals(self, fundamentals: List[Fundamental]) -> List[Symbol]:
        # Sort by dollar volume descending and take top 100
        sorted_by_volume = sorted(
            [f for f in fundamentals if f.has_fundamental_data],
            key=lambda f: f.dollar_volume,
            reverse=True
        )
        selected = [f.symbol for f in sorted_by_volume[:100]]
        # Cache the symbols; leave universe unchanged if cache already populated
        self._liquid_symbols = set(selected)
        return selected

    def _select_cnbc(self, data: List[QuiverCNBCsUniverseSelectionData]) -> List[Symbol]:
        cramer_symbols = []
        for datum in data:
            if datum.traders and "Cramer" in datum.traders:
                # Only keep if it is also in the liquid universe
                if datum.symbol in self._liquid_symbols:
                    cramer_symbols.append(datum.symbol)
        return cramer_symbols

    def on_securities_changed(self, changes: SecurityChanges) -> None:
        for security in changes.added_securities:
            symbol = security.symbol
            # Skip universe symbols (non-equity canonical types)
            if symbol.security_type != SecurityType.EQUITY:
                continue
            # Subscribe to the underlying CNBC data feed for this equity
            self.add_data(QuiverCNBCs, symbol)
            self.log(f"Subscribed to QuiverCNBCs for {symbol.value}")

    def on_data(self, data: Slice) -> None:
        pass

    def on_cnb_cs_data(self, data: QuiverCNBCs) -> None:
        # Log each incoming trader mention
        self.log(
            f"{self.time} | {data.symbol.value} | Traders: {data.traders}"
        )

[AlexCatarino]

Prompt

Build an algorithm that combines two universes. The first is all SPY constituents � cache the symbols and leave the active universe unchanged. The second pulls the Quiver CNBC-trader alternative dataset and keeps only entries whose traders field mentions "Cramer". The final universe is the intersection. When a security is added, subscribe to the underlying CNBC data feed for it, and log each incoming trader mention. Use asynchronous selection and $100,000 starting cash. Backtest from September 1, 2024 to December 31, 2024.

Error

The backtest ended with a runtime error.

Algorithm.Initialize() Error: During the algorithm initialization, the following exception has occurred: Loader.TryCreatePythonAlgorithm(): Unable to import python module ./cache/algorithm/project/main.pyc. AlgorithmPythonWrapper(): name 'QuiverCNBCTrading' is not defined
at CramerTrackerAlgorithm
def _select_cnbc(self, data: List[QuiverCNBCTrading]) -> List[Symbol]:
^^^^^^^^^^^^^^^^^
in main.py: line 42
at
class CramerTrackerAlgorithm(QCAlgorithm):
in main.py: line 5
name 'QuiverCNBCTrading' is not defined Stack Trace: During the algorithm initialization, the following exception has occurred: Loader.TryCreatePythonAlgorithm(): Unable to import python module ./cache/algorithm/project/main.pyc. AlgorithmPythonWrapper(): name 'QuiverCNBCTrading' is not defined
at CramerTrackerAlgorithm
def _select_cnbc(self, data: List[QuiverCNBCTrading]) -> List[Symbol]:
^^^^^^^^^^^^^^^^^
in main.py: line 42
at
class CramerTrackerAlgorithm(QCAlgorithm):
in main.py: line 5
name 'QuiverCNBCTrading' is not defined

During the algorithm initialization, the following exception has occurred: Loader.TryCreatePythonAlgorithm(): Unable to import python module ./cache/algorithm/project/main.pyc. AlgorithmPythonWrapper(): name 'QuiverCNBCTrading' is not defined
at CramerTrackerAlgorithm
def _select_cnbc(self, data: List[QuiverCNBCTrading]) -> List[Symbol]:
^^^^^^^^^^^^^^^^^
in main.py: line 42
at
class CramerTrackerAlgorithm(QCAlgorithm):
in main.py: line 5
name 'QuiverCNBCTrading' is not defined

Code

main.py

from AlgorithmImports import *
from QuantConnect.DataSource import *


class CramerTrackerAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(100_000)

        self.settings.seed_initial_prices = True

        # Cache SPY constituent symbols; selection stays fixed after the first pass
        self._spy_symbols: set[Symbol] = set()

        # ETF constituents universe for SPY
        self._spy_universe = self.add_universe(
            self.universe.etf("SPY", universe_filter_func=self._select_spy)
        )

        # Alternative-data universe based on Quiver CNBC Trading
        self._cnbc_universe = self.add_universe(
            QuiverCNBCTradingUniverse,
            "cnbc-cramer-universe",
            Resolution.DAILY,
            self._select_cnbc,
        )

        self.universe_settings.asynchronous = True

    # ------------------------------------------------------------------ #
    # Universe selectors                                                    #
    # ------------------------------------------------------------------ #

    def _select_spy(self, constituents: List[ETFConstituentData]) -> List[Symbol]:
        """Cache all SPY constituent symbols and return them unchanged."""
        for c in constituents:
            self._spy_symbols.add(c.symbol)
        return [c.symbol for c in constituents]

    def _select_cnbc(self, data: List[QuiverCNBCTrading]) -> List[Symbol]:
        """Keep only symbols whose latest CNBC entry mentions Cramer."""
        cramer_symbols = [
            d.symbol for d in data
            if d.traders is not None and "Cramer" in d.traders
        ]
        # Intersect with the cached SPY symbols
        return [s for s in cramer_symbols if s in self._spy_symbols]

    # ------------------------------------------------------------------ #
    # Security changed handler                                             #
    # ------------------------------------------------------------------ #

    def on_securities_changed(self, changes: SecurityChanges) -> None:
        for security in changes.added_securities:
            # Subscribe to per-security CNBC alternative data feed
            cnbc_symbol = self.add_data(
                QuiverCNBCTrading, security.symbol
            )
            self.log(
                f"Added security {security.symbol.value} | "
                f"CNBC data key: {cnbc_symbol.symbol}"
            )

    # ------------------------------------------------------------------ #
    # Data event handlers                                                  #
    # ------------------------------------------------------------------ #

    def on_data(self, slice: Slice) -> None:
        # Use the typed dataset accessor to iterate CNBC data points
        cnbc_data = slice.get(QuiverCNBCTrading)
        if cnbc_data is None:
            return
        for symbol, data_point in cnbc_data.items():
            traders = data_point.traders if data_point.traders else "N/A"
            self.log(
                f"[CNBC] {self.time} | {symbol.underlying.value} | "
                f"Traders: {traders}"
            )

[AlexCatarino]

Prompt

Build an algorithm using the Brain Sentiment Indicator universe to select the 10 US Equities with the most positive 30-day sentiment, requiring at least one article mention in the past 30 days and a non-null sentiment score. Refresh selection at the start of each month. Cache the sentiment scores per symbol, then on a scheduled event at 9:45 AM on the first market day of each month, build a sentiment-weighted portfolio, skipping any symbol with no price yet. Backtest from September 1, 2024 to December 31, 2024.

Error

The backtest ended with a runtime error.

Runtime Error: 'BrainStockRankingUniverse' object has no attribute 'thirty_day_sentiment'
at _select_universe
score = datum.thirty_day_sentiment
^^^^^^^^^^^^^^^^^^^^^^^^^^
in main.py: line 29

'BrainStockRankingUniverse' object has no attribute 'thirty_day_sentiment'
at _select_universe
score = datum.thirty_day_sentiment
^^^^^^^^^^^^^^^^^^^^^^^^^^
in main.py: line 29

Code

main.py

from AlgorithmImports import *
from QuantConnect.DataSource import *


class BrainSentimentAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(100_000)
        self.settings.seed_initial_prices = True

        self._sentiment_by_symbol = {}

        self._universe = self.add_universe(
            BrainStockRankingUniverse,
            self._select_universe
        )

        self.schedule.on(
            self.date_rules.month_start(days_offset=0),
            self.time_rules.at(9, 45),
            self._rebalance
        )

    def _select_universe(self, data):
        scored = []
        for datum in data:
            score = datum.thirty_day_sentiment
            mentions = datum.thirty_day_article_mentions
            if score is None or mentions is None:
                continue
            if mentions < 1:
                continue
            self._sentiment_by_symbol[datum.symbol] = score
            scored.append((datum.symbol, score))

        scored.sort(key=lambda x: x[1], reverse=True)
        return [s for s, _ in scored[:10]]

    def _rebalance(self):
        targets = []
        total_score = 0.0

        active_symbols = self._universe.selected
        if not active_symbols:
            return

        eligible = []
        for symbol in active_symbols:
            if not self.securities[symbol].has_data:
                continue
            score = self._sentiment_by_symbol.get(symbol)
            if score is None:
                continue
            eligible.append((symbol, score))
            total_score += score

        if total_score <= 0 or not eligible:
            return

        for symbol, score in eligible:
            weight = score / total_score
            targets.append(PortfolioTarget(symbol, weight))

        self.set_holdings(targets, liquidate_existing_holdings=True)

[AlexCatarino]

Prompt

Build an algorithm using the Smart Insider Intention universe to select equities that have announced a buyback program above $10 million. Refresh weekly at the start of each week. Cache each announced buyback amount per symbol. At 1 minute after the SPY market open on week-start, rebalance to a buyback-size-weighted portfolio (weight = amount divided by sum-of-amounts), skipping zero-price securities. When a security leaves the universe, liquidate it and drop it from the cache. Seed initial prices. Backtest from September 1, 2024 to December 31, 2024.

Error

The backtest ended with a runtime error.

During the algorithm initialization, the following exception has occurred: Trying to dynamically access a method that does not exist throws a TypeError exception. To prevent the exception, ensure each parameter type matches those required by the 'method'>) method. Please checkout the API documentation.
at initialize
SmartInsiderIntentionUniverse(
in main.py: line 18
No method matches given arguments for .ctor: (, )

Trying to dynamically access a method that does not exist throws a TypeError exception. To prevent the exception, ensure each parameter type matches those required by the 'method'>) method. Please checkout the API documentation.
at initialize
SmartInsiderIntentionUniverse(
in main.py: line 18
No method matches given arguments for .ctor: (<class 'QuantConnect.Data.UniverseSelection.UniverseSettings'>, <class 'method'>)

Code

main.py

from AlgorithmImports import *
from datetime import timedelta


class SmartInsiderBuybackAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(1_000_000)
        self.settings.seed_initial_prices = True

        # Cache: symbol -> buyback amount
        self._buyback_amounts: dict[Symbol, float] = {}

        # Add Smart Insider Intention universe, refreshed weekly
        self._universe = self.add_universe(
            SmartInsiderIntentionUniverse(
                self.universe_settings,
                self._select_universe
            )
        )
        self.universe_settings.schedule.on(
            self.date_rules.week_start(),
            self.time_rules.midnight
        )

        # Schedule rebalance 1 min after SPY open on week start
        spy = self.add_equity("SPY", Resolution.MINUTE)
        self._spy = spy.symbol

        self.schedule.on(
            self.date_rules.week_start(),
            self.time_rules.after_market_open(self._spy, 1),
            self._rebalance
        )

    def _select_universe(self, intentions: list) -> list[Symbol]:
        selected = []
        for intention in intentions:
            amount = intention.amount
            if amount is not None and amount >= 10_000_000:
                self._buyback_amounts[intention.symbol] = float(amount)
                selected.append(intention.symbol)
        return selected

    def on_securities_changed(self, changes: SecurityChanges):
        for removed in changes.removed_securities:
            sym = removed.symbol
            if sym in self._buyback_amounts:
                del self._buyback_amounts[sym]
            if sym != self._spy:
                self.liquidate(sym)

    def _rebalance(self):
        # Gather active universe symbols with non-zero prices
        active = [
            sym for sym in self._universe.selected
            if sym in self._buyback_amounts
            and self.securities[sym].price > 0
        ]

        if not active:
            return

        total_amount = sum(self._buyback_amounts[sym] for sym in active)
        if total_amount == 0:
            return

        targets = [
            PortfolioTarget(sym, self._buyback_amounts[sym] / total_amount)
            for sym in active
        ]

        self.set_holdings(targets, liquidate_existing_holdings=True)