SubscriptionManager

QuantConnect.Data.SubscriptionManager

SubscriptionManager(time_keeper: ITimeKeeper)

Bases: Object

Enumerable Subscription Management Class

Creates a new instance

subscription_data_config_service

subscription_data_config_service: (
    ISubscriptionDataConfigService
)

Instance that implements ISubscriptionDataConfigService

subscriptions

subscriptions: Iterable[SubscriptionDataConfig]

Returns an IEnumerable of Subscriptions

available_data_types

available_data_types: Dictionary[
    SecurityType, List[TickType]
]

The different TickType each SecurityType supports

count

count: int

Get the count of assets:

add

add(
    symbol: Union[Symbol, str, BaseContract],
    resolution: Resolution,
    time_zone: Any,
    exchange_time_zone: Any,
    is_custom_data: bool = False,
    fill_forward: bool = True,
    extended_market_hours: bool = False,
) -> SubscriptionDataConfig

add(
    data_type: Type,
    tick_type: TickType,
    symbol: Union[Symbol, str, BaseContract],
    resolution: Resolution,
    data_time_zone: Any,
    exchange_time_zone: Any,
    is_custom_data: bool,
    fill_forward: bool = True,
    extended_market_hours: bool = False,
    is_internal_feed: bool = False,
    is_filtered_subscription: bool = True,
    data_normalization_mode: DataNormalizationMode = ...,
) -> SubscriptionDataConfig

Signature descriptions:

Add Market Data Required (Overloaded method for backwards compatibility).
Add Market Data Required - generic data typing support as long as Type implements BaseData.

Parameters:

Name	Type	Description	Default
`symbol`	`Union[Symbol, str, BaseContract]`	Symbol of the asset we're like	required
`resolution`	`Resolution`	Resolution of Asset Required	required
`time_zone`	`Optional[Any]`	The time zone the subscription's data is time stamped in	`None`
`exchange_time_zone`	`Any`	Specifies the time zone of the exchange for the security this subscription is for. This	required
`is_custom_data`	`bool`	True if this is custom user supplied data, false for normal QC data	`False`
`fill_forward`	`bool`	when there is no data pass the last tradebar forward	`True`
`extended_market_hours`	`bool`	Request premarket data as well when true	`False`
`data_type`	`Optional[Type]`	Set the type of the data we're subscribing to.	`None`
`tick_type`	`Optional[TickType]`	Tick type for the subscription.	`None`
`data_time_zone`	`Optional[Any]`	The time zone the subscription's data is time stamped in	`None`
`is_internal_feed`	`Optional[bool]`	Set to true to prevent data from this subscription from being sent into the algorithm's	`False`
`is_filtered_subscription`	`Optional[bool]`	True if this subscription should have filters applied to it (market hours/user	`True`
`data_normalization_mode`	`Optional[DataNormalizationMode]`	Define how data is normalized	`...`

Returns:

Type	Description
`SubscriptionDataConfig`	The newly created SubscriptionDataConfig or existing instance if it already existed.

add_consolidator

add_consolidator(
    symbol: Union[Symbol, str, BaseContract],
    py_consolidator: Any,
) -> None

add_consolidator(
    symbol: Union[Symbol, str, BaseContract],
    consolidator: Union[
        IDataConsolidator, PythonConsolidator, timedelta
    ],
    tick_type: Optional[TickType] = None,
) -> None

Signature descriptions:

Add a custom python consolidator for the symbol
Add a consolidator for the symbol

Parameters:

Name	Type	Description	Default
`symbol`	`Union[Symbol, str, BaseContract]`	Symbol of the asset to consolidate	required
`py_consolidator`	`Optional[Any]`	The custom python consolidator	`None`
`consolidator`	`Optional[Union[IDataConsolidator, PythonConsolidator, timedelta]]`	The consolidator	`None`
`tick_type`	`Optional[Optional[TickType]]`	Desired tick type for the subscription	`None`

remove_consolidator

remove_consolidator(
    symbol: Union[Symbol, str, BaseContract],
    py_consolidator: Any,
) -> None

remove_consolidator(
    symbol: Union[Symbol, str, BaseContract],
    consolidator: Union[
        IDataConsolidator, PythonConsolidator, timedelta
    ],
) -> None

Signature descriptions:

Removes the specified python consolidator for the symbol
Removes the specified consolidator for the symbol

Parameters:

Name	Type	Description	Default
`symbol`	`Union[Symbol, str, BaseContract]`	The symbol the consolidator is receiving data from	required
`py_consolidator`	`Optional[Any]`	The python consolidator instance to be removed	`None`
`consolidator`	`Optional[Union[IDataConsolidator, PythonConsolidator, timedelta]]`	The consolidator instance to be removed	`None`

default_data_types

default_data_types() -> (
    Dictionary[SecurityType, List[TickType]]
)

Hard code the set of default available data feeds

get_data_types_for_security

get_data_types_for_security(
    security_type: SecurityType,
) -> Sequence[TickType]

Get the available data types for a security

is_subscription_valid_for_consolidator

is_subscription_valid_for_consolidator(
    subscription: SubscriptionDataConfig,
    consolidator: Union[
        IDataConsolidator, PythonConsolidator, timedelta
    ],
    desired_tick_type: Optional[TickType] = None,
) -> bool

Checks if the subscription is valid for the consolidator

Parameters:

Name	Type	Description	Default
`subscription`	`SubscriptionDataConfig`	The subscription configuration	required
`consolidator`	`Union[IDataConsolidator, PythonConsolidator, timedelta]`	The consolidator	required
`desired_tick_type`	`Optional[TickType]`	The desired tick type for the subscription. If not given is null.	`None`

Returns:

Type	Description
`bool`	true if the subscription is valid for the consolidator.

lookup_subscription_config_data_types

lookup_subscription_config_data_types(
    symbol_security_type: SecurityType,
    resolution: Resolution,
    is_canonical: bool,
) -> List[Tuple[Type, TickType]]

Get the data feed types for a given SecurityTypeResolution

Parameters:

Name	Type	Description	Default
`symbol_security_type`	`SecurityType`	The SecurityType used to determine the types	required
`resolution`	`Resolution`	The resolution of the data requested	required
`is_canonical`	`bool`	Indicates whether the security is Canonical (future and options)	required

Returns:

Type	Description
`List[Tuple[Type, TickType]]`	Types that should be added to the SubscriptionDataConfig.

scan_past_consolidators

scan_past_consolidators(
    new_utc_time: Union[datetime, date],
    algorithm: IAlgorithm,
) -> None

Will trigger past consolidator scans

Parameters:

Name	Type	Description	Default
`new_utc_time`	`Union[datetime, date]`	The new utc time	required
`algorithm`	`IAlgorithm`	The algorithm instance	required

set_data_manager

set_data_manager(
    subscription_manager: IAlgorithmSubscriptionManager,
) -> None

Sets the Subscription Manager