book
Checkout our new book! Hands on AI Trading with Python, QuantConnect, and AWS Learn More arrow

Drafts

CrunchDAO

Introduction

CrunchDAO is a service that connects strategy developers and capital allocators. With our CrunchDAO integration, you can send trading signals from your live algorithms to the CrunchDAO.

Add Providers

To export signals to CrunchDAO from your algorithm, during initialization, add a CrunchDAO signal export provider.

SignalExport.AddSignalExportProviders(new CrunchDAOSignalExport(apiKey, model, submissionName, comment));
self.signal_export.add_signal_export_providers(CrunchDAOSignalExport(apiKey, model, submissionName, comment))

The CrunchDAOSignalExport constructor accepts the following arguments:

Argument: apiKeyapi_key

Your CrunchDAO API key. To get your API key, see the Profile > Settings page on the CrunchDAO website.

Data Type: stringstr | Default Value: None

Argument: model

The Id of your CrunchDAO model. To create a new model or to get the Id of an existing model, see the Profile > Alpha page on the CrunchDAO website.

Data Type: stringstr | Default Value: None

Argument: submissionName

A name for the submission to distinguish it from your other submissions.

Data Type: stringstr | Default Value: ""

Argument: comment

A comment for the submission.

Data Type: stringstr | Default Value: ""

You can add multiple signal export providers to a single algorithm.

Asset Classes

Our CrunchDAO integration supports signals for US Equities.

Universe Selection

The CrunchDAO skeleton contains the universe constituents. To load it into your algorithm, follow these steps:

  1. Define a custom universe type.
  2. class CrunchDaoSkeleton(PythonData):
        
        def get_source(self, config, date, isLive):
            return SubscriptionDataSource("https://tournament.crunchdao.com/data/skeleton.csv", SubscriptionTransportMedium.REMOTE_FILE)
    
        def reader(self, config, line, date, isLive):
            if not line[0].isdigit(): return None
            skeleton = CrunchDaoSkeleton()
            skeleton.symbol = config.symbol
    
            try:
                csv = line.split(',')
                skeleton.end_time = (datetime.strptime(csv[0], "%Y-%m-%d")).date() 
                skeleton.symbol =  Symbol(SecurityIdentifier.generate_equity(csv[1], Market.USA, mapping_resolve_date=skeleton.time), csv[1])
                skeleton["Ticker"] = csv[1]
    
            except ValueError:
                # Do nothing
                return None
    
            return skeleton
  3. In the Initializeinitialize method, initialize the universe.
  4. self.add_universe(CrunchDaoSkeleton, "CrunchDaoSkeleton", Resolution.DAILY, self.select_symbols)
  5. In the algorithm class, define the selector function to select all of the securities in the skeleton file.
  6. def select_symbols(self, data: list[CrunchDaoSkeleton]) -> list[Symbol]:
        return [x.Symbol for x in data]

Schedule Submissions

CrunchDAO provides a new skeleton every Saturday at 6 PM Coordinated Universal Time (UTC). To schedule your submissions for before the market opens each week, during initialization, you can create a Scheduled Event.

self.week = -1
self.schedule.on(
    self.date_rules.every([DayOfWeek.monday, DayOfWeek.tuesday, DayOfWeek.wednesday, DayOfWeek.thursday, DayOfWeek.friday]), 
    self.time_rules.at(13, 15, TimeZones.UTC), 
    self.submit_signals)

The preceding Scheduled Event calls the submit_signals method every weekday at 1:15 PM UTC. Define this method to create and send portfolio targets to CrunchDAO.

def submit_signals(self):
    if self.IsWarmingUp:
        return
    
    # Submit signals once per week
    week_num = self.Time.isocalendar()[1]
    if self.week == week_num:
        return
    self.week = week_num

    symbols = [security.Symbol for security in self.crunch_universe if security.Price > 0]

    # Get historical price data
    # close_prices = self.History(symbols, 22, Resolution.Daily).close.unstack(0)

    # Create portfolio targets
    weight_by_symbol = {symbol: 1/len(symbols) for symbol in symbols} # Add your logic here
    targets = [PortfolioTarget(symbol, weight) for symbol, weight in weight_by_symbol.items()]

    # (Optional) Place trades
    self.SetHoldings(targets)

    # Send signals to CrunchDAO
    success = self.SignalExport.SetTargetPortfolio(targets)
    if not success:
        self.Debug(f"Couldn't send targets at {self.Time}")

For more information about requesting historical price and alternative data, see History Requests.

Send Portfolio Targets

The signal export manager automatically sends signals when your portfolio holdings change to CrunchDAO. By default, it waits five seconds after an order is filled to aggregate all the state changes into a single post. Set the AutomaticExportTimeSpanautomatic_export_time_span property to change the waiting time.

SignalExport.AutomaticExportTimeSpan = TimeSpan.FromSeconds(1);
self.signal_export.automatic_export_time_span = timedelta(seconds=1)

To send targets that aren't based on your portfolio holdings, see Manual Exports.

To disable automatic exports, set the AutomaticExportTimeSpanautomatic_export_time_span property to nullNone.

SignalExport.AutomaticExportTimeSpan = null;
self.signal_export.automatic_export_time_span = None

Signal Requirements

When you submit signals to CrunchDAO, abide by the following rules:

  • All signals must have a quantity between 0 and 1 (inclusive).
  • Submit one signal per round. If you submit more than one signal in a round, your new submission overwrites the previous one.
  • Send signals for at least half of the investment universe.

You can also see our Videos. You can also get in touch with us via Discord.

Did you find this page helpful?

Contribute to the documentation: