feat: separate realtime display prices from strategy candles

apps/market-gateway/build.gradle.kts

@@ -16,4 +16,5 @@ dependencies {
     testImplementation("org.testcontainers:junit-jupiter")
     testImplementation("org.testcontainers:testcontainers")
     testImplementation("org.java-websocket:Java-WebSocket:1.6.0")
+    testImplementation("io.lettuce:lettuce-core")
 }

apps/market-gateway/src/main/java/com/idea2strategy/trading/gateway/FinalizedCandlePollingWorker.java

@@ -0,0 +1,68 @@
+package com.idea2strategy.trading.gateway;
+
+import com.idea2strategy.trading.market.candle.FinalizedCandleBoundaryPlanner;
+import com.idea2strategy.trading.market.candle.FinalizedCandleCycle;
+import com.idea2strategy.trading.market.session.OfficialMarketSession;
+import com.idea2strategy.trading.market.session.OfficialMarketSessionEvaluator;
+import com.idea2strategy.trading.market.session.OfficialMarketSessionSource;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+
+/** Polls only official 30m boundaries; Redis event IDs make restart catch-up idempotent. */
+final class FinalizedCandlePollingWorker {
+    private static final Logger log = LoggerFactory.getLogger(FinalizedCandlePollingWorker.class);
+
+    private final FinalizedCandleCycle cycle;
+    private final ApprovedInstruments instruments;
+    private final OfficialMarketSessionSource sessions;
+    private final Clock clock;
+    private final Duration grace;
+    private final FinalizedCandleBoundaryPlanner planner = new FinalizedCandleBoundaryPlanner();
+    private final Set<String> completed = new HashSet<>();
+
+    FinalizedCandlePollingWorker(
+            FinalizedCandleCycle cycle,
+            ApprovedInstruments instruments,
+            OfficialMarketSessionSource sessions,
+            Clock clock,
+            Duration grace) {
+        this.cycle = Objects.requireNonNull(cycle, "cycle");
+        this.instruments = Objects.requireNonNull(instruments, "instruments");
+        this.sessions = Objects.requireNonNull(sessions, "sessions");
+        this.clock = Objects.requireNonNull(clock, "clock");
+        this.grace = Objects.requireNonNull(grace, "grace");
+    }
+
+    @Scheduled(fixedDelayString = "${market-gateway.candle-poll-delay:PT5S}")
+    public void poll() {
+        Instant now = clock.instant();
+        LocalDate tradingDate = now.atZone(OfficialMarketSessionEvaluator.NEW_YORK).toLocalDate();
+        OfficialMarketSession session = sessions.session(tradingDate).orElse(null);
+        if (session == null) {
+            return;
+        }
+        for (Instant boundary : planner.readyBoundaries(session, now, grace)) {
+            String key = tradingDate + ":" + boundary;
+            if (completed.contains(key)) {
+                continue;
+            }
+            try {
+                var result = cycle.run(instruments.bySymbol(), session, boundary);
+                completed.add(key);
+                log.info("finalized 30m boundary {}: evaluated={}, missing={}",
+                        boundary, result.evaluatedInstrumentCount(), result.missingInstrumentCount());
+            } catch (RuntimeException failure) {
+                log.error("finalized 30m boundary {} failed and will be retried", boundary, failure);
+                return;
+            }
+        }
+    }
+}

apps/market-gateway/src/main/java/com/idea2strategy/trading/gateway/MarketGatewayConfiguration.java

@@ -12,7 +12,19 @@
 import com.idea2strategy.trading.market.alpaca.ProviderRightsGate;
 import com.idea2strategy.trading.market.alpaca.ReconnectBackoff;
 import com.idea2strategy.trading.market.redis.RedisMarketEventPublisher;
