Trying to Extend a Regression Channel/Trend

Hi LE,

My first love, TV! 😊

You're using the Regression Channel indicator properly. From the doc:

    /// <summary>
   /// The Regression Channel indicator extends the <see cref="LeastSquaresMovingAverage"/>
   /// with the inclusion of two (upper and lower) channel lines that are distanced from
   /// the linear regression line by a user defined number of standard deviations.
   /// Reference: http://www.onlinetradingconcepts.com/TechnicalAnalysis/LinRegChannel.html
   /// </summary>

Least squares moving average will give you curves, not lines. So the QC one is based on classic linear regression, which is fine. Least squares will follow price action closely.

I played around with lin reg in TV of the form y = mx + b (lines). This is certainly doable in Python/QC. You indeed won't see in the future in QC but that's something very good. You don't want to fight “repainting”, believe me. But you can have last bar's position wrt the channel for sure.

You will most likely be flooded with code in the next couple of hours/days but if not, I promise you to look at it. I'll be back with an indie that does that and share to all. Btw, you are just at the verge of playing with zigzags… You might want to take a look too…

Cheers,

Fred

Thank you for the insights Fred! 

I'm just thinking of a simple function since I'm not that advanced. Something like:

1. A RollingWindow that collects the desired amount of bars (example: 24) in OnData or a Consolidator
2. When ready, Calculate the linear regression line of the Close values of these bars
3. Calculate the Upper and Lower Deviation (example: +/-2)
4. “Extend” all these lines so their continuation is available in the next period (like in the above photo).
5. Be able to check or plot the values of the 3 lines at the current time during the next period
6. When a new period ends start over again

Basically like using the TradingView Regression Trend tool. I am not sure how to accomplish most of these steps unfortunately. 

I already got my maths refreshed on lin reg with B_o and B_1 (don't mind - it means lines). We're on the same page.

Except for extending and plotting lines. Lines won't extend as they won't be drawn. But maths will do the prediction. It's already part of lin reg. The indie will also give you the Pearson fit. If you care.

You will be able to do the plotting and drawing as an exercise if you will. The indie will be designed towards algo-trading, i.e, the algo polls the indie to know where is price action wrt the lin reg and acts accordingly, adding other indies in the decision-making.

Gimme a few days… It'll be fun.

Fred

Really appreciate the help Fred! 

Yeah, I understand lines can't be drawn or extend into the future. As you said, I just need to poll the indicator to get the current value of each of the three lines that were generated during the previous period (like doing so at 12:00 on Nov 24 in the above image).  With that, can plot those data points or use in another area of algorithm, etc. 

Looking forward to it, cheers! Thanks again.

Yep. You'll get the points of the projected lines for the current bar (the current being the last closed one in QC).

So you'll set a period B (say, for base) in bars (in whatever res you're in) for the computation of the lin reg over closes. You'll also set another period P (say, for projection) in bars for the period during which you want to “project those lines”. Once the first B bars are passed, the indie is ready and during the next P bars, you have your projected lines as 3 points. After P bars, the indie automatically resets and uses the next B bars after the first B bars to do lin reg again and then you get those new lines from which the 3 points are given for the next P bars. So the best is when B == P. Easy case. Now, if B < P, you have P - B bars where you used “old lines”, so to speak, to do your predictions. If B > P, the indie won't be ready right away once it resets. You will need to wait another B - P bars.

Hope this is clear.

I'll do it but share with all so I am not doing something just for you. And I do it because I know I'll have fun doing it.

Are the B and P above in line with what you had in mind?

Fred

P.S. You'll also get the Pearson fit test value. It is the blue number below the lines in the pic (0.911498…). In periods of volatility, it will be bad…. which means the lines are not a good fit for the closes. But it could also be investigated as a volatility metric.

Yeah, that's clear. So if B = P, the old P period becomes the “source” B period. 

For example using Hourly Resolution, it's just as if you were drawing on the TV chart yourself, using today's bars for tomorrow, and then tomorrow's bars for next day, etc. That's perfect, and you've already thought about if B < P, which might be useful. 

That's exactly what I had in mind. And the Pearson value could be useful for some quick context. Thank you so much for helping me and the community! 

Hi LE,

Attached is the code with a backtest to serve as an example only. Very basic strategy using the indicator… It is very unreliable. It goes long and short, which is very risky. I also don't use the correlation coefficient. Hope you'll have fun playing with it…

