/*
* QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
* Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
using System;
using System.Collections;
using System.Collections.Generic;
using QuantConnect.Data;
using QuantConnect.Data.UniverseSelection;
namespace QuantConnect.Lean.Engine.DataFeeds.Enumerators
{
///
/// Provides an implementation of
/// that aggregates an underlying into a single
/// data packet
///
public class BaseDataCollectionAggregatorEnumerator : IEnumerator
where T : BaseDataCollection, new()
{
private bool _endOfStream;
private bool _needsMoveNext;
private bool _liveMode;
private readonly Symbol _symbol;
private readonly IEnumerator _enumerator;
///
/// Initializes a new instance of the class
/// This will aggregate instances emitted from the underlying enumerator and tag them with the
/// specified symbol
///
/// The underlying enumerator to aggregate
/// The symbol to place on the aggregated collection
/// True if running in live mode
public BaseDataCollectionAggregatorEnumerator(IEnumerator enumerator, Symbol symbol, bool liveMode = false)
{
_symbol = symbol;
_enumerator = enumerator;
_liveMode = liveMode;
_needsMoveNext = true;
}
///
/// Advances the enumerator to the next element of the collection.
///
///
/// true if the enumerator was successfully advanced to the next element; false if the enumerator has passed the end of the collection.
///
/// The collection was modified after the enumerator was created. 2
public bool MoveNext()
{
if (_endOfStream)
{
return false;
}
T collection = null;
while (true)
{
if (_needsMoveNext)
{
// move next if we dequeued the last item last time we were invoked
if (!_enumerator.MoveNext())
{
_endOfStream = true;
if (!IsValid(collection))
{
// we don't emit
collection = null;
}
break;
}
}
if (_enumerator.Current == null)
{
// the underlying returned null, stop here and start again on the next call
_needsMoveNext = true;
break;
}
if (collection == null)
{
// we have new data, set the collection's symbol/times
var current = _enumerator.Current;
collection = CreateCollection(_symbol, current.Time, current.EndTime);
}
if (collection.EndTime != _enumerator.Current.EndTime)
{
// the data from the underlying is at a different time, stop here
_needsMoveNext = false;
if (IsValid(collection))
{
// we emit
break;
}
// we try again
collection = null;
continue;
}
// this data belongs in this collection, keep going until null or bad time
Add(collection, _enumerator.Current);
_needsMoveNext = true;
}
Current = collection;
return _liveMode || collection != null;
}
///
/// Sets the enumerator to its initial position, which is before the first element in the collection.
///
/// The collection was modified after the enumerator was created. 2
public void Reset()
{
_enumerator.Reset();
}
///
/// Gets the element in the collection at the current position of the enumerator.
///
///
/// The element in the collection at the current position of the enumerator.
///
public T Current
{
get; private set;
}
///
/// Gets the current element in the collection.
///
///
/// The current element in the collection.
///
/// 2
object IEnumerator.Current
{
get { return Current; }
}
///
/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
///
/// 2
public void Dispose()
{
_enumerator.Dispose();
}
///
/// Creates a new, empty .
///
/// The base data collection symbol
/// The start time of the collection
/// The end time of the collection
/// A new, empty
protected virtual T CreateCollection(Symbol symbol, DateTime time, DateTime endTime)
{
return new T
{
Symbol = symbol,
Time = time,
EndTime = endTime
};
}
///
/// Adds the specified instance of to the current collection
///
/// The collection to be added to
/// The data to be added
protected virtual void Add(T collection, BaseData current)
{
collection.Data.Add(current);
}
///
/// Adds all specified instances of to the current collection
///
/// The collection to be added to
/// The data collection to be added
protected virtual void SetData(T collection, List current)
{
collection.Data = current;
}
///
/// Determines if a given data point is valid and can be emitted
///
/// The collection to be emitted
/// True if its a valid data point
protected virtual bool IsValid(T collection)
{
return true;
}
}
///
/// Provides a non-generic implementation of
///
public class BaseDataCollectionAggregatorEnumerator : BaseDataCollectionAggregatorEnumerator
{
///
/// Initializes a new instance of the class
///
/// The enumerator to aggregate
/// The output data's symbol
/// True if running in live mode
public BaseDataCollectionAggregatorEnumerator(IEnumerator enumerator, Symbol symbol, bool liveMode = false)
: base(enumerator, symbol, liveMode)
{
}
}
}