[Easy] record quote events for streamed quotes

[fleupold]

Description

CompetitionEstimator::estimate_stream doesn't report participating price estimates to our event bus and also doesn't populate any prometheus metrics.

Streamed quotes therefore show up in the event bus only as a quoteRequested/quoteComputed pair, with no competition behind them, which breaks anything that consumes the competition data (quote analytics, and the upcoming quote-rewards accounting that pays losing quotes).

This makes the streaming path report the competition the same way the one-shot path does.

Changes

• estimate_stream keeps each estimator's name next to its future, publishes a priceEstimate event for every successful answer as it arrives (same emit_quote_event helper and same "only successful estimates" rule as estimates), and once every estimator has answered reports the final best as the winner: winningPriceEstimate event plus the queries_won metric.
• report_winner is split into the index lookup and a name-based report_winner_by_name, shared by both paths.

How to test

New test estimate_stream_counts_the_final_best_as_the_winner checks that only the estimator behind the last forwarded improvement gets the win.

Manually: run the orderbook with an event bus configured, call POST /api/v1/quote/stream, and observe one priceEstimate per solver and one winningPriceEstimate on the bus; competition_price_estimator_queries_won increases for the winning estimator.

[claude]

Claude finished @fleupold's task in 2m 44s —— View job

---

Review: record quote events for streamed quotes

• Read changed files and surrounding context
• Analyze correctness of streamed quote event/metric reporting
• Check for double-counting / winner-consistency issues
• Post review

Verdict: looks correct and well-tested — no changes requested.

I traced the streaming path against the one-shot estimates path and the consumer in shared/src/order_quoting.rs:

• No double counting. estimates backs /quote (compute_quote_data) and estimate_stream backs /quote/stream (calculate_quote_stream) — distinct requests, so queries_won is incremented at most once per quote request in either path.
• Winner matches the one-shot path. best only advances on a strict is_gt, so it's monotonic and the last forwarded improvement is the global best — the same result the one-shot path gets from sorting all quotes. In the None branch, any reasonable Ok would already have been forwarded (so best would be Some), meaning the terminal is always an error; report_winner_by_name correctly logs it without touching the metric, matching estimates.
• Event semantics are consistent. emit_quote_event is attached to each estimator future and fires exactly once whether the future is drained in the take_until buffering phase or the chained live phase, so there's one priceEstimate per successful solver and one winningPriceEstimate. The "only successful estimates" rule matches the one-shot path (both emit before reasonableness filtering).
• report_winner refactor into report_winner_by_name is a clean shared split; the index-based report_winner is still used by estimates.
• Test estimate_stream_counts_the_final_best_as_the_winner uses unique estimator names and before/after deltas, so it's robust to shared global metric state.

One minor, non-blocking observation (no change needed): the two paths break exact ties differently — the one-shot sort keeps the first stage-order quote among equals, while the stream keeps the first-arrived. Since tied quotes rank equally this doesn't affect ranking correctness, only which estimator gets credited on an exact tie.
• branch emit-quote-events-on-stream

[MartinquaXD]

nit: nested matches lean very far to the right which IMO makes reading the code harder. Feel free to ignore if you prefer the matches.

Suggested change

	match best {
	// The last improvement forwarded is the winner of the competition.
	Some((name, estimate)) => {
	let winner: PriceEstimateResult = Ok(estimate);
	report_winner_by_name(&query, query.kind, name, &winner);
	emit_winning_price_estimate_event(name, &query);
	}
	None => {
	let terminal = results
	.into_iter()
	.max_by(|(_, a), (_, b)| {
	compare_quote_result(&query, a, b, &context, self.verification_mode)
	});
	match terminal {
	Some((name, result)) => {
	report_winner_by_name(&query, query.kind, name, &result);
	yield result;
	}
	None => yield Err(unreasonable_estimates_error()),
	}
	}
	}
	if let Some((name, estimate)) = best {
	// The last improvement forwarded is the winner of the competition.
	let winner: PriceEstimateResult = Ok(estimate);
	report_winner_by_name(&query, query.kind, name, &winner);
	emit_winning_price_estimate_event(name, &query);
	} else {
	let worst_error = results
	.into_iter()
	.max_by(|(_, a), (_, b)| {
	compare_quote_result(&query, a, b, &context, self.verification_mode)
	});
	if let Some((name, result)) = worst_error {
	report_winner_by_name(&query, query.kind, name, &result);
	yield result;
	} else {
	yield Err(unreasonable_estimates_error())
	}
	}

[MartinquaXD]

Suggested change

	assert_eq!(wins("stream-fast") - fast_before, 0);
	assert_eq!(wins("stream-fast"), fast_before);