The governed definitions this build depends on. Read them first if a term is unfamiliar.
- PrimarySnapshotA snapshot is a bounded representation of selected market or reference-data state at a stated cutoff.
- PrerequisiteObservation TimeObservation time is the timestamp used to place a measured value on a time axis.
- ImportantAvailability TimeAvailability time is the earliest evidenced instant at which information was usable by a specified system or decision process.
- ImportantEvent TimeEvent time is the timestamp assigned to when a market or business event occurred at its source.
- ImportantLate-Arriving DataLate-arriving data is a record that reaches a processing boundary after the expected time for its event, partition, or decision window.
- ImportantMarket Data FeedA market data feed is a source-defined stream or delivery product carrying market events, reference data, or aggregates.
- ImportantPoint-in-Time DatasetA point-in-time dataset can reconstruct what information was valid and available to a specified user or system at a historical decision time.
- MentionedForward FillForward fill imputes a missing position with the most recent eligible earlier value.
Practical promise. Build a complete multivariate panel from asynchronous streams, preserve when each source became knowable, and measure the updates the synchronized representation leaves behind.
Market streams rarely update together. A liquid instrument may change several times while a slower instrument remains silent. Refresh-time sampling creates an endogenous clock: after the last refresh, wait until every required stream has at least one new usable update, then form one row.
That neat row has a cost. Intermediate fast updates disappear, a sparse stream controls the output rate, and a complete row can still contain an old value. The useful implementation is therefore not just a list of aligned prices. It is an auditable result containing source lineage, ages, controller streams, and loss counts.
This tutorial uses synthetic data. It makes no claim about prediction, profitability, or the statistical suitability of refresh sampling for a particular estimator.
Two clocks, two different questions
Every input record has two timestamps:
- Event time answers, “When does the source say this observation occurred?”
- Availability time answers, “When could this system first have used this record or correction?”
Event time alone is not a causal knowledge boundary. A record stamped 10:00:00
but received at 10:00:03 was unavailable during those three seconds. This
package therefore builds its operational barriers from available_at and
retains event_time for provenance and staleness.
The classical all-refresh construction is recovered when every event is immediately available and there are no revisions. The original multivariate realized-kernel paper documents that construction; the separate availability clock is a conservative engineering convention of this package, not a claim about the paper (primary methodology).
Define the barrier precisely
Inside one declared session partition, let be the prior refresh availability time. For each required instrument , find its first new availability after that barrier:
When every required stream has a first-new update, the next barrier is
The maximum is the earliest moment when the complete panel is knowable. Any stream whose first-new availability equals the maximum is a controller. Ties are retained: there may be more than one controller.
If one stream has no next update, the algorithm stops. It does not invent a partial row or carry an old value forever. Remaining arrivals are counted as an unmatched tail.
The diagram shows why “wait for all” and “select the latest” are separate steps. A may update more than once while the sampler waits for C.
Which value is sampled?
At , apply only revisions whose available_at is no later than the
barrier. For each logical record_id, the highest known contiguous revision is
effective. Then, for each instrument, choose the effective record with the
latest event time.
A correction preserves its record_id, increments revision, and has a later
availability time. It cannot rewrite a row before it became known. A correction
to an old event counts as a new arrival, so it can help meet the wait condition,
but it may not replace the instrument's latest-event source. The diagnostics
make that non-selection visible.
Two different record IDs with the same instrument and event time are an unresolved duplicate. The implementation rejects them rather than guessing a vendor-specific precedence rule.
A five-update example
Suppose A becomes available at seconds 1, 3, and 4; B becomes available at seconds 2 and 4. For this hand calculation, event time equals availability time.
| Row | First new A | First new B | Barrier | Selected values | Unrepresented update |
|---|---|---|---|---|---|
| 1 | 1 | 2 | 2 | A at 1, B at 2 | none |
| 2 | 3 | 4 | 4 | A at 4, B at 4 | A at 3 |
The A update at 3 is real input. It helped make A ready, but it is not present in the row formed at 4. One of five arrivals is absent from the sampled representation, so the loss fraction is .
For each row, the implementation counts arrivals since the prior barrier and subtracts selected source revisions that arrived in that interval. It adds any unmatched tail to the partition loss numerator. This count measures representation loss; it does not estimate information content or statistical bias.
Complete does not mean fresh
For selected source event time , define age
The row is stale when any age is strictly greater than the configured limit
. Equality is accepted. A stale complete row remains in the audit result
with its sources and values; consumers can reject it explicitly. The sampler
does not silently forward-fill, substitute, or repair the record.
In the first worked row, A's event is one second old at the barrier. A 500 ms limit marks that complete row stale. A 1,000 ms limit accepts it.
The sparse-stream failure
Imagine A continues producing ten updates, B produces one, and C stops. After the previous barrier, A and B become ready but C never does. No new synchronized row exists. A and B's remaining updates are an unmatched tail.
If C eventually returns, the barrier can form, but A or B's selected event may be old enough to fail the staleness limit. Refresh-time sampling therefore solves completeness and simultaneity of knowledge, not freshness or data quality.
Open the interactive lab and choose Sparse halt. Step until the wait state appears, then compare it with Slow GAMMA. The stage shows every arriving dot; filled diamonds mark selected sources, hollow dots are unrepresented updates, and the loss counter changes with the state. Back and Reset reproduce the same sequence.
Implementation contract
The Python and TypeScript implementations:
- validate the explicit universe, timestamps, finite values, and revision histories before producing output;
- reset state for each declared
partition; - choose barriers from causal availability times;
- apply only corrections known by the barrier;
- reject ambiguous event-time duplicates across logical records;
- emit both accepted and stale complete rows with full source lineage;
- stop at an incomplete tail and count it;
- report an exact loss numerator and denominator.
The API does not infer sessions, quote side, currency, adjustment convention, or universe changes. Those are upstream contracts. Starting a new session or changing required instruments requires a new partition.
What the tests prove
Shared synthetic fixtures and parity tests cover exact barriers, latest-value selection, tied controllers, delayed availability, the staleness boundary, corrections, sparse tails, partition reset, duplicate rejection, future-dated events, non-finite values, and invalid parameters.
Passing those tests proves that the two implementations follow this package's definition. It does not prove that refresh sampling improves a model, preserves lead-lag information, removes microstructure noise, or is preferable to calendar sampling or nonsynchronizing estimators. Hayashi and Yoshida's primary paper is included specifically because synchronization choices can affect a covariance workflow (methodological alternative).
Practical checklist
- Declare the exact instrument universe and session partition.
- Preserve both event and availability timestamps.
- Require stable record IDs and contiguous correction revisions.
- Resolve true duplicates upstream; do not hide them with sort order.
- Set a staleness limit in the same time unit as the timestamps.
- Retain controller, source, age, discarded, and tail diagnostics.
- Compare loss across instruments and scenarios before consuming results.
- Treat synchronized rows as a representation, not market truth.
Why there is no historical market claim
A trustworthy real case needs a redistributable multi-instrument feed with instrument identity, synchronized event timestamps, receive or availability timestamps, corrections, and session rules. The primary methodology papers support the mathematical context but do not provide that complete reusable operational audit trail. This article therefore labels every teaching record as synthetic and defers a historical case rather than turning an incomplete timeline into a factual example.
Summary and next topic
Refresh-time sampling emits a row only after every required stream has a new usable update. The latest first-new availability controls the barrier; the latest known effective event supplies each value. Those rules are causal only when availability is explicit. The resulting row can still be stale, and fast updates can disappear, so lineage and loss belong in the output.
Continue with Asynchronous Return Alignment to compare an overlap-based method that does not first force every observation onto a common refresh grid.
Primary references
See REFERENCES.md for source roles, dates, and limitations.
Rendered from the canonical Mermaid sources linked by this article.
Availability-time refresh sequence
Purpose: distinguish the first-new readiness updates from the latest effective records selected when the complete barrier becomes knowable.
Takeaway: A1 helps satisfy readiness, but A2 can be the selected A source at the barrier. Intermediate updates must be counted rather than silently hidden.
References3 primary sources and evidence notesExpand the source trail, evidence role, and limitations behind the engineering choices.
Expand the source trail, evidence role, and limitations behind the engineering choices.
Accessed 2026-07-22. Facts from these sources are separated from the package's explicit engineering conventions.
R01 — Multivariate realised kernels
- Organization or authors: Ole E. Barndorff-Nielsen, Peter R. Hansen, Asger Lunde, and Neil Shephard
- Source type: Original methodology paper; Journal of Econometrics 162 (2011), 149–169
- Publication date: 2011; working-paper version dated 25 July 2008
- Version: Author publication record with paper attachment
- URL: https://shephard.scholars.harvard.edu/publications/multivariate-realised-kernels-consistent-positive-semi-definite-estimators
- DOI: https://doi.org/10.1016/j.jeconom.2010.07.009
- Jurisdiction: Statistical methodology; not a market rule
- Supports: All-refresh-time construction for multivariate nonsynchronous observations and its use inside a realized-kernel workflow
- Limitations: Does not define this package's availability-time, correction, staleness, or data-loss API
R02 — On covariance estimation of non-synchronously observed diffusion processes
- Organization or authors: Takaki Hayashi and Nakahiro Yoshida
- Source type: Original peer-reviewed paper; Bernoulli 11(2), 359–379
- Publication date: 2005
- Version: Author-hosted paper
- URL: https://www.ms.u-tokyo.ac.jp/~nakahiro/mypapers_for_personal_use/hayyos03.pdf
- DOI: https://doi.org/10.3150/bj/1116340299
- Jurisdiction: Statistical methodology; not a market rule
- Supports: Synchronization and interpolation choices can be consequential for nonsynchronous covariance estimation; motivates explicit limitations and alternatives
- Limitations: Proposes a nonsynchronizing estimator; it does not prescribe this package's refresh implementation
R03 — RFC 3339
- Organization or authors: Internet Engineering Task Force; Graham Klyne and Chris Newman
- Source type: Standards-track timestamp specification
- Publication date: July 2002
- Version: RFC 3339
- URL: https://www.rfc-editor.org/rfc/rfc3339
- Jurisdiction: Global Internet standard
- Supports: Timestamp syntax and UTC offset representation
- Limitations: Defines no financial-event, session, correction, or availability semantics
Evidence classification
- Sourced facts: the classical refresh-time construction, the existence of synchronization alternatives, and RFC 3339 timestamp syntax.
- Package engineering choices: availability-time barriers, revision policy, duplicate rejection, staleness status, partition reset, and exact loss counts.
- Pedagogical evidence: all numerical records and scenarios are synthetic.
- Not claimed: historical market behavior, predictive value, optimality, profitability, or estimator superiority.
Full dependency-light reference implementations in both supported languages.
export type Observation = {
partition: string;
instrument: string;
record_id: string;
revision: number;
event_time: string;
available_at: string;
value: number;
};
type Internal = Observation & { _event_ms: number; _available_ms: number };
const milliseconds = (value: string): number => {
if (typeof value !== "string" || !value.endsWith("Z")) {
throw new Error("timestamps must be RFC 3339 UTC strings ending in Z");
}
const parsed = Date.parse(value);
if (!Number.isFinite(parsed)) throw new Error(`invalid timestamp: ${value}`);
return parsed;
};
const formatMs = (value: number): string => new Date(value).toISOString();
function validate(
observations: Observation[], requiredInstruments: string[],
): { rows: Internal[]; instruments: string[] } {
const instruments = [...new Set(requiredInstruments)].sort();
if (instruments.length < 2 || instruments.some((item) => !item)) {
throw new Error("requiredInstruments must contain at least two names");
}
const allowed = new Set(instruments);
const rows: Internal[] = observations.map((raw) => {
const required: (keyof Observation)[] = [
"partition", "instrument", "record_id", "revision", "event_time",
"available_at", "value",
];
const missing = required.filter((field) => raw[field] === undefined);
if (missing.length) throw new Error(`missing fields: ${missing.join(", ")}`);
if (!allowed.has(raw.instrument)) throw new Error(`unexpected instrument: ${raw.instrument}`);
if (!raw.partition || !raw.record_id) throw new Error("partition and record_id must be non-empty");
if (!Number.isInteger(raw.revision) || raw.revision < 0) throw new Error("revision must be a non-negative integer");
if (!Number.isFinite(raw.value)) throw new Error("value must be finite");
const event = milliseconds(raw.event_time), available = milliseconds(raw.available_at);
if (event > available) throw new Error("event_time cannot be after available_at");
return { ...raw, _event_ms: event, _available_ms: available };
});
rows.sort((a, b) =>
a.partition.localeCompare(b.partition) || a._available_ms - b._available_ms ||
a.instrument.localeCompare(b.instrument) || a.record_id.localeCompare(b.record_id) ||
a.revision - b.revision
);
const histories = new Map<string, Internal[]>(), eventOwners = new Map<string, string>();
for (const row of rows) {
const eventKey = `${row.partition}\0${row.instrument}\0${row._event_ms}`;
const eventOwner = eventOwners.get(eventKey);
if (eventOwner !== undefined && eventOwner !== row.record_id) {
throw new Error("ambiguous duplicate event_time across distinct record_id values");
}
eventOwners.set(eventKey, row.record_id);
const key = `${row.partition}\0${row.record_id}`, history = histories.get(key) ?? [];
if (!history.length && row.revision !== 0) throw new Error("the first revision of a record must be zero");
if (history.length) {
const previous = history.at(-1)!;
if (row.instrument !== previous.instrument) throw new Error("a correction cannot change instrument");
if (row.revision !== previous.revision + 1) throw new Error("record revisions must be contiguous");
if (row._available_ms <= previous._available_ms) throw new Error("corrections must become available strictly later");
}
history.push(row); histories.set(key, history);
}
return { rows, instruments };
}
function latestSources(known: Internal[], instruments: string[]): Record<string, Internal> {
const latest = new Map<string, Internal>();
for (const row of known) latest.set(row.record_id, row);
const result: Record<string, Internal> = {};
for (const instrument of instruments) {
const candidates = [...latest.values()].filter((row) => row.instrument === instrument);
if (!candidates.length) throw new Error(`no known observation for ${instrument}`);
const owners = new Map<number, string>();
for (const row of candidates) {
const owner = owners.get(row._event_ms);
if (owner !== undefined && owner !== row.record_id) {
throw new Error("ambiguous duplicate event_time across distinct record_id values");
}
owners.set(row._event_ms, row.record_id);
}
result[instrument] = candidates.reduce((best, row) =>
row._event_ms > best._event_ms ||
(row._event_ms === best._event_ms && row._available_ms > best._available_ms) ||
(row._event_ms === best._event_ms && row._available_ms === best._available_ms && row.record_id > best.record_id)
? row : best
);
}
return result;
}
export function refreshTimeSample(
observations: Observation[], requiredInstruments: string[], maxStalenessMs: number,
) {
if (!Number.isInteger(maxStalenessMs) || maxStalenessMs < 0) {
throw new Error("maxStalenessMs must be a non-negative integer");
}
const { rows, instruments } = validate(observations, requiredInstruments);
const partitions = [...new Set(rows.map((row) => row.partition))].sort();
const outputRows: any[] = [], summaries: any[] = [];
for (const partition of partitions) {
const arrivals = rows.filter((row) => row.partition === partition);
const byInstrument = Object.fromEntries(instruments.map((instrument) => [
instrument, arrivals.filter((row) => row.instrument === instrument),
])) as Record<string, Internal[]>;
let cursor = Number.NEGATIVE_INFINITY, sequence = 0, discardedTotal = 0;
let staleCount = 0, acceptedCount = 0;
while (true) {
const firstNew: Record<string, Internal> = {};
let complete = true;
for (const instrument of instruments) {
const found = byInstrument[instrument].find((row) => row._available_ms > cursor);
if (!found) { complete = false; break; }
firstNew[instrument] = found;
}
if (!complete) break;
const refresh = Math.max(...Object.values(firstNew).map((row) => row._available_ms));
const interval = arrivals.filter((row) => row._available_ms > cursor && row._available_ms <= refresh);
const known = arrivals.filter((row) => row._available_ms <= refresh);
const selected = latestSources(known, instruments);
const selectedInterval = new Set(Object.values(selected)
.filter((row) => row._available_ms > cursor && row._available_ms <= refresh)
.map((row) => `${row.record_id}\0${row.revision}`));
const discarded = interval.length - selectedInterval.size;
discardedTotal += discarded;
const ages = Object.fromEntries(instruments.map((instrument) => [
instrument, refresh - selected[instrument]._event_ms,
]));
const status = Math.max(...Object.values(ages)) > maxStalenessMs ? "stale" : "accepted";
staleCount += status === "stale" ? 1 : 0;
acceptedCount += status === "accepted" ? 1 : 0;
outputRows.push({
partition, sequence: ++sequence,
previous_refresh_available_at: Number.isFinite(cursor) ? formatMs(cursor) : null,
refresh_available_at: formatMs(refresh),
controller_instruments: instruments.filter((instrument) => firstNew[instrument]._available_ms === refresh),
status,
values: Object.fromEntries(instruments.map((instrument) => [instrument, selected[instrument].value])),
sources: Object.fromEntries(instruments.map((instrument) => [instrument, {
record_id: selected[instrument].record_id,
revision: selected[instrument].revision,
event_time: selected[instrument].event_time,
available_at: selected[instrument].available_at,
}])),
event_age_ms: ages,
arrivals_by_instrument: Object.fromEntries(instruments.map((instrument) => [
instrument, interval.filter((row) => row.instrument === instrument).length,
])),
discarded_updates: discarded,
});
cursor = refresh;
}
const tail = arrivals.filter((row) => row._available_ms > cursor).length;
summaries.push({
partition, input_updates: arrivals.length, refresh_candidates: sequence,
accepted_rows: acceptedCount, stale_rows: staleCount,
discarded_updates: discardedTotal, unmatched_tail_updates: tail,
loss_fraction: { numerator: discardedTotal + tail, denominator: arrivals.length },
});
}
return { rows: outputRows, partitions: summaries };
}
The embedded lab now expands to its full document height, keeping the article as the only scroll surface.