Sasedev/khadhroony-bobobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.7.42

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-05-24 17:17:26 +02:00
parent 6176c5d4cd
commit 69c8f6c957
25 changed files with 2354 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
kb_lib/src/db.rs
View File

@@ -140,6 +140,7 @@ pub use queries::query_db_metadatas_list;
pub use queries::query_db_metadatas_upsert;
pub use queries::query_db_runtime_events_insert;
pub use queries::query_db_runtime_events_list_recent;
pub use queries::query_dex_decoded_events_delete_by_key;
pub use queries::query_dex_decoded_events_get_by_key;
pub use queries::query_dex_decoded_events_get_latest_pump_fun_create_payload_by_mint;
pub use queries::query_dex_decoded_events_list_by_transaction_id;

1
kb_lib/src/db/queries.rs
View File

@@ -65,6 +65,7 @@ pub use db_runtime_event::query_db_runtime_events_list_recent;
pub use dex::query_dexs_get_by_code;
pub use dex::query_dexs_list;
pub use dex::query_dexs_upsert;
pub use dex_decoded_event::query_dex_decoded_events_delete_by_key;
pub use dex_decoded_event::query_dex_decoded_events_get_by_key;
pub use dex_decoded_event::query_dex_decoded_events_get_latest_pump_fun_create_payload_by_mint;
pub use dex_decoded_event::query_dex_decoded_events_list_by_transaction_id;

39
kb_lib/src/db/queries/dex_decoded_event.rs
View File

