spx

package
v2.5.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 30, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package spx computes S&P 500 breadth measurements locally from a validated constituent universe and daily closes obtained through the daemon's broker connector.

The compute is a sliding window over a stream: for each S&P-500 name keep the last 50 daily closes, count names where the most recent close is ≥ the window mean, divide by member count, multiply by 100. The engine owns refresh concurrency and its in-memory view. In normal daemon operation, snapshots, rolling windows, history, and refreshed membership are persisted as typed daemon.db state and observations. Embedded membership is the cold fallback; JSON file paths remain only for explicit legacy import and isolated codec tests.

Index

Constants

View Source 
const (
	MinMembers = 450
	MaxMembers = 520
)

MinMembers / MaxMembers bound a "looks like the S&P-500" sanity check. The index actually carries ~500–505 names with dual-class entries (BRK.B, GOOG/GOOGL); the bounds are intentionally wider than reality to absorb transient mid-rebalance edits on the source page without rejecting the parse. A result outside the band is treated as "Wikipedia HTML structure broke" and falls back to the previously known list.

View Source 
const CurrentHistorySetVersion = 2

CurrentHistorySetVersion is the history schema version written by the engine. Other versions are not projected into current state.

View Source 
const CurrentWindowSetVersion = 2

CurrentWindowSetVersion is the constituent-window schema version written by the engine. Other versions are not projected into current state.

View Source 
const HTTPTimeout = 15 * time.Second

HTTPTimeout bounds the Wikipedia fetch. 15 s comfortably covers a healthy round-trip; longer would mostly hide a degraded link rather than help the caller. A failed fetch falls back to whatever's already loaded — neither caller blocks user-visible work waiting for retry.

View Source 
const MaxHistoryPoints = 60

MaxHistoryPoints caps how many days of S5FI history the engine retains. The dashboard CLI shows ~30 by default; the engine keeps twice that so a daemon that's been down for a month still ships a useful sparkline on its first call after restart. After cap, oldest points roll off.

View Source 
const MembersFilename = "sp500-members.json"

MembersFilename is the canonical filename for the runtime-refreshed members cache. Lives alongside the rest of Canary's per-feature cache subdirs under $XDG_CACHE_HOME/ibkr/spx-members/.

View Source 
const MethodConstituentFanout = methodConstituentFanout

MethodConstituentFanout is the exported form of the current breadth methodology token for daemon wire envelopes and documentation.

View Source 
const MinCoverageFraction = 0.80

MinCoverageFraction is the minimum fraction of MemberCount that a refresh must cover before the engine will persist its result. Refreshes below this threshold are treated as "did not converge" — typical causes: a connector-not-ready race at cold-start (where every fetch returns "no gateway connector"), an outage mid-fan-out, or a pacing-induced abort. Persisting a below-threshold snapshot would mislead any consumer that reads the cached value, and would poison the scheduler's "today's snapshot exists, skip the next bootstrap" check; refusing to persist forces a retry on the next tick instead. The 0.80 threshold tolerates ordinary per-name fetch errors (e.g. a few delisted-but-not-yet-removed tickers) while rejecting catastrophic fan-out failures.

View Source 
const RollingMaxBars = 252

RollingMaxBars is the lookback for the per-constituent rolling max/min of close used for the new-52-week-highs/lows count. 252 trading bars approximates one calendar year of US sessions (252 = 252 weekday sessions per year, leaving the 9-10 US market holidays out — close enough to "52 weeks" for the reading the renderer wants). A name "makes a new 52-week high today" when today's close strictly exceeds the max of the previous 251 closes.

View Source 
const WikipediaURL = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies"

WikipediaURL is the canonical source the project scrapes for the S&P-500 constituent list. Single constant so the release-time script and the daemon's runtime refresher land on the same page; changing one without the other would silently desync.

View Source 
const WindowSize = 50

