Hi @cinar,

Thanks for the extensive response. I really apprechiate it.

Question 0
I got that there are blocking and non blocking methods and I think in my case it looks like some part of the TestStrategy is not consuming anymore, as soon as i introduce any Indicator or use a blocking function like ChanToSlice.

In general what I do to test all this is:

• setting up the TestStrategy
• prepared a Test Repository which populates the snapshots channel initially
• setup and Run() a backtest aka generate a report

( * additionally I simply created a Operate5 just for testing purposes following the same structure as done for the Operate3)

func Operate5[A any, B any, C any, D any, E any, R any](ac <-chan A, bc <-chan B, cc <-chan C, dc <-chan D, ec <-chan E, o func(A, B, C, D, E) R) <-chan R {
	rc := make(chan R)

	go func() {
		defer close(rc)

		for {
			an, ok := <-ac
			if !ok {
				break
			}

			bn, ok := <-bc
			if !ok {
				break
			}

			cn, ok := <-cc
			if !ok {
				break
			}

			dn, ok := <-dc
			if !ok {
				break
			}

			en, ok := <-ec
			if !ok {
				break
			}

			rc <- o(an, bn, cn, dn, en)
		}

		helper.Drain(ac)
		helper.Drain(bc)
		helper.Drain(cc)
		helper.Drain(dc)
		helper.Drain(ec)
	}()

	return rc
}

when i followed the code correctly it just runs through every strategy and computes the actions and outcomes using the populated snapshots from the repo.

The thing is that depending on what I include in the compute function of the TestStrategy, it either retrieves the snapshots, or it doesnt as soon as I introduce Indicators. (complete test strategy posted below)

// TestStrategy defines a basic test strategy for testing purposes.
type TestStrategy struct {
	Donchian *volatility.DonchianChannel[float64]
}

// NewTestStrategy creates a new TestStrategy instance.
func NewTestStrategy(period int) *TestStrategy {
	return &TestStrategy{
		Donchian: volatility.NewDonchianChannelWithPeriod[float64](period),
	}
}

func (s *TestStrategy) Name() string {
	return "TestStrategy"
}

func (s *TestStrategy) Compute(c <-chan *asset.Snapshot) <-chan strategy.Action {
	snapshots := helper.Duplicate(c, 3)

	volumes := asset.SnapshotsAsVolumes(snapshots[0])
	//_ = volumeSlice

	closings := asset.SnapshotsAsClosings(snapshots[1])

	// Donchian Channels indicator
	u, m, l := s.Donchian.Compute(closings)

	u = helper.Skip(u, s.Donchian.IdlePeriod())
	m = helper.Skip(m, s.Donchian.IdlePeriod())
	l = helper.Skip(l, s.Donchian.IdlePeriod())

	i := 0
	actions := Operate5(snapshots[2], volumes, u, m, l, func(snapshot *asset.Snapshot, volume int64, upperBandValue, middleBandValue, lowerBandValue float64) strategy.Action {
		action := strategy.Hold

		if i%42 == 20 {
			action = strategy.Buy
		} else if i%42 == 41 {
			action = strategy.Sell
		}

		log.Println("Action: ", action, " i: ", i, " Date: ", snapshot.Date, " Volume: ", volume, " Upper Band: ", upperBandValue, " Middle Band: ", middleBandValue, " Lower Band: ", lowerBandValue)

		i++

		return action
	})

	// Shift actions until strategy is ready.
	actions = helper.Shift(actions, s.Donchian.IdlePeriod(), strategy.Hold)

	return actions
}

// Report processes the provided asset snapshots and generates a report annotated with the recommended actions.
func (s *TestStrategy) Report(c <-chan *asset.Snapshot) *helper.Report {
	snapshots := helper.Duplicate(c, 3)

	//prepare dates
	dates := asset.SnapshotsAsDates(snapshots[0])
	dates = helper.Skip(dates, s.Donchian.IdlePeriod())

	// prepare closing prices
	closings := asset.SnapshotsAsClosings(snapshots[1])
	closings = helper.Skip(closings, s.Donchian.IdlePeriod())
	closingSplice := helper.Duplicate(closings, 2)

	// compute the actions and outcomes
	actions, outcomes := strategy.ComputeWithOutcome(s, snapshots[2])
	// create annotations
	annotations := helper.Skip(
		strategy.ActionsToAnnotations(actions),
		s.Donchian.IdlePeriod(),
	)
	// shift the outcomes
	outcomes = helper.Skip(helper.MultiplyBy(
		outcomes,
		100,
	), s.Donchian.IdlePeriod())

	// create the report and the charts
	report := helper.NewReport(s.Name(), dates)
	chart1 := report.AddChart()
	chart2 := report.AddChart()

	// Add the columns to the report for chart 0
	report.AddColumn(helper.NewNumericReportColumn("Close", closingSplice[0]))

	// Add the columns to the report for chart 1
	report.AddColumn(helper.NewNumericReportColumn("Close", closingSplice[1]), chart1)

	// Add the columns to the report for chart 0 and 1
	report.AddColumn(helper.NewAnnotationReportColumn(annotations), 0, chart1)
	report.AddColumn(helper.NewNumericReportColumn("Outcome", outcomes), chart2)

	return report
}