I don't comment too much as I believe you will understand it easily since you are already familiar and also because most of the code is either slightly documented or self-explanatory. The maths are not explained but they have been extensively explained already since the 19th century.

If you have questions, don't hesitate.

Fred

Something to keep in mind here: when measuring linear regression, for every price point on your chart, you have the corresponding LinReg value, the upper channel bound, the lower channel bound, *and* the slope at that moment in time.  

This slope is all you need to extend the three regression values into the future / past, similar to the way that you see it plotted in TradingView.

Hope this helps someone. 

Hi,

Just to straighten this out a bit. You have the slope and the intercept. You need two points to draw a line.

But here, the indicator takes care of that for you. The three returned numbers in the tuple returned by GetProjection() ARE the projections.

So, ok, let's give more insight into this. The indicator calculates the best fit line from a series of close (price) points (param base_period). It then projects that line into the future, to project the “same trend”, for a period of projection_period. It also calculates two other parallel lines, separated from the above-mentioned line by +(channel_width * standard deviation) and -(channel_width * standard deviation). The standard deviation is calculated around the mean of the close points of the same base_period. After projection_period is passed, it resets, and thus moves forward. Close prices in projection_period go into base_period, calculations are redone entirely so lines and standard deviation are re-calculated and you continue.

It thus creates a channel above and below the best fit line for close prices. You control the number of standard deviations above and below, i.e., you control the channel width.

The correlation coefficient is the degree of relationship between x (here time) and y (here close prices). It can range from -1.0 to 1.0 with values equal to 1.0 meaning a perfect positive correlation and values equal to -1.0 meaning a perfect negative correlation. Perfect correlation will not happen often, if ever. A negative correlation is when price decreases with time, instead of increasing. If you square the correlation coefficient you get the coefficient of determination. It is the percentage of variance explained by the independent variable (here x, which is time) with values between 0.0 and 1.0 (since it is the square of a value between -1.0 and 1.0). So if for instance you get 0.75, it means that time explains 75% of the variance in price. The coefficient of determination is the "goodness of fit", how well the line fits the prices. So it is calculated by the user of the indicator by doing GetCorrelationCoefficient() ** 2.

This is provided for informational purposes only.

Fred

Hi Fred,

Thank you so much, this has been a huge help. It's looking great, and I will learn a lot from it. Much appreciated!

I made a basic Algo to create what I see in TradingView. I tried to align the input period as best I could but may have made a mistake. There appears to be a slight discrepancy between the TradingView output and this one, any idea why that might be? I've attached the backtest and some images. On TV, the LinReg appears to be at 1.3315 and in QC it appears to be at 1.3308 at the time of the same close price on both (1.3318).

Hi LE,

Could it be because in TV, you use the current close (and not close[1])? Remember that in QC, the current bar so to speak is the last one to have finished wrt your resolution, AND at all other res. Which is just fantastic for backtesting and making sure that your backtesting results have some sort of relevancy with the same algo live trading. But in TV, the current bar is the real-time one, the one that is building or the one the backtesting engine is at, not the previous one which is now completed.

The indicator also wants to have QC's current bar (the last one finished) inside its update before you call GetProjection(). It uses the number of element in it to see time pass by… to see which x you are looking at….

Other discrepancies could be based on the data itself being used. TV is raw data (to the best of my knowledge). QC by default is adjusted. Of course, different closes will give you different lines and projections.

Fred

Yeah, I was thinking it was something like that but haven't found it yet.

Is there a way to manually reset the indicator or make sure it stays on track? Trying it on some intraday CFDs and the arrival time of the projections can shift if there's a break in trading. For example, I want it to arrive every 30 min on :00 and :30. After a day however, it drifts to :15 and :45 due to a 15 min trading break at the end of the day. 

Otherwise it looks great, will continue playing around with it. 

Hi LE,

TV's lin reg will also be subject to gaps, of course.

You cannot call _reset() manually because that method assumes that P is full. I will add a method Reset() that will check if P is full or not and act accordingly… will repost.

Fred

Hi again LE,

Here is the indicator again with a modified version of the method reset. It is now called Reset() and can be called manually.

# This code is provided for informational purposes only.
# Do NOT trade using it or you WILL loose money.

from collections import deque
from statistics import stdev

