43 problem cannot watch strikes#44
Conversation
…ions Introduce slow/fast watch tiers for options market observation: Slow watch: batch ingestor polls full expiration chain via REST every ~30s. After each poll it calls watchlist.SetStrikes so the realtime ingestor always has fresh contract lists without making HTTP calls inside WebSocket subscription methods. Fast watch: realtime ingestor opens a WebSocket order book subscription per specific contract (WatchInstrument / UnwatchInstrument). Used for pre-trade liquidity assessment and active position monitoring. Key fixes: - floatToString: replace string(rune(f)) with strconv.FormatFloat to prevent colliding store keys for fractional strikes - SubscribeExpirationUpdates no longer makes HTTP calls; contracts are resolved entirely from the watchlist - Stop() cancels context first, waits for goroutines (sync.WaitGroup), then runs connector cleanup to prevent concurrent subscribe/unsubscribe - Pending expirations: if strikes are absent at Start() the expiration is retried when WatchEventStrikesUpdated fires - WebSocketConnector interface extended with SubscribeOrderBook / UnsubscribeOrderBook; mock regenerated - Exported WatchEvent* constants replace bare string literals throughout
There was a problem hiding this comment.
Pull request overview
This PR addresses issue #43 (“cannot watch strikes”) by making options realtime subscriptions contract-aware and by introducing a two-tier watch model (slow expiration-wide updates vs fast per-instrument order book watching) backed by watchlist-provided strike discovery.
Changes:
- Extend the options WebSocket connector interface to subscribe/unsubscribe expiration updates with an explicit contract list, and add per-contract order book subscription methods.
- Add fast-watch instrument tracking + typed watch events to the options watchlist, and wire the realtime ingestor to react to watchlist events (including retrying expirations once strikes arrive).
- Fix options store contract key generation by using a correct float-to-string conversion; batch ingestor now populates watchlist strikes.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/types/connector/options/websocket.go | Updates WS connector API for contract-aware expiration subs + per-contract order books. |
| pkg/markets/options/watchlist.go | Adds fast-watch instrument tracking and emits typed watch events. |
| pkg/markets/options/types/watchlist.go | Documents/defines slow vs fast watch tiers and adds event type constants + new watchlist methods. |
| pkg/markets/options/store/store.go | Fixes float strike-to-string conversion used in contract map keys. |
| pkg/markets/options/ingestor/realtime/ingestor.go | Resolves expiration contracts from watchlist strikes, adds pending retry logic, and handles instrument watch events for order books. |
| pkg/markets/options/ingestor/realtime/ingestor_test.go | Updates mocks/calls for new expiration subscription signature; pre-populates strikes for tests. |
| pkg/markets/options/ingestor/batch/ingestor.go | Collects strikes during REST polling and updates watchlist via SetStrikes. |
| mocks/.../options/WebSocketConnector.go | Regenerated mock to match updated connector interface (incl. new methods). |
| doc.go | Adds top-level package documentation for the module. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| switch ev.Type { | ||
| case optionsTypes.WatchEventInstrumentWatched: | ||
| if ev.Contract == nil { | ||
| continue | ||
| } |
There was a problem hiding this comment.
processWatchlistEvents only reacts to InstrumentWatched/Unwatched and StrikesUpdated, but it ignores ExpirationAdded/ExpirationRemoved events emitted by the watchlist. If a strategy/asset loader requires or releases expirations after the realtime ingestor has started, the connector will never subscribe/unsubscribe for those expirations, so updates won’t flow (and pending-strike retries won’t be set up). Add cases for WatchEventExpirationAdded/Removed to call subscribeExpiration and to unsubscribe/clear any pending state when an expiration is removed.
| // Collect all strikes and store market data for each contract | ||
| strikes := make([]float64, 0, len(expirationData)) | ||
| for strike, optionTypes := range expirationData { | ||
| strikes = append(strikes, strike) | ||
| for optionType, optionData := range optionTypes { |
There was a problem hiding this comment.
strikes is built by ranging over expirationData (a map), so the slice order is nondeterministic. That means SetStrikes may publish a different ordering on each poll even when the strike set is unchanged, which can cause noisy StrikesUpdated events and unstable downstream behavior. Sort (and optionally dedupe) strikes before calling SetStrikes, and consider skipping SetStrikes when the stored strikes are already equal.
| key := instrumentKey{ | ||
| Exchange: exchange, | ||
| Pair: contract.Pair, | ||
| Expiration: contract.Expiration, | ||
| Strike: contract.Strike, | ||
| OptionType: contract.OptionType, | ||
| } |
There was a problem hiding this comment.
instrumentKey uses contract.OptionType as-is. Elsewhere the code normalizes option types to uppercase (e.g., store keys use strings.ToUpper and realtime uses "CALL"/"PUT"), so passing "call" vs "CALL" would create duplicate fast-watch entries and make UnwatchInstrument fail to match. Normalize OptionType (e.g., strings.ToUpper) when building instrumentKey (and ideally when emitting/storing) to make the behavior consistent.
| key := instrumentKey{ | ||
| Exchange: exchange, | ||
| Pair: contract.Pair, | ||
| Expiration: contract.Expiration, | ||
| Strike: contract.Strike, | ||
| OptionType: contract.OptionType, | ||
| } |
There was a problem hiding this comment.
Same OptionType normalization issue as WatchInstrument: UnwatchInstrument builds instrumentKey from contract.OptionType without normalization, so callers using different casing than what was originally watched won’t be able to remove the instrument. Normalize OptionType when constructing the key so watch/unwatch are consistent.
| // Update the watchlist with the discovered strikes so the realtime | ||
| // ingestor can resolve instruments without making REST calls | ||
| o.watchlist.SetStrikes(exchangeName, pair, expiration, strikes) | ||
| } |
There was a problem hiding this comment.
CollectNow now updates the watchlist via SetStrikes, but the batch ingestor tests only assert store writes. Add assertions that SetStrikes results in watchlist.GetAvailableStrikes returning the discovered strikes (ideally in sorted order) so the realtime ingestor contract-resolution path is exercised.
| func (o *optionsRealtimeIngestor) processWatchlistEvents() { | ||
| defer o.wg.Done() | ||
|
|
||
| exchangeName := o.connector.GetConnectorInfo().Name | ||
| events := o.watchlist.Subscribe(exchangeName) | ||
| defer o.watchlist.Unsubscribe(exchangeName) | ||
|
|
||
| for { | ||
| select { | ||
| case <-o.ctx.Done(): | ||
| return | ||
| case ev, ok := <-events: | ||
| if !ok { | ||
| return | ||
| } | ||
| switch ev.Type { | ||
| case optionsTypes.WatchEventInstrumentWatched: | ||
| if ev.Contract == nil { | ||
| continue | ||
| } | ||
| c := toConnectorContract(*ev.Contract) | ||
| if err := o.connector.SubscribeOrderBook(&c); err != nil { | ||
| o.logger.Errorf("failed to subscribe order book for %v: %v", ev.Contract, err) | ||
| } | ||
|
|
||
| case optionsTypes.WatchEventInstrumentUnwatched: | ||
| if ev.Contract == nil { | ||
| continue | ||
| } | ||
| c := toConnectorContract(*ev.Contract) | ||
| if err := o.connector.UnsubscribeOrderBook(&c); err != nil { | ||
| o.logger.Warnf("failed to unsubscribe order book for %v: %v", ev.Contract, err) | ||
| } | ||
|
|
||
| case optionsTypes.WatchEventStrikesUpdated: | ||
| // Retry any expirations that were skipped at Start() due to absent strikes. | ||
| o.pendingMu.Lock() | ||
| ref := expirationRef{pair: ev.Pair, expiration: ev.Expiration} | ||
| _, isPending := o.pendingExpirations[ref] | ||
| o.pendingMu.Unlock() | ||
|
|
||
| if isPending { | ||
| o.subscribeExpiration(exchangeName, ev.Pair, ev.Expiration) | ||
| } | ||
| } |
There was a problem hiding this comment.
New behavior (pending expirations retry on StrikesUpdated + fast-watch order book subscribe/unsubscribe on instrument events) isn’t covered by tests. Add tests that (1) Start() with a watched expiration but no strikes does not call SubscribeExpirationUpdates, then after SetStrikes triggers StrikesUpdated it does; and (2) WatchInstrument/UnwatchInstrument events lead to SubscribeOrderBook/UnsubscribeOrderBook calls.
| SubscribeExpirationUpdates(pair portfolio.Pair, expiration time.Time) error | ||
| UnsubscribeExpirationUpdates(pair portfolio.Pair, expiration time.Time) error | ||
| // Slow watch — subscribe to ticker updates for all contracts in an expiration. | ||
| // contracts is resolved by the realtime ingestor from the watchlist (no REST calls). |
There was a problem hiding this comment.
Grammar: “contracts is resolved” → “contracts are resolved”.
| // contracts is resolved by the realtime ingestor from the watchlist (no REST calls). | |
| // contracts are resolved by the realtime ingestor from the watchlist (no REST calls). |
Resolves: #43