WindowSize is the 50-day SMA lookback (S&P DJI's S5FI is the 50-day variant). The window holds the 50 most recent daily closes chronologically; the most recent close is window[len-1]. SMA = mean(window). A name is "above 50DMA" when window[len-1] >= mean(window). Today's close participates in its own SMA — this matches the convention used by $SPXA50R / StockCharts and S&P DJI's published S5FI methodology.

View Source 
const WindowSize200 = 200

WindowSize200 is the 200-day SMA lookback ($SPXA200R). Catches cyclical tops cleanly (1999, 2021) — slower-moving than the 50-day reading but a meaningful complement. Computed in the same pass over each constituent's daily bars, so the cold-start cost is unchanged (IBKR's pacing limit is per-request, not per-bar; pulling 200 days instead of 50 doesn't cost more requests).

Variables

This section is empty.

Functions

func CompletedSessionKey 

func CompletedSessionKey(now time.Time) string

CompletedSessionKey returns the latest US-equity session whose close plus the breadth settlement pad has passed at now. During weekends, holidays, and pre-close trading hours this stays on the previous completed session, which is the only daily-bar set the breadth cache can publish without racing partial data.

func DefaultDir 

func DefaultDir() (string, error)

DefaultDir returns the on-disk cache root the daemon uses by default: $XDG_CACHE_HOME/ibkr/breadth-spx/, falling back to $HOME/.cache/ibkr/breadth-spx/ when XDG_CACHE_HOME is unset (the XDG spec's documented default).

Returns an error only if neither XDG_CACHE_HOME nor HOME is set, which on a real OS user account doesn't happen. Tests that need a deterministic path should construct NewStore directly with t.TempDir() rather than relying on this function.

func FetchAndParse 

func FetchAndParse(ctx context.Context, url, version string) ([]string, time.Time, error)

FetchAndParse pulls the constituent list from url (typically WikipediaURL) and parses it. Returns the symbols plus the wall-clock time the fetch completed (UTC) — the latter is what callers stamp into their on-disk envelope as `as_of`.

Network errors, non-200 responses, and parse failures all surface as errors; sanity-bound enforcement (MinMembers ≤ N ≤ MaxMembers) is the CALLER's job because release-time and runtime want different behaviour on bounds-fail. The version argument is folded into the User-Agent so Wikipedia ops can correlate scrapes to releases.

func ImportLegacyHistory 

func ImportLegacyHistory(ctx context.Context, authority *corestore.Store, payload, metadata []byte, observedAt time.Time) error

ImportLegacyHistory preserves legacy history JSON as a non-authorizing observation without publishing it as current state.

func ImportLegacyMembersObservation 

func ImportLegacyMembersObservation(ctx context.Context, authority *corestore.Store, payload, metadata []byte, observedAt time.Time) error

ImportLegacyMembersObservation preserves exact legacy bytes as non-authorizing evidence. The caller supplies metadata containing the decision_eligible=false cutover marker.

func ImportLegacySnapshot 

func ImportLegacySnapshot(ctx context.Context, authority *corestore.Store, payload, metadata []byte, observedAt time.Time) error

ImportLegacySnapshot/Windows/History preserve exact legacy JSON bytes as non-authorizing observations. They intentionally do not publish current state: clean-slate cutover starts every live cache cold, and only a current-code fetch may create a state document.

func ImportLegacyWindows 

func ImportLegacyWindows(ctx context.Context, authority *corestore.Store, payload, metadata []byte, observedAt time.Time) error

ImportLegacyWindows preserves legacy window JSON as a non-authorizing observation without publishing it as current state.

func LoadExternal 

func LoadExternal(path string) (members []string, asOf time.Time, ok bool)

LoadExternal reads the attached daemon.db projection, or the legacy file when no authority has been attached, and returns (members, asOf, true) when it passes every gate, or (nil, zero, false) otherwise. Gates (any failure → cached file is treated as absent):

  • File missing or unreadable.
  • Corrupt JSON.
  • Version mismatch (future schema bump triggers cold rebuild).
  • Sanity bounds: MinMembers ≤ count ≤ MaxMembers. A 600-name list or a 200-name list means the parser tripped and we'd rather keep computing against the embedded baseline than publish nonsense.

The function returns no error — every gate-fail collapses to "use embedded". The daemon's refresh path logs WHY a file was rejected before falling through; this helper is intentionally silent so it stays usable from non-daemon contexts (CLI, tests).

func MemberList 

func MemberList() (members []string, asOf time.Time)

MemberList returns the embedded S&P-500 membership baked into the binary at release time (members_data.go, regenerated by `make refresh-spx-members` on every release). asOf is the release-time timestamp.

MemberList is the embedded-fallback accessor — it does not consult the attached daemon.db membership projection or the legacy import file. Callers that want runtime-current membership should:

  • try LoadExternal first, fall back to MemberList on miss (this is what the daemon's deferred members resolver does, see server.go's resolveBreadthMembers), or
  • read the live engine via engine.Members() (which the runtime refresher updates via engine.SetMembers as reconstitutions land).

The split keeps MemberList usable from cold contexts (CLI, startup, tests) without forcing them to know about daemon persistence.

NOTE: new constituents from reconstitution are admitted to breadth with full weight from day 1 of inclusion. A "pending until 50d accrue" exclusion was scoped out as disproportionate complexity vs. the <1% noise on 1-3 names per quarter. Revisit if a user complains breadth reads materially off.

func MembersDefaultPath 

func MembersDefaultPath() (string, error)

MembersDefaultPath resolves the canonical legacy members-cache path. During normal daemon runtime the path is only the stable binding key for daemon.db; it is not read or written after UseCoreMembersStore succeeds.

func MembersFileExists 

func MembersFileExists(path string) bool

MembersFileExists reports whether path exists. Used by status rendering to decide between the "cache:DATE" and "embedded:DATE" source token without re-loading the file.

func ParseHTML 

func ParseHTML(html []byte) ([]string, error)

ParseHTML extracts the S&P-500 ticker list from a Wikipedia "List of S&P 500 companies" page body. The result is uppercase, deduplicated, and sorted ascending — the form members_data.go's generator and the daemon's runtime refresher both expect.

Returns an error when the constituents table can't be located or when no candidate rows survive (Wikipedia page restructure). The MinMembers / MaxMembers sanity bound is NOT enforced here — that's the caller's job because the two callers (release script, runtime refresher) handle a bounds-fail differently (script log.Fatals; daemon warns and falls back).

func PublicationDeadline 

func PublicationDeadline(sessionKey string) (time.Time, bool)

PublicationDeadline returns the bounded deadline for publishing one session's breadth snapshot. The start follows the official session close (including known early closes) plus the normal settlement delay; the end is sized for one normally paced full-universe HMDS pass.

func PublicationPending 

func PublicationPending(lastGoodSessionKey string, refreshActive bool, now time.Time) bool

PublicationPending reports whether lastGoodSessionKey is the immediately prior session and an active refresh is still inside the current session's bounded publication window. Callers may keep that prior last-good as typed not-due context only while this returns true; once the deadline passes (or the engine is no longer active), the older session is overdue.

func SaveExternal 

func SaveExternal(path string, members []string, asOf time.Time) error

SaveExternal writes members + asOf atomically to daemon.db after attachment. Its file branch remains for the cutover codec and isolated legacy tests.

func UseCoreMembersStore 

func UseCoreMembersStore(path string, store *corestore.Store) error

UseCoreMembersStore binds the exact runtime cache path to daemon.db. The existing document is fully validated before the binding is published. Once bound, LoadExternal, SaveExternal, and MembersFileExists never touch the legacy JSON path. A missing document is the intended clean-slate state and lets callers use the embedded release-time fallback.

func UserAgent 

func UserAgent(version string) string

UserAgent identifies our scraper to Wikipedia ops. Their bot policy wants a descriptive UA with a contact path; anonymous python-requests UAs get rate-limited or blocked outright. The shared constant lets the release script and the daemon present a unified identity (only the version segment varies — the caller passes it).

Format: "canary/<version> (https://github.com/osauer/canary; +breadth indicator)"

func ValidateLegacyMembersObservation 

func ValidateLegacyMembersObservation(payload []byte) (time.Time, int, error)

ValidateLegacyMembersObservation validates a legacy members payload for observation-only cutover without publishing it as current state.

Types

type Bar 

type Bar struct {
	Date  string // "YYYY-MM-DD"
	Close float64
}

Bar is the engine's view of one daily price bar — just the date and close, since 50-DMA breadth needs nothing else. Decoupling from ibkr.HistoricalBar (which carries open/high/low/volume the engine doesn't use) keeps the spx package free of any gateway dependency, so tests can fake the fetcher without importing the connector.

type BarFetcher 

type BarFetcher interface {
	FetchDaily(ctx context.Context, symbol string, lookbackDays int) ([]Bar, error)
}

BarFetcher is the daily-bar source the engine pulls from. The production implementation in internal/daemon/breadth_fetcher.go wraps *ibkr.Connector.FetchHistoricalDailyBars; unit tests use FakeBarFetcher below.

Contract:

  • FetchDaily returns bars in chronological order, oldest first.
  • lookbackDays is a soft hint: the fetcher may return more or fewer than the requested count (holidays, half-days, listing date). Callers slice the result themselves.
  • Errors per-symbol are non-fatal to the engine: a refresh that loses some names still returns a partial result rather than failing the whole call.
  • Cancellation honours the supplied context. A long-running fetch must bail when ctx.Done() fires.

type ConstituentWindow 

type ConstituentWindow struct {
	Symbol    string    `json:"symbol"`
	Closes    []float64 `json:"closes"`
	LastBarAt string    `json:"last_bar_at"`
	// HighWindow is the trailing 252-bar rolling max of close
	// (~1 year), updated each refresh from the bars merged in. Kept
	// separately from Closes so the persisted footprint stays bounded:
	// we don't need 252 floats per name; the rolling max compressed
	// into a single value + the count of bars contributing is enough
	// to detect today's-close-above-prior-252-day-max. RollingMax is
	// the max of the previous N closes (excluding today); the daemon
	// compares today's close vs RollingMax to decide whether today is
	// a new high.
	HighRollingMax     float64 `json:"high_rolling_max,omitempty"`
	HighRollingBarsHad int     `json:"high_rolling_bars_had,omitempty"`
	LowRollingMin      float64 `json:"low_rolling_min,omitempty"`
	LowRollingBarsHad  int     `json:"low_rolling_bars_had,omitempty"`
}

ConstituentWindow holds the sliding window of daily closes for one S&P-500 name. Closes is chronological (oldest first); when the window is full len(Closes) == WindowSize200. LastBarAt is the date string of the most recent close in YYYY-MM-DD form — used to decide whether the next refresh needs to fetch new bars for this name.

The window holds the trailing 200 closes per constituent, enough to cover the 200-day SMA. The 50-day reading slices the last 50 closes; the 200-day reading uses the full window; the rolling-max/min for new-highs/lows uses a separate field tracked outside the close window because the lookback (252 bars) exceeds what we keep in Closes.

func SlideWindow 

func SlideWindow(w ConstituentWindow, close float64, barDate string) ConstituentWindow

SlideWindow folds today's close into a constituent window. It does three things in one pass:

  1. Append today's close to the chronological Closes slice and trim to the v2 cap of WindowSize200 entries.
  2. Update the rolling max/min over the previous RollingMaxBars closes (excluding today's), so the next Compute can detect "today made a new 252-bar high".
  3. Track HighRollingBarsHad / LowRollingBarsHad so a name with fewer than RollingMaxBars of history doesn't get counted as making a new high on its 30th day of trading.

Same-day idempotency: if barDate matches w.LastBarAt, the existing tail close is overwritten and counters are not double-bumped. The rolling-max state for that name doesn't change on a same-day re-fetch — late prints settling shouldn't kick a new-high.

The input is not mutated; callers assign the result back if they want persistence.

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine is the breadth-spx state machine: it loads persisted state, drives a background refresh against a BarFetcher when asked, and serves the most recent Snapshot to callers. Safe for concurrent use.

Lifecycle:

  • New() loads persisted state. If the cache is fresh, Get() returns it immediately and no fetch is needed.
  • Refresh(ctx) is the long-running operation. Serialised against concurrent calls (the second caller waits behind the first).
  • Get() / Status() are fast read-only views; safe to call during a Refresh in progress.

State is held in memory and successful window progress is persisted in bounded batches during a refresh, then once more when the pass completes. A crash mid-refresh therefore resumes from the last committed daemon.db checkpoint without publishing an incomplete snapshot.

func New 

func New(store *Store, fetcher BarFetcher, opts Options) *Engine

New constructs an Engine. Loads any persisted state from store (best effort — a corrupted or missing cache results in a cold start, not an error). Members come from Options: an explicit Members list, a deferred MembersFn resolver, or the checked-in embedded list in members_data.go as the fallback. Runtime updates arrive only via SetMembers (the daemon's members refresher).

func (*Engine) Get 

func (e *Engine) Get() (*Snapshot, bool)

Get returns the most recent successful snapshot, or (nil, false) if the engine hasn't computed one yet (cold start). Fast: holds only a read lock; safe during an in-flight Refresh.

The returned snapshot is a defensive copy — the Excluded slice is cloned so a caller iterating its result cannot race against an in-flight refresh that's appending exclusions to the engine's canonical state.

func (*Engine) History 

func (e *Engine) History(limit int) []HistoryPoint

History returns up to `limit` trailing history points, oldest first. A non-positive limit returns the full retained series (bounded by MaxHistoryPoints). The returned slice is a defensive copy — callers can mutate freely without affecting engine state.

func (*Engine) IsBusy 

func (e *Engine) IsBusy() bool

IsBusy reports whether the engine has refresh work in progress or a scheduled below-threshold retry that should keep the owning daemon alive.

A breadth cold-start often converges over multiple refresh attempts as IBKR's contract-details bucket refills. The refresh itself may finish quickly, then Run sleeps belowThresholdRetryDelay before continuing. From the daemon's lifecycle point of view that sleep is still active bootstrap work, even though IsRefreshing is false.

func (*Engine) IsRefreshing 

func (e *Engine) IsRefreshing() bool

IsRefreshing reports whether a Refresh is currently in flight. The daemon's handleBreadthSPX uses this to decide between returning a cached snapshot and surfacing status="computing".

func (*Engine) LastRefreshCoverage 

func (e *Engine) LastRefreshCoverage() (coverage, memberCount int)

LastRefreshCoverage returns (coverage, memberCount) from the most recent finalise. Zero values indicate "no refresh has completed yet" — distinct from "refresh completed with zero coverage" because the scheduler treats the latter as a signal to retry, not give up.

The scheduler reads this to decide whether the previous refresh converged (coverage ≥ MinCoverageFraction × memberCount) or whether to schedule a retry sooner than the daily cadence.

func (*Engine) MarkPendingBootstrap 

func (e *Engine) MarkPendingBootstrap()

MarkPendingBootstrap pre-sets refreshing=true if Run() would fire a bootstrap refresh on entry — i.e. iff shouldRefreshOnStartup is true against the current snapshot and clock. The caller MUST spawn Run() immediately after; otherwise the flag stays true forever.

The point is to close the race in postConnectSetup where the daemon reports Connected=true (handshake done) before `go e.Run()` has scheduled and called Refresh — a status RPC in that window would otherwise see Connected=true but no breadth-spx background task, even though one is about to fire. Refresh() itself sets refreshing to true again under e.mu (idempotent) and clears it via defer, so the canonical lifetime tracking inside Refresh stays unchanged.

No-op when no bootstrap would fire (snapshot is fresh) — that's also the correctness guard against a stuck flag.

func (*Engine) Members 

func (e *Engine) Members() []string

Members returns the constituent list the engine is currently using. Defensive copy: callers cannot mutate engine state by editing the returned slice. Used by the daemon's status renderer to surface member count, and by the runtime refresher to compare its newly fetched list against the in-process snapshot before swapping.

func (*Engine) Progress 

func (e *Engine) Progress() (RefreshProgress, bool)

Progress returns the current or most recently completed refresh attempt. The bool is false before the engine has started its first pass.

func (*Engine) Refresh 

func (e *Engine) Refresh(ctx context.Context) error

Refresh runs one pass of the constituent-fanout compute: decide which names need new bars, fetch them in parallel, slide each window forward, recompute S5FI, persist.

Cold start is ~74 min wall-clock: IBKR's historical-data pacing limit caps each gateway connection at 60 requests per 10-minute sliding window, so 503 constituents land at ~6 names/min sustained after the initial 60-name burst. Adding workers above the default 6 doesn't help — the gateway throttles the second any pacing budget is exceeded. Warm refresh (same daemon, populated cache) is ~1–10 min: only today's bar per name needs fetching.

Concurrent calls serialise via refreshMu — the second caller waits behind the first and then sees the updated snapshot. A returned error means the compute didn't complete; partial fetch failures (some names succeeded, some didn't) do NOT return an error — they surface as Excluded entries in the resulting Snapshot.

func (*Engine) Run 

func (e *Engine) Run(ctx context.Context)

Run starts the engine's scheduler. Returns when ctx is cancelled. Designed to be called once from the daemon's startup sequence inside a goroutine; multiple concurrent Run loops on the same Engine would compete on the refresh mutex without crashing, but the caller shouldn't do that.

Lifecycle:

  • On entry, check shouldRefreshOnStartup. If true, call Refresh immediately so handleBreadthSPX has data ASAP.
  • After each refresh, check coverage. If it converged (≥ threshold), sleep until nextRefreshAt (daily cadence). If it didn't and we're under the retry limit, sleep belowThresholdRetryDelay and retry — letting accumulated windows + IBKR's refilled reqContractDetails bucket push coverage higher on the next pass.
  • On ctx.Done at any point: return cleanly. An in-flight Refresh is cancelled via its context.

Errors from Refresh are logged but do not stop the loop — the engine retries on the next tick. A transient gateway disconnect shouldn't take the daily cadence offline.

func (*Engine) SetMembers 

func (e *Engine) SetMembers(members []string) bool

SetMembers swaps the constituent list. Returns true when the new list differs from the existing one (caller may want to invalidate downstream state); returns false when the lists are identical (no-op, no need to touch the cache or kick a recompute).

On change, the in-memory windows map is NOT cleared: names dropped from the list become irrelevant to Compute (which iterates over members), and names added are picked up by the next Refresh which sees them missing from cached and triggers a cold fetch. The existing windows for surviving names stay warm — a reconstitution of 1-3 names per quarter shouldn't invalidate ~500 cached windows.

New constituents will be excluded from the next Compute pass with Reason="thin_history" until their cold-fetch lands and their window populates. Per design decision (b) — "pending until 50d accrue" — full inclusion in the breadth reading is deferred until the new name has 50 trading days of post-inclusion history. Today the engine doesn't track per-symbol inclusion dates, so the approximation we ship is: the name appears in the exclusion list as "thin_history" until its window naturally exceeds WindowSize. A follow-up can add per-symbol inclusion dates and the strict "exclude for 50d regardless of bar count" semantics.

func (*Engine) UseCoreStore 

func (e *Engine) UseCoreStore(store *corestore.Store) error

UseCoreStore attaches daemon.db and loads the engine projection from it. Production constructs the engine with Options.DeferStoreLoad so no legacy cache is read before the daemon acquires its persistence lock.

type ExcludedMember 

type ExcludedMember struct {
	Symbol string `json:"symbol"`
	Reason string `json:"reason"`
}

ExcludedMember explains why a constituent did not contribute to the compute. The codebase logs this so the verification scrape can attribute small divergences to known causes (new listing, missing data feed, etc.) rather than algorithm bugs.

type FakeBarFetcher 

type FakeBarFetcher struct {

	// Bars is the canned response per symbol. The fetcher trims to
	// the requested lookback length so test inputs can be a single
	// long series shared across cases.
	Bars map[string][]Bar
	// Errors injects per-symbol failures (e.g. simulate a gateway
	// throttle). When a symbol is in both Bars and Errors, Errors
	// wins.
	Errors map[string]error
	// Latency makes FetchDaily sleep before returning. Used to test
	// the worker-pool concurrency and context cancellation paths.
	Latency time.Duration
	// Calls records every (symbol, lookbackDays) pair the engine
	// has invoked, so tests can assert what the refresh planner
	// asked for without instrumenting the engine.
	Calls []FakeCall
	// contains filtered or unexported fields
}

FakeBarFetcher is a test-only BarFetcher. Routes calls to a canned map of bars per symbol; missing symbols return an error so tests can assert engine behaviour when individual fetches fail. Safe for concurrent use — refresh tests fan calls out across workers.

func (*FakeBarFetcher) CallCount 

func (f *FakeBarFetcher) CallCount() int

CallCount returns the number of recorded fetch attempts. Convenience for test assertions on planner behaviour ("expected exactly N fetches for a cold start, got M").

func (*FakeBarFetcher) FetchDaily 

func (f *FakeBarFetcher) FetchDaily(ctx context.Context, symbol string, lookbackDays int) ([]Bar, error)

FetchDaily satisfies the BarFetcher interface. ctx-aware: returns early if ctx is cancelled during the simulated Latency.

type FakeCall 

type FakeCall struct {
	Symbol       string
	LookbackDays int
}

FakeCall is one (symbol, lookbackDays) the engine asked for.

type FetchFunc 

type FetchFunc func(ctx context.Context) ([]string, time.Time, error)

FetchFunc abstracts the Wikipedia round-trip so tests can inject a canned response without standing up an httptest server. Production passes a closure around FetchAndParse with the daemon's version stamp.

type HistoryPoint 

type HistoryPoint struct {
	Date           string  `json:"date"`
	PctAbove50DMA  float64 `json:"pct_above_50dma"`
	PctAbove200DMA float64 `json:"pct_above_200dma,omitempty"`
	NewHighs       int     `json:"new_highs,omitempty"`
	NewLows        int     `json:"new_lows,omitempty"`
}

HistoryPoint is one session's breadth reading in rolling history. The renderer pulls the trailing N for the dashboard sparkline. Date is the NY-tz session key (YYYY-MM-DD); the four numbers carry the 50-DMA reading, the 200-DMA reading, and the constituent counts for new 52-week highs and lows.

type HistorySet 

type HistorySet struct {
	Version int            `json:"version"`
	Points  []HistoryPoint `json:"points"`
}

HistorySet is the versioned rolling-history persistence shape. Points are stored chronologically, oldest first, and capped at MaxHistoryPoints.

type Logger 

type Logger interface {
	Warnf(format string, args ...any)
	Infof(format string, args ...any)
}

Logger is the minimal logging surface the engine needs. The daemon passes its standard logger; tests can pass nil to silence output.

type Options 

type Options struct {
	// Logger receives non-fatal refresh events (per-symbol fetch
	// errors, persistence failures). nil silences all logging — fine
	// for tests, not recommended for production.
	Logger Logger
	// Clock injects a synthetic time source for tests. Production
	// callers pass nil and get time.Now.
	Clock func() time.Time
	// Workers caps refresh concurrency. Each worker calls
	// BarFetcher.FetchDaily for one symbol at a time. Defaults to 6,
	// matching the IBKR-side historical-data pacing headroom. Setting
	// to 1 serialises fetches — useful in tests that want
	// deterministic ordering.
	Workers int
	// ColdLookbackDays is how many trailing daily bars to fetch for
	// a name with no cached history. Defaults to WindowSize + 10 to
	// absorb holiday gaps in the trailing 50 trading days.
	ColdLookbackDays int
	// WarmLookbackDays is how many trailing daily bars to fetch for
	// a name whose cached window is current except for today.
	// Defaults to 2 — today's bar plus one for duplicate-detection
	// during the same-session retry path.
	WarmLookbackDays int
	// Members lets the caller seed the engine with a non-embedded
	// constituent list. nil/empty falls back to MembersFn when set,
	// else to MemberList()'s embedded list, preserving every existing
	// caller.
	Members []string
	// MembersFn defers constituent-list resolution to first actual
	// use. When Members is empty and MembersFn is non-nil, the engine
	// calls it exactly once — behind a sync.Once, from the first
	// operation that touches the list (Refresh, Members, SetMembers) —
	// instead of resolving at construction. The daemon uses this to
	// keep the persisted-members read and its INFO log line out of
	// daemon.New, which runs before Server.Start acquires the
	// single-instance lock: autospawn race losers build a full Server
	// but never serve a call, so a deferred load keeps them off the
	// persistence authority and out of the shared log. An empty return falls back
	// to MemberList()'s embedded list. Ignored when Members is set.
	MembersFn func() []string
	// DeferStoreLoad constructs the engine cold without reading Store. The
	// daemon uses this before it owns the persistence lock, then calls
	// Engine.UseCoreStore to attach and load daemon.db before serving. Legacy
	// and isolated callers keep the historical eager-load default.
	DeferStoreLoad bool
}

Options configures Engine construction. All fields are optional — the zero value picks sensible defaults documented per-field.

type RefreshFailure 

type RefreshFailure string

RefreshFailure is a redacted machine-readable reason for the latest breadth refresh problem. Raw broker and transport errors remain local logs. 

const (
	RefreshFailureFetch     RefreshFailure = "fetch_failed"
	RefreshFailurePersist   RefreshFailure = "persist_failed"
	RefreshFailureCancelled RefreshFailure = "cancelled"
)

Breadth refresh failure values keep raw broker and storage text local.

type RefreshProgress 

type RefreshProgress struct {
	SessionKey  string
	StartedAt   time.Time
	Processed   int
	Total       int
	Deadline    time.Time
	LastFailure RefreshFailure
}

RefreshProgress is the current or most recently completed fan-out attempt. Processed includes both successful and failed symbol fetches; Total is the plan size at StartedAt. Deadline is the calendar-based publication SLA for SessionKey, not an ETA.

type RefreshState 

type RefreshState string

RefreshState reflects the current health of the members-list refresher. Surfaced on the wire by the daemon's status handler so `canary status` can flag silent parser rot or a long-disabled auto-refresh. 

const (
	// RefreshHealthy is the steady-state: the most recent fetch
	// landed parseable HTML inside the sanity bounds.
	RefreshHealthy RefreshState = "healthy"
	// RefreshNetworkFailed means the most recent fetch failed at the
	// transport layer (DNS, connect, timeout). Wikipedia
	// unreachable, captive portal, etc.
	RefreshNetworkFailed RefreshState = "network_failed"
	// RefreshParseFailed means we fetched but couldn't extract a
	// usable list — the HTML didn't contain the constituents table
	// or the parse landed outside the sanity bounds. Surfaces a
	// Wikipedia-side restructure or a regex regression.
	RefreshParseFailed RefreshState = "parse_failed"
	// RefreshDisabledConfig means the daemon's config.toml has
	// `[spx] members_auto_refresh = false`.
	RefreshDisabledConfig RefreshState = "disabled (config)"
	// RefreshDisabledEnv means the CANARY_SPX_MEMBERS_AUTO_REFRESH env
	// var force-disabled refresh (=0), regardless of TOML.
	RefreshDisabledEnv RefreshState = "disabled (env)"
)

func (RefreshState) IsDisabled 

func (s RefreshState) IsDisabled() bool

IsDisabled reports whether the refresher is intentionally off (via config or env). Used by status to render "disabled" distinctly from the failure states.

func (RefreshState) IsHealthy 

func (s RefreshState) IsHealthy() bool

IsHealthy reports whether the state is the steady-state. Wraps the constant comparison so external callers don't depend on string equality.

type Refresher 

type Refresher struct {
	// contains filtered or unexported fields
}

Refresher manages the daemon's runtime membership refresh: three triggers (daily 02:30 ET ticker, startup catch-up, opportunistic post-rollover) all converge on one singleflighted fetch goroutine. On a successful fetch the new list is written atomically to disk and pushed into the engine; failures fall back to whatever's already loaded — breadth never goes silent because the network is down.

Construction is via NewRefresher; the daemon stands one of these up per Server lifetime and runs Run() in a goroutine. Tests can drive it via TriggerNow() and inspect via State().

func NewRefresher 

func NewRefresher(opts RefresherOptions) *Refresher

NewRefresher constructs a refresher. Engine and Fetch are required; everything else is optional with sensible defaults.

func (*Refresher) Run 

func (r *Refresher) Run(ctx context.Context)

Run starts the daemon-internal refresh loop: a daily 02:30 ET ticker plus a startup catch-up if the loaded file's session date is earlier than today. Returns when ctx is cancelled. A no-op when the refresher is disabled — Run() returns immediately so the caller's goroutine exits cleanly.

The opportunistic post-rollover trigger is exposed via TriggerIfRolledOver(); the daemon's breadth handler calls it on the first request of a new NY session. Three triggers, one singleflighted fetcher — concurrent triggers join the in-flight job rather than racing it.

func (*Refresher) State 

func (r *Refresher) State() RefreshState

State returns the refresher's current health. Read by the daemon's status renderer; cheap to call (short mutex).

func (*Refresher) TriggerIfRolledOver 

func (r *Refresher) TriggerIfRolledOver(ctx context.Context)

TriggerIfRolledOver is the opportunistic third trigger: the daemon's breadth handler calls it on the first request after the NY-date rolls over. Belt-and-suspenders against the 02:30 ET ticker missing (network outage, daemon paused). No-op when the loaded file is already from today, when disabled, or when a fetch is already in flight (singleflighted by triggerAsync).

func (*Refresher) TriggerNow 

func (r *Refresher) TriggerNow(ctx context.Context)

TriggerNow forces an immediate fetch attempt for tests. Skips the disabled check so the caller can verify pinned-mode behaviour directly; in production the disabled check in Run() and TriggerIfRolledOver gates entry.

type RefresherOptions 

type RefresherOptions struct {
	// Engine is the breadth engine whose members list this refresher
	// updates. Required.
	Engine *Engine
	// CachePath is where the refresher writes / reads the cached
	// members JSON. Typically MembersDefaultPath().
	CachePath string
	// Fetch is the Wikipedia round-trip. Required.
	Fetch FetchFunc
	// Logger receives non-fatal events. nil silences output.
	Logger Logger
	// Clock injects a synthetic time source for tests. nil → time.Now.
	Clock func() time.Time
	// PinnedByConfig is true when config.toml has
	// [spx] members_auto_refresh = false AND the env var did not
	// force-enable. The refresher renders the state as
	// "disabled (config)" and Run() returns immediately.
	PinnedByConfig bool
	// PinnedByEnv is true when CANARY_SPX_MEMBERS_AUTO_REFRESH=0 is
	// set. Takes precedence over PinnedByConfig in the status
	// surface. Run() returns immediately. Env=1 force-enables and
	// leaves both Pinned* flags false even when TOML says false —
	// see internal/daemon/server.go installMembersRefresher for the
	// resolution rules.
	PinnedByEnv bool
}

RefresherOptions configures NewRefresher. The Pinned* fields are resolved by the caller (config layer) before construction so the refresher doesn't have to know about TOML / env semantics.

type Snapshot 

type Snapshot struct {
	// Value is the 50-DMA reading: percentage of constituents trading
	// above their own 50-day SMA, in [0, 100].
	Value float64 `json:"value"`
	// PctAbove50DMA is the 50-day reading exposed under the canonical
	// long-form name (renderer-friendly). Equal to Value; kept
	// alongside Value so the wire shape is self-documenting.
	PctAbove50DMA float64 `json:"pct_above_50dma"`
	// PctAbove200DMA is the 200-day reading. Below 40% = red /
	// 40–60% = yellow / above 60% = green per the locked plan
	// (calibrated to the post-Mag-7 era; StockCharts' 70/30 default
	// would have read red far too often through 2024-2025).
	PctAbove200DMA float64 `json:"pct_above_200dma"`
	// NewHighsToday is the count of S&P 500 constituents whose latest
	// close strictly exceeded their rolling 252-bar max (~1 year of
	// trading sessions ≈ "52-week high"). Coverage-aware: a name with
	// < RollingMaxBars history is skipped, not counted as either.
	NewHighsToday int `json:"new_highs_today"`
	// NewLowsToday is the symmetric count for new 252-bar lows.
	NewLowsToday int `json:"new_lows_today"`
	// NetNewHighsPct is (NewHighs - NewLows) / coverage × 100. A
	// positive number means more names making new highs than new
	// lows; a deeply-negative one is the textbook divergence pattern
	// (SPX near highs but few constituents leading it).
	NetNewHighsPct float64 `json:"net_new_highs_pct"`
	// AsOf is the wall-clock instant the compute finished. Distinct
	// from SessionKey: a snapshot may be refreshed multiple times
	// against the same trading session as late prints settle.
	AsOf time.Time `json:"as_of"`
	// SessionKey is the New-York date of the trading session the
	// snapshot represents (YYYY-MM-DD). Resilient to UTC vs local
	// timezone confusion when the daemon runs outside the US.
	SessionKey string `json:"session_key"`
	// Method is a stable token identifying the compute methodology.
	// Renderers and tests pin this string to detect silent algorithm
	// changes.
	Method string `json:"method"`
	// MemberCount is the size of the membership list used in the
	// compute. Should track the S&P-500 cardinality (~500–505 with
	// the dual-class names).
	MemberCount int `json:"member_count"`
	// Coverage is the count of members that had enough 50-DMA history
	// (≥ WindowSize closes) to contribute to the headline. The
	// denominator in Value is Coverage, not MemberCount, so a recent
	// listing with thin history doesn't push the percentage downward.
	Coverage int `json:"coverage"`
	// Coverage200 is the analogous denominator for PctAbove200DMA —
	// names with ≥ WindowSize200 closes. Smaller than Coverage when
	// some constituents have between 50 and 200 days of history (post-
	// IPO names, recent index additions).
	Coverage200 int `json:"coverage_200"`
	// CoverageHighsLows is the denominator for the new-highs/lows
	// count — names with ≥ RollingMaxBars closes. Smaller than
	// Coverage200 when some constituents have between 200 and 252
	// days of history. Used as the denominator in NetNewHighsPct.
	CoverageHighsLows int `json:"coverage_highs_lows"`
	// Excluded lists members dropped from the compute and the reason
	// — useful when verifying against $SPXA50R divergence. Empty in
	// the steady state.
	Excluded []ExcludedMember `json:"excluded,omitempty"`
}

Snapshot is one breadth reading: the computed values, represented trading session, and provenance carried by the breadth.spx RPC envelope and runtime persistence record.

func Compute 

func Compute(members []string, windows map[string]ConstituentWindow, sessionKey string, asOf time.Time) Snapshot

Compute reduces a set of constituent windows to a single snapshot carrying the 50-DMA reading, the 200-DMA reading, and the new-highs/lows counts. Pure: no I/O, no clock dependency beyond the wall-clock stamp the caller supplies. Deterministic — same inputs in, same outputs out — which is what makes verification against public breadth indices meaningful.

members is the authoritative S&P-500 list for this session. Only names appearing in members count; windows for delisted names are silently ignored. Names in members but missing from windows are excluded with reason "no_window". A cached window contributes only when its LastBarAt matches sessionKey; this prevents yesterday's complete windows from being relabelled and published as today's snapshot after a failed warm refresh. Names with thin history are counted toward whichever readings their history supports (a name with 75 bars contributes to the 50-DMA reading and is excluded from the 200-DMA and new-highs/lows counts).

sessionKey is the New-York trading-day date string the snapshot represents (YYYY-MM-DD). The caller derives it; Compute does not inspect clocks. asOf is the wall-clock stamp that goes into Snapshot.AsOf — typically time.Now() at the call site, but injectable so tests can pin it.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store persists the engine's current snapshot, constituent windows, and rolling history. UseCoreStore binds normal daemon operation to typed daemon.db state and observations. The directory supplied to NewStore remains for explicit legacy import and isolated codec use.

func NewStore 

func NewStore(dir string) *Store

NewStore returns a Store rooted at dir. The directory is created on first write (lazy mkdir keeps tests that pass an unwritable dir from failing at construction time). dir must be an absolute path or relative to the daemon's working directory; the store does not resolve XDG paths itself — that's the caller's job (see DefaultDir).

func (*Store) LoadHistory 

func (s *Store) LoadHistory() ([]HistoryPoint, error)

LoadHistory returns the persisted rolling-history series or (nil, nil) when no file exists yet. Like the other loaders, an unknown schema version triggers a cold rebuild rather than an error so a future format bump doesn't poison startup.

func (*Store) LoadSnapshot 

func (s *Store) LoadSnapshot() (*Snapshot, error)

LoadSnapshot returns the persisted snapshot or (nil, nil) when no current state exists. A methodology-token mismatch is also treated as no state so an incompatible payload cannot publish zero-valued measurements as current.

func (*Store) LoadWindows 

func (s *Store) LoadWindows() (map[string]ConstituentWindow, error)

LoadWindows returns the persisted constituent windows, or (nil, nil) when no file exists or the on-disk schema version doesn't match. The version-mismatch case is intentionally non-fatal: a future format bump triggers a cold-rebuild rather than a daemon error.

func (*Store) SaveHistory 

func (s *Store) SaveHistory(points []HistoryPoint) error

SaveHistory persists the rolling history. Pass an empty slice to wipe.

func (*Store) SaveSnapshot 

func (s *Store) SaveSnapshot(snap Snapshot) error

SaveSnapshot writes snap atomically. Existing file is replaced.

func (*Store) SaveWindows 

func (s *Store) SaveWindows(windows map[string]ConstituentWindow, asOf time.Time) error

SaveWindows persists the full window map. Pass nil to wipe.

func (*Store) UseCoreStore 

func (s *Store) UseCoreStore(store *corestore.Store) error

UseCoreStore makes daemon.db the sole runtime persistence path. NewStore's directory remains available only to the explicit cutover importer and isolated legacy codec tests.

type WindowSet 

type WindowSet struct {
	Version int                          `json:"version"`
	AsOf    time.Time                    `json:"as_of"`
	Windows map[string]ConstituentWindow `json:"windows"`
}

WindowSet is the versioned persistence shape for constituent windows. An incompatible Version is treated as no state and triggers a cold rebuild.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.