class SimpleLinearRegressionChannel(PythonIndicator):
    def __init__(self, algorithm: QCAlgorithm, base_period: int, projection_period: int, channel_width: float):
        super().__init__()
        assert base_period > 0, f"{self.__init__.__qualname__}: base_period must be greater than 0."
        assert projection_period > 0, f"{self.__init__.__qualname__}: projection_period must be greater than 0."
        assert channel_width >= 0.0, f"{self.__init__.__qualname__}: channel_width must be greater than or equal to 0.0."
        if base_period < 10:
            algorithm.Log(f"Warning - {self.__init__.__qualname__}: base_period is less than 10. This is very few data points to compute a simple linear regression.")
        self._algorithm = algorithm
        self._base_period = base_period
        self._x = list(range(1, base_period + 1))
        self._x_sum = sum(self._x)
        self._x_mean = self._x_sum / base_period
        self._diffs_x_mean = [(x_i - self._x_mean) for x_i in self._x]
        self._B1_den = sum(pow(x_i, 2) for x_i in self._diffs_x_mean)
        self._projection_period = projection_period
        self._channel_width = channel_width
        self._stdev = None
        self.Value = None

        self._base_window = deque(maxlen=base_period)
        self._B0 = None
        self._B1 = None
        self._projection_window = deque(maxlen=projection_period)

        self._R_den_x = (base_period * sum(pow(x_i, 2) for x_i in self._x)) - pow(self._x_sum, 2)

        self.WarmUpPeriod = base_period

    @property
    def IsReady(self):
        return (len(self._base_window) == self._base_window.maxlen) and (len(self._projection_window) >= 1)

    def Update(self, _input):
        if len(self._base_window) != self._base_window.maxlen:
            self._base_window.append(_input.Close)
        else:
            projection_size = len(self._projection_window)
            if projection_size == 0:
                self._simple_linreg()
            if projection_size != self._projection_window.maxlen:
                self._projection_window.append(_input.Close)
            else:
                self.Reset(_input)

    def _simple_linreg(self):
        y_mean = sum(self._base_window) / self._base_period

        B1_num = sum((x_j * y_j) for x_j, y_j in zip(self._diffs_x_mean, [(y_i - y_mean) for y_i in self._base_window]))
        self._B1 = B1_num / self._B1_den
        self._B0 = y_mean - (self._B1 * self._x_mean)

        self._stdev = stdev(self._base_window)

    def Reset(self, _input = None):
        self._base_window.clear()
        if self._base_period > self._projection_period:
            self._base_window.extend(self._projection_window)
            if _input is not None:
                self._base_window.append(_input.Close)
            self._projection_window.clear()
        else:
            while (len(self._projection_window) > 0) and (len(self._base_window) != self._base_window.maxlen):
                self._base_window.append(self._projection_window.popleft())
            if len(self._base_window) == self._base_window.maxlen:
                self._simple_linreg()
                if _input is not None:
                    self._projection_window.append(_input.Close)
            else:
                if _input is not None:
                    self._base_window.append(_input.Close)
    
    def GetSlope(self):
        if self.IsReady:
            return self._B1
        else:
            return None

    def GetProjection(self):
        if self.IsReady:
            x = self._base_period + len(self._projection_window)
            y = self._B0 + (self._B1 * x)
            channel = self._channel_width * self._stdev
            return (y - channel, y, y + channel)
        else:
            return (None, None, None)

    def GetCorrelationCoefficient(self):
        if self.IsReady:
            num = (self._base_period * sum((x_i * y_i) for x_i, y_i in zip(self._x, self._base_window))) - (self._x_sum * sum(self._base_window))
            den = math.sqrt(self._R_den_x * ((self._base_period * sum(pow(y_i, 2) for y_i in self._base_window)) - pow(sum(self._base_window), 2)))
            return num / den
        else:
            return None

When you manually call Reset() however, be aware of the current bar so you make sure not to skip it. If you are in OnData and there is a bar in “data” and you manually call Reset() without that bar, then the indie will reset, your OnData code will eventually terminate and it will be called again for the next bar. Thus, you will skip a bar. In a nutshell, if there is a current bar, call Reset and pass it to it. If the current bar is empty or None, such as when a bar is skipped because of a pause or data is missing, you can call Reset() without a param (None by default).

Fred

Hi Fred,

Thank you for the update. Unfortunately I'm not able to get it working, I call Reset(data) at 18:30 and expect the next bar to arrive at 19:00 - however it arrives at 18:45. Not sure what I'm doing wrong! I've attached the backtest, where you can see it happen at 18:30 on Dec 9. 

