Universes
Custom Universes
Define Custom Universe Types
Custom universes should extend the BaseData
PythonData
class. Extensions of the BaseData
PythonData
class must implement a GetSource
and Reader
method.
The GetSource
method in your custom data class instructs LEAN where to find the data. This method must return a SubscriptionDataSource
object, which contains the data location and format (SubscriptionTransportMedium
). You can even change source locations for backtesting and live modes. We support many different data sources.
The Reader
method of your custom data class takes one line of data from the source location and parses it into one of your custom objects. You can add as many properties to your custom data objects as you need, but must set Symbol
and EndTime
properties. When there is no useable data in a line, the method should return null
None
. LEAN repeatedly calls the Reader
method until the date/time advances or it reaches the end of the file.
//Example custom universe data; it is virtually identical to other custom data types. public class MyCustomUniverseDataClass : BaseData { public int CustomAttribute1; public decimal CustomAttribute2; public override DateTime EndTime { // define end time as exactly 1 day after Time get { return Time + QuantConnect.Time.OneDay; } set { Time = value - QuantConnect.Time.OneDay; } } public override SubscriptionDataSource GetSource(SubscriptionDataConfig config, DateTime date, bool isLiveMode) { return new SubscriptionDataSource(@"your-remote-universe-data", SubscriptionTransportMedium.RemoteFile); } public override BaseData Reader(SubscriptionDataConfig config, string line, DateTime date, bool isLiveMode) { var items = line.Split(","); // Generate required data, then return an instance of your class. return new MyCustomUniverseDataClass { EndTime = Parse.DateTimeExact(items[0], "yyyy-MM-dd"), Symbol = Symbol.Create(items[1], SecurityType.Equity, Market.USA), CustomAttribute1 = int.Parse(items[2]), CustomAttribute2 = decimal.Parse(items[3], NumberStyles.Any, CultureInfo.Invariant) }; } }
# Example custom universe data; it is virtually identical to other custom data types. class MyCustomUniverseDataClass(PythonData): def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource: return SubscriptionDataSource(@"your-remote-universe-data", SubscriptionTransportMedium.RemoteFile) def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData: items = line.split(",") # Generate required data, then return an instance of your class. data = MyCustomUniverseDataClass() data.EndTime = datetime.strptime(items[0], "%Y-%m-%d") # define Time as exactly 1 day earlier Time data.Time = data.EndTime - timedelta(1) data.Symbol = Symbol.Create(items[1], SecurityType.Crypto, Market.Bitfinex) data["CustomAttribute1"] = int(items[2]) data["CustomAttribute2"] = float(items[3]) return data
Your Reader
method should return objects in chronological order. If an object has a timestamp that is the same or earlier than the timestamp of the previous object, LEAN ignores it.
If you need to create multiple objects in your Reader
method from a single line
, follow these steps:
- In the
GetSource
method, passFileFormat.UnfoldingCollection
as the third argument to theSubscriptionDataSource
constructor. - In the
Reader
method, order the objects by their timestamp and then return aBaseDataCollection(endTime, config.Symbol, objects)
whereobjects
is a list of your custom data objects.
public class MyCustomUniverseDataClass : BaseData { [JsonProperty(PropertyName = "Attr1")] public int CustomAttribute1 { get; set; } [JsonProperty(PropertyName = "Ticker")] public string Ticker { get; set; } [JsonProperty(PropertyName = "date")] public DateTime Date { get; set; } public override DateTime EndTime { // define end time as exactly 1 day after Time get { return Time + QuantConnect.Time.OneDay; } set { Time = value - QuantConnect.Time.OneDay; } } public MyCustomUniverseDataClass() { Symbol = Symbol.Empty; DataType = MarketDataType.Base; } public override SubscriptionDataSource GetSource(SubscriptionDataConfig config, DateTime date, bool isLiveMode) { return new SubscriptionDataSource(@"your-data-source-url", SubscriptionTransportMedium.RemoteFile, FileFormat.UnfoldingCollection); } public override BaseData Reader(SubscriptionDataConfig config, string line, DateTime date, bool isLiveMode) { var items = JsonConvert.DeserializeObject<List<MyCustomUniverseDataClass>>(line); var endTime = items.Last().Date; foreach (var item in items) { item.Symbol = Symbol.Create(item.Ticker, SecurityType.Equity, Market.USA); item.Time = item.Date; item.Value = (decimal) item.CustomAttribute1; } return new BaseDataCollection(endTime, config.Symbol, items); } }
class MyCustomUniverseDataClass(PythonData): def GetSource(self, config, date, isLive): return SubscriptionDataSource("your-data-source-url", SubscriptionTransportMedium.RemoteFile, FileFormat.UnfoldingCollection) def Reader(self, config, line, date, isLive): json_response = json.loads(line) endTime = datetime.strptime(json_response[-1]["date"], '%Y-%m-%d') + timedelta(1) data = list() for json_datum in json_response: datum = MyCustomUniverseDataClass() datum.Symbol = Symbol.Create(json_datum["Ticker"], SecurityType.Equity, Market.USA) datum.Time = datetime.strptime(json_datum["date"], '%Y-%m-%d') datum.EndTime = datum.Time + timedelta(1) datum['CustomAttribute1'] = int(json_datum['Attr1']) datum.Value = float(json_datum['Attr1']) data.append(datum) return BaseDataCollection(endTime, config.Symbol, data)
Initialize Custom Universes
To add a custom universe to your algorithm, in the Initialize
method, pass your universe type and a selector function to the AddUniverse
method. The selector function receives a list of your custom objects and must return a list of Symbol
objects. In the selector function definition, you can use any of the properties of your custom data type. The Symbol
objects that you return from the selector function set the constituents of the universe.
// In Initialize AddUniverse<MyCustomUniverseDataClass>("myCustomUniverse", Resolution.Daily, data => { return (from singleStockData in data where singleStockData.CustomAttribute1 > 0 orderby singleStockData.CustomAttribute2 descending select singleStockData.Symbol).Take(5); });
# In Initialize self.AddUniverse(MyCustomUniverseDataClass, "myCustomUniverse", Resolution.Daily, self.selector_function) # Define the selector function def selector_function(self, data: List[MyCustomUniverseDataClass]) -> List[Symbol]: sorted_data = sorted([ x for x in data if x["CustomAttribute1"] > 0 ], key=lambda x: x["CustomAttribute2"], reverse=True) return [x.Symbol for x in sorted_data[:5]]