Can you see anything strange in here that would lead to a blocking aka not consuming issue?

As soon as I remove the indicator out of the equasion like in the initial example or only passing one snapshot chan & the volumes to a helper.Operate() (or newly created Operate4/5/6/7) , it just works (=snapshots are there within the compute and it runs through the Operate). It is pretty hard to debug since it just does nothing (treats the snapshots as empty) in the incorrect version.

edit: I just reconfirmed, that when including "any" indicator and passing its results to an OperateN() function, it doesnt block, but it also doesnt consume the snapshots populated or duplicated on the way there...

One more example....

WORKING

func (s *TestStrategy) Compute(c <-chan *asset.Snapshot) <-chan strategy.Action {
	snapshots := helper.Duplicate(c, 5)

	volumes := helper.Duplicate(asset.SnapshotsAsVolumes(snapshots[0]), 1)
	//_ = volumeSlice

	// 5 closings needed
	closings := helper.Duplicate(asset.SnapshotsAsClosings(snapshots[1]), 1)
	highPrices := asset.SnapshotsAsHighs(snapshots[2])
	lowPrices := asset.SnapshotsAsLows(snapshots[3])

	// // MACD indicator
	//macdLine, signalLine := s.Macd.Compute(closings[1])

	//macdLine = helper.Skip(macdLine, s.Macd.IdlePeriod())
	//signalLine = helper.Skip(signalLine, s.Macd.IdlePeriod())

	// // VWMA indicator
	// vwma := s.Vwma.Compute(closings[0], TransformChannel(volumes[0], func(i int64) float64 {
	// 	return float64(i)
	// }))
	//vwma = helper.Skip(vwma, s.Vwma.IdlePeriod())

	// // Donchian Channels indicator
	// u, m, l := s.Donchian.Compute(closings)

	// u = helper.Skip(u, s.Donchian.IdlePeriod())
	// m = helper.Skip(m, s.Donchian.IdlePeriod())
	// l = helper.Skip(l, s.Donchian.IdlePeriod())

	i := 0
	actions := Operate5(snapshots[4], volumes[0], highPrices, lowPrices, closings[0], func(snapshot *asset.Snapshot, volume int64, highValue, lowValue float64, closingValue float64) strategy.Action {
		action := strategy.Hold

		if i%42 == 20 {
			action = strategy.Buy
		} else if i%42 == 41 {
			action = strategy.Sell
		}

		log.Println("Action: ", action, " i: ", i, " Date: ", snapshot.Date, " Volume: ", volume, " High: ", highValue, " Low: ", lowValue, " Closing: ", closingValue)

		i++

		return action
	})

	// Shift actions until strategy is ready.
	//actions = helper.Shift(actions, s.Macd.IdlePeriod(), strategy.Hold)

	return actions
}

NOT WORKING

func (s *TestStrategy) Compute(c <-chan *asset.Snapshot) <-chan strategy.Action {
	snapshots := helper.Duplicate(c, 5)

	volumes := helper.Duplicate(asset.SnapshotsAsVolumes(snapshots[0]), 1)
	//_ = volumeSlice

	// 5 closings needed
	closings := helper.Duplicate(asset.SnapshotsAsClosings(snapshots[1]), 2)
	highPrices := asset.SnapshotsAsHighs(snapshots[2])
	lowPrices := asset.SnapshotsAsLows(snapshots[3])

	// // MACD indicator
	macdLine, signalLine := s.Macd.Compute(closings[1])

	//macdLine = helper.Skip(macdLine, s.Macd.IdlePeriod())
	//signalLine = helper.Skip(signalLine, s.Macd.IdlePeriod())

	// // VWMA indicator
	// vwma := s.Vwma.Compute(closings[0], TransformChannel(volumes[0], func(i int64) float64 {
	// 	return float64(i)
	// }))
	//vwma = helper.Skip(vwma, s.Vwma.IdlePeriod())

	// // Donchian Channels indicator
	// u, m, l := s.Donchian.Compute(closings)

	// u = helper.Skip(u, s.Donchian.IdlePeriod())
	// m = helper.Skip(m, s.Donchian.IdlePeriod())
	// l = helper.Skip(l, s.Donchian.IdlePeriod())

	i := 0
	actions := Operate7(snapshots[4], volumes[0], highPrices, lowPrices, closings[0], macdLine, signalLine, func(snapshot *asset.Snapshot, volume int64, highValue, lowValue float64, closingValue, macdValue, signalValue float64) strategy.Action {
		action := strategy.Hold

		if i%42 == 20 {
			action = strategy.Buy
		} else if i%42 == 41 {
			action = strategy.Sell
		}

		log.Println("Action: ", action, " i: ", i, " Date: ", snapshot.Date, " Volume: ", volume, " High: ", highValue, " Low: ", lowValue, " Closing: ", closingValue)

		i++

		return action
	})

	// Shift actions until strategy is ready.
	//actions = helper.Shift(actions, s.Macd.IdlePeriod(), strategy.Hold)

	return actions
}