@@ -89,6 +89,45 @@ LIMIT 1
}
}
/// Deletes one decoded DEX event row by its natural key.
pub async fn query_dex_decoded_events_delete_by_key(
database: &crate::Database,
transaction_id: i64,
instruction_id: std::option::Option<i64>,
event_kind: &str,
) -> Result<u64, crate::Error> {
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result = sqlx::query(
r#"
DELETE FROM k_sol_dex_decoded_events
WHERE transaction_id = ?
AND (
(instruction_id IS NULL AND ? IS NULL)
OR instruction_id = ?
)
AND event_kind = ?
"#,
)
.bind(transaction_id)
.bind(instruction_id)
.bind(instruction_id)
.bind(event_kind)
.execute(pool)
.await;
match query_result {
Ok(result) => return Ok(result.rows_affected()),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot delete k_sol_dex_decoded_events by key on sqlite: {}",
error
)));
},
}
},
}
}
/// Reads one decoded DEX event by its natural key.
pub async fn query_dex_decoded_events_get_by_key(
database: &crate::Database,

26
kb_lib/src/db/queries/local_pipeline_diagnostics.rs
View File

@@ -584,14 +584,11 @@ ORDER BY dex_code
}
}
/// Lists observed Raydium program instruction diagnostics.
pub async fn query_local_raydium_program_instruction_diagnostic_list_summaries(
database: &crate::Database,
) -> Result<
std::vec::Vec<crate::LocalRaydiumProgramInstructionDiagnosticSummaryDto>,
crate::Error,
> {
) -> Result<std::vec::Vec<crate::LocalRaydiumProgramInstructionDiagnosticSummaryDto>, crate::Error>
{
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let rows_result = sqlx::query_as::<
@@ -2025,7 +2022,7 @@ HAVING COUNT(DISTINCT CASE
AND COUNT(DISTINCT pc.bucket_start_unix || ':' || pc.timeframe_seconds) = 0
"#
};
let sql = format!(
let mut builder = sqlx::QueryBuilder::<sqlx::Sqlite>::new(format!(
r#"
SELECT
pair.id AS pair_id,
@@ -2070,17 +2067,14 @@ GROUP BY
pair.symbol
{}
ORDER BY decoded_trade_candidate_count DESC, pair.id
LIMIT ?
"#,
LIMIT "#,
having_clause
);
let rows_result = sqlx::query_as::<
sqlx::Sqlite,
crate::db::dtos::LocalPairGapDiagnosticSampleRow,
>(sql.as_str())
.bind(limit)
.fetch_all(pool)
.await;
));
builder.push_bind(limit);
let rows_result = builder
.build_query_as::<crate::db::dtos::LocalPairGapDiagnosticSampleRow>()
.fetch_all(pool)
.await;
let rows = match rows_result {
Ok(rows) => rows,
Err(error) => {

2
kb_lib/src/db/schema.rs
View File

@@ -439,7 +439,7 @@ pub(crate) async fn ensure_schema(database: &crate::Database) -> Result<(), crat
async fn execute_sqlite_schema_statement(
pool: &sqlx::SqlitePool,
statement_name: &str,
statement_sql: &str,
statement_sql: &'static str,
) -> Result<(), crate::Error> {
let execute_result = sqlx::query(statement_sql).execute(pool).await;
match execute_result {

3
kb_lib/src/dex.rs
View File

@@ -57,6 +57,9 @@ pub use raydium_amm_v4::RaydiumAmmV4Decoder;
pub use raydium_amm_v4::RaydiumAmmV4Initialize2PoolDecoded;
pub use raydium_amm_v4::RaydiumAmmV4SwapDecoded;
pub use raydium_clmm::RaydiumClmmDecodedEvent;
pub use raydium_clmm::RaydiumClmmDecodedInstructionEvent;
pub use raydium_clmm::RaydiumClmmDecoder;
pub use raydium_clmm::RaydiumClmmSwapLegacyDecoded;
pub use raydium_clmm::RaydiumClmmSwapV2Decoded;
pub use raydium_clmm::decode_raydium_clmm_instruction;
pub use raydium_cpmm::RaydiumCpmmDecodedEvent;

6
kb_lib/src/dex/orca_whirlpools.rs
View File

@@ -219,7 +219,11 @@ impl OrcaWhirlpoolsDecoder {
"poolAccount": pool_account,
"tokenAMint": token_a_mint,
"tokenBMint": token_b_mint,
"tradeSide": format!("{:?}", trade_side)
"tradeSide": format!("{:?}", trade_side),
"tradeCandidate": false,
"candleCandidate": false,
"skipTradeReason": "orca_whirlpools_swap_amount_payload_not_resolved",
"skipCandleReason": "orca_whirlpools_swap_amount_payload_not_resolved"
});
decoded_events.push(crate::OrcaWhirlpoolsDecodedEvent::Swap(
crate::OrcaWhirlpoolsSwapDecoded {

689
kb_lib/src/dex/raydium_clmm.rs
View File

@@ -6,9 +6,13 @@ const RAYDIUM_CLMM_SWAP_V2_DISCRIMINATOR: [u8; 8] = [43, 4, 237, 11, 26, 201, 30
const RAYDIUM_CLMM_SWAP_LEGACY_DISCRIMINATOR: [u8; 8] = [248, 198, 158, 145, 225, 117, 135, 200];
const OBSERVED_JUPITER_V6_PROGRAM_ID: &str = "JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4";
/// Decoded Raydium CLMM event.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
pub enum RaydiumClmmDecodedEvent {
/// Raydium CLMM legacy swap event.
Swap(crate::RaydiumClmmSwapLegacyDecoded),
/// Raydium CLMM swap_v2 event.
SwapV2(crate::RaydiumClmmSwapV2Decoded),
}
@@ -17,6 +21,7 @@ impl RaydiumClmmDecodedEvent {
/// Returns the normalized event kind.
pub fn event_kind(&self) -> &'static str {
match self {
crate::RaydiumClmmDecodedEvent::Swap(_) => return "raydium_clmm.swap",
crate::RaydiumClmmDecodedEvent::SwapV2(_) => return "raydium_clmm.swap_v2",
}
}
@@ -24,6 +29,7 @@ impl RaydiumClmmDecodedEvent {
/// Returns the pool account.
pub fn pool_account(&self) -> &str {
match self {
crate::RaydiumClmmDecodedEvent::Swap(event) => return event.pool_state.as_str(),
crate::RaydiumClmmDecodedEvent::SwapV2(event) => return event.pool_state.as_str(),
}
}
@@ -31,6 +37,7 @@ impl RaydiumClmmDecodedEvent {
/// Returns the normalized base mint.
pub fn base_mint(&self) -> &str {
match self {
crate::RaydiumClmmDecodedEvent::Swap(event) => return event.base_mint.as_str(),
crate::RaydiumClmmDecodedEvent::SwapV2(event) => return event.base_mint.as_str(),
}
}
@@ -38,6 +45,7 @@ impl RaydiumClmmDecodedEvent {
/// Returns the normalized quote mint.
pub fn quote_mint(&self) -> &str {
match self {
crate::RaydiumClmmDecodedEvent::Swap(event) => return event.quote_mint.as_str(),
crate::RaydiumClmmDecodedEvent::SwapV2(event) => return event.quote_mint.as_str(),
}
}
@@ -45,6 +53,13 @@ impl RaydiumClmmDecodedEvent {
/// Converts the decoded event to JSON payload.
pub fn to_payload_json(&self) -> std::option::Option<std::string::String> {
match self {
crate::RaydiumClmmDecodedEvent::Swap(event) => {
let result = serde_json::to_string(event);
match result {
Ok(payload_json) => return Some(payload_json),
Err(_) => return None,
}
},
crate::RaydiumClmmDecodedEvent::SwapV2(event) => {
let result = serde_json::to_string(event);
match result {
@@ -56,6 +71,71 @@ impl RaydiumClmmDecodedEvent {
}
}
/// One decoded Raydium CLMM instruction associated with its projected instruction id.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
pub struct RaydiumClmmDecodedInstructionEvent {
/// Projected chain instruction id.
pub instruction_id: i64,
/// Decoded Raydium CLMM event.
pub decoded_event: crate::RaydiumClmmDecodedEvent,
}
/// Decoded Raydium CLMM legacy swap instruction.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
pub struct RaydiumClmmSwapLegacyDecoded {
/// User performing the swap.
pub payer: std::string::String,
/// AMM config account.
pub amm_config: std::string::String,
/// CLMM pool state account.
pub pool_state: std::string::String,
/// User input token account.
pub input_token_account: std::string::String,
/// User output token account.
pub output_token_account: std::string::String,
/// Pool input vault.
pub input_vault: std::string::String,
/// Pool output vault.
pub output_vault: std::string::String,
/// Pool oracle observation state.
pub observation_state: std::string::String,
/// Input vault mint inferred from transaction token balances.
pub input_vault_mint: std::string::String,
/// Output vault mint inferred from transaction token balances.
pub output_vault_mint: std::string::String,
/// Canonical base mint.
pub base_mint: std::string::String,
/// Canonical quote mint.
pub quote_mint: std::string::String,
/// Canonical base vault.
pub base_vault: std::string::String,
/// Canonical quote vault.
pub quote_vault: std::string::String,
/// Trade side relative to the canonical base mint.
#[serde(rename = "tradeSide")]
pub trade_side: std::string::String,
/// Optional raw base amount inferred from vault balance deltas.
#[serde(rename = "baseAmountRaw")]
pub base_amount_raw: std::option::Option<std::string::String>,
/// Optional raw quote amount inferred from vault balance deltas.
#[serde(rename = "quoteAmountRaw")]
pub quote_amount_raw: std::option::Option<std::string::String>,
/// Amount argument.
pub amount: u64,
/// Other amount threshold argument.
pub other_amount_threshold: u64,
/// Sqrt price limit as decimal string.
pub sqrt_price_limit_x64: std::string::String,
/// Whether the instruction uses exact input mode.
pub is_base_input: bool,
/// Whether this decoded event comes from the legacy CLMM swap discriminator.
#[serde(rename = "legacyInstruction")]
pub legacy_instruction: bool,
/// Optional top-level caller inferred from parent instruction.
#[serde(rename = "routeSource")]
pub route_source: std::option::Option<std::string::String>,
}
/// Decoded Raydium CLMM swap_v2 instruction.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
pub struct RaydiumClmmSwapV2Decoded {
@@ -100,6 +180,80 @@ pub struct RaydiumClmmSwapV2Decoded {
pub is_base_input: bool,
}
/// Raydium CLMM transaction decoder.
#[derive(Clone, Debug, Default)]
pub struct RaydiumClmmDecoder;
impl RaydiumClmmDecoder {
/// Creates a new Raydium CLMM decoder.
pub fn new() -> Self {
return Self;
}
/// Decodes one projected transaction into Raydium CLMM instruction events.
pub fn decode_transaction(
&self,
transaction: &crate::ChainTransactionDto,
instructions: &[crate::ChainInstructionDto],
) -> Result<std::vec::Vec<crate::RaydiumClmmDecodedInstructionEvent>, crate::Error> {
let transaction_json_result =
serde_json::from_str::<serde_json::Value>(transaction.transaction_json.as_str());
let transaction_json = match transaction_json_result {
Ok(transaction_json) => transaction_json,
Err(error) => {
return Err(crate::Error::Json(format!(
"cannot parse transaction_json for signature '{}': {}",
transaction.signature, error
)));
},
};
let meta_value_result = parse_transaction_meta_value(transaction, &transaction_json);
let meta_value = match meta_value_result {
Ok(meta_value) => meta_value,
Err(error) => return Err(error),
};
let account_keys = extract_transaction_account_keys(&transaction_json, meta_value.as_ref());
let token_balances =
extract_token_balance_records(meta_value.as_ref(), account_keys.as_slice());
let mut decoded = std::vec::Vec::new();
for instruction in instructions {
let program_id = match instruction.program_id.as_ref() {
Some(program_id) => program_id,
None => continue,
};
if program_id.as_str() != crate::RAYDIUM_CLMM_PROGRAM_ID {
continue;
}
let instruction_id = match instruction.id {
Some(instruction_id) => instruction_id,
None => continue,
};
let data_json = match instruction.data_json.as_ref() {
Some(data_json) => data_json,
None => continue,
};
let decoded_events_result = decode_raydium_clmm_instruction_with_token_balances(
instruction.accounts_json.as_str(),
data_json.as_str(),
instructions,
instruction,
token_balances.as_slice(),
);
let decoded_events = match decoded_events_result {
Ok(decoded_events) => decoded_events,
Err(error) => return Err(error),
};
for decoded_event in decoded_events {
decoded.push(crate::RaydiumClmmDecodedInstructionEvent {
instruction_id,
decoded_event,
});
}
}
return Ok(decoded);
}
}
/// Decodes a Raydium CLMM instruction.
pub fn decode_raydium_clmm_instruction(
accounts_json: &str,
@@ -144,6 +298,200 @@ pub fn decode_raydium_clmm_instruction(
return decoded;
}
fn decode_raydium_clmm_instruction_with_token_balances(
accounts_json: &str,
data_json: &str,
transaction_instructions: &[crate::ChainInstructionDto],
instruction: &crate::ChainInstructionDto,
token_balances: &[TokenBalanceRecord],
) -> Result<std::vec::Vec<crate::RaydiumClmmDecodedEvent>, crate::Error> {
let mut decoded = std::vec::Vec::new();
let accounts_result = serde_json::from_str::<std::vec::Vec<std::string::String>>(accounts_json);
let accounts = match accounts_result {
Ok(accounts) => accounts,
Err(error) => {
return Err(crate::Error::Json(format!(
"cannot parse raydium clmm accounts json: {}",
error
)));
},
};
let data_base58_result = serde_json::from_str::<std::string::String>(data_json);
let data_base58 = match data_base58_result {
Ok(data_base58) => data_base58,
Err(_) => data_json.to_string(),
};
let data_option = decode_base58(data_base58.as_str());
let data = match data_option {
Some(data) => data,
None => return Ok(decoded),
};
if data.len() < 41 {
return Ok(decoded);
}
let discriminator_option = read_discriminator(data.as_slice());
let discriminator = match discriminator_option {
Some(discriminator) => discriminator,
None => return Ok(decoded),
};
if discriminator == RAYDIUM_CLMM_SWAP_V2_DISCRIMINATOR {
let event_option = decode_swap_v2(accounts.as_slice(), data.as_slice());
if let Some(event) = event_option {
decoded.push(crate::RaydiumClmmDecodedEvent::SwapV2(event));
}
return Ok(decoded);
}
if discriminator == RAYDIUM_CLMM_SWAP_LEGACY_DISCRIMINATOR {
let event_option = decode_swap_legacy(
accounts.as_slice(),
data.as_slice(),
transaction_instructions,
instruction,
token_balances,
);
if let Some(event) = event_option {
decoded.push(crate::RaydiumClmmDecodedEvent::Swap(event));
}
return Ok(decoded);
}
return Ok(decoded);
}
fn decode_swap_legacy(
accounts: &[std::string::String],
data: &[u8],
transaction_instructions: &[crate::ChainInstructionDto],
instruction: &crate::ChainInstructionDto,
token_balances: &[TokenBalanceRecord],
) -> std::option::Option<crate::RaydiumClmmSwapLegacyDecoded> {
let payer = match clone_account(accounts, 0) {
Some(value) => value,
None => return None,
};
let amm_config = match clone_account(accounts, 1) {
Some(value) => value,
None => return None,
};
let pool_state = match clone_account(accounts, 2) {
Some(value) => value,
None => return None,
};
let input_token_account = match clone_account(accounts, 3) {
Some(value) => value,
None => return None,
};
let output_token_account = match clone_account(accounts, 4) {
Some(value) => value,
None => return None,
};
let input_vault = match clone_account(accounts, 5) {
Some(value) => value,
None => return None,
};
let output_vault = match clone_account(accounts, 6) {
Some(value) => value,
None => return None,
};
let observation_state = match clone_account(accounts, 7) {
Some(value) => value,
None => return None,
};
let input_vault_mint = match mint_for_account(token_balances, input_vault.as_str()) {
Some(value) => value,
None => return None,
};
let output_vault_mint = match mint_for_account(token_balances, output_vault.as_str()) {
Some(value) => value,
None => return None,
};
if input_vault_mint == output_vault_mint {
return None;
}
let amount = match read_u64_le(data, 8) {
Some(value) => value,
None => return None,
};
let other_amount_threshold = match read_u64_le(data, 16) {
Some(value) => value,
None => return None,
};
let sqrt_price_limit_x64 = match read_u128_le(data, 24) {
Some(value) => value,
None => return None,
};
let is_base_input = match read_bool(data, 40) {
Some(value) => value,
None => return None,
};
let mut base_mint = input_vault_mint.clone();
let mut quote_mint = output_vault_mint.clone();
let mut base_vault = input_vault.clone();
let mut quote_vault = output_vault.clone();
if is_quote_mint(input_vault_mint.as_str()) && !is_quote_mint(output_vault_mint.as_str()) {
base_mint = output_vault_mint.clone();
quote_mint = input_vault_mint.clone();
base_vault = output_vault.clone();
quote_vault = input_vault.clone();
} else if !is_quote_mint(input_vault_mint.as_str()) && is_quote_mint(output_vault_mint.as_str())
{
base_mint = input_vault_mint.clone();
quote_mint = output_vault_mint.clone();
base_vault = input_vault.clone();
quote_vault = output_vault.clone();
} else if output_vault_mint.as_str() < input_vault_mint.as_str() {
base_mint = output_vault_mint.clone();
quote_mint = input_vault_mint.clone();
base_vault = output_vault.clone();
quote_vault = input_vault.clone();
}
let base_amount_raw = amount_delta_abs_for_account(token_balances, base_vault.as_str());
let quote_amount_raw = amount_delta_abs_for_account(token_balances, quote_vault.as_str());
if base_amount_raw.is_none() || quote_amount_raw.is_none() {
return None;
}
let trade_side = match infer_trade_side_from_vault_deltas(
token_balances,
base_vault.as_str(),
quote_vault.as_str(),
) {
Some(trade_side) => trade_side,
None => {
if base_vault == input_vault {
"SellBase".to_string()
} else {
"BuyBase".to_string()
}
},
};
let parent_program = parent_program_id_for_instruction(instruction, transaction_instructions);
let route_source = route_source_from_parent(parent_program.as_deref());
return Some(crate::RaydiumClmmSwapLegacyDecoded {
payer,
amm_config,
pool_state,
input_token_account,
output_token_account,
input_vault,
output_vault,
observation_state,
input_vault_mint,
output_vault_mint,
base_mint,
quote_mint,
base_vault,
quote_vault,
trade_side,
base_amount_raw,
quote_amount_raw,
amount,
other_amount_threshold,
sqrt_price_limit_x64: sqrt_price_limit_x64.to_string(),
is_base_input,
legacy_instruction: true,
route_source,
});
}
fn decode_swap_v2(
accounts: &[std::string::String],
data: &[u8],
@@ -307,6 +655,346 @@ fn decode_base58(input: &str) -> std::option::Option<std::vec::Vec<u8>> {
}
}
#[derive(Debug, Clone)]
struct TokenBalanceRecord {
account_address: std::option::Option<std::string::String>,
mint: std::string::String,
pre_amount_raw: std::option::Option<std::string::String>,
post_amount_raw: std::option::Option<std::string::String>,
}
#[derive(Debug, Clone)]
struct TokenBalanceAccumulator {
account_index: std::option::Option<i64>,
account_address: std::option::Option<std::string::String>,
mint: std::string::String,
pre_amount_raw: std::option::Option<std::string::String>,
post_amount_raw: std::option::Option<std::string::String>,
}
#[derive(Debug, Clone)]
struct AccountKeyInfo {
index: i64,
address: std::string::String,
}
fn parse_transaction_meta_value(
transaction: &crate::ChainTransactionDto,
transaction_json: &serde_json::Value,
) -> Result<std::option::Option<serde_json::Value>, crate::Error> {
if let Some(meta_json) = transaction.meta_json.as_deref() {
let meta_result = serde_json::from_str::<serde_json::Value>(meta_json);
match meta_result {
Ok(meta) => return Ok(Some(meta)),
Err(error) => {
return Err(crate::Error::Json(format!(
"cannot parse meta_json for signature '{}': {}",
transaction.signature, error
)));
},
}
}
let meta = transaction_json.get("meta");
match meta {
Some(meta) => return Ok(Some(meta.clone())),
None => return Ok(None),
}
}
fn extract_transaction_account_keys(
transaction: &serde_json::Value,
meta: std::option::Option<&serde_json::Value>,
) -> std::vec::Vec<AccountKeyInfo> {
let mut account_keys = std::vec::Vec::new();
let values = transaction
.get("transaction")
.and_then(|value| value.get("message"))
.and_then(|value| value.get("accountKeys"))
.and_then(serde_json::Value::as_array);
if let Some(values) = values {
let mut index = 0usize;
for value in values {
let parsed = parse_account_key_info(value, index as i64);
if let Some(parsed) = parsed {
account_keys.push(parsed);
}
index += 1;
}
}
append_loaded_addresses(&mut account_keys, meta, "writable");
append_loaded_addresses(&mut account_keys, meta, "readonly");
return account_keys;
}
fn parse_account_key_info(
value: &serde_json::Value,
index: i64,
) -> std::option::Option<AccountKeyInfo> {
if let Some(address) = value.as_str() {
return Some(AccountKeyInfo { index, address: address.to_string() });
}
let address = match value.get("pubkey").and_then(serde_json::Value::as_str) {
Some(address) => address.to_string(),
None => return None,
};
return Some(AccountKeyInfo { index, address });
}
fn append_loaded_addresses(
account_keys: &mut std::vec::Vec<AccountKeyInfo>,
meta: std::option::Option<&serde_json::Value>,
key: &str,
) {
let addresses = meta
.and_then(|value| value.get("loadedAddresses"))
.and_then(|value| value.get(key))
.and_then(serde_json::Value::as_array);
let addresses = match addresses {
Some(addresses) => addresses,
None => return,
};
for value in addresses {
let address = match value.as_str() {
Some(address) => address,
None => continue,
};
let index = account_keys.len() as i64;
account_keys.push(AccountKeyInfo { index, address: address.to_string() });
}
}
fn extract_token_balance_records(
meta: std::option::Option<&serde_json::Value>,
account_keys: &[AccountKeyInfo],
) -> std::vec::Vec<TokenBalanceRecord> {
let mut accumulators = std::vec::Vec::new();
collect_token_balance_side(meta, account_keys, "preTokenBalances", true, &mut accumulators);
collect_token_balance_side(meta, account_keys, "postTokenBalances", false, &mut accumulators);
let mut records = std::vec::Vec::new();
for accumulator in accumulators {
records.push(TokenBalanceRecord {
account_address: accumulator.account_address,
mint: accumulator.mint,
pre_amount_raw: accumulator.pre_amount_raw,
post_amount_raw: accumulator.post_amount_raw,
});
}
return records;
}
fn collect_token_balance_side(
meta: std::option::Option<&serde_json::Value>,
account_keys: &[AccountKeyInfo],
key: &str,
is_pre: bool,
accumulators: &mut std::vec::Vec<TokenBalanceAccumulator>,
) {
let values = meta.and_then(|value| value.get(key)).and_then(serde_json::Value::as_array);
let values = match values {
Some(values) => values,
None => return,
};
for value in values {
let account_index = value.get("accountIndex").and_then(serde_json::Value::as_i64);
let mint = match value.get("mint").and_then(serde_json::Value::as_str) {
Some(mint) => mint.to_string(),
None => continue,
};
let amount = value
.get("uiTokenAmount")
.and_then(|amount| amount.get("amount"))
.and_then(serde_json::Value::as_str)
.map(|text| text.to_string());
let account_address = match account_index {
Some(account_index) => account_address_by_index(account_keys, account_index),
None => None,
};
let accumulator_index =
find_token_balance_accumulator(accumulators.as_slice(), account_index, mint.as_str());
let index = match accumulator_index {
Some(index) => index,
None => {
accumulators.push(TokenBalanceAccumulator {
account_index,
account_address,
mint,
pre_amount_raw: None,
post_amount_raw: None,
});
accumulators.len() - 1
},
};
if is_pre {
accumulators[index].pre_amount_raw = amount;
} else {
accumulators[index].post_amount_raw = amount;
}
}
}
fn find_token_balance_accumulator(
accumulators: &[TokenBalanceAccumulator],
account_index: std::option::Option<i64>,
mint: &str,
) -> std::option::Option<usize> {
let mut index = 0usize;
while index < accumulators.len() {
let accumulator = &accumulators[index];
if accumulator.account_index == account_index && accumulator.mint == mint {
return Some(index);
}
index += 1;
}
return None;
}
fn account_address_by_index(
account_keys: &[AccountKeyInfo],
account_index: i64,
) -> std::option::Option<std::string::String> {
for account_key in account_keys {
if account_key.index == account_index {
return Some(account_key.address.clone());
}
}
return None;
}
impl TokenBalanceRecord {
fn delta_raw(&self) -> std::option::Option<i128> {
let pre = parse_i128_or_zero(self.pre_amount_raw.as_deref());
let post = parse_i128_or_zero(self.post_amount_raw.as_deref());
match (pre, post) {
(Some(pre), Some(post)) => return Some(post - pre),
_ => return None,
}
}
}
fn mint_for_account(
token_balances: &[TokenBalanceRecord],
account: &str,
) -> std::option::Option<std::string::String> {
let record = token_balance_record_for_account(token_balances, account);
match record {
Some(record) => return Some(record.mint.clone()),
None => return None,
}
}
fn token_balance_record_for_account<'a>(
token_balances: &'a [TokenBalanceRecord],
account: &str,
) -> std::option::Option<&'a TokenBalanceRecord> {
for record in token_balances {
if record.account_address.as_deref() == Some(account) {
return Some(record);
}
}
return None;
}
fn amount_delta_abs_for_account(
token_balances: &[TokenBalanceRecord],
account: &str,
) -> std::option::Option<std::string::String> {
let record = token_balance_record_for_account(token_balances, account);
let record = match record {
Some(record) => record,
None => return None,
};
let delta = record.delta_raw();
let delta = match delta {
Some(delta) => delta,
None => return None,
};
if delta < 0 {
return Some((-delta).to_string());
}
return Some(delta.to_string());
}
fn infer_trade_side_from_vault_deltas(
token_balances: &[TokenBalanceRecord],
base_vault: &str,
quote_vault: &str,
) -> std::option::Option<std::string::String> {
let base_record = token_balance_record_for_account(token_balances, base_vault);
let base_record = match base_record {
Some(base_record) => base_record,
None => return None,
};
let quote_record = token_balance_record_for_account(token_balances, quote_vault);
let quote_record = match quote_record {
Some(quote_record) => quote_record,
None => return None,
};
let base_delta = base_record.delta_raw();
let quote_delta = quote_record.delta_raw();
match (base_delta, quote_delta) {
(Some(base_delta), Some(quote_delta)) => {
if base_delta < 0 && quote_delta > 0 {
return Some("BuyBase".to_string());
}
if base_delta > 0 && quote_delta < 0 {
return Some("SellBase".to_string());
}
return None;
},
_ => return None,
}
}
fn is_quote_mint(mint: &str) -> bool {
return mint == crate::WSOL_MINT_ID
|| mint == crate::USDC_MINT_ID
|| mint == crate::USDT_MINT_ID;
}
fn parent_program_id_for_instruction(
instruction: &crate::ChainInstructionDto,
instructions: &[crate::ChainInstructionDto],
) -> std::option::Option<std::string::String> {
let parent_id = match instruction.parent_instruction_id {
Some(parent_id) => parent_id,
None => return None,
};
for candidate in instructions {
if candidate.id == Some(parent_id) {
return candidate.program_id.clone();
}
}
return None;
}
fn route_source_from_parent(
parent_program: std::option::Option<&str>,
) -> std::option::Option<std::string::String> {
let parent_program = match parent_program {
Some(parent_program) => parent_program,
None => return None,
};
if parent_program == OBSERVED_JUPITER_V6_PROGRAM_ID {
return Some("jupiter".to_string());
}
return Some(parent_program.to_string());
}
fn parse_i128_or_zero(value: std::option::Option<&str>) -> std::option::Option<i128> {
let value = match value {
Some(value) => value.trim(),
None => return Some(0),
};
if value.is_empty() {
return Some(0);
}
let parsed_result = value.parse::<i128>();
match parsed_result {
Ok(parsed) => return Some(parsed),
Err(_) => return None,
}
}
#[cfg(test)]
mod tests {
fn sample_swap_v2_accounts_json() -> &'static str {
@@ -358,6 +1046,7 @@ mod tests {
assert_eq!(event.sqrt_price_limit_x64, "0");
assert!(event.is_base_input);
},
crate::RaydiumClmmDecodedEvent::Swap(_) => panic!("expected swap_v2 event"),
}
}

760
kb_lib/src/dex_decode.rs
View File

@@ -8,6 +8,7 @@ pub struct DexDecodeService {
database: std::sync::Arc<crate::Database>,
persistence: crate::DetectionPersistenceService,
raydium_amm_v4_decoder: crate::RaydiumAmmV4Decoder,
raydium_clmm_decoder: crate::RaydiumClmmDecoder,
pump_fun_decoder: crate::PumpFunDecoder,
pump_swap_decoder: crate::PumpSwapDecoder,
orca_whirlpools_decoder: crate::OrcaWhirlpoolsDecoder,
@@ -27,6 +28,7 @@ impl DexDecodeService {
database,
persistence,
raydium_amm_v4_decoder: crate::RaydiumAmmV4Decoder::new(),
raydium_clmm_decoder: crate::RaydiumClmmDecoder::new(),
pump_fun_decoder: crate::PumpFunDecoder::new(),
pump_swap_decoder: crate::PumpSwapDecoder::new(),
orca_whirlpools_decoder: crate::OrcaWhirlpoolsDecoder::new(),
@@ -77,6 +79,14 @@ impl DexDecodeService {
if let Err(error) = append_result {
return Err(error);
}
let append_result = append_persisted_events_result(
&mut persisted,
self.preserve_unmatched_raydium_instruction_audits(&transaction, &instructions)
.await,
);
if let Err(error) = append_result {
return Err(error);
}
let append_result = append_persisted_events_result(
&mut persisted,
self.decode_and_persist_pump_fun_events(&transaction, &instructions).await,
@@ -181,8 +191,52 @@ impl DexDecodeService {
signal_kind: format!("signal.dex.{event_kind}"),
missing_after_upsert_message: "decoded event disappeared after upsert".to_string(),
};
return crate::dex_decoded_event_materialization::materialize_dex_decoded_event(input)
let materialized_result =
crate::dex_decoded_event_materialization::materialize_dex_decoded_event(input).await;
let materialized = match materialized_result {
Ok(materialized) => materialized,
Err(error) => return Err(error),
};
let cleanup_result = self
.delete_replaced_raydium_instruction_audit(
transaction_id,
instruction_id,
protocol_name,
event_kind,
)
.await;
if let Err(error) = cleanup_result {
return Err(error);
}
return Ok(materialized);
}
async fn delete_replaced_raydium_instruction_audit(
&self,
transaction_id: i64,
instruction_id: i64,
protocol_name: &str,
event_kind: &str,
) -> Result<(), crate::Error> {
if event_kind.ends_with(".instruction_audit") {
return Ok(());
}
let audit_event_kind = match raydium_instruction_audit_event_kind_by_protocol(protocol_name)
{
Some(audit_event_kind) => audit_event_kind,
None => return Ok(()),
};
let delete_result = crate::query_dex_decoded_events_delete_by_key(
self.database.as_ref(),
transaction_id,
Some(instruction_id),
audit_event_kind,
)
.await;
match delete_result {
Ok(_) => return Ok(()),
Err(error) => return Err(error),
}
}
async fn persist_dexlab_event(
@@ -586,7 +640,7 @@ impl DexDecodeService {
async fn persist_raydium_clmm_event(
&self,
transaction: &crate::ChainTransactionDto,
instruction: &crate::ChainInstructionDto,
instruction_id: i64,
decoded_event: &crate::RaydiumClmmDecodedEvent,
) -> Result<crate::DexDecodedEventDto, crate::Error> {
let transaction_id = match transaction.id {
@@ -598,15 +652,6 @@ impl DexDecodeService {
)));
},
};
let instruction_id = match instruction.id {
Some(instruction_id) => instruction_id,
None => {
return Err(crate::Error::InvalidState(format!(
"raydium clmm instruction for transaction '{}' has no internal id",
transaction.signature
)));
},
};
let event_kind = decoded_event.event_kind().to_string();
let raw_payload_json = match decoded_event.to_payload_json() {
Some(payload_json) => payload_json,
@@ -889,32 +934,26 @@ impl DexDecodeService {
transaction: &crate::ChainTransactionDto,
instructions: &[crate::ChainInstructionDto],
) -> Result<std::vec::Vec<crate::DexDecodedEventDto>, crate::Error> {
let decoded_result =
self.raydium_clmm_decoder.decode_transaction(transaction, instructions);
let decoded_events = match decoded_result {
Ok(decoded_events) => decoded_events,
Err(error) => return Err(error),
};
let mut persisted = std::vec::Vec::new();
for instruction in instructions {
let program_id = match instruction.program_id.as_ref() {
Some(program_id) => program_id,
None => continue,
for decoded_event in &decoded_events {
let persist_result = self
.persist_raydium_clmm_event(
transaction,
decoded_event.instruction_id,
&decoded_event.decoded_event,
)
.await;
let persisted_event = match persist_result {
Ok(persisted_event) => persisted_event,
Err(error) => return Err(error),
};
if program_id.as_str() != crate::RAYDIUM_CLMM_PROGRAM_ID {
continue;
}
let data_json = match instruction.data_json.as_ref() {
Some(data_json) => data_json,
None => continue,
};
let decoded_events = crate::decode_raydium_clmm_instruction(
instruction.accounts_json.as_str(),
data_json.as_str(),
);
for decoded_event in &decoded_events {
let persist_result =
self.persist_raydium_clmm_event(transaction, instruction, decoded_event).await;
let persisted_event = match persist_result {
Ok(persisted_event) => persisted_event,
Err(error) => return Err(error),
};
persisted.push(persisted_event);
}
persisted.push(persisted_event);
}
return Ok(persisted);
}
@@ -943,6 +982,129 @@ impl DexDecodeService {
return Ok(persisted);
}
async fn preserve_unmatched_raydium_instruction_audits(
&self,
transaction: &crate::ChainTransactionDto,
instructions: &[crate::ChainInstructionDto],
) -> Result<std::vec::Vec<crate::DexDecodedEventDto>, crate::Error> {
let transaction_id = match transaction.id {
Some(transaction_id) => transaction_id,
None => {
return Err(crate::Error::InvalidState(format!(
"transaction '{}' has no internal id",
transaction.signature
)));
},
};
let decoded_events_result = crate::query_dex_decoded_events_list_by_transaction_id(
self.database.as_ref(),
transaction_id,
)
.await;
let decoded_events = match decoded_events_result {
Ok(decoded_events) => decoded_events,
Err(error) => return Err(error),
};
let mut decoded_instruction_ids = std::collections::HashSet::<i64>::new();
for decoded_event in &decoded_events {
if !decoded_event.protocol_name.starts_with("raydium_") {
continue;
}
if decoded_event.event_kind.ends_with(".instruction_audit") {
continue;
}
let instruction_id = match decoded_event.instruction_id {
Some(instruction_id) => instruction_id,
None => continue,
};
decoded_instruction_ids.insert(instruction_id);
}
let mut persisted = std::vec::Vec::new();
for instruction in instructions {
let program_id = match instruction.program_id.as_ref() {
Some(program_id) => program_id,
None => continue,
};
let audit_spec = match raydium_instruction_audit_spec(program_id.as_str()) {
Some(audit_spec) => audit_spec,
None => continue,
};
let instruction_id = match instruction.id {
Some(instruction_id) => instruction_id,
None => continue,
};
if decoded_instruction_ids.contains(&instruction_id) {
continue;
}
let accounts = parse_instruction_accounts_vec(instruction.accounts_json.as_str());
let data_base58 = parse_instruction_data_base58(instruction.data_json.as_deref());
let discriminator_hex = discriminator_hex_from_base58(data_base58.as_deref());
let mapped_spec = raydium_mapped_non_trade_instruction_spec(
audit_spec.protocol_name,
discriminator_hex.as_deref(),
accounts.len(),
);
let event_kind = match mapped_spec {
Some(mapped_spec) => mapped_spec.event_kind,
None => audit_spec.event_kind,
};
let mut payload = build_raydium_instruction_audit_payload(
transaction,
instruction,
audit_spec.protocol_name,
event_kind,
program_id.as_str(),
);
if let Some(mapped_spec) = mapped_spec {
payload = enrich_raydium_mapped_non_trade_payload(
payload,
mapped_spec,
data_base58.as_deref(),
);
}
let pool_account = candidate_raydium_mapped_pool_account(
mapped_spec,
accounts.as_slice(),
audit_spec.protocol_name,
instruction.accounts_json.as_str(),
);
let token_a_mint = candidate_raydium_mapped_account(
mapped_spec.and_then(|spec| spec.token_a_mint_index),
accounts.as_slice(),
);
let token_b_mint = candidate_raydium_mapped_account(
mapped_spec.and_then(|spec| spec.token_b_mint_index),
accounts.as_slice(),
);
let lp_mint = candidate_raydium_mapped_account(
mapped_spec.and_then(|spec| spec.lp_mint_index),
accounts.as_slice(),
);
let persist_result = self
.materialize_named_dex_event(
transaction,
transaction_id,
instruction_id,
audit_spec.protocol_name,
program_id.clone(),
event_kind,
pool_account,
None,
token_a_mint,
token_b_mint,
lp_mint,
payload,
)
.await;
let persisted_event = match persist_result {
Ok(persisted_event) => persisted_event,
Err(error) => return Err(error),
};
persisted.push(persisted_event);
}
return Ok(persisted);
}
async fn decode_and_persist_pump_fun_events(
&self,
transaction: &crate::ChainTransactionDto,
@@ -1150,6 +1312,453 @@ impl DexDecodeService {
}
}
struct RaydiumInstructionAuditSpec {
protocol_name: &'static str,
event_kind: &'static str,
candidate_pool_account_index: usize,
}
#[derive(Clone, Copy)]
struct RaydiumMappedNonTradeInstructionSpec {
instruction_name: &'static str,
event_kind: &'static str,
pool_account_index: std::option::Option<usize>,
token_a_mint_index: std::option::Option<usize>,
token_b_mint_index: std::option::Option<usize>,
lp_mint_index: std::option::Option<usize>,
amount_layout: RaydiumMappedNonTradeAmountLayout,
}
#[derive(Clone, Copy)]
enum RaydiumMappedNonTradeAmountLayout {
None,
ClmmLiquidityV2,
CpmmWithdraw,
}
fn raydium_instruction_audit_spec(
program_id: &str,
) -> std::option::Option<RaydiumInstructionAuditSpec> {
if program_id == crate::RAYDIUM_AMM_V4_PROGRAM_ID {
return Some(RaydiumInstructionAuditSpec {
protocol_name: "raydium_amm_v4",
event_kind: "raydium_amm_v4.instruction_audit",
candidate_pool_account_index: 1,
});
}
if program_id == crate::RAYDIUM_CLMM_PROGRAM_ID {
return Some(RaydiumInstructionAuditSpec {
protocol_name: "raydium_clmm",
event_kind: "raydium_clmm.instruction_audit",
candidate_pool_account_index: 2,
});
}
if program_id == crate::RAYDIUM_CPMM_PROGRAM_ID {
return Some(RaydiumInstructionAuditSpec {
protocol_name: "raydium_cpmm",
event_kind: "raydium_cpmm.instruction_audit",
candidate_pool_account_index: 3,
});
}
return None;
}
fn raydium_mapped_non_trade_instruction_spec(
protocol_name: &str,
discriminator_hex: std::option::Option<&str>,
account_count: usize,
) -> std::option::Option<RaydiumMappedNonTradeInstructionSpec> {
let discriminator_hex = match discriminator_hex {
Some(discriminator_hex) => discriminator_hex,
None => return None,
};
if protocol_name == "raydium_clmm" {
if discriminator_hex == "3a7fbc3e4f52c460" && account_count >= 16 {
return Some(RaydiumMappedNonTradeInstructionSpec {
instruction_name: "decrease_liquidity_v2",
event_kind: "raydium_clmm.decrease_liquidity_v2",
pool_account_index: Some(3),
token_a_mint_index: Some(14),
token_b_mint_index: Some(15),
lp_mint_index: None,
amount_layout: RaydiumMappedNonTradeAmountLayout::ClmmLiquidityV2,
});
}
if discriminator_hex == "851d59df45eeb00a" && account_count >= 15 {
return Some(RaydiumMappedNonTradeInstructionSpec {
instruction_name: "increase_liquidity_v2",
event_kind: "raydium_clmm.increase_liquidity_v2",
pool_account_index: Some(2),
token_a_mint_index: Some(13),
token_b_mint_index: Some(14),
lp_mint_index: None,
amount_layout: RaydiumMappedNonTradeAmountLayout::ClmmLiquidityV2,
});
}
if discriminator_hex == "4dffae527d1dc92e" && account_count >= 20 {
return Some(RaydiumMappedNonTradeInstructionSpec {
instruction_name: "open_position_with_token22_nft",
event_kind: "raydium_clmm.open_position_with_token22_nft",
pool_account_index: Some(4),
token_a_mint_index: Some(18),
token_b_mint_index: Some(19),
lp_mint_index: Some(2),
amount_layout: RaydiumMappedNonTradeAmountLayout::None,
});
}
if discriminator_hex == "7b86510031446262" && account_count >= 6 {
return Some(RaydiumMappedNonTradeInstructionSpec {
instruction_name: "close_position",
event_kind: "raydium_clmm.close_position",
pool_account_index: None,
token_a_mint_index: None,
token_b_mint_index: None,
lp_mint_index: Some(1),
amount_layout: RaydiumMappedNonTradeAmountLayout::None,
});
}
}
if protocol_name == "raydium_cpmm" {
if discriminator_hex == "1416567bc61cdb84" && account_count >= 14 {
return Some(RaydiumMappedNonTradeInstructionSpec {
instruction_name: "collect_creator_fee",
event_kind: "raydium_cpmm.collect_creator_fee",
pool_account_index: Some(3),
token_a_mint_index: None,
token_b_mint_index: None,
lp_mint_index: None,
amount_layout: RaydiumMappedNonTradeAmountLayout::None,
});
}
if discriminator_hex == "b712469c946da122" && account_count >= 14 {
return Some(RaydiumMappedNonTradeInstructionSpec {
instruction_name: "withdraw",
event_kind: "raydium_cpmm.withdraw",
pool_account_index: Some(3),
token_a_mint_index: None,
token_b_mint_index: None,
lp_mint_index: None,
amount_layout: RaydiumMappedNonTradeAmountLayout::CpmmWithdraw,
});
}
if discriminator_hex == "afaf6d1f0d989bed" && account_count >= 20 {
return Some(RaydiumMappedNonTradeInstructionSpec {
instruction_name: "initialize",
event_kind: "raydium_cpmm.initialize",
pool_account_index: Some(3),
token_a_mint_index: Some(4),
token_b_mint_index: Some(5),
lp_mint_index: Some(13),
amount_layout: RaydiumMappedNonTradeAmountLayout::None,
});
}
}
return None;
}
fn candidate_raydium_mapped_pool_account(
mapped_spec: std::option::Option<RaydiumMappedNonTradeInstructionSpec>,
accounts: &[std::string::String],
protocol_name: &str,
accounts_json: &str,
) -> std::option::Option<std::string::String> {
if let Some(mapped_spec) = mapped_spec {
if let Some(pool_account_index) = mapped_spec.pool_account_index {
return candidate_raydium_mapped_account(Some(pool_account_index), accounts);
}
return None;
}
return candidate_raydium_audit_pool_account(protocol_name, accounts_json);
}
fn candidate_raydium_mapped_account(
index: std::option::Option<usize>,
accounts: &[std::string::String],
) -> std::option::Option<std::string::String> {
let index = match index {
Some(index) => index,
None => return None,
};
return accounts.get(index).cloned();
}
fn enrich_raydium_mapped_non_trade_payload(
payload: serde_json::Value,
mapped_spec: RaydiumMappedNonTradeInstructionSpec,
data_base58: std::option::Option<&str>,
) -> serde_json::Value {
let mut object = match payload {
serde_json::Value::Object(object) => object,
other => {
let mut object = serde_json::Map::new();
object.insert("rawPayload".to_string(), other);
object
},
};
object.remove("tradeCandidate");
object.remove("candleCandidate");
object.remove("nonTradeUseful");
object.remove("skipTradeReason");
object.remove("skipCandleReason");
object.insert(
"instructionName".to_string(),
serde_json::Value::String(mapped_spec.instruction_name.to_string()),
);
object.insert("decodedFromAudit".to_string(), serde_json::Value::Bool(true));
object.insert(
"auditReason".to_string(),
serde_json::Value::String("raydium_non_swap_instruction_mapped_from_corpus".to_string()),
);
object.insert(
"proofSource".to_string(),
serde_json::Value::String(
"local_corpus_discriminator_and_raydium_idl_instruction_name".to_string(),
),
);
let data_bytes = instruction_data_bytes_from_base58(data_base58);
if let Some(data_bytes) = data_bytes {
insert_raydium_mapped_amounts(
&mut object,
mapped_spec.amount_layout,
data_bytes.as_slice(),
);
}
return serde_json::Value::Object(object);
}
fn insert_raydium_mapped_amounts(
object: &mut serde_json::Map<std::string::String, serde_json::Value>,
amount_layout: RaydiumMappedNonTradeAmountLayout,
data: &[u8],
) {
match amount_layout {
RaydiumMappedNonTradeAmountLayout::None => return,
RaydiumMappedNonTradeAmountLayout::ClmmLiquidityV2 => {
if let Some(liquidity) = read_u128_le_from_bytes(data, 8) {
object.insert(
"liquidity".to_string(),
serde_json::Value::String(liquidity.to_string()),
);
object.insert(
"lpAmountRaw".to_string(),
serde_json::Value::String(liquidity.to_string()),
);
}
if let Some(amount_0) = read_u64_le_from_bytes(data, 24) {
object.insert(
"tokenAAmount".to_string(),
serde_json::Value::String(amount_0.to_string()),
);
}
if let Some(amount_1) = read_u64_le_from_bytes(data, 32) {
object.insert(
"tokenBAmount".to_string(),
serde_json::Value::String(amount_1.to_string()),
);
}
},
RaydiumMappedNonTradeAmountLayout::CpmmWithdraw => {
if let Some(lp_amount) = read_u64_le_from_bytes(data, 8) {
object.insert(
"lpAmountRaw".to_string(),
serde_json::Value::String(lp_amount.to_string()),
);
object.insert(
"liquidity".to_string(),
serde_json::Value::String(lp_amount.to_string()),
);
}
if let Some(amount_0) = read_u64_le_from_bytes(data, 16) {
object.insert(
"tokenAAmount".to_string(),
serde_json::Value::String(amount_0.to_string()),
);
}
if let Some(amount_1) = read_u64_le_from_bytes(data, 24) {
object.insert(
"tokenBAmount".to_string(),
serde_json::Value::String(amount_1.to_string()),
);
}
},
}
}
fn instruction_data_bytes_from_base58(
data_base58: std::option::Option<&str>,
) -> std::option::Option<std::vec::Vec<u8>> {
let data_base58 = match data_base58 {
Some(data_base58) => data_base58,
None => return None,
};
let bytes_result = bs58::decode(data_base58).into_vec();
match bytes_result {
Ok(bytes) => return Some(bytes),
Err(_) => return None,
}
}
fn read_u64_le_from_bytes(data: &[u8], offset: usize) -> std::option::Option<u64> {
if data.len() < offset + 8 {
return None;
}
let mut bytes = [0_u8; 8];
let mut index = 0_usize;
while index < 8 {
bytes[index] = data[offset + index];
index += 1;
}
return Some(u64::from_le_bytes(bytes));
}
fn read_u128_le_from_bytes(data: &[u8], offset: usize) -> std::option::Option<u128> {
if data.len() < offset + 16 {
return None;
}
let mut bytes = [0_u8; 16];
let mut index = 0_usize;
while index < 16 {
bytes[index] = data[offset + index];
index += 1;
}
return Some(u128::from_le_bytes(bytes));
}
fn raydium_instruction_audit_event_kind_by_protocol(
protocol_name: &str,
) -> std::option::Option<&'static str> {
match protocol_name {
"raydium_amm_v4" => return Some("raydium_amm_v4.instruction_audit"),
"raydium_clmm" => return Some("raydium_clmm.instruction_audit"),
"raydium_cpmm" => return Some("raydium_cpmm.instruction_audit"),
_ => return None,
}
}
fn build_raydium_instruction_audit_payload(
transaction: &crate::ChainTransactionDto,
instruction: &crate::ChainInstructionDto,
protocol_name: &str,
event_kind: &str,
program_id: &str,
) -> serde_json::Value {
let accounts = parse_instruction_accounts_value(instruction.accounts_json.as_str());
let account_count = match accounts.as_array() {
Some(items) => items.len(),
None => 0,
};
let data_base58 = parse_instruction_data_base58(instruction.data_json.as_deref());
let discriminator_hex = discriminator_hex_from_base58(data_base58.as_deref());
return serde_json::json!({
"decoder": protocol_name,
"eventKind": event_kind,
"signature": transaction.signature,
"instructionId": instruction.id,
"instructionIndex": instruction.instruction_index,
"innerInstructionIndex": instruction.inner_instruction_index,
"innerInstruction": instruction.inner_instruction_index.is_some(),
"parentInstructionId": instruction.parent_instruction_id,
"programId": program_id,
"accounts": accounts,
"accountCount": account_count,
"data": data_base58,
"discriminatorHex": discriminator_hex,
"auditReason": "raydium_instruction_not_decoded_by_specific_decoder",
"tradeCandidate": false,
"candleCandidate": false,
"nonTradeUseful": false,
"skipTradeReason": "instruction_audit_only",
"skipCandleReason": "instruction_audit_only"
});
}
fn candidate_raydium_audit_pool_account(
protocol_name: &str,
accounts_json: &str,
) -> std::option::Option<std::string::String> {
let spec = match protocol_name {
"raydium_amm_v4" => RaydiumInstructionAuditSpec {
protocol_name: "raydium_amm_v4",
event_kind: "raydium_amm_v4.instruction_audit",
candidate_pool_account_index: 1,
},
"raydium_clmm" => RaydiumInstructionAuditSpec {
protocol_name: "raydium_clmm",
event_kind: "raydium_clmm.instruction_audit",
candidate_pool_account_index: 2,
},
"raydium_cpmm" => RaydiumInstructionAuditSpec {
protocol_name: "raydium_cpmm",
event_kind: "raydium_cpmm.instruction_audit",
candidate_pool_account_index: 3,
},
_ => return None,
};
let accounts_result = serde_json::from_str::<std::vec::Vec<std::string::String>>(accounts_json);
let accounts = match accounts_result {
Ok(accounts) => accounts,
Err(_) => return None,
};
return accounts.get(spec.candidate_pool_account_index).cloned();
}
fn parse_instruction_accounts_vec(accounts_json: &str) -> std::vec::Vec<std::string::String> {
let accounts_result = serde_json::from_str::<std::vec::Vec<std::string::String>>(accounts_json);
match accounts_result {
Ok(accounts) => return accounts,
Err(_) => return std::vec::Vec::new(),
}
}
fn parse_instruction_accounts_value(accounts_json: &str) -> serde_json::Value {
let accounts_result = serde_json::from_str::<serde_json::Value>(accounts_json);
match accounts_result {
Ok(accounts) => return accounts,
Err(_) => return serde_json::Value::Array(std::vec::Vec::new()),
}
}
fn parse_instruction_data_base58(
data_json: std::option::Option<&str>,
) -> std::option::Option<std::string::String> {
let data_json = match data_json {
Some(data_json) => data_json,
None => return None,
};
let data_result = serde_json::from_str::<std::string::String>(data_json);
match data_result {
Ok(data) => return Some(data),
Err(_) => {
if data_json.trim().is_empty() {
return None;
}
return Some(data_json.to_string());
},
}
}
fn discriminator_hex_from_base58(
data_base58: std::option::Option<&str>,
) -> std::option::Option<std::string::String> {
let data_base58 = match data_base58 {
Some(data_base58) => data_base58,
None => return None,
};
let bytes_result = bs58::decode(data_base58).into_vec();
let bytes = match bytes_result {
Ok(bytes) => bytes,
Err(_) => return None,
};
if bytes.len() < 8 {
return None;
}
let mut text = std::string::String::new();
for byte in bytes.iter().take(8) {
text.push_str(format!("{byte:02x}").as_str());
}
return Some(text);
}
fn append_persisted_events(
target: &mut std::vec::Vec<crate::DexDecodedEventDto>,
source: std::vec::Vec<crate::DexDecodedEventDto>,
@@ -2073,6 +2682,22 @@ mod tests {
crate::classify_dex_event_category_code("raydium_cpmm.initialize"),
"pool_lifecycle"
);
assert_eq!(
crate::classify_dex_event_category_code("raydium_clmm.instruction_audit"),
"informational"
);
assert_eq!(
crate::classify_dex_event_lifecycle_kind_code("raydium_clmm.instruction_audit"),
"instruction_audit"
);
assert_eq!(
crate::classify_dex_event_actionability_code(
"raydium_clmm.instruction_audit",
false,
false
),
"informational"
);
}
#[test]
@@ -2133,4 +2758,69 @@ mod tests {
Some(&serde_json::Value::String("non_trade_event".to_owned()))
);
}
#[test]
fn maps_observed_raydium_clmm_non_swap_discriminators() {
let decrease = super::raydium_mapped_non_trade_instruction_spec(
"raydium_clmm",
Some("3a7fbc3e4f52c460"),
19,
);
let decrease = match decrease {
Some(decrease) => decrease,
None => panic!("decrease_liquidity_v2 discriminator must be mapped"),
};
assert_eq!(decrease.event_kind, "raydium_clmm.decrease_liquidity_v2");
assert_eq!(decrease.pool_account_index, Some(3));
assert_eq!(decrease.token_a_mint_index, Some(14));
assert_eq!(decrease.token_b_mint_index, Some(15));
let increase = super::raydium_mapped_non_trade_instruction_spec(
"raydium_clmm",
Some("851d59df45eeb00a"),
15,
);
let increase = match increase {
Some(increase) => increase,
None => panic!("increase_liquidity_v2 discriminator must be mapped"),
};
assert_eq!(increase.event_kind, "raydium_clmm.increase_liquidity_v2");
assert_eq!(increase.pool_account_index, Some(2));
}
#[test]
fn maps_observed_raydium_cpmm_non_swap_discriminators() {
let collect_creator_fee = super::raydium_mapped_non_trade_instruction_spec(
"raydium_cpmm",
Some("1416567bc61cdb84"),
14,
);
let collect_creator_fee = match collect_creator_fee {
Some(collect_creator_fee) => collect_creator_fee,
None => panic!("collect_creator_fee discriminator must be mapped"),
};
assert_eq!(collect_creator_fee.event_kind, "raydium_cpmm.collect_creator_fee");
let withdraw = super::raydium_mapped_non_trade_instruction_spec(
"raydium_cpmm",
Some("b712469c946da122"),
14,
);
let withdraw = match withdraw {
Some(withdraw) => withdraw,
None => panic!("withdraw discriminator must be mapped"),
};
assert_eq!(withdraw.event_kind, "raydium_cpmm.withdraw");
let initialize = super::raydium_mapped_non_trade_instruction_spec(
"raydium_cpmm",
Some("afaf6d1f0d989bed"),
20,
);
let initialize = match initialize {
Some(initialize) => initialize,
None => panic!("initialize discriminator must be mapped"),
};
assert_eq!(initialize.event_kind, "raydium_cpmm.initialize");
}
}

3
kb_lib/src/dex_detection_route.rs
View File

@@ -56,6 +56,9 @@ pub(crate) fn dex_detection_route(
("raydium_cpmm", "raydium_cpmm.swap_base_output") => {
return Some(crate::dex_detection_route::DexDetectionRoute::RaydiumCpmmTrade);
},
("raydium_clmm", "raydium_clmm.swap") => {
return Some(crate::dex_detection_route::DexDetectionRoute::RaydiumClmmTrade);
},
("raydium_clmm", "raydium_clmm.swap_v2") => {
return Some(crate::dex_detection_route::DexDetectionRoute::RaydiumClmmTrade);
},

24
kb_lib/src/dex_event_classification.rs
View File

@@ -23,6 +23,8 @@ pub enum DexEventCategory {
PoolLifecycle,
/// Protocol administration, configuration or permission update event.
Admin,
/// Informational or audit-only decoded event retained for corpus analysis.
Informational,
/// Event kind that is not classified yet.
Unknown,
}
@@ -37,6 +39,7 @@ impl DexEventCategory {
Self::Reward => return "reward",
Self::PoolLifecycle => return "pool_lifecycle",
Self::Admin => return "admin",
Self::Informational => return "informational",
Self::Unknown => return "unknown",
}
}
@@ -73,6 +76,8 @@ pub enum DexEventLifecycleKind {
Reward,
/// Administration, configuration or permission update event.
AdminConfig,
/// Instruction-level audit event retained for corpus analysis.
InstructionAudit,
/// Event kind that is not classified yet.
Unknown,
}
@@ -95,6 +100,7 @@ impl DexEventLifecycleKind {
Self::FeeCollection => return "fee_collection",
Self::Reward => return "reward",
Self::AdminConfig => return "admin_config",
Self::InstructionAudit => return "instruction_audit",
Self::Unknown => return "unknown",
}
}
@@ -133,6 +139,9 @@ impl DexEventActionability {
/// Classifies a DEX event kind into a stable business category.
pub fn classify_dex_event_category(event_kind: &str) -> DexEventCategory {
if is_dex_informational_event_kind(event_kind) {
return DexEventCategory::Informational;
}
if is_dex_reward_event_kind(event_kind) {
return DexEventCategory::Reward;
}
@@ -161,6 +170,9 @@ pub fn classify_dex_event_category_code(event_kind: &str) -> &'static str {
/// Classifies a DEX event kind into a fine-grained lifecycle kind.
pub fn classify_dex_event_lifecycle_kind(event_kind: &str) -> DexEventLifecycleKind {
if is_dex_informational_event_kind(event_kind) {
return DexEventLifecycleKind::InstructionAudit;
}
if is_dex_token_burn_event_kind(event_kind) {
return DexEventLifecycleKind::Burn;
}
@@ -233,6 +245,7 @@ pub fn classify_dex_event_actionability(
DexEventCategory::Reward => return DexEventActionability::NonTradeUseful,
DexEventCategory::PoolLifecycle => return DexEventActionability::NonTradeUseful,
DexEventCategory::Admin => return DexEventActionability::NonTradeUseful,
DexEventCategory::Informational => return DexEventActionability::Informational,
DexEventCategory::Trade => return DexEventActionability::NonActionableTrade,
DexEventCategory::Unknown => return DexEventActionability::Unknown,
}
@@ -248,6 +261,17 @@ pub fn classify_dex_event_actionability_code(
.as_str();
}
/// Returns true for decoded audit-only events retained for corpus analysis.
pub fn is_dex_informational_event_kind(event_kind: &str) -> bool {
if event_kind.contains(".instruction_audit") {
return true;
}
if event_kind.contains(".unknown_instruction") {
return true;
}
return false;
}
/// Returns true when the event kind represents a swap-like event.
pub fn is_dex_trade_event_kind(event_kind: &str) -> bool {
if event_kind.ends_with(".buy") {

10
kb_lib/src/http_client.rs
View File

@@ -978,7 +978,11 @@ fn http_classify_method(method: &str) -> HttpMethodClass {
if method == "sendTransaction" || method == "sendRawTransaction" {
return HttpMethodClass::SendTransaction;
}
if method == "getProgramAccounts" || method == "getLargestAccounts" {
if method == "getProgramAccounts"
|| method == "getLargestAccounts"
|| method == "getTransaction"
|| method == "getBlock"
{
return HttpMethodClass::HeavyRead;
}
return HttpMethodClass::GeneralRpc;
@@ -1385,6 +1389,10 @@ mod tests {
crate::HttpClient::classify_method("getProgramAccounts"),
crate::HttpMethodClass::HeavyRead
);
assert_eq!(
crate::HttpClient::classify_method("getTransaction"),
crate::HttpMethodClass::HeavyRead
);
}
#[tokio::test]

9
kb_lib/src/lib.rs
View File

@@ -567,6 +567,8 @@ pub use db::query_db_metadatas_upsert;
pub use db::query_db_runtime_events_insert;
/// Lists recent runtime events ordered from newest to oldest.
pub use db::query_db_runtime_events_list_recent;
/// Deletes one decoded DEX event row by its natural key.
pub use db::query_dex_decoded_events_delete_by_key;
/// Reads one decoded DEX event by its natural key.
pub use db::query_dex_decoded_events_get_by_key;
/// Returns the latest Pump.fun create payload associated with a token mint.
@@ -927,6 +929,12 @@ pub use dex::RaydiumAmmV4Initialize2PoolDecoded;
pub use dex::RaydiumAmmV4SwapDecoded;
/// Decoded Raydium CLMM event.
pub use dex::RaydiumClmmDecodedEvent;
/// Decoded Raydium CLMM instruction event with projected instruction id.
pub use dex::RaydiumClmmDecodedInstructionEvent;
/// Raydium CLMM transaction decoder.
pub use dex::RaydiumClmmDecoder;
/// Decoded Raydium CLMM legacy swap event.
pub use dex::RaydiumClmmSwapLegacyDecoded;
/// Decoded Raydium CLMM swap_v2 instruction.
pub use dex::RaydiumClmmSwapV2Decoded;
/// Raydium CPMM decoded event.
@@ -979,6 +987,7 @@ pub use dex_event_classification::is_dex_admin_event_kind;
pub use dex_event_classification::is_dex_candle_candidate_event_kind;
/// Returns true for fee collection DEX events.
pub use dex_event_classification::is_dex_fee_event_kind;
pub use dex_event_classification::is_dex_informational_event_kind;
/// Returns true for launch or bonding-curve creation DEX events.
pub use dex_event_classification::is_dex_launch_event_kind;
/// Returns true for liquidity add-like DEX events.

548
kb_lib/src/local_pipeline_diagnostics.rs
View File

@@ -14,6 +14,72 @@ impl LocalPipelineDiagnosticsService {
return Self { database };
}
/// Builds a validation-oriented diagnostics summary from already persisted data.
///
/// This path intentionally skips row-level pair summaries and diagnostic
/// samples. Those sections are useful for UI inspection, but they can be
/// expensive on larger SQLite corpora because pair-level joins multiply
/// decoded events, trades and candle buckets. Validation only needs global
/// counters, DEX summaries, Raydium surface summaries and classification
/// summaries.
pub async fn diagnose_for_validation(
&self,
) -> Result<crate::LocalPipelineDiagnosticSummaryDto, crate::Error> {
let counters_result = query_lightweight_validation_counters(self.database.as_ref()).await;
let counters = match counters_result {
Ok(counters) => counters,
Err(error) => return Err(error),
};
let dex_summaries_result =
crate::query_local_pipeline_diagnostic_list_summaries(self.database.as_ref()).await;
let dex_summaries = match dex_summaries_result {
Ok(dex_summaries) => dex_summaries,
Err(error) => return Err(error),
};
let raydium_program_instruction_summaries_result =
crate::query_local_raydium_program_instruction_diagnostic_list_summaries(
self.database.as_ref(),
)
.await;
let raydium_program_instruction_summaries =
match raydium_program_instruction_summaries_result {
Ok(summaries) => summaries,
Err(error) => return Err(error),
};
let raydium_surface_summaries =
build_raydium_surface_summaries(&dex_summaries, &raydium_program_instruction_summaries);
let decoded_event_summaries_result =
crate::query_local_decoded_event_diagnostic_list_summaries(self.database.as_ref())
.await;
let decoded_event_summaries = match decoded_event_summaries_result {
Ok(decoded_event_summaries) => decoded_event_summaries,
Err(error) => return Err(error),
};
let event_classification_summaries_result =
crate::query_local_event_classification_diagnostic_list_summaries(
self.database.as_ref(),
)
.await;
let event_classification_summaries = match event_classification_summaries_result {
Ok(summaries) => summaries,
Err(error) => return Err(error),
};
let blocking_issue_count = counters.actionable_missing_trade_event_count
+ counters.invalid_trade_event_count
+ counters.duplicate_decoded_event_trade_count
+ counters.duplicate_candle_bucket_count;
let diagnostics_clean = blocking_issue_count == 0;
return Ok(build_lightweight_diagnostic_summary(
counters,
diagnostics_clean,
blocking_issue_count,
dex_summaries,
raydium_surface_summaries,
decoded_event_summaries,
event_classification_summaries,
));
}
/// Builds a local pipeline diagnostics summary from already persisted data.
pub async fn diagnose(&self) -> Result<crate::LocalPipelineDiagnosticSummaryDto, crate::Error> {
let sample_limit = 25_i64;
@@ -39,10 +105,8 @@ impl LocalPipelineDiagnosticsService {
Ok(summaries) => summaries,
Err(error) => return Err(error),
};
let raydium_surface_summaries = build_raydium_surface_summaries(
&dex_summaries,
&raydium_program_instruction_summaries,
);
let raydium_surface_summaries =
build_raydium_surface_summaries(&dex_summaries, &raydium_program_instruction_summaries);
let pair_summaries_result =
crate::query_local_pair_diagnostic_list_summaries(self.database.as_ref()).await;
let pair_summaries = match pair_summaries_result {
@@ -264,6 +328,481 @@ impl LocalPipelineDiagnosticsService {
}
}
async fn query_lightweight_validation_counters(
database: &crate::Database,
) -> Result<crate::LocalPipelineDiagnosticCountersDto, crate::Error> {
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let transaction_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_chain_transactions", "transaction_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let ok_transaction_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_chain_transactions WHERE err_json IS NULL", "ok_transaction_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let failed_transaction_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_chain_transactions WHERE err_json IS NOT NULL", "failed_transaction_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events", "decoded_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_trade_candidate_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events WHERE json_extract(payload_json, '$.tradeCandidate') = 1", "decoded_trade_candidate_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_candle_candidate_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events WHERE json_extract(payload_json, '$.candleCandidate') = 1", "decoded_candle_candidate_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_non_trade_useful_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events WHERE COALESCE(json_extract(payload_json, '$.nonTradeUseful'), 0) = 1 OR COALESCE(json_extract(payload_json, '$.eventActionability'), '') = 'non_trade_useful'", "decoded_non_trade_useful_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_non_actionable_trade_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events WHERE COALESCE(json_extract(payload_json, '$.eventActionability'), '') = 'non_actionable_trade' OR (COALESCE(json_extract(payload_json, '$.eventActionability'), '') = '' AND COALESCE(json_extract(payload_json, '$.eventCategory'), '') = 'trade' AND COALESCE(json_extract(payload_json, '$.tradeCandidate'), 0) = 0 AND COALESCE(json_extract(payload_json, '$.transactionFailed'), 0) = 0)", "decoded_non_actionable_trade_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_unknown_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events WHERE COALESCE(json_extract(payload_json, '$.eventCategory'), 'unknown') = 'unknown'", "decoded_unknown_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let liquidity_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_liquidity_events", "liquidity_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let pool_lifecycle_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pool_lifecycle_events", "pool_lifecycle_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let fee_event_count =
{
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_fee_events", "fee_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let reward_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_reward_events", "reward_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let pool_admin_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pool_admin_events", "pool_admin_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let missing_trade_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events dde WHERE json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.decoded_event_id = dde.id)", "missing_trade_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_trade_candidate_without_trade_event_count = missing_trade_event_count;
let decoded_trade_candidate_without_trade_event_on_ok_transaction_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events dde JOIN k_sol_chain_transactions ct ON ct.id = dde.transaction_id WHERE json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.decoded_event_id = dde.id) AND ct.err_json IS NULL AND dde.pool_account IS NOT NULL AND dde.token_a_mint IS NOT NULL AND dde.token_b_mint IS NOT NULL AND EXISTS (SELECT 1 FROM k_sol_pools p JOIN k_sol_pairs pair ON pair.pool_id = p.id WHERE p.address = dde.pool_account)", "decoded_trade_candidate_without_trade_event_on_ok_transaction_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let decoded_trade_candidate_without_trade_event_on_failed_transaction_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events dde JOIN k_sol_chain_transactions ct ON ct.id = dde.transaction_id WHERE json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.decoded_event_id = dde.id) AND ct.err_json IS NOT NULL", "decoded_trade_candidate_without_trade_event_on_failed_transaction_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let actionable_missing_trade_event_count =
decoded_trade_candidate_without_trade_event_on_ok_transaction_count;
let ignored_failed_transaction_trade_candidate_count =
decoded_trade_candidate_without_trade_event_on_failed_transaction_count;
let decoded_trade_candidate_without_amount_payload_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_dex_decoded_events dde WHERE json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.decoded_event_id = dde.id) AND ((json_extract(dde.payload_json, '$.baseAmountRaw') IS NULL AND json_extract(dde.payload_json, '$.base_amount_raw') IS NULL) OR (json_extract(dde.payload_json, '$.quoteAmountRaw') IS NULL AND json_extract(dde.payload_json, '$.quote_amount_raw') IS NULL))", "decoded_trade_candidate_without_amount_payload_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let trade_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_trade_events", "trade_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let invalid_trade_event_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_trade_events WHERE base_amount_raw IS NULL OR quote_amount_raw IS NULL OR price_quote_per_base IS NULL OR CAST(base_amount_raw AS INTEGER) <= 0 OR CAST(quote_amount_raw AS INTEGER) <= 0 OR price_quote_per_base <= 0", "invalid_trade_event_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let pair_candle_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pair_candles", "pair_candle_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let duplicate_decoded_event_trade_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM (SELECT decoded_event_id FROM k_sol_trade_events WHERE decoded_event_id IS NOT NULL GROUP BY decoded_event_id HAVING COUNT(*) > 1)", "duplicate_decoded_event_trade_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let multi_trade_signature_pair_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM (SELECT signature, pair_id FROM k_sol_trade_events GROUP BY signature, pair_id HAVING COUNT(*) > 1)", "multi_trade_signature_pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let duplicate_candle_bucket_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM (SELECT pair_id, timeframe_seconds, bucket_start_unix FROM k_sol_pair_candles GROUP BY pair_id, timeframe_seconds, bucket_start_unix HAVING COUNT(*) > 1)", "duplicate_candle_bucket_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let token_count =
{
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_tokens", "token_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let token_metadata_missing_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_tokens WHERE symbol IS NULL OR TRIM(symbol) = '' OR name IS NULL OR TRIM(name) = ''", "token_metadata_missing_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let tradable_token_metadata_missing_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(DISTINCT token.id) FROM k_sol_tokens token JOIN (SELECT pair.base_token_id AS token_id FROM k_sol_pairs pair JOIN k_sol_trade_events te ON te.pair_id = pair.id UNION SELECT pair.quote_token_id AS token_id FROM k_sol_pairs pair JOIN k_sol_trade_events te ON te.pair_id = pair.id) tradable_pair_token ON tradable_pair_token.token_id = token.id WHERE token.symbol IS NULL OR TRIM(token.symbol) = '' OR token.name IS NULL OR TRIM(token.name) = ''", "tradable_token_metadata_missing_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let quote_token_metadata_missing_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(DISTINCT quote_token.id) FROM k_sol_pairs pair JOIN k_sol_tokens quote_token ON quote_token.id = pair.quote_token_id WHERE quote_token.symbol IS NULL OR TRIM(quote_token.symbol) = '' OR quote_token.name IS NULL OR TRIM(quote_token.name) = ''", "quote_token_metadata_missing_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let pair_symbol_fallback_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_tokens base_token ON base_token.id = pair.base_token_id JOIN k_sol_tokens quote_token ON quote_token.id = pair.quote_token_id WHERE pair.symbol IS NULL OR TRIM(pair.symbol) = '' OR pair.symbol = base_token.mint || '/' || quote_token.mint OR instr(pair.symbol, base_token.mint) > 0 OR instr(pair.symbol, quote_token.mint) > 0", "pair_symbol_fallback_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let pair_symbol_resolved_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_tokens base_token ON base_token.id = pair.base_token_id JOIN k_sol_tokens quote_token ON quote_token.id = pair.quote_token_id WHERE pair.symbol IS NOT NULL AND TRIM(pair.symbol) != '' AND pair.symbol != base_token.mint || '/' || quote_token.mint AND instr(pair.symbol, base_token.mint) = 0 AND instr(pair.symbol, quote_token.mint) = 0", "pair_symbol_resolved_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let wsol_quote_pair_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_tokens quote_token ON quote_token.id = pair.quote_token_id WHERE quote_token.mint = 'So11111111111111111111111111111111111111112'", "wsol_quote_pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let stable_quote_pair_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_tokens quote_token ON quote_token.id = pair.quote_token_id WHERE quote_token.mint IN ('EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v', 'Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB', 'USD1ttGY1N17NEEHLmELoaybftRBUSErhqYiQzvEmuB', 'JuprjznTrTSp2UFa3ZBUFgwdAmtZCq4MQCwysN55USD')", "stable_quote_pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let pool_count =
{
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pools", "pool_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let pair_count =
{
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs", "pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let literal_pair_without_trade_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair WHERE NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.pair_id = pair.id)", "literal_pair_without_trade_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let literal_pair_without_candle_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair WHERE NOT EXISTS (SELECT 1 FROM k_sol_pair_candles pc WHERE pc.pair_id = pair.id)", "literal_pair_without_candle_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let trade_materialized_pair_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(DISTINCT pair_id) FROM k_sol_trade_events", "trade_materialized_pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let candle_materialized_pair_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(DISTINCT pair_id) FROM k_sol_pair_candles", "candle_materialized_pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let actionable_pair_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_pools p ON p.id = pair.pool_id WHERE EXISTS (SELECT 1 FROM k_sol_dex_decoded_events dde JOIN k_sol_chain_transactions ct ON ct.id = dde.transaction_id WHERE dde.pool_account = p.address AND json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND ct.err_json IS NULL)", "actionable_pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let candle_bucket_timeframe_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(DISTINCT timeframe_seconds) FROM k_sol_pair_candles", "candle_bucket_timeframe_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let non_actionable_pair_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_pools p ON p.id = pair.pool_id WHERE EXISTS (SELECT 1 FROM k_sol_dex_decoded_events dde WHERE dde.pool_account = p.address AND json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.decoded_event_id = dde.id)) AND NOT EXISTS (SELECT 1 FROM k_sol_dex_decoded_events dde JOIN k_sol_chain_transactions ct ON ct.id = dde.transaction_id WHERE dde.pool_account = p.address AND json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND ct.err_json IS NULL AND NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.decoded_event_id = dde.id))", "non_actionable_pair_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let blocking_pair_without_trade_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_pools p ON p.id = pair.pool_id WHERE EXISTS (SELECT 1 FROM k_sol_dex_decoded_events dde JOIN k_sol_chain_transactions ct ON ct.id = dde.transaction_id WHERE dde.pool_account = p.address AND json_extract(dde.payload_json, '$.tradeCandidate') = 1 AND ct.err_json IS NULL) AND NOT EXISTS (SELECT 1 FROM k_sol_trade_events te WHERE te.pair_id = pair.id)", "blocking_pair_without_trade_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
let blocking_pair_without_candle_count = {
let counter_result = query_validation_i64(pool, "SELECT COUNT(*) FROM k_sol_pairs pair JOIN k_sol_pools p ON p.id = pair.pool_id WHERE EXISTS (SELECT 1 FROM k_sol_dex_decoded_events dde JOIN k_sol_chain_transactions ct ON ct.id = dde.transaction_id WHERE dde.pool_account = p.address AND json_extract(dde.payload_json, '$.candleCandidate') = 1 AND ct.err_json IS NULL) AND NOT EXISTS (SELECT 1 FROM k_sol_pair_candles pc WHERE pc.pair_id = pair.id)", "blocking_pair_without_candle_count").await;
match counter_result {
Ok(value) => value,
Err(error) => return Err(error),
}
};
return Ok(crate::LocalPipelineDiagnosticCountersDto {
transaction_count,
ok_transaction_count,
failed_transaction_count,
decoded_event_count,
decoded_trade_candidate_count,
decoded_candle_candidate_count,
decoded_non_trade_useful_event_count,
decoded_non_actionable_trade_event_count,
decoded_unknown_event_count,
liquidity_event_count,
pool_lifecycle_event_count,
fee_event_count,
reward_event_count,
pool_admin_event_count,
missing_trade_event_count,
decoded_trade_candidate_without_trade_event_count,
decoded_trade_candidate_without_trade_event_on_ok_transaction_count,
decoded_trade_candidate_without_trade_event_on_failed_transaction_count,
actionable_missing_trade_event_count,
ignored_failed_transaction_trade_candidate_count,
decoded_trade_candidate_without_amount_payload_count,
trade_event_count,
invalid_trade_event_count,
pair_candle_count,
duplicate_decoded_event_trade_count,
multi_trade_signature_pair_count,
duplicate_candle_bucket_count,
token_count,
token_metadata_missing_count,
tradable_token_metadata_missing_count,
quote_token_metadata_missing_count,
pair_symbol_fallback_count,
pair_symbol_resolved_count,
wsol_quote_pair_count,
stable_quote_pair_count,
pool_count,
pair_count,
literal_pair_without_trade_count,
literal_pair_without_candle_count,
trade_materialized_pair_count,
candle_materialized_pair_count,
actionable_pair_count,
candle_bucket_timeframe_count,
non_actionable_pair_count,
blocking_pair_without_trade_count,
blocking_pair_without_candle_count,
pair_without_trade_count: blocking_pair_without_trade_count,
pair_without_candle_count: blocking_pair_without_candle_count,
});
},
}
}
async fn query_validation_i64(
pool: &sqlx::Pool<sqlx::Sqlite>,
sql: &'static str,
counter_name: &str,
) -> Result<i64, crate::Error> {
let result = sqlx::query_scalar::<sqlx::Sqlite, i64>(sql).fetch_one(pool).await;
match result {
Ok(value) => return Ok(value),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot read local pipeline validation counter '{}' on sqlite: {}",
counter_name, error
)));
},
}
}
fn build_lightweight_diagnostic_summary(
counters: crate::LocalPipelineDiagnosticCountersDto,
diagnostics_clean: bool,
blocking_issue_count: i64,
dex_summaries: std::vec::Vec<crate::LocalDexDiagnosticSummaryDto>,
raydium_surface_summaries: std::vec::Vec<crate::LocalRaydiumSurfaceDiagnosticSummaryDto>,
decoded_event_summaries: std::vec::Vec<crate::LocalDecodedEventDiagnosticSummaryDto>,
event_classification_summaries: std::vec::Vec<
crate::LocalEventClassificationDiagnosticSummaryDto,
>,
) -> crate::LocalPipelineDiagnosticSummaryDto {
return crate::LocalPipelineDiagnosticSummaryDto {
transaction_count: counters.transaction_count,
ok_transaction_count: counters.ok_transaction_count,
failed_transaction_count: counters.failed_transaction_count,
decoded_event_count: counters.decoded_event_count,
decoded_trade_candidate_count: counters.decoded_trade_candidate_count,
decoded_candle_candidate_count: counters.decoded_candle_candidate_count,
decoded_non_trade_useful_event_count: counters.decoded_non_trade_useful_event_count,
decoded_non_actionable_trade_event_count: counters.decoded_non_actionable_trade_event_count,
decoded_unknown_event_count: counters.decoded_unknown_event_count,
liquidity_event_count: counters.liquidity_event_count,
pool_lifecycle_event_count: counters.pool_lifecycle_event_count,
fee_event_count: counters.fee_event_count,
reward_event_count: counters.reward_event_count,
pool_admin_event_count: counters.pool_admin_event_count,
diagnostics_clean,
blocking_issue_count,
missing_trade_event_count: counters.missing_trade_event_count,
decoded_trade_candidate_without_trade_event_count: counters
.decoded_trade_candidate_without_trade_event_count,
decoded_trade_candidate_without_trade_event_on_ok_transaction_count: counters
.decoded_trade_candidate_without_trade_event_on_ok_transaction_count,
decoded_trade_candidate_without_trade_event_on_failed_transaction_count: counters
.decoded_trade_candidate_without_trade_event_on_failed_transaction_count,
actionable_missing_trade_event_count: counters.actionable_missing_trade_event_count,
ignored_failed_transaction_trade_candidate_count: counters
.ignored_failed_transaction_trade_candidate_count,
decoded_trade_candidate_without_amount_payload_count: counters
.decoded_trade_candidate_without_amount_payload_count,
trade_event_count: counters.trade_event_count,
invalid_trade_event_count: counters.invalid_trade_event_count,
pair_candle_count: counters.pair_candle_count,
duplicate_decoded_event_trade_count: counters.duplicate_decoded_event_trade_count,
multi_trade_signature_pair_count: counters.multi_trade_signature_pair_count,
duplicate_candle_bucket_count: counters.duplicate_candle_bucket_count,
token_count: counters.token_count,
token_metadata_missing_count: counters.token_metadata_missing_count,
tradable_token_metadata_missing_count: counters.tradable_token_metadata_missing_count,
quote_token_metadata_missing_count: counters.quote_token_metadata_missing_count,
pair_symbol_fallback_count: counters.pair_symbol_fallback_count,
pair_symbol_resolved_count: counters.pair_symbol_resolved_count,
wsol_quote_pair_count: counters.wsol_quote_pair_count,
stable_quote_pair_count: counters.stable_quote_pair_count,
pool_count: counters.pool_count,
pair_count: counters.pair_count,
pair_gap_counter_semantics: "blocking_actionable_pairs_only".to_string(),
literal_pair_without_trade_count: counters.literal_pair_without_trade_count,
literal_pair_without_candle_count: counters.literal_pair_without_candle_count,
trade_materialized_pair_count: counters.trade_materialized_pair_count,
candle_materialized_pair_count: counters.candle_materialized_pair_count,
actionable_pair_count: counters.actionable_pair_count,
candle_bucket_timeframe_count: counters.candle_bucket_timeframe_count,
candles_are_bucketed: counters.candle_bucket_timeframe_count > 0,
blocking_pair_without_trade_count: counters.blocking_pair_without_trade_count,
blocking_pair_without_candle_count: counters.blocking_pair_without_candle_count,
pair_without_trade_count: counters.pair_without_trade_count,
pair_without_candle_count: counters.pair_without_candle_count,
dex_summaries,
raydium_surface_summaries,
pair_summaries: std::vec::Vec::new(),
pair_actionability_summaries: std::vec::Vec::new(),
pair_trading_readiness_summaries: std::vec::Vec::new(),
decoded_event_summaries,
event_classification_summaries,
missing_trade_event_reason_summaries: std::vec::Vec::new(),
launch_origin_samples: std::vec::Vec::new(),
pool_origin_samples: std::vec::Vec::new(),
token_metadata_gap_samples: std::vec::Vec::new(),
non_actionable_pair_count: counters.non_actionable_pair_count,
non_actionable_pair_summaries: std::vec::Vec::new(),
missing_trade_event_samples: std::vec::Vec::new(),
duplicate_decoded_event_trade_samples: std::vec::Vec::new(),
multi_trade_signature_pair_samples: std::vec::Vec::new(),
pair_without_trade_samples: std::vec::Vec::new(),
pair_without_candle_samples: std::vec::Vec::new(),
};
}
fn build_raydium_surface_summaries(
dex_summaries: &[crate::LocalDexDiagnosticSummaryDto],
program_instruction_summaries: &[crate::LocalRaydiumProgramInstructionDiagnosticSummaryDto],
@@ -358,4 +897,3 @@ fn find_raydium_program_summary<'a>(
}
return None;
}

57
kb_lib/src/local_pipeline_validation.rs
View File

@@ -324,6 +324,26 @@ impl LocalPipelineValidationConfig {
return config;
}
/// Builds the `0.7.42` Raydium family event-coverage validation config.
///
/// This profile expects the three effective Raydium trade surfaces to remain
/// present while allowing audit-only instruction preservation and future
/// non-swap event coverage to be added without turning absent Stable Swap or
/// router activity into blockers.
pub fn v0_7_42_raydium_family_event_coverage() -> Self {
let mut config = Self::v0_7_41_raydium_amm_v4_swap_decoder();
config.profile_code = "0.7.42_raydium_family_event_coverage".to_string();
config.expected_dex_codes = vec![
"raydium_cpmm".to_string(),
"raydium_clmm".to_string(),
"raydium_amm_v4".to_string(),
];
config.require_all_expected_dexes = true;
config.allow_unexpected_dexes = true;
config.require_pair_trading_readiness_semantics = false;
return config;
}
/// Builds the legacy `0.7.39` launch-surface validation alias.
///
/// The implementation now delegates to the DEX-first profile so callers that
@@ -454,7 +474,11 @@ impl LocalPipelineValidationService {
) -> Result<crate::LocalPipelineValidationRunDto, crate::Error> {
let diagnostics_service =
crate::LocalPipelineDiagnosticsService::new(self.database.clone());
let summary_result = diagnostics_service.diagnose().await;
let summary_result = if config.profile_code == "0.7.42_raydium_family_event_coverage" {
diagnostics_service.diagnose_for_validation().await
} else {
diagnostics_service.diagnose().await
};
let summary = match summary_result {
Ok(summary) => summary,
Err(error) => return Err(error),
@@ -585,6 +609,14 @@ impl LocalPipelineValidationService {
let config = crate::LocalPipelineValidationConfig::v0_7_41_raydium_amm_v4_swap_decoder();
return self.validate_current_database(&config).await;
}
/// Diagnoses the current database with the `0.7.42` Raydium family profile.
pub async fn validate_v0_7_42_current_database(
&self,
) -> Result<crate::LocalPipelineValidationRunDto, crate::Error> {
let config = crate::LocalPipelineValidationConfig::v0_7_42_raydium_family_event_coverage();
return self.validate_current_database(&config).await;
}
}
/// Validates a diagnostics summary without performing database access.
@@ -1544,6 +1576,29 @@ mod tests {
assert!(report.expected_dex_codes.contains(&"raydium_amm_v4".to_string()));
}
#[test]
fn validation_accepts_0_7_42_raydium_family_event_coverage_profile() {
let mut summary = make_0_7_28_summary_with_meteora();
summary.dex_summaries.push(crate::LocalDexDiagnosticSummaryDto {
dex_code: "raydium_amm_v4".to_string(),
pool_count: 11,
pair_count: 11,
decoded_event_count: 58,
decoded_trade_candidate_count: 58,
decoded_candle_candidate_count: 58,
trade_event_count: 58,
pair_candle_count: 147,
});
let config = crate::LocalPipelineValidationConfig::v0_7_42_raydium_family_event_coverage();
let report = crate::validate_local_pipeline_diagnostics_summary(&summary, &config);
assert!(report.validation_passed);
assert_eq!(report.validation_profile_code, "0.7.42_raydium_family_event_coverage");
assert_eq!(report.blocking_issue_count, 0);
assert!(report.expected_dex_codes.contains(&"raydium_cpmm".to_string()));
assert!(report.expected_dex_codes.contains(&"raydium_clmm".to_string()));
assert!(report.expected_dex_codes.contains(&"raydium_amm_v4".to_string()));
}
#[test]
fn validation_rejects_0_7_33_pair_trading_readiness_mismatch() {
let mut summary = make_0_7_28_summary_with_meteora();

31
kb_lib/src/onchain_dex_pair_discovery.rs
View File

@@ -497,6 +497,37 @@ fn decode_raydium_clmm_candidate(
crate::decode_raydium_clmm_instruction(accounts_json.as_str(), data_json.as_str());
for decoded in decoded_events {
match decoded {
crate::RaydiumClmmDecodedEvent::Swap(event) => {
return Some(crate::OnchainDexPairCandidateDto {
signature: signature.to_string(),
slot,
block_time,
failed,
program_id: program_id.to_string(),
dex_code,
candidate_kind: "swap".to_string(),
confidence: "high".to_string(),
instruction_index: instruction.instruction_index,
inner_instruction_index: instruction.inner_instruction_index,
instruction_name: Some("raydium_clmm.swap".to_string()),
pool_address: Some(event.pool_state.clone()),
token_a_mint: Some(event.base_mint),
token_b_mint: Some(event.quote_mint),
verified_pool_address: Some(event.pool_state.clone()),
observed_token_mints: std::vec::Vec::new(),
token_balance_deltas: std::vec::Vec::new(),
candidate_pool_accounts: std::vec::Vec::new(),
candidate_token_vault_accounts: std::vec::Vec::new(),
candidate_program_accounts: std::vec::Vec::new(),
account_samples: sample_strings(instruction.accounts.as_slice(), 12),
log_samples: sample_logs(logs, 8),
backfill_hint: build_backfill_hint(
"pool",
Some(event.pool_state.as_str()),
signature,
),
});
},
crate::RaydiumClmmDecodedEvent::SwapV2(event) => {
return Some(crate::OnchainDexPairCandidateDto {
signature: signature.to_string(),

129
kb_lib/src/token_backfill.rs
View File

@@ -19,6 +19,10 @@ pub struct TokenBackfillResult {
pub resolved_transaction_count: usize,
/// Number of signatures whose `getTransaction` lookup returned `null`.
pub missing_transaction_count: usize,
/// Number of signatures whose `getTransaction` lookup failed after retries.
pub transaction_fetch_error_count: usize,
/// Last transaction fetch error observed during this run, if any.
pub last_transaction_fetch_error: std::option::Option<std::string::String>,
/// Total number of decoded DEX events replayed during this run.
pub decoded_event_count: usize,
/// Total number of DEX detection results produced during this run.
@@ -58,6 +62,10 @@ pub struct PoolBackfillResult {
pub resolved_transaction_count: usize,
/// Number of signatures whose `getTransaction` lookup returned `null`.
pub missing_transaction_count: usize,
/// Number of signatures whose `getTransaction` lookup failed after retries.
pub transaction_fetch_error_count: usize,
/// Last transaction fetch error observed during this run, if any.
pub last_transaction_fetch_error: std::option::Option<std::string::String>,
/// Total number of decoded DEX events replayed during this run.
pub decoded_event_count: usize,
/// Total number of DEX detection results produced during this run.
@@ -93,6 +101,10 @@ pub struct SignatureBackfillResult {
pub resolved_transaction_count: usize,
/// Number of signatures whose `getTransaction` lookup returned `null`.
pub missing_transaction_count: usize,
/// Number of signatures whose `getTransaction` lookup failed after retries.
pub transaction_fetch_error_count: usize,
/// Last transaction fetch error observed during this run, if any.
pub last_transaction_fetch_error: std::option::Option<std::string::String>,
/// Total number of decoded DEX events replayed during this run.
pub decoded_event_count: usize,
/// Total number of DEX detection results produced during this run.
@@ -142,6 +154,9 @@ pub struct TokenBackfillService {
token_metadata_service: crate::TokenMetadataBackfillService,
}
const TOKEN_BACKFILL_GET_TRANSACTION_MAX_ATTEMPTS: usize = 4;
const TOKEN_BACKFILL_GET_TRANSACTION_RETRY_BASE_DELAY_MS: u64 = 500;
impl TokenBackfillService {
/// Creates a new token-backfill service.
pub fn new(
@@ -202,6 +217,8 @@ impl TokenBackfillService {
unique_signature_count: 0,
resolved_transaction_count: 0,
missing_transaction_count: 0,
transaction_fetch_error_count: 0,
last_transaction_fetch_error: None,
decoded_event_count: 0,
detection_count: 0,
launch_attribution_count: 0,
@@ -279,6 +296,8 @@ impl TokenBackfillService {
"uniqueSignatureCount": result.unique_signature_count,
"resolvedTransactionCount": result.resolved_transaction_count,
"missingTransactionCount": result.missing_transaction_count,
"transactionFetchErrorCount": result.transaction_fetch_error_count,
"lastTransactionFetchError": result.last_transaction_fetch_error,
"decodedEventCount": result.decoded_event_count,
"detectionCount": result.detection_count,
"launchAttributionCount": result.launch_attribution_count,
@@ -410,18 +429,42 @@ impl TokenBackfillService {
"encoding": "jsonParsed",
"maxSupportedTransactionVersion": 0
}));
let transaction_value_result = self
.http_pool
.get_transaction_raw_for_role(self.http_role.as_str(), signature.clone(), config)
.await;
let transaction_value_result =
self.fetch_transaction_value_with_retry(signature.as_str(), config).await;
let transaction_value = match transaction_value_result {
Ok(transaction_value) => transaction_value,
Err(error) => return Err(error),
Err(error) => {
tracing::warn!(
signature = %signature,
error = %error,
"skipping signature after getTransaction retries failed during backfill"
);
return Ok(TokenBackfillSignatureResult {
resolved_transaction_count: 0,
missing_transaction_count: 0,
transaction_fetch_error_count: 1,
last_transaction_fetch_error: Some(error.to_string()),
decoded_event_count: 0,
detection_count: 0,
launch_attribution_count: 0,
pool_origin_count: 0,
wallet_participation_count: 0,
trade_event_count: 0,
liquidity_event_count: 0,
pool_lifecycle_event_count: 0,
fee_event_count: 0,
reward_event_count: 0,
pool_admin_event_count: 0,
pair_candle_count: 0,
});
},
};
if transaction_value.is_null() {
return Ok(TokenBackfillSignatureResult {
resolved_transaction_count: 0,
missing_transaction_count: 1,
transaction_fetch_error_count: 0,
last_transaction_fetch_error: None,
decoded_event_count: 0,
detection_count: 0,
launch_attribution_count: 0,
@@ -532,6 +575,8 @@ impl TokenBackfillService {
return Ok(TokenBackfillSignatureResult {
resolved_transaction_count: 1,
missing_transaction_count: 0,
transaction_fetch_error_count: 0,
last_transaction_fetch_error: None,
decoded_event_count: decoded.len(),
detection_count: detections.len(),
launch_attribution_count: launch_attributions.len(),
@@ -560,6 +605,8 @@ impl TokenBackfillService {
unique_signature_count: 0,
resolved_transaction_count: 0,
missing_transaction_count: 0,
transaction_fetch_error_count: 0,
last_transaction_fetch_error: None,
decoded_event_count: 0,
detection_count: 0,
launch_attribution_count: 0,
@@ -648,6 +695,11 @@ impl TokenBackfillService {
};
result.resolved_transaction_count += replay_result.resolved_transaction_count;
result.missing_transaction_count += replay_result.missing_transaction_count;
result.transaction_fetch_error_count += replay_result.transaction_fetch_error_count;
if replay_result.last_transaction_fetch_error.is_some() {
result.last_transaction_fetch_error =
replay_result.last_transaction_fetch_error.clone();
}
result.decoded_event_count += replay_result.decoded_event_count;
result.detection_count += replay_result.detection_count;
result.launch_attribution_count += replay_result.launch_attribution_count;
@@ -669,6 +721,8 @@ impl TokenBackfillService {
"uniqueSignatureCount": result.unique_signature_count,
"resolvedTransactionCount": result.resolved_transaction_count,
"missingTransactionCount": result.missing_transaction_count,
"transactionFetchErrorCount": result.transaction_fetch_error_count,
"lastTransactionFetchError": result.last_transaction_fetch_error,
"decodedEventCount": result.decoded_event_count,
"detectionCount": result.detection_count,
"launchAttributionCount": result.launch_attribution_count,
@@ -735,6 +789,8 @@ impl TokenBackfillService {
signature: trimmed_signature.clone(),
resolved_transaction_count: replay.resolved_transaction_count,
missing_transaction_count: replay.missing_transaction_count,
transaction_fetch_error_count: replay.transaction_fetch_error_count,
last_transaction_fetch_error: replay.last_transaction_fetch_error.clone(),
decoded_event_count: replay.decoded_event_count,
detection_count: replay.detection_count,
launch_attribution_count: replay.launch_attribution_count,
@@ -752,6 +808,8 @@ impl TokenBackfillService {
"signature": result.signature.clone(),
"resolvedTransactionCount": result.resolved_transaction_count,
"missingTransactionCount": result.missing_transaction_count,
"transactionFetchErrorCount": result.transaction_fetch_error_count,
"lastTransactionFetchError": result.last_transaction_fetch_error,
"decodedEventCount": result.decoded_event_count,
"detectionCount": result.detection_count,
"launchAttributionCount": result.launch_attribution_count,
@@ -797,6 +855,44 @@ impl TokenBackfillService {
return Ok(result);
}
async fn fetch_transaction_value_with_retry(
&self,
signature: &str,
config: std::option::Option<serde_json::Value>,
) -> Result<serde_json::Value, crate::Error> {
let mut attempt_index = 1usize;
loop {
let transaction_value_result = self
.http_pool
.get_transaction_raw_for_role(
self.http_role.as_str(),
signature.to_string(),
config.clone(),
)
.await;
match transaction_value_result {
Ok(transaction_value) => return Ok(transaction_value),
Err(error) => {
if !token_backfill_should_retry_http_error(&error)
|| attempt_index >= TOKEN_BACKFILL_GET_TRANSACTION_MAX_ATTEMPTS
{
return Err(error);
}
let delay_ms = token_backfill_retry_delay_ms(attempt_index);
tracing::warn!(
signature = %signature,
attempt = attempt_index,
delay_ms = delay_ms,
error = %error,
"getTransaction failed during backfill; retrying"
);
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
attempt_index += 1;
},
}
}
}
async fn backfill_missing_token_metadata_best_effort(&self, limit: i64) {
let metadata_result =
self.token_metadata_service.backfill_missing_token_metadata(Some(limit)).await;
@@ -824,10 +920,29 @@ impl TokenBackfillService {
}
}
fn token_backfill_should_retry_http_error(error: &crate::Error) -> bool {
match error {
crate::Error::Http(_) => return true,
_ => return false,
}
}
fn token_backfill_retry_delay_ms(attempt_index: usize) -> u64 {
let multiplier = match attempt_index {
0 => 1,
1 => 1,
2 => 3,
_ => 6,
};
return TOKEN_BACKFILL_GET_TRANSACTION_RETRY_BASE_DELAY_MS * multiplier;
}
#[derive(Debug, Clone, Default)]
struct TokenBackfillSignatureResult {
resolved_transaction_count: usize,
missing_transaction_count: usize,
transaction_fetch_error_count: usize,
last_transaction_fetch_error: std::option::Option<std::string::String>,
decoded_event_count: usize,
detection_count: usize,
launch_attribution_count: usize,
@@ -848,6 +963,10 @@ fn merge_token_backfill_signature_result(
) {
aggregate.resolved_transaction_count += value.resolved_transaction_count;
aggregate.missing_transaction_count += value.missing_transaction_count;
aggregate.transaction_fetch_error_count += value.transaction_fetch_error_count;
if value.last_transaction_fetch_error.is_some() {
aggregate.last_transaction_fetch_error = value.last_transaction_fetch_error.clone();
}
aggregate.decoded_event_count += value.decoded_event_count;
aggregate.detection_count += value.detection_count;
aggregate.launch_attribution_count += value.launch_attribution_count;