+import com.idea2strategy.trading.market.candle.AlpacaThirtyMinuteBarsJsonParser;
+import com.idea2strategy.trading.market.candle.FinalizedCandleCycle;
+import com.idea2strategy.trading.market.candle.HttpAlpacaThirtyMinuteBarsClient;
+import com.idea2strategy.trading.market.display.LatestTradeCoalescer;
+import com.idea2strategy.trading.market.display.RedisDisplayPricePublisher;
+import com.idea2strategy.trading.market.display.RedisDisplayTradeSubscriptionSource;
+import com.idea2strategy.trading.market.session.HttpAlpacaOfficialMarketSessionSource;
+import com.idea2strategy.trading.market.session.OfficialMarketSessionSource;
+import com.idea2strategy.trading.market.availability.MarketDataAvailabilityResult;
+import com.idea2strategy.trading.market.availability.MarketDataAvailabilityStatus;
+import com.idea2strategy.trading.messaging.market.MarketEventType;
 import java.net.URI;
+import java.net.http.HttpClient;
 import java.nio.file.Path;
 import java.time.Clock;
 import java.time.Duration;
@@ -22,9 +34,11 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @Configuration(proxyBeanMethods = false)
 @ConditionalOnProperty(prefix = "market-gateway", name = "redis-uri")