Another finding:

It also works if i only compute the macd like so:

func (s *TestStrategy) Compute(c <-chan *asset.Snapshot) <-chan strategy.Action {
	snapshots := helper.Duplicate(c, 1)

	closings := asset.SnapshotsAsClosings(snapshots[0])

	// MACD indicator
	macdLine, signalLine := s.Macd.Compute(closings)

	macdLine = helper.Skip(macdLine, s.Macd.IdlePeriod())
	signalLine = helper.Skip(signalLine, s.Macd.IdlePeriod())

	actions := helper.Operate(macdLine, signalLine, func(macdValue, signalValue float64) strategy.Action {
		action := strategy.Hold

		if macdValue > signalValue {
			action = strategy.Buy
		} else if macdValue < signalValue {
			action = strategy.Sell
		}

		return action
	})

	return actions
}

thats great, thank you!

Current state to what I could break it down to is, that as soon as I Duplicate snapshots to retrieve highs, lows and closings

        snapshots := helper.Duplicate(c, 3)

	highPrices := asset.SnapshotsAsHighs(snapshots[2])
	lowPrices := asset.SnapshotsAsLows(snapshots[1])

	closings := asset.SnapshotsAsClosings(snapshots[0])

	// // MACD indicator
	macdLine, signalLine := s.Macd.Compute(closings)

and introduce then also pass them on to the Operate4

actions := Operate4(highPrices, lowPrices, macdLine, signalLine, func(high, low, macdValue, signalValue float64) strategy.Action {
		action := strategy.Hold

		if i%42 == 20 {
			action = strategy.Buy
		} else if i%42 == 41 {
			action = strategy.Sell
		}

		log.Println("Action: ", action, " i: ", i)

		i++

		return action
	})

it doesnt work anymore.

It works without the duplication of the snapshots and also when duplication is set to 1.

I had also hit some issues where things blocked. I love the clean ‘everything is a channel’ design but it does make it quite hard to determine the cause of the blocking.

In my code I had been reporting on 23 strategies for each of about 20 assets and I found that certain strategies would cause the blocking with certain assets. It turned out to be that the assets that caused the problems were ones that didn’t have sufficient data points to satisfy the requirements of one or two strategies.

My rather crude solution was to test each asset/strategy combination for sufficient data before running the strategy report. I’m not proud of the code, but since I introduced it I have had no blocking issues at all. Unfortunately, in order to find out how much data is upstream, it breaks the channel model by converting it to a slice so that the length of data can be determined.

So my function to run a strategy report against an asset looks like this:

func runReport(st strategy.Strategy, assetName string, data <-chan *asset.Snapshot) {
	data, _, datalen := duplicateChan(data)
	// Detect certain strategies that require a minimum amount of data.
	if notEnoughData(st, assetName, datalen) {
		return
	}
	rep := st.Report(data)
	…
}

Where duplicateChan and notEnoughData are:

func duplicateChan(cin <-chan *asset.Snapshot) (<-chan *asset.Snapshot, <-chan *asset.Snapshot, int) {
	slice := helper.ChanToSlice(cin)
	cout0 := helper.SliceToChan(slice)
	cout1 := helper.SliceToChan(slice)
	return cout0, cout1, len(slice)
}

func notEnoughData(st strategy.Strategy, assetName string, datalen int) bool {
	msg := fmt.Sprintf("Ignoring strategy %s for %s due to insufficient data: %s", st.Name(), assetName, st.Name())
	if st.Name()[0:4] == "MACD" {
		if datalen < trend.NewMacdStrategy().Macd.Ema2.Period {
			fmt.Println(msg)
			return true
		}
	}
	if st.Name() == "Awesome Oscillator Strategy" {
		if datalen < momentum.NewAwesomeOscillatorStrategy().AwesomeOscillator.LongSma.Period {
			fmt.Println(msg)
			return true
		}
	}
	return false
}