All the trouble starts when there is a break in data at 16:15-16:30. The day before runs on :00s and :30s, but after that it shifts to :15s and :45s. I could avoid taking in data at 16:00 since I wouldn't use that time period anyway, but the Reset function is great for more control. 

Sure. That's not a case that has been considered so far.

Let's consider the cycles:

• The indicator warms up between 23h30 and 23h59 on Dec 8th.
• Thus at 00:00 on Dec 9th, once the first minute bar is added, it is ready to project.
• It projects between 00:00 and 00:29.
• Then, it resets automatically with that projection period, and projects again between 00:30 to 00:59.
• It goes on.
• Between 16h00 and 16h15, it projects, as usual (appending minute bars into the projection window).
• There is no data between 16h15 and 16h30.
• At 16h30 (16h31 bar), it resets, manually. So the projection window gets copies into the base window (after that one was cleared). But then, the indicator is not ready anymore as it only has 15 bars in the base window. It needs 16 more before it is ready. The 16th one is the current one, the first projection. In that case, the indicator is ready at the bar you receive at 16h46.

So what you need is some sort of HARD reset. One that drops data. I had no use case for that until now. So then, the following will arrive:

• At the 16h31 bar, your HARD reset. That bar (16h30 to 16h31) will still be passed to the hard reset method because you want to keep that one. What you want to drop is the data between 16h00 and 16h15.
• The indicator will not be ready anymore. It will need 30 more bars to be. At 17h01, it will be ready to project for that bar…

Thus, your 30 minute intervals won't be :15 - :45 anymore, they will stay :00 - :30. But you will basically skip the projection of 16h30-17h00. And of course, you will only project between 16h00 and 16h15 (not 16h15 - 16h30, there's no data there to compare to).

Is that what you are trying to do?

Fred

Yes, the hard reset sounds like what I'm trying to do. Sorry about the confusion, I mistakenly thought that's what the current Reset() was doing. The hard reset would solve the issue I think. One could do it at the end or beginning of the trading day to make sure the time periods remain consistent. 

I'm also looking for a way to Log the first and last Close price that the B period is using, so I can make sure I'm aligning it with TradingView. Would this be accurate: 

self.Log(f"first: {self._base_window[-1]} latest:{self._base_window[0]}")

Is there any way to Log from the SimpleLinearRegressionChannel.py? I would like to put it above the self._base_window.clear() in Reset(), to easily log it at the new cycle, but of course the Class doesn't have the Log function. 

😊

Here is the new class, LE, with the HardReset() method added. Call it passing the current bar if that bar is relevant, like in the above explanation.

# This code is provided for informational purposes only.
# Do NOT trade using it or you WILL loose money.

from collections import deque
from statistics import stdev

