Class AbstractSubscriptionStrategy<T>

java.lang.Object
com.wualabs.qtsurfer.engine.core.AbstractNameable
com.wualabs.qtsurfer.engine.strategy.AbstractStrategy
com.wualabs.qtsurfer.engine.strategy.AbstractSubscriptionStrategy<T>
All Implemented Interfaces:
Nameable, Named, Unsubscribable, Strategy, SubscriptionStrategy<T>
Direct Known Subclasses:
AbstractFundingRateStrategy, AbstractKlineStrategy, AbstractTickerStrategy
public abstract class AbstractSubscriptionStrategy<T> extends AbstractStrategy implements SubscriptionStrategy<T>, Unsubscribable
Base class for all subscribable strategies with included RT indicators group support.

  • Constructor Details

    • AbstractSubscriptionStrategy

      protected AbstractSubscriptionStrategy()

  • Method Details

    • subscribeSource

      public void subscribeSource(@NonNull @NonNull io.reactivex.rxjava3.core.Observable<T> source)
      Subscribes to a market data source under supervision. Unsubscribes any previous subscription first, filters events via SubscriptionStrategy.accept(T), and processes them on the IO scheduler to avoid blocking the source thread. Source-agnostic: the same path runs whether the stream comes from an internal exchange link or an external SPI provider (e.g. NATS).

      Supervision (this used to be a bare subscribe(this::update)):

      • Per-event error isolation — an exception thrown by strategy/indicator code for one event no longer terminates the subscription (which silently blinded the strategy, possibly with open positions). The event is logged and counted (getUpdateErrorCount()); the feed keeps flowing.
      • Source failure visibility — an upstream onError is logged loudly and flips isSourceFailed() so health checks can see a blind strategy. Auto-resubscription is deliberately out of scope (it needs the provider lifecycle).
      • Bounded buffering — the source is consumed as a Flowable with the sourceBackpressure() policy instead of an unbounded observeOn buffer; a slow strategy no longer grows the heap and (for conflatable feeds) no longer acts on stale events.
      Specified by:
      subscribeSource in interface SubscriptionStrategy<T>
      Parameters:
      source - the observable market data stream

    • sourceBackpressure

      protected io.reactivex.rxjava3.core.BackpressureStrategy sourceBackpressure()
      Backpressure policy applied when consuming the market-data source. Default is BackpressureStrategy.LATEST: for conflatable feeds (tickers) a stale event is worthless once a fresher one exists, so a slow strategy skips ahead instead of buffering without bound. Subclasses whose events are discrete facts that must not be dropped (closed klines, funding settlements) override to BackpressureStrategy.BUFFER — safe there because those feeds emit at low frequency.

    • getUpdateErrorCount

      public long getUpdateErrorCount()
      Count of events whose SubscriptionStrategy.update(T) threw and were isolated (subscription survived).

    • isSourceFailed

      public boolean isSourceFailed()
      Whether the market-data source terminated with an error (the strategy receives no data).

    • unsubscribe

      public void unsubscribe()
      Specified by:
      unsubscribe in interface Unsubscribable

    • reset

      public void reset(@NonNull @NonNull Instrument instrument)
      Specified by:
      reset in interface Strategy
      Overrides:
      reset in class AbstractStrategy

    • createInfoStrategySignal

      public InfoStrategySignal createInfoStrategySignal(Instrument instrument)

    • fillSignal

      public void fillSignal(InstrumentMapRTIndicator indicators, InfoStrategySignal signal)

    • updateInstrument

      protected void updateInstrument(@NonNull @NonNull Instrument instrument, long timestamp)
      Tracks an instrument update: registers the instrument, records the timestamp for backtesting, and increments the stats counter.

      Every update(T) override calls this first, before touching indicators or its own fields — which is what makes it the right place to apply @StrategyProperty defaults (idempotent; a no-op after the first event). The compile-time dry run drives a strategy through this same path without ever calling AbstractStrategy.init(ExchangeSupport), so relying on init alone would leave it seeing undefaulted fields.

      Parameters:
      instrument - the instrument that was updated
      timestamp - the event timestamp (used as virtual clock in backtest mode)

    • updateIndicators

      protected InstrumentMapRTIndicator updateIndicators(Instrument instrument, T source)
      Updates the RT indicator group for an instrument from the market data source, triggering recalculation of all indicators in the group.
      Parameters:
      instrument - the instrument to update indicators for
      source - the raw market data to feed into the indicators
      Returns:
      the updated indicator map

    • getUpdatedSignal

      protected InfoStrategySignal getUpdatedSignal(InstrumentMapRTIndicator indicators)
      Creates a new info signal populated with the current values of all public indicators.
      Parameters:
      indicators - the indicator map to read values from
      Returns:
      a signal carrying all public indicator values

    • emitSignal

      public void emitSignal(@NonNull @NonNull StrategySignal signal)
      Emits a signal, replacing the system timestamp with the backtest virtual clock when backtesting is enabled.
      Overrides:
      emitSignal in class AbstractStrategy

    • emitYield

      public void emitYield(@NonNull @NonNull StrategyYield yield)
      Emits a yield, replacing the system timestamp with the backtest virtual clock when backtesting is enabled.
      Overrides:
      emitYield in class AbstractStrategy

    • createBuySignal

      protected BuySignal createBuySignal(Instrument instrument, Number price)

    • createSellSignal

      protected SellSignal createSellSignal(Instrument instrument, Number price)

    • emitBuy

      protected void emitBuy(Instrument instrument, Number price)

    • emitSell

      protected void emitSell(Instrument instrument, Number price)

    • emitInfo

      protected void emitInfo(Instrument instrument)

    • overrideTimestamp

      protected void overrideTimestamp(@NonNull @NonNull AbstractStrategyEvent event)
      Replaces the event's wall-clock timestamp with the last market data timestamp during backtesting, ensuring time-consistent replay.

    • setStateStoreProvider

      public void setStateStoreProvider(@NonNull @NonNull StateStoreProvider stateStoreProvider)
      Injects the backend behind this strategy's per-instrument state. Everything the strategy persists for an instrument — update() state, every window listener's state, and the execution callbacks' state — comes from the store this provider returns, so swapping it is enough to move state onto a persistent backend. The provider is consulted once per instrument and the result memoized; set it before the strategy sees its first tick.
      Parameters:
      stateStoreProvider - supplies the store for an instrument

    • getStateStore

      public Optional<StateStore> getStateStore(@NonNull @NonNull Instrument instrument)
      Specified by:
      getStateStore in interface Strategy

    • getStateStoreForResult

      protected StateStore getStateStoreForResult(ExecutionResult result)

    • getRTIndicator

      protected Optional<RTIndicator> getRTIndicator(Instrument instrument, String key)

    • clearIndicators

      protected void clearIndicators()

    • getNotices

      public List<Notice> getNotices()
      Adds this strategy's per-instrument indicator groups to the strategy-wide notices, which come first. No deduplication is applied here and none is needed: each scope already reports a given condition once — the groups by (code, indicator) and the strategy by code — and the groups partition by instrument, which every indicator notice carries.

      The result is reproducible, not chronological. Each half is in first-raised order, but concatenating them puts every strategy-wide notice before every indicator one whenever it was actually raised — a source failure at the last event still precedes an indicator notice from the first. True cross-scope chronology would need a sequence shared with qtsurfer-engine-indicators, which holds no reference back to the strategy. What callers need is that two calls agree, and that the API's notice cap keeps the same 50 twice.

      Groups are therefore visited in Instrument.symbol() order rather than in instrumentIndicatorGroups' iteration order: that map is a ConcurrentHashMap carrying computeIfAbsent on the dispatch path, which it should stay, so the ordering is imposed here where it costs one sort per read instead of on every event.

      Specified by:
      getNotices in interface Strategy
      Overrides:
      getNotices in class AbstractStrategy
      Returns:
      the notices raised so far, empty if none (the default for strategies without indicators)

    • getInstrumentIndicatorGroup

      protected InstrumentGroupRTIndicator getInstrumentIndicatorGroup(Instrument instrument)

    • setupInstrumentGroupRTIndicator

      protected InstrumentGroupRTIndicator setupInstrumentGroupRTIndicator(Instrument instrument)
      Creates and configures the RT indicator group for an instrument, including backtest mode propagation and subclass-specific indicator setup.
      Parameters:
      instrument - the instrument to build indicators for
      Returns:
      the fully configured indicator group

    • setupIndicators

      protected void setupIndicators(InstrumentGroupRTIndicator indicators)

    • createInstrumentGroupRTIndicator

      protected abstract InstrumentGroupRTIndicator createInstrumentGroupRTIndicator(Instrument instrument)

    • getMarketDataSourceName

      protected final String getMarketDataSourceName()

    • toString

      protected void toString(com.google.common.base.MoreObjects.ToStringHelper ts)
      Overrides:
      toString in class AbstractStrategy

    • getLastEventTimestamp

      public long getLastEventTimestamp()

    • setLastEventTimestamp

      public void setLastEventTimestamp(long lastEventTimestamp)

    • isEnableIndicatorsSignals

      public boolean isEnableIndicatorsSignals()

    • setEnableIndicatorsSignals

      public void setEnableIndicatorsSignals(boolean enableIndicatorsSignals)

    • setIndicatorBankSession

      public void setIndicatorBankSession(com.wualabs.qtsurfer.engine.indicators.cache.IndicatorBank.Session indicatorBankSession)
      This strategy instance's view of a sweep's indicator memoization bank, or null — which is every case except a batch of backtest runs over one dataset.

      Set it before the indicator groups are built, i.e. before the strategy sees its first data point; the groups pick it up as they register their indicators.