Price and Bar-Index Regression with Candle Breakout Signals
Article Strategy library · Author: QuantConnect
Summary
This strategy fits a rolling linear relationship between closing price and bar index. Over a configurable window, it calculates averages, standard deviations, and correlation to derive a regression slope and intercept, then evaluates whether the resulting line is rising or falling. With a rising line, a bullish candle that closes above the previous bar’s high triggers a long entry. With a falling line, a bearish candle that closes below the previous bar’s low closes the long position. The published window parameter is 64.
The description presents the method as an adjustable trend indicator and notes that window choice can make it too smooth or too sensitive, while short-term price noise may undermine signals. It suggests adding volume confirmation or dynamic exits, but the supplied code enters and closes longs without a stated stop or short entry. Backtest settings specify BTC_USDT futures and a date range, yet no performance results are given. The material is therefore a rule description, not evidence of robust returns across markets or periods.
Key ideas
- The strategy estimates a rolling regression of closing price against bar index.
- A rising regression line plus a close above the prior high triggers a long entry.
- A falling line plus a bearish close below the prior low closes the long position.
- The window length affects how smooth or responsive the regression signal is.
- The supplied material gives backtest settings but no performance results.
Tags
Full text
# EmaCrossAlphaModel
# EmaCrossAlphaModel
Alpha model that uses an EMA cross to create insights
## Source (Apache-2.0)
```python
# 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.
from AlgorithmImports import *
class EmaCrossAlphaModel(AlphaModel):
'''Alpha model that uses an EMA cross to create insights'''
def __init__(self,
fast_period = 12,
slow_period = 26,
resolution = Resolution.DAILY):
'''Initializes a new instance of the EmaCrossAlphaModel class
Args:
fast_period: The fast EMA period
slow_period: The slow EMA period'''
self.fast_period = fast_period
self.slow_period = slow_period
self.resolution = resolution
self.prediction_interval = Time.multiply(Extensions.to_time_span(resolution), fast_period)
self.symbol_data_by_symbol = {}
self.name = '{}({},{},{})'.format(self.__class__.__name__, fast_period, slow_period, resolution)
def update(self, algorithm, data):
'''Updates this alpha model with the latest data from the algorithm.
This is called each time the algorithm receives data for subscribed securities
Args:
algorithm: The algorithm instance
data: The new data available
Returns:
The new insights generated'''
insights = []
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if symbol_data.fast.is_ready and symbol_data.slow.is_ready:
if symbol_data.fast_is_over_slow:
if symbol_data.slow > symbol_data.fast:
insights.append(Insight.price(symbol_data.symbol, self.prediction_interval, InsightDirection.DOWN))
elif symbol_data.slow_is_over_fast:
if symbol_data.fast > symbol_data.slow:
insights.append(Insight.price(symbol_data.symbol, self.prediction_interval, InsightDirection.UP))
symbol_data.fast_is_over_slow = symbol_data.fast > symbol_data.slow
return insights
def on_securities_changed(self, algorithm, changes):
'''Event fired each time the we add/remove securities from the data feed
Args:
algorithm: The algorithm instance that experienced the change in securities
changes: The security additions and removals from the algorithm'''
for added in changes.added_securities:
symbol_data = self.symbol_data_by_symbol.get(added.symbol)
if symbol_data is None:
symbol_data = SymbolData(added, self.fast_period, self.slow_period, algorithm, self.resolution)
self.symbol_data_by_symbol[added.symbol] = symbol_data
else:
# a security that was already initialized was re-added, reset the indicators
symbol_data.fast.reset()
symbol_data.slow.reset()
for removed in changes.removed_securities:
data = self.symbol_data_by_symbol.pop(removed.symbol, None)
if data is not None:
# clean up our consolidators
data.remove_consolidators()
class SymbolData:
'''Contains data specific to a symbol required by this model'''
def __init__(self, security, fast_period, slow_period, algorithm, resolution):
self.security = security
self.symbol = security.symbol
self.algorithm = algorithm
self.fast_consolidator = algorithm.resolve_consolidator(security.symbol, resolution)
self.slow_consolidator = algorithm.resolve_consolidator(security.symbol, resolution)
algorithm.subscription_manager.add_consolidator(security.symbol, self.fast_consolidator)
algorithm.subscription_manager.add_consolidator(security.symbol, self.slow_consolidator)
# create fast/slow EMAs
self.fast = ExponentialMovingAverage(security.symbol, fast_period, ExponentialMovingAverage.smoothing_factor_default(fast_period))
self.slow = ExponentialMovingAverage(security.symbol, slow_period, ExponentialMovingAverage.smoothing_factor_default(slow_period))
algorithm.register_indicator(security.symbol, self.fast, self.fast_consolidator);
algorithm.register_indicator(security.symbol, self.slow, self.slow_consolidator);
algorithm.warm_up_indicator(security.symbol, self.fast, resolution);
algorithm.warm_up_indicator(security.symbol, self.slow, resolution);
# True if the fast is above the slow, otherwise false.
# This is used to prevent emitting the same signal repeatedly
self.fast_is_over_slow = False
def remove_consolidators(self):
self.algorithm.subscription_manager.remove_consolidator(self.security.symbol, self.fast_consolidator)
self.algorithm.subscription_manager.remove_consolidator(self.security.symbol, self.slow_consolidator)
@property
def slow_is_over_fast(self):
return not self.fast_is_over_slow
```Shown in full with attribution under the source's licence. Licence: Apache-2.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.