class SimpleLinearRegressionChannel(PythonIndicator):
    def __init__(self, algorithm: QCAlgorithm, base_period: int, projection_period: int, channel_width: float):
        super().__init__()
        assert base_period > 0, f"{self.__init__.__qualname__}: base_period must be greater than 0."
        assert projection_period > 0, f"{self.__init__.__qualname__}: projection_period must be greater than 0."
        assert channel_width >= 0.0, f"{self.__init__.__qualname__}: channel_width must be greater than or equal to 0.0."
        if base_period < 10:
            algorithm.Log(f"Warning - {self.__init__.__qualname__}: base_period is less than 10. This is very few data points to compute a simple linear regression.")
        self._algorithm = algorithm
        self._base_period = base_period
        self._x = list(range(1, base_period + 1))
        self._x_sum = sum(self._x)
        self._x_mean = self._x_sum / base_period
        self._diffs_x_mean = [(x_i - self._x_mean) for x_i in self._x]
        self._B1_den = sum(pow(x_i, 2) for x_i in self._diffs_x_mean)
        self._projection_period = projection_period
        self._channel_width = channel_width
        self._stdev = None
        self.Value = None

        self._base_window = deque(maxlen=base_period)
        self._B0 = None
        self._B1 = None
        self._projection_window = deque(maxlen=projection_period)

        self._R_den_x = (base_period * sum(pow(x_i, 2) for x_i in self._x)) - pow(self._x_sum, 2)

        self.WarmUpPeriod = base_period

    @property
    def IsReady(self):
        return (len(self._base_window) == self._base_window.maxlen) and (len(self._projection_window) >= 1)

    def Update(self, _input):
        if len(self._base_window) != self._base_window.maxlen:
            self._base_window.append(_input.Close)
        else:
            projection_size = len(self._projection_window)
            if projection_size == 0:
                self._simple_linreg()
            if projection_size != self._projection_window.maxlen:
                self._projection_window.append(_input.Close)
            else:
                self.Reset(_input)

    def _simple_linreg(self):
        y_mean = sum(self._base_window) / self._base_period

        B1_num = sum((x_j * y_j) for x_j, y_j in zip(self._diffs_x_mean, [(y_i - y_mean) for y_i in self._base_window]))
        self._B1 = B1_num / self._B1_den
        self._B0 = y_mean - (self._B1 * self._x_mean)

        self._stdev = stdev(self._base_window)

    def Reset(self, _input = None):
        self._base_window.clear()
        if self._base_period > self._projection_period:
            self._base_window.extend(self._projection_window)
            if _input is not None:
                self._base_window.append(_input.Close)
            self._projection_window.clear()
        else:
            while (len(self._projection_window) > 0) and (len(self._base_window) != self._base_window.maxlen):
                self._base_window.append(self._projection_window.popleft())
            if len(self._base_window) == self._base_window.maxlen:
                self._simple_linreg()
                if _input is not None:
                    self._projection_window.append(_input.Close)
            else:
                if _input is not None:
                    self._base_window.append(_input.Close)

    def HardReset(self, _input = None):
        self._base_window.clear()
        if _input is not None:
            self._base_window.append(_input.Close)
        self._projection_window.clear()

    def GetSlope(self):
        if self.IsReady:
            return self._B1
        else:
            return None

    def GetProjection(self):
        if self.IsReady:
            x = self._base_period + len(self._projection_window)
            y = self._B0 + (self._B1 * x)
            channel = self._channel_width * self._stdev
            return (y - channel, y, y + channel)
        else:
            return (None, None, None)

    def GetCorrelationCoefficient(self):
        if self.IsReady:
            num = (self._base_period * sum((x_i * y_i) for x_i, y_i in zip(self._x, self._base_window))) - (self._x_sum * sum(self._base_window))
            den = math.sqrt(self._R_den_x * ((self._base_period * sum(pow(y_i, 2) for y_i in self._base_window)) - pow(sum(self._base_window), 2)))
            return num / den
        else:
            return None

If you want to add some logging before the call to clear() in HardReset(), you can do it using “self._algorithm”. That's one of the reasons why the algo is passed to the indicator. So, it would look like this:

    def HardReset(self, _input = None):
        self._algorithm.Log(f"first: {self._base_window[0]} latest: {self._base_window[-1]}")
        self._base_window.clear()
        if _input is not None:
            self._base_window.append(_input.Close)
        self._projection_window.clear()

self._base_window, as implemented in that code, stores the close prices in chronological order. So [0] is the older one and [-1] the most recent.

Cheers!

Fred

HardReset is working perfectly now! I reset it at 16:59 (no data input) and it's aligned for the next day (starting at 18:00 for CFDs, indicator ready at 18:30). Thanks for the help with the Log, I'm learning from this.

Using the log, I found that the correct cadence to match the bars on TradingView is -1 at the cycle start, and -2 at the end. For example, you must set the TV Regression Trend tool at 2:59 and 3:28, to achieve the same output as the algo 3:00-3:30. 

The Linear Regression then looks perfectly accurate, but the Upper and Lower Channels look a little bit wider? In the images below you can see the LinReg matches in relation to the price (see breaks above / below) but the channels are a little bit wider. I'm not sure it's accepting my 2.0 std dev input? Or maybe it's TradingView that's off?

Hi LE,

Very happy you like it! Again, beware. I don't want you to loose too much of your money……

-1 at the cycle start looks like TV uses close but QC uses TV's close[1]. Which would put TV at -1 yes.

-2 at the end. I'd say you have another -1 there, logically (same reason than above). And the additional -1 might be because I start my lin reg's ‘x’ at 1, not 0. Because then, if 0, the first point counts for nothing. Dunno how TV's built-in indicator is coded as the code is not available for that one. So I may be wrong here wrt the reason.

As for the width of the channels. It seems like you're looking at a TV channel that does not reset after P vs one that does (in QC). I see 3 peeks in TV in the same continuous channel and then the 3 peeks in QC are across two periods… no? Each time P goes into B automatically, all three lines and the channel absolute width change…. I'm pretty sure the TV indicator does not do that, you set it once and you need to change the settings if you want another period…

I had fun coding that!

Fred