This was my first time using the indicator package and I was in a hurry to make things work, but after trying lots of things this was the solution to my particular blocking issue. I need to go back to look more closely at it. But it did feel like certain strategies required a minimum number of data points to Compute or Report without blocking. In looking at it now I think that my use of the Period is almost certainly not the right value to use - but it does feel like there is perhaps a minimum count of data below which certain strategies can not be run without blocking.

Yes, unfortunately, this is a common problem now with the channels. I will think if I can generalize the solution here and see if I can at least find a way to incorporate it to help troubleshooting.

Yes, instead of using the Periods, you can use the IdlePeriod() method. It will give you after how many period the indicator will be generating values. I will most likely rename that method very soon.

I have played around a little further with things and I have one other observation.

If I replace helper.Duplicate with this code:

func Duplicate[T any](input <-chan T, count int) []<-chan T {
	outputs := make([]chan T, count)
	result := make([]<-chan T, count)

	inputSlice := ChanToSlice[T](input)

	for i := range outputs {
		outputs[i] = make(chan T, cap(input))
		result[i] = outputs[i]
	}

	for _, output := range outputs {
		go func() {
			defer close(output)

			for _, n := range inputSlice {
				output <- n
			}
		}()
	}

	return result
}

Then my blocking problems go away and I don't need my duplicateChan and my notEnoughData functions above.
The two main differences between this version of helper.Duplicate and the current one are that a) this one does create a slice from the channel, and b) it creates a goroutine to supply each output channel and so doesn't require that none of the output channels block before we have finished distributing all of the input data.

Unfortunately, the ChanToSlice there is a blocking function. It will consume the entire input channel first before moving to the rest. So I agree that it will solve the problem if you have a finite data, but this will break the real-time use case. I do like your idea of having separate go routines for each duplicate. Let me see if I can think of a way to do this.

One other suggestion for you is to use the helper.Buffered() function. Ideally, you can buffer the output channels and it should have a similar impact for you.

Good day everybody..

Does that all mean that the library like it is right now is not really built to combine multiple indicators into larger custom strategies?
I also really like how this lib is making use of the channels, but the side effects of that (figuring out the correct periods.. not being able to debug or test the channel population..) make it currently unusable for my purpose of combining multiple indicators into one custom strategy.

A reference to what i mean would be this here: https://github.com/twopirllc/pandas-ta
There you can set up cutom strategies by simply defining all the contained indicators for example.

Is this similar to how it was planned for version 2?

Thank you for considering these principles! It sounds exactly like what is needed. An addition like this would improve it significantly 👍🏻

Cant wait to test it.

Starting to work on it, let's see how it will go.

#192

I got most of them updated (#193). All strategies are now returning denormalized actions. The reports/charts are still using normalized actions. I updated And, Or, Majority, and Split strategies as well. However, they are currently using the ActionsSource, which does a DenormalizeActions, and it should be replacing Hold with either Buy or Sell currently. I am not fully clear on how to handle that portion yet, so that's the only remaining change right now.

Could you take a quick look and let me know?

Super cool thanks, i will take a look and let you know when i went through it. Will happen in the next few hours.

Using an AndStrategy does a denormalization on the actions and should cluster them.
-> the denormalization indeed creates these clusters of the same actions

The strategies should produce nonormalized actions, when executed seperately. ( No AndStrategy aka. no denormalization)
-> produced actions are also clustered (reminding me a bit of how the denormalized actions also look like).

All in all things seem to work properly on the strategy composition side.

To me it is not 100% clear though, how the actions (produced when the ActionsSource is NOT used) differ from the the ones produced via Denormalize within the ActionSource. Both seem to cluster the actions together or maybe it is just a "data coincidence".
My expected outcome for running a single strategy would have been "non clustered" actions similar to how the actions were produced in V1. But like mentioned that could also be a coincidence.

I definitely would like to do that. I just forgot about it. Adding an issue on this #211 . I will try to put together a solution over the weekend.

great to hear that! Thx

It turned out to be much simpler than I anticipated, but not sure if this will work for your use case. Could you take a look at the #213?

Ok wow, that was quick! Thanks for looking into that

I was moving on to another topic within my project for a while, so in order to tell you if that would work in my use case, I need a few days and get back into it.. Will answer you that question soon.

So I am trying to wrap my head around this..

Could you please provide an example on how one could use this to synchronize multiple indicators?

Maybe the following szenario:

• One CustomStrategy that combines macd, bollingerbands and vwma indicators and uses an Operate function to operate synced channels which all should have the same length if the sync worked correctly.

Actually this is a simplified version of what I am trying to achieve, but I am not sure how to use the Sync function in this szenario.