+@EnableScheduling
 public class MarketGatewayConfiguration {
     @Bean
     VerifiedGatewayMaterialization verifiedGatewayMaterialization(
@@ -47,6 +61,77 @@ RedisMarketEventPublisher marketEventPublisher(
         return RedisMarketEventPublisher.connect(redisUri, keyPrefix, recentBarCapacity);
     }
 
+    @Bean(destroyMethod = "close")
+    RedisDisplayPricePublisher displayPricePublisher(
+            @Value("${market-gateway.redis-uri}") String redisUri,
+            @Value("${market-gateway.redis-key-prefix}") String keyPrefix) {
+        return RedisDisplayPricePublisher.connect(redisUri, keyPrefix);
+    }
+
+    @Bean(destroyMethod = "close")
+    RedisDisplayTradeSubscriptionSource displayTradeSubscriptionSource(
+            @Value("${market-gateway.redis-uri}") String redisUri,
+            @Value("${market-gateway.redis-key-prefix}") String keyPrefix) {
+        return RedisDisplayTradeSubscriptionSource.connect(redisUri, keyPrefix);
+    }
+
+    @Bean
+    @ConditionalOnProperty(
+            prefix = "market-gateway", name = "strategy-candles-enabled", havingValue = "true", matchIfMissing = true)
+    OfficialMarketSessionSource officialMarketSessionSource(
+            @Value("${market-gateway.alpaca-calendar-endpoint:https://api.alpaca.markets/v2/calendar}")
+                    String endpoint,
+            AlpacaCredentialsProvider credentialsProvider,
+            Clock marketGatewayClock) {
+        return new HttpAlpacaOfficialMarketSessionSource(
+                HttpClient.newHttpClient(), URI.create(endpoint), credentialsProvider, marketGatewayClock);
+    }
+
+    @Bean
+    @ConditionalOnProperty(
+            prefix = "market-gateway", name = "strategy-candles-enabled", havingValue = "true", matchIfMissing = true)
+    FinalizedCandleCycle finalizedCandleCycle(
+            @Value("${market-gateway.alpaca-bars-endpoint:https://data.alpaca.markets/v2/stocks/bars}")
+                    String endpoint,
+            @Value("${market-gateway.candle-fetch-batch-size:200}") int batchSize,
+            AlpacaCredentialsProvider credentialsProvider,
+            RedisMarketEventPublisher publisher,
+            Clock marketGatewayClock) {
+        var client = new HttpAlpacaThirtyMinuteBarsClient(
+                HttpClient.newHttpClient(),
+                URI.create(endpoint),
+                credentialsProvider,
+                new AlpacaThirtyMinuteBarsJsonParser());
+        var ordering = new MarketEventOrderingProcessor();
+        return new FinalizedCandleCycle(client, event -> {
+            publisher.publish(ordering.process(event));
+            if (event.eventType() == MarketEventType.MARKET_EVALUATION_READY) {
+                publisher.publishAvailability(
+                        event.instrumentId(),
+                        event.sequence(),
+                        event.receivedAt(),
+                        new MarketDataAvailabilityResult(
+                                MarketDataAvailabilityStatus.AVAILABLE,
+                                true,
+                                true,
+                                java.util.Set.of(),
+                                java.util.List.of()));
+            }
+        }, marketGatewayClock, batchSize);
+    }
+
+    @Bean
+    @ConditionalOnProperty(
+            prefix = "market-gateway", name = "strategy-candles-enabled", havingValue = "true", matchIfMissing = true)
+    FinalizedCandlePollingWorker finalizedCandlePollingWorker(
+            FinalizedCandleCycle cycle,
+            ApprovedInstruments instruments,
+            OfficialMarketSessionSource sessions,
+            Clock marketGatewayClock,
+            @Value("${market-gateway.candle-finalization-grace:PT2S}") Duration grace) {
+        return new FinalizedCandlePollingWorker(cycle, instruments, sessions, marketGatewayClock, grace);
+    }
+
     @Bean
     ApprovedInstruments approvedInstruments(
             @Value("${market-gateway.instrument-mapping-path}") String mappingPath,
@@ -102,6 +187,9 @@ MarketGatewayRunner marketGatewayRunner(
             AlpacaCredentialsProvider credentialsProvider,
             AlpacaMarketEventNormalizer normalizer,
             RedisMarketEventPublisher publisher,
+            ApprovedInstruments approvedInstruments,
+            RedisDisplayPricePublisher displayPricePublisher,
+            RedisDisplayTradeSubscriptionSource displayTradeSubscriptionSource,
             FileReadinessMarker readinessMarker,
             Clock marketGatewayClock) {
         AlpacaDataFeed feed = AlpacaDataFeed.parse(feedValue);
@@ -117,6 +205,9 @@ MarketGatewayRunner marketGatewayRunner(
                 normalizer,
                 new MarketEventOrderingProcessor(),
                 publisher,
+                new LatestTradeCoalescer(approvedInstruments.bySymbol(), marketGatewayClock),
+                displayPricePublisher,
+                displayTradeSubscriptionSource,
                 readinessMarker,
                 new ReconnectBackoff(reconnectInitialDelay, reconnectMaxDelay),
                 marketGatewayClock);

apps/market-gateway/src/main/java/com/idea2strategy/trading/gateway/MarketGatewayRunner.java

@@ -19,6 +19,9 @@
 import com.idea2strategy.trading.market.availability.MarketDataAvailabilityResult;
 import com.idea2strategy.trading.market.availability.MarketDataAvailabilityStatus;
 import com.idea2strategy.trading.market.availability.MarketDataDegradationReason;
+import com.idea2strategy.trading.market.display.DisplayTradeSubscriptionSource;
+import com.idea2strategy.trading.market.display.LatestTradeCoalescer;
+import com.idea2strategy.trading.market.display.RedisDisplayPricePublisher;
 import com.idea2strategy.trading.market.redis.MarketEventPublishResult;
 import com.idea2strategy.trading.market.redis.RedisMarketEventPublisher;
 import com.idea2strategy.trading.messaging.market.MarketEventEnvelope;
@@ -56,6 +59,9 @@ public final class MarketGatewayRunner implements SmartLifecycle {
     private final AlpacaMarketEventNormalizer normalizer;
     private final MarketEventOrderingProcessor orderingProcessor;
     private final RedisMarketEventPublisher publisher;
+    private final LatestTradeCoalescer tradeCoalescer;
+    private final RedisDisplayPricePublisher displayPricePublisher;
+    private final DisplayTradeSubscriptionSource displaySubscriptions;
     private final FileReadinessMarker readinessMarker;
     private final ReconnectBackoff backoff;
     private final Clock clock;
@@ -68,6 +74,7 @@ public final class MarketGatewayRunner implements SmartLifecycle {
     });
     private final AtomicInteger failedAttempts = new AtomicInteger();
     private final AtomicReference<WebSocket> activeSocket = new AtomicReference<>();
+    private final AtomicReference<AlpacaSipSubscriptionManager> activeSubscription = new AtomicReference<>();
     private final Map<String, LongAdder> unpublishedFrames = new ConcurrentHashMap<>();
     private final Map<UUID, Long> latestSequenceByInstrument = new ConcurrentHashMap<>();
     private final Set<UUID> instrumentsWithSequenceGap = ConcurrentHashMap.newKeySet();
@@ -83,6 +90,9 @@ public MarketGatewayRunner(
             AlpacaMarketEventNormalizer normalizer,
             MarketEventOrderingProcessor orderingProcessor,
             RedisMarketEventPublisher publisher,
+            LatestTradeCoalescer tradeCoalescer,
+            RedisDisplayPricePublisher displayPricePublisher,
+            DisplayTradeSubscriptionSource displaySubscriptions,
             FileReadinessMarker readinessMarker,
             ReconnectBackoff backoff,
             Clock clock) {
@@ -95,6 +105,9 @@ public MarketGatewayRunner(
         this.normalizer = Objects.requireNonNull(normalizer, "normalizer");
         this.orderingProcessor = Objects.requireNonNull(orderingProcessor, "orderingProcessor");
         this.publisher = Objects.requireNonNull(publisher, "publisher");
+        this.tradeCoalescer = Objects.requireNonNull(tradeCoalescer, "tradeCoalescer");
+        this.displayPricePublisher = Objects.requireNonNull(displayPricePublisher, "displayPricePublisher");
+        this.displaySubscriptions = Objects.requireNonNull(displaySubscriptions, "displaySubscriptions");
         this.readinessMarker = Objects.requireNonNull(readinessMarker, "readinessMarker");
         this.backoff = Objects.requireNonNull(backoff, "backoff");
         this.clock = Objects.requireNonNull(clock, "clock");
@@ -106,6 +119,8 @@ public void start() {
         credentialsProvider.load();
         running = true;
         log.info("market-gateway connecting to {} for {} symbols", endpoint, universe.symbols().size());
+        scheduler.scheduleAtFixedRate(this::flushDisplayPrices, 250, 250, TimeUnit.MILLISECONDS);
+        scheduler.scheduleAtFixedRate(this::reconcileDisplaySubscriptions, 0, 1, TimeUnit.SECONDS);
         scheduler.execute(this::connect);
     }
 
@@ -151,6 +166,26 @@ private void scheduleReconnect() {
         scheduler.schedule(this::connect, delay.toMillis(), TimeUnit.MILLISECONDS);
     }
 
+    private void flushDisplayPrices() {
+        try {
+            tradeCoalescer.flush().forEach(displayPricePublisher::publish);
+        } catch (RuntimeException failure) {
+            log.error("display price coalescing failed", failure);
+        }
+    }
+
+    private void reconcileDisplaySubscriptions() {
+        AlpacaSipSubscriptionManager manager = activeSubscription.get();
+        if (manager == null || !manager.isAuthenticated()) {
+            return;
+        }
+        try {
+            manager.replaceTradeSubscriptions(displaySubscriptions.desiredSymbols(clock.instant()));
+        } catch (RuntimeException failure) {
+            log.error("display trade subscription reconciliation failed", failure);
+        }
+    }
+
     private void stopForRightsFailure(ProviderRightsUnavailableException failure) {
         log.error("Alpaca {} rights are no longer verified; the gateway stays down until restarted "
                 + "with current rights evidence", feed.eventValue(), failure);
@@ -174,6 +209,7 @@ public void onOpen(WebSocket webSocket) {
                     credentialsProvider,
                     AlpacaSipWebSocketTransport.connected(webSocket),
                     feed);
+            activeSubscription.set(subscription);
             webSocket.request(1);
         }
 
@@ -217,48 +253,39 @@ private void handle(WebSocket webSocket, String text) {
         private void dispatch(AlpacaSipInboundMessage message) {
             switch (message) {
                 case AlpacaSipInboundMessage.Connected ignored -> subscription.onConnected();
-                case AlpacaSipInboundMessage.Authenticated ignored -> subscription.onAuthenticationApproved();
+                case AlpacaSipInboundMessage.Authenticated ignored -> {
+                    subscription.onAuthenticationApproved();
+                    failedAttempts.set(0);
+                    readinessMarker.markReady();
+                    reconcileDisplaySubscriptions();
+                }
                 case AlpacaSipInboundMessage.SubscriptionConfirmed confirmed -> {
-                    subscription.onSubscriptionApproved(confirmed.barSymbols());
+                    subscription.onSubscriptionApproved(confirmed.tradeSymbols());
                     failedAttempts.set(0);
                     readinessMarker.markReady();
-                    log.info("Alpaca {} subscription active for {} symbols",
-                            feed.eventValue(), confirmed.barSymbols().size());
+                    log.info("Alpaca {} display trade subscription active for {} symbols",
+                            feed.eventValue(), confirmed.tradeSymbols().size());
                 }
                 case AlpacaSipInboundMessage.ProviderError error -> {
                     log.warn("Alpaca {} error {}: {}", feed.eventValue(), error.code(), error.message());
                     readinessMarker.markNotReady();
                     publishUnavailable(MarketDataDegradationReason.PROVIDER_DISCONNECTED);
                 }
-                case AlpacaSipInboundMessage.MinuteBar bar -> publishBar(bar);
+                case AlpacaSipInboundMessage.TradeTick tick -> tradeCoalescer.accept(tick);
                 case AlpacaSipInboundMessage.UnsupportedFrame unsupported ->
                         unpublishedFrames.computeIfAbsent(unsupported.frameType(), key -> new LongAdder())
                                 .increment();
             }
         }
 
-        private void publishBar(AlpacaSipInboundMessage.MinuteBar bar) {
-            MarketEventEnvelope envelope;
-            try {
-                envelope = normalizer.normalize(bar.input());
-            } catch (UnsupportedInstrumentException exception) {
-                unpublishedFrames.computeIfAbsent("unsupported-instrument", key -> new LongAdder()).increment();
-                return;
-            }
-            MarketEventHandlingResult handling = orderingProcessor.process(envelope);
-            MarketEventPublishResult result = publisher.publish(handling);
-            publishAvailability(envelope, handling);
-            log.debug("bar {} {} handling={} publish={}",
-                    envelope.instrumentId(), envelope.occurredAt(), handling.status(), result.status());
-        }
-
         private void handleDisconnect(String reason) {
             readinessMarker.markNotReady();
             AlpacaSipSubscriptionManager manager = subscription;
             if (manager != null) {
                 manager.onDisconnected();
             }
             activeSocket.set(null);
+            activeSubscription.compareAndSet(subscription, null);
             publishUnavailable(MarketDataDegradationReason.PROVIDER_DISCONNECTED);
             if (!unpublishedFrames.isEmpty()) {
                 log.info("Alpaca {} frames received without a publishing path this connection: {}",

apps/market-gateway/src/main/resources/application.yaml

@@ -17,6 +17,9 @@ spring:
 #   instrument-mapping-path: /etc/market-gateway/instruments.json
 #   minimum-instrument-count: 500
 #   recent-bar-capacity: 390
+#   candle-fetch-batch-size: 200
+#   candle-finalization-grace: PT2S
+#   candle-poll-delay: PT5S
 #   alpaca-feed: sip
 #   rights-evidence-path: /etc/market-gateway/alpaca-sip-rights.json
 #   materialization-receipt-path: /etc/market-gateway/materialization.properties