blockworks-foundation
/
solana-accountsdb-connector
mirror of https://github.com/blockworks-foundation/solana-accountsdb-connector.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add Serum support for orderbook, fix pnl main

This commit is contained in:
Riordan Panayides 2022-12-24 12:49:17 +00:00
parent df562a3e42
commit fad9f7bae3
6 changed files with 341 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
Cargo.lock generated
View File

@ -5076,6 +5076,30 @@ dependencies = [
"without-alloc",
]
[[package]]
name = "serum_dex"
version = "0.5.10"
source = "git+ssh://git@github.com/jup-ag/openbook-program?branch=feat/expose-things#d287391bd7ed8111078fd1afb8f7b6d55cdd1d1f"
dependencies = [
"anchor-lang",
"arrayref",
"bincode",
"bytemuck",
"byteorder",
"enumflags2",
"field-offset",
"itertools 0.9.0",
"num-traits",
"num_enum",
"safe-transmute",
"serde",
"solana-program",
"spl-token",
"static_assertions",
"thiserror",
"without-alloc",
]
[[package]]
name = "serum_dex"
version = "0.5.10"
@ -5121,7 +5145,7 @@ dependencies = [
"serde",
"serde_derive",
"serde_json",
"serum_dex 0.5.10",
"serum_dex 0.5.10 (git+ssh://git@github.com/jup-ag/openbook-program?branch=feat/expose-things)",
"solana-geyser-connector-lib",
"solana-logger",
"tokio",
@ -5149,7 +5173,7 @@ dependencies = [
"serde",
"serde_derive",
"serde_json",
"serum_dex 0.5.10",
"serum_dex 0.5.10 (git+ssh://git@github.com/openbook-dex/program?branch=master)",
"solana-geyser-connector-lib",
"solana-logger",
"tokio",
@ -5728,7 +5752,7 @@ dependencies = [
"serde",
"serde_derive",
"serde_json",
"serum_dex 0.5.10",
"serum_dex 0.5.10 (git+ssh://git@github.com/jup-ag/openbook-program?branch=feat/expose-things)",
"solana-account-decoder",
"solana-client",
"solana-rpc",

2
lib/Cargo.toml
View File

@ -58,7 +58,7 @@ warp = "0.3"
anchor-lang = "0.25.0"
serum_dex = { git = "ssh://git@github.com/openbook-dex/program", branch = "master" }
serum_dex = { git = "ssh://git@github.com/jup-ag/openbook-program", branch = "feat/expose-things" }
[build-dependencies]
tonic-build = { version = "0.6", features = ["compression"] }

316
lib/src/orderbook_filter.rs
View File

@ -3,8 +3,10 @@ use crate::{
metrics::{MetricType, Metrics},
AccountWrite, SlotUpdate,
};
use itertools::Itertools;
use log::*;
use serde::{ser::SerializeStruct, Serialize, Serializer};
use serum_dex::critbit::Slab;
use solana_sdk::{
account::{ReadableAccount, WritableAccount},
clock::Epoch,
@ -12,13 +14,13 @@ use solana_sdk::{
};
use std::{
borrow::BorrowMut,
collections::{HashMap, HashSet}, time::{UNIX_EPOCH, SystemTime},
collections::{HashMap, HashSet},
time::{SystemTime, UNIX_EPOCH}, mem::size_of,
};
use itertools::Itertools;
use crate::metrics::MetricU64;
use anchor_lang::AccountDeserialize;
use mango_v4::state::{BookSide, OrderTreeType};
use mango_v4::{state::{BookSide, OrderTreeType}, serum3_cpi::OrderBookStateHeader};
#[derive(Clone, Debug)]
pub enum OrderbookSide {
@ -29,19 +31,19 @@ pub enum OrderbookSide {
impl Serialize for OrderbookSide {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer
S: Serializer,
{
match *self {
OrderbookSide::Bid => serializer.serialize_unit_variant("Side", 0, "bid"),
OrderbookSide::Ask => serializer.serialize_unit_variant("Side", 1, "ask"),
}
}
}
}
#[derive(Clone, Debug)]
pub struct OrderbookLevel {
pub price: i64,
pub size: i64,
pub price: f64,
pub size: f64,
}
impl Serialize for OrderbookLevel {
@ -112,48 +114,34 @@ pub enum OrderbookFilterMessage {
Checkpoint(OrderbookCheckpoint),
}
#[derive(Clone, Debug)]
pub struct MarketConfig {
pub name: String,
pub bids: Pubkey,
pub asks: Pubkey,
pub base_decimals: u8,
pub quote_decimals: u8,
}
pub fn native_to_ui(native: i64, decimals: u8) -> f64 {
native as f64 / (10u64.pow(decimals.into())) as f64
}
fn publish_changes(
slot: u64,
write_version: u64,
mkt: &(Pubkey, MarketConfig),
bookside: &BookSide,
old_bookside: &BookSide,
other_bookside: Option<&BookSide>,
side: OrderbookSide,
current_bookside: &Vec<OrderbookLevel>,
previous_bookside: &Vec<OrderbookLevel>,
maybe_other_bookside: Option<&Vec<OrderbookLevel>>,
orderbook_update_sender: &async_channel::Sender<OrderbookFilterMessage>,
metric_updates: &mut MetricU64,
) {
let time_now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
let oracle_price_lots = 0; // todo: does this matter? where to find it?
let side = match bookside.nodes.order_tree_type() {
OrderTreeType::Bids => OrderbookSide::Bid,
OrderTreeType::Asks => OrderbookSide::Ask,
};
let current_l2_snapshot: Vec<OrderbookLevel> = bookside
.iter_valid(time_now, oracle_price_lots)
.map(|item| (item.node.price_data() as i64, item.node.quantity))
.group_by(|(price, _)| *price)
.into_iter()
.map(|(price, group)| OrderbookLevel { price, size: group.map(|(_, quantity)| quantity).fold(0, |acc, x| acc + x)})
.collect();
let previous_l2_snapshot: Vec<OrderbookLevel> = old_bookside
.iter_valid(time_now, oracle_price_lots)
.map(|item| (item.node.price_data() as i64, item.node.quantity))
.group_by(|(price, _)| *price)
.into_iter()
.map(|(price, group)| OrderbookLevel { price, size: group.map(|(_, quantity)| quantity).fold(0, |acc, x| acc + x)})
.collect();
let mut update: Vec<OrderbookLevel> = vec!();
let mut update: Vec<OrderbookLevel> = vec![];
// push diff for levels that are no longer present
for previous_order in previous_l2_snapshot.iter() {
let peer = current_l2_snapshot
for previous_order in previous_bookside.iter() {
let peer = current_bookside
.iter()
.find(|level| previous_order.price == level.price);
@ -161,17 +149,17 @@ fn publish_changes(
None => {
info!("level removed {}", previous_order.price);
update.push(OrderbookLevel {
price: previous_order.price,
size: 0,
price: previous_order.price,
size: 0f64,
});
},
_ => continue
}
_ => continue,
}
}
// push diff where there's a new level or size has changed
for current_order in &current_l2_snapshot {
let peer = previous_l2_snapshot
for current_order in current_bookside {
let peer = previous_bookside
.iter()
.find(|item| item.price == current_order.price);
@ -180,9 +168,12 @@ fn publish_changes(
if previous_order.size == current_order.size {
continue;
}
debug!("size changed {} -> {}", previous_order.size, current_order.size);
debug!(
"size changed {} -> {}",
previous_order.size, current_order.size
);
update.push(current_order.clone());
},
}
None => {
debug!("new level {},{}", current_order.price, current_order.size);
update.push(current_order.clone())
@ -190,29 +181,22 @@ fn publish_changes(
}
}
match other_bookside {
match maybe_other_bookside {
Some(other_bookside) => {
let other_l2_snapshot = other_bookside
.iter_valid(time_now, oracle_price_lots)
.map(|item| (item.node.price_data() as i64, item.node.quantity))
.group_by(|(price, _)| *price)
.into_iter()
.map(|(price, group)| OrderbookLevel { price, size: group.map(|(_, quantity)| quantity).fold(0, |acc, x| acc + x)})
.collect();
let (bids, asks) = match side {
OrderbookSide::Bid => (current_l2_snapshot, other_l2_snapshot),
OrderbookSide::Ask => (other_l2_snapshot, current_l2_snapshot)
OrderbookSide::Bid => (current_bookside, other_bookside),
OrderbookSide::Ask => (other_bookside, current_bookside),
};
orderbook_update_sender
.try_send(OrderbookFilterMessage::Checkpoint(OrderbookCheckpoint {
slot,
write_version,
bids,
asks,
market: mkt.0.to_string(),
}))
.unwrap()
},
.try_send(OrderbookFilterMessage::Checkpoint(OrderbookCheckpoint {
slot,
write_version,
bids: bids.clone(),
asks: asks.clone(),
market: mkt.0.to_string(),
}))
.unwrap()
}
None => info!("other bookside not in cache"),
}
@ -230,11 +214,100 @@ fn publish_changes(
}))
.unwrap(); // TODO: use anyhow to bubble up error
metric_updates.increment();
}
fn publish_changes_serum(
slot: u64,
write_version: u64,
mkt: &(Pubkey, MarketConfig),
side: OrderbookSide,
current_bookside: &Vec<OrderbookLevel>,
previous_bookside: &Vec<OrderbookLevel>,
maybe_other_bookside: Option<&Vec<OrderbookLevel>>,
orderbook_update_sender: &async_channel::Sender<OrderbookFilterMessage>,
metric_updates: &mut MetricU64,
) {
let mut update: Vec<OrderbookLevel> = vec![];
// push diff for levels that are no longer present
for previous_order in previous_bookside.iter() {
let peer = current_bookside
.iter()
.find(|level| previous_order.price == level.price);
match peer {
None => {
info!("level removed {}", previous_order.price);
update.push(OrderbookLevel {
price: previous_order.price,
size: 0f64,
});
}
_ => continue,
}
}
// push diff where there's a new level or size has changed
for current_order in current_bookside {
let peer = previous_bookside
.iter()
.find(|item| item.price == current_order.price);
match peer {
Some(previous_order) => {
if previous_order.size == current_order.size {
continue;
}
debug!(
"size changed {} -> {}",
previous_order.size, current_order.size
);
update.push(current_order.clone());
}
None => {
debug!("new level {},{}", current_order.price, current_order.size);
update.push(current_order.clone())
}
}
}
match maybe_other_bookside {
Some(other_bookside) => {
let (bids, asks) = match side {
OrderbookSide::Bid => (current_bookside, other_bookside),
OrderbookSide::Ask => (other_bookside, current_bookside),
};
orderbook_update_sender
.try_send(OrderbookFilterMessage::Checkpoint(OrderbookCheckpoint {
slot,
write_version,
bids: bids.clone(),
asks: asks.clone(),
market: mkt.0.to_string(),
}))
.unwrap()
}
None => info!("other bookside not in cache"),
}
info!("diff {} {:?}", mkt.1.name, update);
if update.len() > 0 {
orderbook_update_sender
.try_send(OrderbookFilterMessage::Update(OrderbookUpdate {
market: mkt.0.to_string(),
side: side.clone(),
update,
slot,
write_version,
}))
.unwrap(); // TODO: use anyhow to bubble up error
metric_updates.increment();
}
}
pub async fn init(
market_configs: Vec<(Pubkey, MarketConfig)>,
serum_market_configs: Vec<(Pubkey, MarketConfig)>,
metrics_sender: Metrics,
) -> anyhow::Result<(
async_channel::Sender<AccountWrite>,
@ -261,12 +334,17 @@ pub async fn init(
let account_write_queue_receiver_c = account_write_queue_receiver.clone();
let mut chain_cache = ChainData::new();
let mut bookside_cache: HashMap<String, BookSide> = HashMap::new();
let mut bookside_cache: HashMap<String, Vec<OrderbookLevel>> = HashMap::new();
let mut serum_bookside_cache: HashMap<String, Vec<OrderbookLevel>> = HashMap::new();
let mut last_write_versions = HashMap::<String, (u64, u64)>::new();
let relevant_pubkeys = market_configs.iter().flat_map(|m| [m.1.bids, m.1.asks]).collect::<HashSet<Pubkey>>();
let relevant_pubkeys = [market_configs.clone(), serum_market_configs.clone()]
.concat()
.iter()
.flat_map(|m| [m.1.bids, m.1.asks])
.collect::<HashSet<Pubkey>>();
info!("relevant_pubkeys {:?}", relevant_pubkeys);
// update handling thread, reads both sloths and account updates
// update handling thread, reads both slots and account updates
tokio::spawn(async move {
loop {
tokio::select! {
@ -304,16 +382,8 @@ pub async fn init(
for mkt in market_configs.iter() {
for side in 0..2 {
let mkt_pk = mkt.0;
let side_pk = if side == 0 {
mkt.1.bids
} else {
mkt.1.asks
};
let other_side_pk = if side == 0 {
mkt.1.asks
} else {
mkt.1.bids
};
let side_pk = if side == 0 { mkt.1.bids } else { mkt.1.asks };
let other_side_pk = if side == 0 { mkt.1.asks } else { mkt.1.bids };
let last_write_version = last_write_versions
.get(&side_pk.to_string())
.unwrap_or(&(0, 0));
@ -332,6 +402,28 @@ pub async fn init(
let account = &account_info.account;
let bookside =
BookSide::try_deserialize(account.data().borrow_mut()).unwrap();
let side = match bookside.nodes.order_tree_type() {
OrderTreeType::Bids => OrderbookSide::Bid,
OrderTreeType::Asks => OrderbookSide::Ask,
};
let time_now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
let oracle_price_lots = 0; // todo: does this matter? where to find it?
let bookside = bookside
.iter_valid(time_now, oracle_price_lots)
.map(|item| (item.node.price_data() as i64, item.node.quantity))
.group_by(|(price, _)| *price)
.into_iter()
.map(|(price, group)| OrderbookLevel {
price: native_to_ui(price, mkt.1.quote_decimals),
size: native_to_ui(group
.map(|(_, quantity)| quantity)
.fold(0, |acc, x| acc + x), mkt.1.base_decimals),
})
.collect();
let other_bookside = bookside_cache.get(&other_side_pk.to_string());
match bookside_cache.get(&side_pk_string) {
@ -339,6 +431,7 @@ pub async fn init(
account_info.slot,
account_info.write_version,
mkt,
side,
&bookside,
&old_bookside,
other_bookside,
@ -349,11 +442,82 @@ pub async fn init(
}
bookside_cache.insert(side_pk_string.clone(), bookside.clone());
}
}
Err(_) => info!("chain_cache could not find {}", mkt_pk),
}
}
}
for mkt in serum_market_configs.iter() {
for side in 0..2 {
let side_pk = if side == 0 { mkt.1.bids } else { mkt.1.asks };
let other_side_pk = if side == 0 { mkt.1.asks } else { mkt.1.bids };
let last_write_version = last_write_versions
.get(&side_pk.to_string())
.unwrap_or(&(0, 0));
match chain_cache.account(&side_pk) {
Ok(account_info) => {
let side_pk_string = side_pk.to_string();
let write_version = (account_info.slot, account_info.write_version);
// todo: should this be <= so we don't overwrite with old data received late?
if write_version == *last_write_version {
continue;
}
last_write_versions.insert(side_pk_string.clone(), write_version);
let account = &mut account_info.account.clone();
let data = account.data_as_mut_slice();
let len = data.len();
info!("side pk {} side {}", side_pk, side);
let inner = &mut data[5..len - 7];
let slab = Slab::new(&mut inner[size_of::<OrderBookStateHeader>()..]);
let bookside: Vec<OrderbookLevel> = slab
.iter(side == 0)
.map(|item| (u64::from(item.price()) as i64, item.quantity() as i64))
.group_by(|(price, _)| *price)
.into_iter()
.map(|(price, group)| OrderbookLevel {
price: native_to_ui(price, mkt.1.quote_decimals),
size: native_to_ui(group
.map(|(_, quantity)| quantity)
.fold(0, |acc, x| acc + x), mkt.1.base_decimals),
})
.collect();
let other_bookside =
serum_bookside_cache.get(&other_side_pk.to_string());
match serum_bookside_cache.get(&side_pk_string) {
Some(old_bookside) => publish_changes_serum(
account_info.slot,
account_info.write_version,
mkt,
if side == 0 {
OrderbookSide::Bid
} else {
OrderbookSide::Ask
},
&bookside,
old_bookside,
other_bookside,
&fill_update_sender,
&mut metric_events_new,
),
_ => info!("bookside_cache could not find {}", side_pk_string),
}
serum_bookside_cache.insert(
side_pk_string.clone(),
bookside,
);
}
Err(_) => info!("chain_cache could not find {}", side_pk),
}
}
}
}
});

2
service-mango-fills/Cargo.toml
View File

@ -26,6 +26,6 @@ bytemuck = "1.7.2"
mango-v4 = { git = "ssh://git@github.com/blockworks-foundation/mango-v4", branch = "dev" }
client = { git = "ssh://git@github.com/blockworks-foundation/mango-v4", branch = "dev" }
serum_dex = { git = "ssh://git@github.com/openbook-dex/program", branch = "master" }
serum_dex = { git = "ssh://git@github.com/jup-ag/openbook-program", branch = "feat/expose-things" }
anchor-lang = "0.25.0"
anchor-client = "0.25.0"

94
service-mango-orderbook/src/main.rs
View File

@ -7,8 +7,7 @@ use client::{Client, MangoGroupContext};
use futures_channel::mpsc::{unbounded, UnboundedSender};
use futures_util::{
future::{self, Ready},
pin_mut,
SinkExt, StreamExt, TryStreamExt,
pin_mut, SinkExt, StreamExt, TryStreamExt,
};
use log::*;
use std::{
@ -27,15 +26,15 @@ use tokio::{
};
use tokio_tungstenite::tungstenite::{protocol::Message, Error};
use serde::{Deserialize};
use serde::Deserialize;
use solana_geyser_connector_lib::{
grpc_plugin_source, metrics, websocket_source, MetricsConfig, SourceConfig,
};
use solana_geyser_connector_lib::{
metrics::{MetricType, MetricU64},
orderbook_filter::{self, MarketConfig, OrderbookCheckpoint, OrderbookFilterMessage},
FilterConfig, StatusResponse,
};
use solana_geyser_connector_lib::{
grpc_plugin_source, metrics, websocket_source, MetricsConfig, SourceConfig,
};
type CheckpointMap = Arc<Mutex<HashMap<String, OrderbookCheckpoint>>>;
type PeerMap = Arc<Mutex<HashMap<SocketAddr, Peer>>>;
@ -81,7 +80,7 @@ pub struct Config {
async fn handle_connection_error(
checkpoint_map: CheckpointMap,
peer_map: PeerMap,
market_ids: Vec<String>,
market_ids: HashMap<String, String>,
raw_stream: TcpStream,
addr: SocketAddr,
metrics_opened_connections: MetricU64,
@ -109,7 +108,7 @@ async fn handle_connection_error(
async fn handle_connection(
checkpoint_map: CheckpointMap,
peer_map: PeerMap,
market_ids: Vec<String>,
market_ids: HashMap<String, String>,
raw_stream: TcpStream,
addr: SocketAddr,
) -> Result<(), Error> {
@ -153,7 +152,7 @@ fn handle_commands(
msg: Message,
peer_map: PeerMap,
checkpoint_map: CheckpointMap,
market_ids: Vec<String>,
market_ids: HashMap<String, String>,
) -> Ready<Result<(), Error>> {
let msg_str = msg.clone().into_text().unwrap();
let command: Result<Command, serde_json::Error> = serde_json::from_str(&msg_str);
@ -162,17 +161,20 @@ fn handle_commands(
match command {
Ok(Command::Subscribe(cmd)) => {
let market_id = cmd.clone().market_id;
let subscribed = peer.subscriptions.insert(market_id.clone());
if !market_ids.contains(&market_id) {
let res = StatusResponse {
success: false,
message: "market not found",
};
peer.sender
.unbounded_send(Message::Text(serde_json::to_string(&res).unwrap()))
.unwrap();
return future::ok(());
match market_ids.get(&market_id) {
None => {
let res = StatusResponse {
success: false,
message: "market not found",
};
peer.sender
.unbounded_send(Message::Text(serde_json::to_string(&res).unwrap()))
.unwrap();
return future::ok(());
}
_ => {}
}
let subscribed = peer.subscriptions.insert(market_id.clone());
let res = if subscribed {
StatusResponse {
@ -221,13 +223,13 @@ fn handle_commands(
peer.sender
.unbounded_send(Message::Text(serde_json::to_string(&res).unwrap()))
.unwrap();
},
}
Ok(Command::GetMarkets) => {
info!("getMarkets");
peer.sender
.unbounded_send(Message::Text(serde_json::to_string(&market_ids).unwrap()))
.unwrap();
},
}
Err(err) => {
info!("error deserializing user input {:?}", err);
let res = StatusResponse {
@ -287,31 +289,65 @@ async fn main() -> anyhow::Result<()> {
Pubkey::from_str(&config.mango_group).unwrap(),
client.cluster.clone(),
client.commitment,
)?);
)?);
// todo: reload markets at intervals
let market_pubkey_strings: Vec<String> = group_context
.perp_markets
.iter()
.map(|(_, context)| context.address.to_string())
.collect();
let market_configs: Vec<(Pubkey, MarketConfig)> = group_context
.perp_markets
.iter()
.map(|(_, context)| {
let quote_decimals = match group_context.tokens.get(&context.market.settle_token_index) {
Some(token) => token.decimals,
None => panic!("token not found for market") // todo: default to 6 for usdc?
};
(
context.address,
MarketConfig {
name: context.market.name().to_owned(),
bids: context.market.bids,
asks: context.market.asks,
base_decimals: context.market.base_decimals,
quote_decimals,
},
)
})
.collect();
let serum_market_configs: Vec<(Pubkey, MarketConfig)> = group_context
.serum3_markets
.iter()
.map(|(_, context)| {
let base_decimals = match group_context.tokens.get(&context.market.base_token_index) {
Some(token) => token.decimals,
None => panic!("token not found for market") // todo: default?
};
let quote_decimals = match group_context.tokens.get(&context.market.quote_token_index) {
Some(token) => token.decimals,
None => panic!("token not found for market") // todo: default to 6 for usdc?
};
(
context.address,
MarketConfig {
name: context.market.name().to_owned(),
bids: context.bids,
asks: context.asks,
base_decimals,
quote_decimals,
},
)
})
.collect();
let market_pubkey_strings: HashMap<String, String> = [market_configs.clone(), serum_market_configs.clone()]
.concat()
.iter()
.map(|market| (market.0.to_string(), market.1.name.clone()))
.collect::<Vec<(String, String)>>()
.into_iter()
.collect();
let (account_write_queue_sender, slot_queue_sender, orderbook_receiver) =
orderbook_filter::init(market_configs, metrics_tx.clone()).await?;
orderbook_filter::init(market_configs, serum_market_configs, metrics_tx.clone()).await?;
let checkpoints_ref_thread = checkpoints.clone();
let peers_ref_thread = peers.clone();
@ -381,7 +417,7 @@ async fn main() -> anyhow::Result<()> {
let filter_config = FilterConfig {
program_ids: vec![
"4MangoMjqJ2firMokCjjGgoK8d4MXcrgL7XJaL3w6fVg".into(),
//"srmqPvymJeFKQ4zGQed1GFppgkRHL9kaELCbyksJtPX".into(),
"srmqPvymJeFKQ4zGQed1GFppgkRHL9kaELCbyksJtPX".into(),
],
};
grpc_plugin_source::process_events(

8
service-mango-pnl/src/main.rs
View File

@ -288,11 +288,17 @@ async fn main() -> anyhow::Result<()> {
// start filling chain_data from the grpc plugin source
let (account_write_queue_sender, slot_queue_sender) = memory_target::init(chain_data).await?;
let filter_config = FilterConfig {
program_ids: vec![
"4MangoMjqJ2firMokCjjGgoK8d4MXcrgL7XJaL3w6fVg".into(),
],
};
grpc_plugin_source::process_events(
&config.source,
&filter_config,
account_write_queue_sender,
slot_queue_sender,
metrics_tx,
metrics_tx.clone(),
)
.await;