use super::{DataContent, DataItem, StreamService};
use crate::common::{AssetMainType, AssetSubType};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use std::collections::HashMap;
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub struct Field {
pub number: u32,
pub name: &'static str,
pub nested: Option<&'static [Field]>,
}
impl Field {
const fn new(number: u32, name: &'static str) -> Self {
Self {
number,
name,
nested: None,
}
}
const fn nested(number: u32, name: &'static str, table: &'static [Field]) -> Self {
Self {
number,
name,
nested: Some(table),
}
}
}
pub(super) fn csv(numbers: impl Iterator<Item = u32>) -> String {
numbers
.map(|number| number.to_string())
.collect::<Vec<_>>()
.join(",")
}
fn lookup<'t>(key: &str, table: &'t [Field]) -> Option<&'t Field> {
let number = key.parse::<u32>().ok()?;
table
.get(number as usize)
.filter(|field| field.number == number)
.or_else(|| table.iter().find(|field| field.number == number))
}
fn rename(
entries: impl IntoIterator<Item = (String, Value)>,
table: &[Field],
) -> Map<String, Value> {
entries
.into_iter()
.map(|(key, value)| match lookup(&key, table) {
Some(field) => {
let value = match field.nested {
Some(nested) => rename_nested(value, nested),
None => value,
};
(field.name.to_owned(), value)
}
None => (key, value),
})
.collect()
}
fn rename_nested(value: Value, table: &[Field]) -> Value {
match value {
Value::Array(items) => Value::Array(
items
.into_iter()
.map(|item| rename_nested(item, table))
.collect(),
),
Value::Object(map) => Value::Object(rename(map, table)),
other => other,
}
}
impl DataContent {
pub fn named_fields(&self, table: &[Field]) -> Map<String, Value> {
rename(
self.fields.iter().map(|(k, v)| (k.clone(), v.clone())),
table,
)
}
pub fn to_named_value(&self, table: &[Field]) -> Value {
self.clone().into_named_value(table)
}
pub fn into_named_value(self, table: &[Field]) -> Value {
let mut map = rename(self.fields, table);
map.insert("key".into(), Value::String(self.key));
if let Some(delayed) = self.delayed {
map.insert("delayed".into(), Value::Bool(delayed));
}
if let Some(Ok(asset)) = self.asset_main_type.map(serde_json::to_value) {
map.insert("assetMainType".into(), asset);
}
if let Some(Ok(asset)) = self.asset_sub_type.map(serde_json::to_value) {
map.insert("assetSubType".into(), asset);
}
if let Some(cusip) = self.cusip {
map.insert("cusip".into(), Value::String(cusip));
}
Value::Object(map)
}
}
macro_rules! field_table {
(
$(#[$doc:meta])*
$table:ident => $payload:ident {
key = $key:ident;
envelope = { $( $env:ident : $env_ty:ty ),* $(,)? };
$( default = [ $( $default:ident ),* $(,)? ]; )?
$(
$(#[$field_doc:meta])*
$number:literal $konst:ident $field:ident $name:literal : $ty:ty
$([ $nested:ident ])? ,
)*
}
) => {
#[doc = concat!("Field numbers and names for [`", stringify!($payload), "`].")]
#[derive(Debug, Clone, Copy)]
pub struct $table;
impl $table {
pub const $key: u32 = 0;
$( pub const $konst: u32 = $number; )*
pub const FIELDS: &'static [Field] = &[
Field::new(0, "key"),
$( field_table!(@field $number $name $($nested)?), )*
];
pub fn name(number: u32) -> Option<&'static str> {
Self::FIELDS
.iter()
.find(|field| field.number == number)
.map(|field| field.name)
}
pub fn all_fields() -> String {
csv(Self::FIELDS.iter().map(|field| field.number))
}
$(
pub fn default_fields() -> String {
csv([$( Self::$default ),*].into_iter())
}
)?
}
$(#[$doc])*
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
pub struct $payload {
/// The subscription key (field 0).
pub key: String,
$(
#[serde(default, skip_serializing_if = "Option::is_none")]
pub $env: Option<$env_ty>,
)*
$(
$(#[$field_doc])*
#[serde(rename = $name, default, skip_serializing_if = "Option::is_none")]
pub $field: Option<$ty>,
)*
#[serde(flatten, default, skip_serializing_if = "HashMap::is_empty")]
pub extra: HashMap<String, Value>,
}
impl $payload {
#[doc = concat!(
"Decode a content item using [`", stringify!($table), "::FIELDS`]."
)]
pub fn from_content(content: &DataContent) -> Result<Self, serde_json::Error> {
serde_json::from_value(content.to_named_value($table::FIELDS))
}
}
};
(@field $number:literal $name:literal) => {
Field::new($number, $name)
};
(@field $number:literal $name:literal $nested:ident) => {
Field::nested($number, $name, $nested::FIELDS)
};
}
field_table! {
LevelOneEquityFields => LevelOneEquity {
key = SYMBOL;
envelope = {
delayed: bool,
asset_main_type: AssetMainType,
asset_sub_type: AssetSubType,
cusip: String,
};
default = [
SYMBOL, BID_PRICE, ASK_PRICE, LAST_PRICE, BID_SIZE, ASK_SIZE, TOTAL_VOLUME,
LAST_SIZE, HIGH_PRICE, LOW_PRICE, CLOSE_PRICE, OPEN_PRICE, NET_CHANGE,
NET_CHANGE_PERCENT, QUOTE_TIME_MILLIS, TRADE_TIME_MILLIS,
];
1 BID_PRICE bid_price "bidPrice": f64,
2 ASK_PRICE ask_price "askPrice": f64,
3 LAST_PRICE last_price "lastPrice": f64,
4 BID_SIZE bid_size "bidSize": i64,
5 ASK_SIZE ask_size "askSize": i64,
6 ASK_ID ask_id "askId": String,
7 BID_ID bid_id "bidId": String,
8 TOTAL_VOLUME total_volume "totalVolume": i64,
9 LAST_SIZE last_size "lastSize": i64,
10 HIGH_PRICE high_price "highPrice": f64,
11 LOW_PRICE low_price "lowPrice": f64,
12 CLOSE_PRICE close_price "closePrice": f64,
13 EXCHANGE_ID exchange_id "exchangeId": String,
14 MARGINABLE marginable "marginable": bool,
15 DESCRIPTION description "description": String,
16 LAST_ID last_id "lastId": String,
17 OPEN_PRICE open_price "openPrice": f64,
18 NET_CHANGE net_change "netChange": f64,
19 HIGH_52_WEEK fifty_two_week_high "52WeekHigh": f64,
20 LOW_52_WEEK fifty_two_week_low "52WeekLow": f64,
21 PE_RATIO pe_ratio "peRatio": f64,
22 DIV_AMOUNT div_amount "divAmount": f64,
23 DIV_YIELD div_yield "divYield": f64,
24 NAV nav "nav": f64,
25 EXCHANGE_NAME exchange_name "exchangeName": String,
26 DIV_DATE div_date "divDate": String,
27 IS_REGULAR_MARKET_QUOTE regular_market_quote "regularMarketQuote": bool,
28 IS_REGULAR_MARKET_TRADE regular_market_trade "regularMarketTrade": bool,
29 REGULAR_MARKET_LAST_PRICE regular_market_last_price "regularMarketLastPrice": f64,
30 REGULAR_MARKET_LAST_SIZE regular_market_last_size "regularMarketLastSize": i64,
31 REGULAR_MARKET_NET_CHANGE regular_market_net_change "regularMarketNetChange": f64,
32 SECURITY_STATUS security_status "securityStatus": String,
33 MARK mark "mark": f64,
34 QUOTE_TIME_MILLIS quote_time "quoteTime": i64,
35 TRADE_TIME_MILLIS trade_time "tradeTime": i64,
36 REGULAR_MARKET_TRADE_MILLIS regular_market_trade_time "regularMarketTradeTime": i64,
37 BID_TIME_MILLIS bid_time "bidTime": i64,
38 ASK_TIME_MILLIS ask_time "askTime": i64,
39 ASK_MIC_ID ask_mic_id "askMICId": String,
40 BID_MIC_ID bid_mic_id "bidMICId": String,
41 LAST_MIC_ID last_mic_id "lastMICId": String,
42 NET_CHANGE_PERCENT net_percent_change "netPercentChange": f64,
43 REGULAR_MARKET_CHANGE_PERCENT regular_market_percent_change
"regularMarketPercentChange": f64,
44 MARK_CHANGE mark_change "markChange": f64,
45 MARK_CHANGE_PERCENT mark_percent_change "markPercentChange": f64,
46 HTB_QUANTITY htb_quantity "htbQuantity": i64,
47 HTB_RATE htb_rate "htbRate": f64,
48 HARD_TO_BORROW hard_to_borrow "hardToBorrow": i64,
49 IS_SHORTABLE shortable "shortable": i64,
50 POST_MARKET_NET_CHANGE post_market_change "postMarketChange": f64,
51 POST_MARKET_CHANGE_PERCENT post_market_percent_change "postMarketPercentChange": f64,
}
}
field_table! {
LevelOneOptionFields => LevelOneOption {
key = SYMBOL;
envelope = { delayed: bool, asset_main_type: AssetMainType };
1 DESCRIPTION description "description": String,
2 BID_PRICE bid_price "bidPrice": f64,
3 ASK_PRICE ask_price "askPrice": f64,
4 LAST_PRICE last_price "lastPrice": f64,
5 HIGH_PRICE high_price "highPrice": f64,
6 LOW_PRICE low_price "lowPrice": f64,
7 CLOSE_PRICE close_price "closePrice": f64,
8 TOTAL_VOLUME total_volume "totalVolume": i64,
9 OPEN_INTEREST open_interest "openInterest": i64,
10 VOLATILITY volatility "volatility": f64,
11 INTRINSIC_VALUE money_intrinsic_value "moneyIntrinsicValue": f64,
12 EXP_YEAR expiration_year "expirationYear": i64,
13 MULTIPLIER multiplier "multiplier": f64,
14 DIGITS digits "digits": i64,
15 OPEN_PRICE open_price "openPrice": f64,
16 BID_SIZE bid_size "bidSize": i64,
17 ASK_SIZE ask_size "askSize": i64,
18 LAST_SIZE last_size "lastSize": i64,
19 NET_CHANGE net_change "netChange": f64,
20 STRIKE_PRICE strike_price "strikePrice": f64,
21 CONTRACT_TYPE contract_type "contractType": String,
22 UNDERLYING underlying "underlying": String,
23 EXP_MONTH expiration_month "expirationMonth": i64,
24 DELIVERABLES deliverables "deliverables": String,
25 TIME_VALUE time_value "timeValue": f64,
26 EXP_DAY expiration_day "expirationDay": i64,
27 DAYS_TO_EXP days_to_expiration "daysToExpiration": i64,
28 DELTA delta "delta": f64,
29 GAMMA gamma "gamma": f64,
30 THETA theta "theta": f64,
31 VEGA vega "vega": f64,
32 RHO rho "rho": f64,
33 SECURITY_STATUS security_status "securityStatus": String,
34 THEORETICAL_VALUE theoretical_option_value "theoreticalOptionValue": f64,
35 UNDERLYING_PRICE underlying_price "underlyingPrice": f64,
36 UV_EXPIRATION_TYPE uv_expiration_type "uvExpirationType": String,
37 MARK mark "mark": f64,
38 QUOTE_TIME_MILLIS quote_time "quoteTime": i64,
39 TRADE_TIME_MILLIS trade_time "tradeTime": i64,
40 EXCHANGE exchange "exchange": String,
41 EXCHANGE_NAME exchange_name "exchangeName": String,
42 LAST_TRADING_DAY last_trading_day "lastTradingDay": i64,
43 SETTLEMENT_TYPE settlement_type "settlementType": String,
44 NET_PERCENT_CHANGE net_percent_change "netPercentChange": f64,
45 MARK_NET_CHANGE mark_change "markChange": f64,
46 MARK_PERCENT_CHANGE mark_percent_change "markPercentChange": f64,
47 IMPLIED_YIELD implied_yield "impliedYield": f64,
48 IS_PENNY_PILOT is_penny_pilot "isPennyPilot": bool,
49 OPTION_ROOT option_root "optionRoot": String,
50 HIGH_52_WEEK fifty_two_week_high "52WeekHigh": f64,
51 LOW_52_WEEK fifty_two_week_low "52WeekLow": f64,
52 INDICATIVE_ASK_PRICE indicative_ask_price "indAskPrice": f64,
53 INDICATIVE_BID_PRICE indicative_bid_price "indBidPrice": f64,
54 INDICATIVE_QUOTE_TIME indicative_quote_time "indQuoteTime": i64,
55 EXERCISE_TYPE exercise_type "exerciseType": String,
}
}
field_table! {
LevelOneFuturesFields => LevelOneFutures {
key = SYMBOL;
envelope = { delayed: bool, asset_main_type: AssetMainType };
1 BID_PRICE bid_price "bidPrice": f64,
2 ASK_PRICE ask_price "askPrice": f64,
3 LAST_PRICE last_price "lastPrice": f64,
4 BID_SIZE bid_size "bidSize": i64,
5 ASK_SIZE ask_size "askSize": i64,
6 BID_ID bid_id "bidId": String,
7 ASK_ID ask_id "askId": String,
8 TOTAL_VOLUME total_volume "totalVolume": f64,
9 LAST_SIZE last_size "lastSize": i64,
10 QUOTE_TIME quote_time "quoteTime": i64,
11 TRADE_TIME trade_time "tradeTime": i64,
12 HIGH_PRICE high_price "highPrice": f64,
13 LOW_PRICE low_price "lowPrice": f64,
14 CLOSE_PRICE close_price "closePrice": f64,
15 EXCHANGE_ID exchange_id "exchangeId": String,
16 DESCRIPTION description "description": String,
17 LAST_ID last_id "lastId": String,
18 OPEN_PRICE open_price "openPrice": f64,
19 NET_CHANGE net_change "netChange": f64,
20 FUTURE_PERCENT_CHANGE future_percent_change "futurePercentChange": f64,
21 EXCHANGE_NAME exchange_name "exchangeName": String,
22 SECURITY_STATUS security_status "securityStatus": String,
23 OPEN_INTEREST open_interest "openInterest": i64,
24 MARK mark "mark": f64,
25 TICK tick "tick": f64,
26 TICK_AMOUNT tick_amount "tickAmount": f64,
27 PRODUCT product "product": String,
28 FUTURE_PRICE_FORMAT future_price_format "futurePriceFormat": String,
29 FUTURE_TRADING_HOURS future_trading_hours "futureTradingHours": String,
30 FUTURE_IS_TRADABLE future_is_tradable "futureIsTradable": bool,
31 FUTURE_MULTIPLIER future_multiplier "futureMultiplier": f64,
32 FUTURE_IS_ACTIVE future_is_active "futureIsActive": bool,
33 FUTURE_SETTLEMENT_PRICE future_settlement_price "futureSettlementPrice": f64,
34 FUTURE_ACTIVE_SYMBOL future_active_symbol "futureActiveSymbol": String,
35 FUTURE_EXPIRATION_DATE future_expiration_date "futureExpirationDate": i64,
36 EXPIRATION_STYLE expiration_style "expirationStyle": String,
37 ASK_TIME ask_time "askTime": i64,
38 BID_TIME bid_time "bidTime": i64,
39 QUOTED_IN_SESSION quoted_in_session "quotedInSession": bool,
40 SETTLEMENT_DATE settlement_date "settlementDate": i64,
}
}
field_table! {
LevelOneFuturesOptionFields => LevelOneFuturesOption {
key = SYMBOL;
envelope = { delayed: bool, asset_main_type: AssetMainType };
1 BID_PRICE bid_price "bidPrice": f64,
2 ASK_PRICE ask_price "askPrice": f64,
3 LAST_PRICE last_price "lastPrice": f64,
4 BID_SIZE bid_size "bidSize": i64,
5 ASK_SIZE ask_size "askSize": i64,
6 BID_ID bid_id "bidId": String,
7 ASK_ID ask_id "askId": String,
8 TOTAL_VOLUME total_volume "totalVolume": i64,
9 LAST_SIZE last_size "lastSize": i64,
10 QUOTE_TIME quote_time "quoteTime": i64,
11 TRADE_TIME trade_time "tradeTime": i64,
12 HIGH_PRICE high_price "highPrice": f64,
13 LOW_PRICE low_price "lowPrice": f64,
14 CLOSE_PRICE close_price "closePrice": f64,
15 LAST_ID last_id "lastId": String,
16 DESCRIPTION description "description": String,
17 OPEN_PRICE open_price "openPrice": f64,
18 OPEN_INTEREST open_interest "openInterest": f64,
19 MARK mark "mark": f64,
20 TICK tick "tick": f64,
21 TICK_AMOUNT tick_amount "tickAmount": f64,
22 FUTURE_MULTIPLIER future_multiplier "futureMultiplier": f64,
23 FUTURE_SETTLEMENT_PRICE future_settlement_price "futureSettlementPrice": f64,
24 UNDERLYING_SYMBOL underlying_symbol "underlyingSymbol": String,
25 STRIKE_PRICE strike_price "strikePrice": f64,
26 FUTURE_EXPIRATION_DATE future_expiration_date "futureExpirationDate": i64,
27 EXPIRATION_STYLE expiration_style "expirationStyle": String,
28 CONTRACT_TYPE contract_type "contractType": String,
29 SECURITY_STATUS security_status "securityStatus": String,
30 EXCHANGE exchange "exchange": String,
31 EXCHANGE_NAME exchange_name "exchangeName": String,
}
}
field_table! {
LevelOneForexFields => LevelOneForex {
key = SYMBOL;
envelope = { delayed: bool, asset_main_type: AssetMainType };
1 BID_PRICE bid_price "bidPrice": f64,
2 ASK_PRICE ask_price "askPrice": f64,
3 LAST_PRICE last_price "lastPrice": f64,
4 BID_SIZE bid_size "bidSize": i64,
5 ASK_SIZE ask_size "askSize": i64,
6 TOTAL_VOLUME total_volume "totalVolume": f64,
7 LAST_SIZE last_size "lastSize": i64,
8 QUOTE_TIME quote_time "quoteTime": i64,
9 TRADE_TIME trade_time "tradeTime": i64,
10 HIGH_PRICE high_price "highPrice": f64,
11 LOW_PRICE low_price "lowPrice": f64,
12 CLOSE_PRICE close_price "closePrice": f64,
13 EXCHANGE exchange "exchange": String,
14 DESCRIPTION description "description": String,
15 OPEN_PRICE open_price "openPrice": f64,
16 NET_CHANGE net_change "netChange": f64,
17 PERCENT_CHANGE percent_change "percentChange": f64,
18 EXCHANGE_NAME exchange_name "exchangeName": String,
19 DIGITS digits "digits": i64,
20 SECURITY_STATUS security_status "securityStatus": String,
21 TICK tick "tick": f64,
22 TICK_AMOUNT tick_amount "tickAmount": f64,
23 PRODUCT product "product": String,
24 TRADING_HOURS trading_hours "tradingHours": String,
25 IS_TRADABLE is_tradable "isTradable": bool,
26 MARKET_MAKER market_maker "marketMaker": String,
27 HIGH_52_WEEK fifty_two_week_high "52WeekHigh": f64,
28 LOW_52_WEEK fifty_two_week_low "52WeekLow": f64,
29 MARK mark "mark": f64,
}
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct BookMarketMaker {
pub market_maker_id: String,
pub size: i64,
pub quote_time: i64,
}
impl BookMarketMaker {
pub const FIELDS: &'static [Field] = &[
Field::new(0, "marketMakerId"),
Field::new(1, "size"),
Field::new(2, "quoteTime"),
];
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct BookLevel {
pub price: f64,
pub aggregate_size: i64,
pub market_maker_count: i64,
#[serde(default)]
pub market_makers: Vec<BookMarketMaker>,
}
impl BookLevel {
pub const FIELDS: &'static [Field] = &[
Field::new(0, "price"),
Field::new(1, "aggregateSize"),
Field::new(2, "marketMakerCount"),
Field::nested(3, "marketMakers", BookMarketMaker::FIELDS),
];
}
field_table! {
BookFields => Book {
key = SYMBOL;
envelope = {};
1 MARKET_SNAPSHOT_TIME market_snapshot_time "marketSnapshotTime": i64,
2 BID_SIDE_LEVELS bid_side_levels "bidSideLevels": Vec<BookLevel> [BookLevel],
3 ASK_SIDE_LEVELS ask_side_levels "askSideLevels": Vec<BookLevel> [BookLevel],
}
}
field_table! {
ChartEquityFields => ChartEquity {
key = KEY;
envelope = { seq: i64 };
1 SEQUENCE sequence "sequence": i64,
2 OPEN_PRICE open "open": f64,
3 HIGH_PRICE high "high": f64,
4 LOW_PRICE low "low": f64,
5 CLOSE_PRICE close "close": f64,
6 VOLUME volume "volume": f64,
7 CHART_TIME chart_time "chartTime": i64,
8 CHART_DAY chart_day "chartDay": i64,
}
}
field_table! {
ChartFuturesFields => ChartFutures {
key = KEY;
envelope = { seq: i64 };
1 CHART_TIME chart_time "chartTime": i64,
2 OPEN_PRICE open "open": f64,
3 HIGH_PRICE high "high": f64,
4 LOW_PRICE low "low": f64,
5 CLOSE_PRICE close "close": f64,
6 VOLUME volume "volume": f64,
}
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct ScreenerItem {
pub symbol: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_price: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub net_change: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub net_percent_change: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub market_share: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub total_volume: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub volume: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub trades: Option<i64>,
#[serde(flatten, default, skip_serializing_if = "HashMap::is_empty")]
pub extra: HashMap<String, Value>,
}
field_table! {
ScreenerFields => Screener {
key = SYMBOL;
envelope = {};
1 TIMESTAMP timestamp "timestamp": i64,
2 SORT_FIELD sort_field "sortField": String,
3 FREQUENCY frequency "frequency": i64,
4 ITEMS items "items": Vec<ScreenerItem>,
}
}
field_table! {
AcctActivityFields => AccountActivity {
key = KEY;
envelope = { seq: i64 };
1 ACCOUNT account_number "accountNumber": String,
2 MESSAGE_TYPE message_type "messageType": String,
3 MESSAGE_DATA message_data "messageData": String,
}
}
impl AccountActivity {
pub fn message(&self) -> Option<Map<String, Value>> {
match serde_json::from_str(self.message_data.as_deref()?) {
Ok(Value::Object(map)) => Some(map),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum StreamPayload {
LevelOneEquity(LevelOneEquity),
LevelOneOption(LevelOneOption),
LevelOneFutures(LevelOneFutures),
LevelOneFuturesOption(LevelOneFuturesOption),
LevelOneForex(LevelOneForex),
Book(Book),
ChartEquity(ChartEquity),
ChartFutures(ChartFutures),
Screener(Screener),
AccountActivity(AccountActivity),
}
impl StreamService {
pub fn fields(self) -> &'static [Field] {
match self {
Self::Admin => &[],
Self::LeveloneEquities => LevelOneEquityFields::FIELDS,
Self::LeveloneOptions => LevelOneOptionFields::FIELDS,
Self::LeveloneFutures => LevelOneFuturesFields::FIELDS,
Self::LeveloneFuturesOptions => LevelOneFuturesOptionFields::FIELDS,
Self::LeveloneForex => LevelOneForexFields::FIELDS,
Self::NyseBook | Self::NasdaqBook | Self::OptionsBook => BookFields::FIELDS,
Self::ChartEquity => ChartEquityFields::FIELDS,
Self::ChartFutures => ChartFuturesFields::FIELDS,
Self::ScreenerEquity | Self::ScreenerOption => ScreenerFields::FIELDS,
Self::AcctActivity => AcctActivityFields::FIELDS,
}
}
pub fn decode(self, content: &DataContent) -> Result<StreamPayload, serde_json::Error> {
use serde::de::Error as _;
Ok(match self {
Self::Admin => {
return Err(serde_json::Error::custom("ADMIN frames carry no data"));
}
Self::LeveloneEquities => {
StreamPayload::LevelOneEquity(LevelOneEquity::from_content(content)?)
}
Self::LeveloneOptions => {
StreamPayload::LevelOneOption(LevelOneOption::from_content(content)?)
}
Self::LeveloneFutures => {
StreamPayload::LevelOneFutures(LevelOneFutures::from_content(content)?)
}
Self::LeveloneFuturesOptions => {
StreamPayload::LevelOneFuturesOption(LevelOneFuturesOption::from_content(content)?)
}
Self::LeveloneForex => {
StreamPayload::LevelOneForex(LevelOneForex::from_content(content)?)
}
Self::NyseBook | Self::NasdaqBook | Self::OptionsBook => {
StreamPayload::Book(Book::from_content(content)?)
}
Self::ChartEquity => StreamPayload::ChartEquity(ChartEquity::from_content(content)?),
Self::ChartFutures => StreamPayload::ChartFutures(ChartFutures::from_content(content)?),
Self::ScreenerEquity | Self::ScreenerOption => {
StreamPayload::Screener(Screener::from_content(content)?)
}
Self::AcctActivity => {
StreamPayload::AccountActivity(AccountActivity::from_content(content)?)
}
})
}
}
impl DataItem {
pub fn service(&self) -> Option<StreamService> {
StreamService::from_api_str(&self.service)
}
pub fn decode(&self) -> Result<Vec<StreamPayload>, serde_json::Error> {
use serde::de::Error as _;
let service = self.service().ok_or_else(|| {
serde_json::Error::custom(format!("unknown streamer service `{}`", self.service))
})?;
self.content
.iter()
.map(|content| service.decode(content))
.collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn content(value: Value) -> DataContent {
serde_json::from_value(value).unwrap()
}
#[test]
fn tables_are_dense_and_unique() {
for service in [
StreamService::LeveloneEquities,
StreamService::LeveloneOptions,
StreamService::LeveloneFutures,
StreamService::LeveloneFuturesOptions,
StreamService::LeveloneForex,
StreamService::NyseBook,
StreamService::ChartEquity,
StreamService::ChartFutures,
StreamService::ScreenerEquity,
StreamService::AcctActivity,
] {
let table = service.fields();
for (index, field) in table.iter().enumerate() {
assert_eq!(field.number as usize, index, "{service:?} {}", field.name);
}
let mut names: Vec<_> = table.iter().map(|field| field.name).collect();
names.sort_unstable();
names.dedup();
assert_eq!(names.len(), table.len(), "{service:?} names repeat");
}
assert_eq!(LevelOneEquityFields::FIELDS.len(), 52);
assert_eq!(LevelOneOptionFields::FIELDS.len(), 56);
assert_eq!(LevelOneFuturesFields::FIELDS.len(), 41);
assert_eq!(LevelOneFuturesOptionFields::FIELDS.len(), 32);
assert_eq!(LevelOneForexFields::FIELDS.len(), 30);
assert!(StreamService::Admin.fields().is_empty());
}
#[test]
fn payload_names_match_the_table() {
let numbered: Map<String, Value> = LevelOneEquityFields::FIELDS
.iter()
.skip(1)
.map(|field| {
let value = match field.name {
"askId" | "bidId" | "lastId" | "exchangeId" | "description"
| "exchangeName" | "divDate" | "securityStatus" | "askMICId" | "bidMICId"
| "lastMICId" => json!("x"),
"marginable" | "regularMarketQuote" | "regularMarketTrade" => json!(true),
_ => json!(1),
};
(field.number.to_string(), value)
})
.collect();
let mut wire = numbered.clone();
wire.insert("key".into(), json!("AAPL"));
let quote = LevelOneEquity::from_content(&content(Value::Object(wire))).unwrap();
assert!(quote.extra.is_empty(), "unmapped: {:?}", quote.extra);
assert_eq!(quote.post_market_percent_change, Some(1.0));
let named = serde_json::to_value("e).unwrap();
for field in LevelOneEquityFields::FIELDS {
assert!(
named.get(field.name).is_some(),
"{} not serialized",
field.name
);
}
}
#[test]
fn decodes_deltas_and_keeps_unknown_fields() {
let content = content(json!({
"key": "AMD", "delayed": false, "assetMainType": "EQUITY",
"assetSubType": "COE", "cusip": "007903107",
"1": 217.86, "2": 217.95, "6": "P", "8": 33292396, "99": "future"
}));
let quote = LevelOneEquity::from_content(&content).unwrap();
assert_eq!(quote.key, "AMD");
assert_eq!(quote.delayed, Some(false));
assert_eq!(quote.asset_main_type, Some(AssetMainType::Equity));
assert_eq!(quote.asset_sub_type, Some(AssetSubType::Coe));
assert_eq!(quote.cusip.as_deref(), Some("007903107"));
assert_eq!(quote.ask_price, Some(217.95));
assert_eq!(quote.ask_id.as_deref(), Some("P"));
assert_eq!(quote.last_price, None);
assert_eq!(quote.extra["99"], "future");
let named = content.to_named_value(LevelOneEquityFields::FIELDS);
assert_eq!(named["bidPrice"], 217.86);
assert_eq!(named["cusip"], "007903107");
assert_eq!(named["99"], "future");
assert!(named.get("1").is_none());
}
#[test]
fn option_field_one_is_the_description() {
let quote = LevelOneOption::from_content(&content(json!({
"key": "SPXW 251208C06880000", "delayed": false, "assetMainType": "OPTION",
"1": "SPXW 12/08/2025 6880.00 C", "2": 10.2, "3": 10.6
})))
.unwrap();
assert_eq!(
quote.description.as_deref(),
Some("SPXW 12/08/2025 6880.00 C")
);
assert_eq!(quote.bid_price, Some(10.2));
}
#[test]
fn envelope_classes_outside_the_enum_decode_as_unknown() {
let quote = LevelOneEquity::from_content(&content(json!({
"key": "X", "assetMainType": "CRYPTO", "assetSubType": "NEW", "1": 1.0
})))
.unwrap();
assert_eq!(quote.asset_main_type, Some(AssetMainType::Unknown));
assert_eq!(quote.asset_sub_type, Some(AssetSubType::Unknown));
assert_eq!(quote.bid_price, Some(1.0));
}
#[test]
fn book_levels_decode_recursively() {
let book = Book::from_content(&content(json!({
"key": "F",
"1": 1765306322349i64,
"2": [{"0": 13.12, "1": 31600, "2": 2, "3": [
{"0": "NYSE", "1": 8800, "2": 49786299},
{"0": "ARCX", "1": 4200, "2": 49904606}
]}],
"3": [{"0": 13.13, "1": 900, "2": 0, "3": []}]
})))
.unwrap();
let bids = book.bid_side_levels.as_ref().unwrap();
assert_eq!(bids[0].price, 13.12);
assert_eq!(bids[0].market_makers[1].market_maker_id, "ARCX");
assert_eq!(book.ask_side_levels.unwrap()[0].aggregate_size, 900);
}
#[test]
fn charts_screeners_and_activity_decode() {
let candle = ChartEquity::from_content(&content(json!({
"seq": 202, "key": "AMD", "1": 425, "2": 221.133, "5": 221.02,
"6": 11730, "7": 1765307100000i64, "8": 20431
})))
.unwrap();
assert_eq!(candle.seq, Some(202));
assert_eq!(candle.sequence, Some(425));
assert_eq!(candle.volume, Some(11730.0));
let screener = Screener::from_content(&content(json!({
"key": "NASDAQ_VOLUME_30", "1": 1765307297706i64, "2": "VOLUME", "3": 30,
"4": [{"symbol": "WBD", "lastPrice": 28.095, "trades": 21302}]
})))
.unwrap();
assert_eq!(screener.items.unwrap()[0].trades, Some(21302));
let activity = AccountActivity::from_content(&content(json!({
"seq": 1, "key": "Account Activity",
"1": "123456789", "2": "OrderCreated", "3": "{\"SchwabOrderID\":\"1\"}"
})))
.unwrap();
assert_eq!(activity.account_number.as_deref(), Some("123456789"));
assert_eq!(activity.message_type.as_deref(), Some("OrderCreated"));
}
#[test]
fn items_decode_by_service() {
let item: DataItem = serde_json::from_value(json!({
"service": "LEVELONE_FOREX", "timestamp": 1, "command": "SUBS",
"content": [{"key": "EUR/USD", "1": 1.1643, "2": 1.16443}]
}))
.unwrap();
assert_eq!(item.service(), Some(StreamService::LeveloneForex));
match item.decode().unwrap().as_slice() {
[StreamPayload::LevelOneForex(quote)] => assert_eq!(quote.ask_price, Some(1.16443)),
other => panic!("{other:?}"),
}
let unknown = DataItem {
service: "LEVELTWO".into(),
..item.clone()
};
assert!(
unknown
.decode()
.unwrap_err()
.to_string()
.contains("LEVELTWO")
);
assert!(StreamService::Admin.decode(&item.content[0]).is_err());
let wrong_type = content(json!({"key": "EUR/USD", "1": "not a price"}));
assert!(LevelOneForex::from_content(&wrong_type).is_err());
}
#[test]
fn subscription_field_lists_come_from_the_tables() {
assert_eq!(
LevelOneEquityFields::default_fields(),
"0,1,2,3,4,5,8,9,10,11,12,17,18,42,34,35"
);
assert_eq!(LevelOneEquityFields::all_fields(), csv(0..=51));
assert_eq!(
StreamService::LeveloneEquities.default_fields(),
LevelOneEquityFields::default_fields()
);
assert_eq!(
StreamService::LeveloneOptions.default_fields(),
LevelOneOptionFields::all_fields()
);
assert_eq!(StreamService::NasdaqBook.default_fields(), "0,1,2,3");
assert_eq!(StreamService::AcctActivity.default_fields(), "0,1,2,3");
assert_eq!(StreamService::Admin.default_fields(), "");
assert_eq!(LevelOneEquityFields::name(0), Some("key"));
assert_eq!(LevelOneEquityFields::name(52), None);
}
}