From 0b36caf77d032c29ce48f3c071d8d0c1256926ca Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Wed, 29 Apr 2026 17:18:35 +0200 Subject: [PATCH] 0.7.16 --- CHANGELOG.md | 1 + Cargo.toml | 2 +- ROADMAP.md | 14 +- kb_lib/src/db.rs | 10 + kb_lib/src/db/dtos.rs | 4 + kb_lib/src/db/dtos/pair_metric.rs | 102 ++++ kb_lib/src/db/dtos/trade_event.rs | 153 ++++++ kb_lib/src/db/entities.rs | 4 + kb_lib/src/db/entities/pair_metric.rs | 36 ++ kb_lib/src/db/entities/trade_event.rs | 46 ++ kb_lib/src/db/queries.rs | 8 + kb_lib/src/db/queries/pair_metric.rs | 186 +++++++ kb_lib/src/db/queries/trade_event.rs | 208 ++++++++ kb_lib/src/db/schema.rs | 105 ++++ kb_lib/src/lib.rs | 13 + kb_lib/src/trade_aggregation.rs | 693 ++++++++++++++++++++++++++ kb_lib/src/tx_resolution.rs | 19 +- 17 files changed, 1593 insertions(+), 11 deletions(-) create mode 100644 kb_lib/src/db/dtos/pair_metric.rs create mode 100644 kb_lib/src/db/dtos/trade_event.rs create mode 100644 kb_lib/src/db/entities/pair_metric.rs create mode 100644 kb_lib/src/db/entities/trade_event.rs create mode 100644 kb_lib/src/db/queries/pair_metric.rs create mode 100644 kb_lib/src/db/queries/trade_event.rs create mode 100644 kb_lib/src/trade_aggregation.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0472253..17b4c35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,3 +46,4 @@ 0.7.13 - Extension de la couche launch origins avec Bags et Moonit, ajout d’un enregistrement programmatique des mappings Bags, et détection automatique Moonit via suffixe de token mint 0.7.14 - Ajout d’une couche consolidée de traçabilité fondatrice multi-DEX avec enregistrement des pool origins, rattachement au decoded event, au pool/pair/listing et à l’éventuelle launch attribution 0.7.15 - Ajout d’une couche wallets observés et participations observées, avec extraction des rôles depuis les payloads décodés et rattachement transaction / decoded event / pool / pair +0.7.16 - Ajout d’une première couche trade events et pair metrics avec normalisation des swaps, agrégation par paire et branchement automatique dans le pipeline de résolution transactionnelle diff --git a/Cargo.toml b/Cargo.toml index b855704..2496965 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ members = [ ] [workspace.package] -version = "0.7.15" +version = "0.7.16" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-bobobot" diff --git a/ROADMAP.md b/ROADMAP.md index c2126c7..7b60318 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -605,14 +605,14 @@ Réalisé : - report des holdings à l’étape suivante afin de conserver une séparation nette entre acteurs observés et balances observées. ### 6.048. Version `0.7.16` — Séries de prix, volumes et agrégats DEX -Objectif : préparer la couche analytique fine à partir des événements métier normalisés. +Réalisé : -À faire : - -- préparer des agrégations de prix/volume par paire, -- introduire la base des futures candles et séries temporelles, -- permettre plus tard le calcul d’OHLCV, volume, nombre de trades et liquidité par fenêtre, -- préparer le terrain pour la couche analytique `0.8.x`. +- ajout d’une première table `trade events` pour normaliser les swaps observés, +- ajout d’une première table `pair metrics` pour agréger les swaps par paire, +- prise en charge des compteurs `trade_count`, `buy_count`, `sell_count`, +- prise en charge optionnelle des volumes bruts `base` / `quote` et du dernier prix dérivé `quote_per_base`, +- branchement automatique dans le pipeline de résolution transactionnelle, +- conservation d’un modèle simple et idempotent en préparation de futures candles / séries temporelles. ### 6.049. Version `0.7.x` — Couverture DEX v1 Objectif : structurer les connecteurs DEX autour d’un pipeline complet de résolution, décodage et normalisation métier. diff --git a/kb_lib/src/db.rs b/kb_lib/src/db.rs index 74fc40e..4db195e 100644 --- a/kb_lib/src/db.rs +++ b/kb_lib/src/db.rs @@ -32,6 +32,7 @@ pub use dtos::KbLiquidityEventDto; pub use dtos::KbObservedTokenDto; pub use dtos::KbOnchainObservationDto; pub use dtos::KbPairDto; +pub use dtos::KbPairMetricDto; pub use dtos::KbPoolDto; pub use dtos::KbPoolListingDto; pub use dtos::KbPoolOriginDto; @@ -40,6 +41,7 @@ pub use dtos::KbSwapDto; pub use dtos::KbTokenBurnEventDto; pub use dtos::KbTokenDto; pub use dtos::KbTokenMintEventDto; +pub use dtos::KbTradeEventDto; pub use dtos::KbWalletDto; pub use dtos::KbWalletParticipationDto; pub use entities::KbAnalysisSignalEntity; @@ -59,6 +61,7 @@ pub use entities::KbLiquidityEventEntity; pub use entities::KbObservedTokenEntity; pub use entities::KbOnchainObservationEntity; pub use entities::KbPairEntity; +pub use entities::KbPairMetricEntity; pub use entities::KbPoolEntity; pub use entities::KbPoolListingEntity; pub use entities::KbPoolOriginEntity; @@ -67,6 +70,7 @@ pub use entities::KbSwapEntity; pub use entities::KbTokenBurnEventEntity; pub use entities::KbTokenEntity; pub use entities::KbTokenMintEventEntity; +pub use entities::KbTradeEventEntity; pub use entities::KbWalletEntity; pub use entities::KbWalletParticipationEntity; pub use queries::delete_chain_instructions_by_transaction_id; @@ -82,10 +86,12 @@ pub use queries::get_launch_surface_by_code; pub use queries::get_launch_surface_key_by_match; pub use queries::get_observed_token_by_mint; pub use queries::get_pair_by_pool_id; +pub use queries::get_pair_metric_by_pair_id; pub use queries::get_pool_by_address; pub use queries::get_pool_listing_by_pool_id; pub use queries::get_pool_origin_by_pool_id; pub use queries::get_token_by_mint; +pub use queries::get_trade_event_by_decoded_event_id; pub use queries::get_wallet_by_address; pub use queries::get_wallet_participation_by_unique_key; pub use queries::insert_analysis_signal; @@ -102,6 +108,7 @@ pub use queries::list_launch_attributions_by_pool_id; pub use queries::list_launch_surface_keys_by_surface_id; pub use queries::list_launch_surfaces; pub use queries::list_observed_tokens; +pub use queries::list_pair_metrics; pub use queries::list_pairs; pub use queries::list_pool_listings; pub use queries::list_pool_origins; @@ -116,6 +123,7 @@ pub use queries::list_recent_onchain_observations; pub use queries::list_recent_swaps; pub use queries::list_recent_token_burn_events; pub use queries::list_recent_token_mint_events; +pub use queries::list_trade_events_by_pair_id; pub use queries::list_wallet_participations_by_pool_id; pub use queries::list_wallet_participations_by_wallet_id; pub use queries::list_wallets; @@ -132,6 +140,7 @@ pub use queries::upsert_launch_surface_key; pub use queries::upsert_liquidity_event; pub use queries::upsert_observed_token; pub use queries::upsert_pair; +pub use queries::upsert_pair_metric; pub use queries::upsert_pool; pub use queries::upsert_pool_listing; pub use queries::upsert_pool_origin; @@ -140,6 +149,7 @@ pub use queries::upsert_swap; pub use queries::upsert_token; pub use queries::upsert_token_burn_event; pub use queries::upsert_token_mint_event; +pub use queries::upsert_trade_event; pub use queries::upsert_wallet; pub use queries::upsert_wallet_participation; pub use types::KbAnalysisSignalSeverity; diff --git a/kb_lib/src/db/dtos.rs b/kb_lib/src/db/dtos.rs index dbfbb7f..8dcaaf2 100644 --- a/kb_lib/src/db/dtos.rs +++ b/kb_lib/src/db/dtos.rs @@ -19,6 +19,7 @@ mod liquidity_event; mod observed_token; mod onchain_observation; mod pair; +mod pair_metric; mod pool; mod pool_listing; mod pool_origin; @@ -27,6 +28,7 @@ mod swap; mod token; mod token_burn_event; mod token_mint_event; +mod trade_event; mod wallet; mod wallet_participation; @@ -47,6 +49,7 @@ pub use liquidity_event::KbLiquidityEventDto; pub use observed_token::KbObservedTokenDto; pub use onchain_observation::KbOnchainObservationDto; pub use pair::KbPairDto; +pub use pair_metric::KbPairMetricDto; pub use pool::KbPoolDto; pub use pool_listing::KbPoolListingDto; pub use pool_origin::KbPoolOriginDto; @@ -55,5 +58,6 @@ pub use swap::KbSwapDto; pub use token::KbTokenDto; pub use token_burn_event::KbTokenBurnEventDto; pub use token_mint_event::KbTokenMintEventDto; +pub use trade_event::KbTradeEventDto; pub use wallet::KbWalletDto; pub use wallet_participation::KbWalletParticipationDto; diff --git a/kb_lib/src/db/dtos/pair_metric.rs b/kb_lib/src/db/dtos/pair_metric.rs new file mode 100644 index 0000000..9b78d34 --- /dev/null +++ b/kb_lib/src/db/dtos/pair_metric.rs @@ -0,0 +1,102 @@ +// file: kb_lib/src/db/dtos/pair_metric.rs + +//! Pair-metric DTO. + +/// Application-facing pair-metric DTO. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbPairMetricDto { + /// Optional numeric primary key. + pub id: std::option::Option, + /// Related pair id. + pub pair_id: i64, + /// Optional first observed slot. + pub first_slot: std::option::Option, + /// Optional last observed slot. + pub last_slot: std::option::Option, + /// Optional first observed signature. + pub first_signature: std::option::Option, + /// Optional last observed signature. + pub last_signature: std::option::Option, + /// Total trade count. + pub trade_count: i64, + /// Total buy count. + pub buy_count: i64, + /// Total sell count. + pub sell_count: i64, + /// Optional cumulative raw base volume. + pub cumulative_base_amount_raw: std::option::Option, + /// Optional cumulative raw quote volume. + pub cumulative_quote_amount_raw: std::option::Option, + /// Optional last derived quote-per-base price. + pub last_price_quote_per_base: std::option::Option, + /// Creation timestamp. + pub created_at: chrono::DateTime, + /// Update timestamp. + pub updated_at: chrono::DateTime, +} + +impl KbPairMetricDto { + /// Creates a new pair-metric DTO. + pub fn new(pair_id: i64) -> Self { + let now = chrono::Utc::now(); + Self { + id: None, + pair_id, + first_slot: None, + last_slot: None, + first_signature: None, + last_signature: None, + trade_count: 0, + buy_count: 0, + sell_count: 0, + cumulative_base_amount_raw: None, + cumulative_quote_amount_raw: None, + last_price_quote_per_base: None, + created_at: now, + updated_at: now, + } + } +} + +impl TryFrom for KbPairMetricDto { + type Error = crate::KbError; + + fn try_from(entity: crate::KbPairMetricEntity) -> Result { + let created_at_result = chrono::DateTime::parse_from_rfc3339(&entity.created_at); + let created_at = match created_at_result { + Ok(created_at) => created_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse pair_metric created_at '{}': {}", + entity.created_at, error + ))); + } + }; + let updated_at_result = chrono::DateTime::parse_from_rfc3339(&entity.updated_at); + let updated_at = match updated_at_result { + Ok(updated_at) => updated_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse pair_metric updated_at '{}': {}", + entity.updated_at, error + ))); + } + }; + Ok(Self { + id: Some(entity.id), + pair_id: entity.pair_id, + first_slot: entity.first_slot, + last_slot: entity.last_slot, + first_signature: entity.first_signature, + last_signature: entity.last_signature, + trade_count: entity.trade_count, + buy_count: entity.buy_count, + sell_count: entity.sell_count, + cumulative_base_amount_raw: entity.cumulative_base_amount_raw, + cumulative_quote_amount_raw: entity.cumulative_quote_amount_raw, + last_price_quote_per_base: entity.last_price_quote_per_base, + created_at, + updated_at, + }) + } +} diff --git a/kb_lib/src/db/dtos/trade_event.rs b/kb_lib/src/db/dtos/trade_event.rs new file mode 100644 index 0000000..7fa1b79 --- /dev/null +++ b/kb_lib/src/db/dtos/trade_event.rs @@ -0,0 +1,153 @@ +// file: kb_lib/src/db/dtos/trade_event.rs + +//! Trade-event DTO. + +/// Application-facing trade-event DTO. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbTradeEventDto { + /// Optional numeric primary key. + pub id: std::option::Option, + /// Related DEX id. + pub dex_id: i64, + /// Related pool id. + pub pool_id: i64, + /// Related pair id. + pub pair_id: i64, + /// Related transaction id. + pub transaction_id: i64, + /// Related decoded event id. + pub decoded_event_id: i64, + /// Related transaction signature. + pub signature: std::string::String, + /// Optional observed slot. + pub slot: std::option::Option, + /// Stable trade side. + pub trade_side: crate::KbSwapTradeSide, + /// Related base token id. + pub base_token_id: i64, + /// Related quote token id. + pub quote_token_id: i64, + /// Optional raw base amount. + pub base_amount_raw: std::option::Option, + /// Optional raw quote amount. + pub quote_amount_raw: std::option::Option, + /// Optional derived quote-per-base price. + pub price_quote_per_base: std::option::Option, + /// Observation source kind. + pub source_kind: crate::KbObservationSourceKind, + /// Optional logical source endpoint name. + pub source_endpoint_name: std::option::Option, + /// Persisted payload JSON. + pub payload_json: std::string::String, + /// Creation timestamp. + pub created_at: chrono::DateTime, + /// Update timestamp. + pub updated_at: chrono::DateTime, +} + +impl KbTradeEventDto { + /// Creates a new trade-event DTO. + pub fn new( + dex_id: i64, + pool_id: i64, + pair_id: i64, + transaction_id: i64, + decoded_event_id: i64, + signature: std::string::String, + slot: std::option::Option, + trade_side: crate::KbSwapTradeSide, + base_token_id: i64, + quote_token_id: i64, + base_amount_raw: std::option::Option, + quote_amount_raw: std::option::Option, + price_quote_per_base: std::option::Option, + source_kind: crate::KbObservationSourceKind, + source_endpoint_name: std::option::Option, + payload_json: std::string::String, + ) -> Self { + let now = chrono::Utc::now(); + Self { + id: None, + dex_id, + pool_id, + pair_id, + transaction_id, + decoded_event_id, + signature, + slot, + trade_side, + base_token_id, + quote_token_id, + base_amount_raw, + quote_amount_raw, + price_quote_per_base, + source_kind, + source_endpoint_name, + payload_json, + created_at: now, + updated_at: now, + } + } +} + +impl TryFrom for KbTradeEventDto { + type Error = crate::KbError; + + fn try_from(entity: crate::KbTradeEventEntity) -> Result { + let trade_side = kb_trade_side_from_string(entity.trade_side.as_str()); + let source_kind_result = crate::KbObservationSourceKind::from_i16(entity.source_kind); + let source_kind = match source_kind_result { + Ok(source_kind) => source_kind, + Err(error) => return Err(error), + }; + let created_at_result = chrono::DateTime::parse_from_rfc3339(&entity.created_at); + let created_at = match created_at_result { + Ok(created_at) => created_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse trade_event created_at '{}': {}", + entity.created_at, error + ))); + } + }; + let updated_at_result = chrono::DateTime::parse_from_rfc3339(&entity.updated_at); + let updated_at = match updated_at_result { + Ok(updated_at) => updated_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse trade_event updated_at '{}': {}", + entity.updated_at, error + ))); + } + }; + Ok(Self { + id: Some(entity.id), + dex_id: entity.dex_id, + pool_id: entity.pool_id, + pair_id: entity.pair_id, + transaction_id: entity.transaction_id, + decoded_event_id: entity.decoded_event_id, + signature: entity.signature, + slot: entity.slot, + trade_side, + base_token_id: entity.base_token_id, + quote_token_id: entity.quote_token_id, + base_amount_raw: entity.base_amount_raw, + quote_amount_raw: entity.quote_amount_raw, + price_quote_per_base: entity.price_quote_per_base, + source_kind, + source_endpoint_name: entity.source_endpoint_name, + payload_json: entity.payload_json, + created_at, + updated_at, + }) + } +} + +fn kb_trade_side_from_string(value: &str) -> crate::KbSwapTradeSide { + match value { + "BuyBase" => crate::KbSwapTradeSide::BuyBase, + "SellBase" => crate::KbSwapTradeSide::SellBase, + _ => crate::KbSwapTradeSide::Unknown, + } +} diff --git a/kb_lib/src/db/entities.rs b/kb_lib/src/db/entities.rs index 4f13d96..c1557f9 100644 --- a/kb_lib/src/db/entities.rs +++ b/kb_lib/src/db/entities.rs @@ -21,6 +21,7 @@ mod liquidity_event; mod observed_token; mod onchain_observation; mod pair; +mod pair_metric; mod pool; mod pool_listing; mod pool_origin; @@ -29,6 +30,7 @@ mod swap; mod token; mod token_burn_event; mod token_mint_event; +mod trade_event; mod wallet; mod wallet_participation; @@ -49,6 +51,7 @@ pub use liquidity_event::KbLiquidityEventEntity; pub use observed_token::KbObservedTokenEntity; pub use onchain_observation::KbOnchainObservationEntity; pub use pair::KbPairEntity; +pub use pair_metric::KbPairMetricEntity; pub use pool::KbPoolEntity; pub use pool_listing::KbPoolListingEntity; pub use pool_origin::KbPoolOriginEntity; @@ -57,5 +60,6 @@ pub use swap::KbSwapEntity; pub use token::KbTokenEntity; pub use token_burn_event::KbTokenBurnEventEntity; pub use token_mint_event::KbTokenMintEventEntity; +pub use trade_event::KbTradeEventEntity; pub use wallet::KbWalletEntity; pub use wallet_participation::KbWalletParticipationEntity; diff --git a/kb_lib/src/db/entities/pair_metric.rs b/kb_lib/src/db/entities/pair_metric.rs new file mode 100644 index 0000000..d7acf03 --- /dev/null +++ b/kb_lib/src/db/entities/pair_metric.rs @@ -0,0 +1,36 @@ +// file: kb_lib/src/db/entities/pair_metric.rs + +//! Pair-metric entity. + +/// Persisted pair-metric row. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)] +pub struct KbPairMetricEntity { + /// Numeric primary key. + pub id: i64, + /// Related pair id. + pub pair_id: i64, + /// Optional first observed slot. + pub first_slot: std::option::Option, + /// Optional last observed slot. + pub last_slot: std::option::Option, + /// Optional first observed signature. + pub first_signature: std::option::Option, + /// Optional last observed signature. + pub last_signature: std::option::Option, + /// Total trade count. + pub trade_count: i64, + /// Total buy count. + pub buy_count: i64, + /// Total sell count. + pub sell_count: i64, + /// Optional cumulative raw base volume. + pub cumulative_base_amount_raw: std::option::Option, + /// Optional cumulative raw quote volume. + pub cumulative_quote_amount_raw: std::option::Option, + /// Optional last derived quote-per-base price. + pub last_price_quote_per_base: std::option::Option, + /// Creation timestamp encoded as RFC3339 UTC text. + pub created_at: std::string::String, + /// Update timestamp encoded as RFC3339 UTC text. + pub updated_at: std::string::String, +} diff --git a/kb_lib/src/db/entities/trade_event.rs b/kb_lib/src/db/entities/trade_event.rs new file mode 100644 index 0000000..f545b04 --- /dev/null +++ b/kb_lib/src/db/entities/trade_event.rs @@ -0,0 +1,46 @@ +// file: kb_lib/src/db/entities/trade_event.rs + +//! Trade-event entity. + +/// Persisted trade-event row. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)] +pub struct KbTradeEventEntity { + /// Numeric primary key. + pub id: i64, + /// Related DEX id. + pub dex_id: i64, + /// Related pool id. + pub pool_id: i64, + /// Related pair id. + pub pair_id: i64, + /// Related transaction id. + pub transaction_id: i64, + /// Related decoded event id. + pub decoded_event_id: i64, + /// Related transaction signature. + pub signature: std::string::String, + /// Optional observed slot. + pub slot: std::option::Option, + /// Stable trade side. + pub trade_side: std::string::String, + /// Related base token id. + pub base_token_id: i64, + /// Related quote token id. + pub quote_token_id: i64, + /// Optional raw base amount. + pub base_amount_raw: std::option::Option, + /// Optional raw quote amount. + pub quote_amount_raw: std::option::Option, + /// Optional derived quote-per-base price. + pub price_quote_per_base: std::option::Option, + /// Observation source kind. + pub source_kind: i16, + /// Optional logical source endpoint name. + pub source_endpoint_name: std::option::Option, + /// Persisted payload JSON. + pub payload_json: std::string::String, + /// Creation timestamp encoded as RFC3339 UTC text. + pub created_at: std::string::String, + /// Update timestamp encoded as RFC3339 UTC text. + pub updated_at: std::string::String, +} diff --git a/kb_lib/src/db/queries.rs b/kb_lib/src/db/queries.rs index 81b55b4..77b041c 100644 --- a/kb_lib/src/db/queries.rs +++ b/kb_lib/src/db/queries.rs @@ -23,6 +23,7 @@ mod liquidity_event; mod observed_token; mod onchain_observation; mod pair; +mod pair_metric; mod pool; mod pool_listing; mod pool_origin; @@ -31,6 +32,7 @@ mod swap; mod token; mod token_burn_event; mod token_mint_event; +mod trade_event; mod wallet; mod wallet_participation; @@ -81,6 +83,9 @@ pub use onchain_observation::list_recent_onchain_observations; pub use pair::get_pair_by_pool_id; pub use pair::list_pairs; pub use pair::upsert_pair; +pub use pair_metric::get_pair_metric_by_pair_id; +pub use pair_metric::list_pair_metrics; +pub use pair_metric::upsert_pair_metric; pub use pool::get_pool_by_address; pub use pool::list_pools; pub use pool::upsert_pool; @@ -100,6 +105,9 @@ pub use token_burn_event::list_recent_token_burn_events; pub use token_burn_event::upsert_token_burn_event; pub use token_mint_event::list_recent_token_mint_events; pub use token_mint_event::upsert_token_mint_event; +pub use trade_event::get_trade_event_by_decoded_event_id; +pub use trade_event::list_trade_events_by_pair_id; +pub use trade_event::upsert_trade_event; pub use wallet::get_wallet_by_address; pub use wallet::list_wallets; pub use wallet::upsert_wallet; diff --git a/kb_lib/src/db/queries/pair_metric.rs b/kb_lib/src/db/queries/pair_metric.rs new file mode 100644 index 0000000..3dc33d0 --- /dev/null +++ b/kb_lib/src/db/queries/pair_metric.rs @@ -0,0 +1,186 @@ +// file: kb_lib/src/db/queries/pair_metric.rs + +//! Queries for `kb_pair_metrics`. + +/// Inserts or updates one pair-metric row and returns its stable internal id. +pub async fn upsert_pair_metric( + database: &crate::KbDatabase, + dto: &crate::KbPairMetricDto, +) -> Result { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query( + r#" +INSERT INTO kb_pair_metrics ( + pair_id, + first_slot, + last_slot, + first_signature, + last_signature, + trade_count, + buy_count, + sell_count, + cumulative_base_amount_raw, + cumulative_quote_amount_raw, + last_price_quote_per_base, + created_at, + updated_at +) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(pair_id) DO UPDATE SET + first_slot = excluded.first_slot, + last_slot = excluded.last_slot, + first_signature = excluded.first_signature, + last_signature = excluded.last_signature, + trade_count = excluded.trade_count, + buy_count = excluded.buy_count, + sell_count = excluded.sell_count, + cumulative_base_amount_raw = excluded.cumulative_base_amount_raw, + cumulative_quote_amount_raw = excluded.cumulative_quote_amount_raw, + last_price_quote_per_base = excluded.last_price_quote_per_base, + updated_at = excluded.updated_at + "#, + ) + .bind(dto.pair_id) + .bind(dto.first_slot) + .bind(dto.last_slot) + .bind(dto.first_signature.clone()) + .bind(dto.last_signature.clone()) + .bind(dto.trade_count) + .bind(dto.buy_count) + .bind(dto.sell_count) + .bind(dto.cumulative_base_amount_raw.clone()) + .bind(dto.cumulative_quote_amount_raw.clone()) + .bind(dto.last_price_quote_per_base) + .bind(dto.created_at.to_rfc3339()) + .bind(dto.updated_at.to_rfc3339()) + .execute(pool) + .await; + if let Err(error) = query_result { + return Err(crate::KbError::Db(format!( + "cannot upsert kb_pair_metrics on sqlite: {}", + error + ))); + } + let id_result = sqlx::query_scalar::( + r#" +SELECT id +FROM kb_pair_metrics +WHERE pair_id = ? +LIMIT 1 + "#, + ) + .bind(dto.pair_id) + .fetch_one(pool) + .await; + match id_result { + Ok(id) => Ok(id), + Err(error) => Err(crate::KbError::Db(format!( + "cannot fetch kb_pair_metrics id for pair_id '{}' on sqlite: {}", + dto.pair_id, error + ))), + } + } + } +} + +/// Returns one pair-metric row identified by its pair id, if it exists. +pub async fn get_pair_metric_by_pair_id( + database: &crate::KbDatabase, + pair_id: i64, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + pair_id, + first_slot, + last_slot, + first_signature, + last_signature, + trade_count, + buy_count, + sell_count, + cumulative_base_amount_raw, + cumulative_quote_amount_raw, + last_price_quote_per_base, + created_at, + updated_at +FROM kb_pair_metrics +WHERE pair_id = ? +LIMIT 1 + "#, + ) + .bind(pair_id) + .fetch_optional(pool) + .await; + let entity_option = match query_result { + Ok(entity_option) => entity_option, + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot read kb_pair_metrics by pair_id '{}' on sqlite: {}", + pair_id, error + ))); + } + }; + match entity_option { + Some(entity) => crate::KbPairMetricDto::try_from(entity).map(Some), + None => Ok(None), + } + } + } +} + +/// Lists all pair-metric rows ordered by pair id. +pub async fn list_pair_metrics( + database: &crate::KbDatabase, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + pair_id, + first_slot, + last_slot, + first_signature, + last_signature, + trade_count, + buy_count, + sell_count, + cumulative_base_amount_raw, + cumulative_quote_amount_raw, + last_price_quote_per_base, + created_at, + updated_at +FROM kb_pair_metrics +ORDER BY pair_id ASC + "#, + ) + .fetch_all(pool) + .await; + let entities = match query_result { + Ok(entities) => entities, + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot list kb_pair_metrics on sqlite: {}", + error + ))); + } + }; + let mut dtos = std::vec::Vec::new(); + for entity in entities { + let dto_result = crate::KbPairMetricDto::try_from(entity); + let dto = match dto_result { + Ok(dto) => dto, + Err(error) => return Err(error), + }; + dtos.push(dto); + } + Ok(dtos) + } + } +} diff --git a/kb_lib/src/db/queries/trade_event.rs b/kb_lib/src/db/queries/trade_event.rs new file mode 100644 index 0000000..5580d68 --- /dev/null +++ b/kb_lib/src/db/queries/trade_event.rs @@ -0,0 +1,208 @@ +// file: kb_lib/src/db/queries/trade_event.rs + +//! Queries for `kb_trade_events`. + +/// Inserts or updates one trade-event row and returns its stable internal id. +pub async fn upsert_trade_event( + database: &crate::KbDatabase, + dto: &crate::KbTradeEventDto, +) -> Result { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query( + r#" +INSERT INTO kb_trade_events ( + dex_id, + pool_id, + pair_id, + transaction_id, + decoded_event_id, + signature, + slot, + trade_side, + base_token_id, + quote_token_id, + base_amount_raw, + quote_amount_raw, + price_quote_per_base, + source_kind, + source_endpoint_name, + payload_json, + created_at, + updated_at +) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(decoded_event_id) DO UPDATE SET + updated_at = excluded.updated_at + "#, + ) + .bind(dto.dex_id) + .bind(dto.pool_id) + .bind(dto.pair_id) + .bind(dto.transaction_id) + .bind(dto.decoded_event_id) + .bind(dto.signature.clone()) + .bind(dto.slot) + .bind(kb_trade_side_to_string(dto.trade_side)) + .bind(dto.base_token_id) + .bind(dto.quote_token_id) + .bind(dto.base_amount_raw.clone()) + .bind(dto.quote_amount_raw.clone()) + .bind(dto.price_quote_per_base) + .bind(dto.source_kind.to_i16()) + .bind(dto.source_endpoint_name.clone()) + .bind(dto.payload_json.clone()) + .bind(dto.created_at.to_rfc3339()) + .bind(dto.updated_at.to_rfc3339()) + .execute(pool) + .await; + if let Err(error) = query_result { + return Err(crate::KbError::Db(format!( + "cannot upsert kb_trade_events on sqlite: {}", + error + ))); + } + + let id_result = sqlx::query_scalar::( + r#" +SELECT id +FROM kb_trade_events +WHERE decoded_event_id = ? +LIMIT 1 + "#, + ) + .bind(dto.decoded_event_id) + .fetch_one(pool) + .await; + match id_result { + Ok(id) => Ok(id), + Err(error) => Err(crate::KbError::Db(format!( + "cannot fetch kb_trade_events id for decoded_event_id '{}' on sqlite: {}", + dto.decoded_event_id, error + ))), + } + } + } +} + +/// Returns one trade-event row identified by its decoded-event id, if it exists. +pub async fn get_trade_event_by_decoded_event_id( + database: &crate::KbDatabase, + decoded_event_id: i64, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + dex_id, + pool_id, + pair_id, + transaction_id, + decoded_event_id, + signature, + slot, + trade_side, + base_token_id, + quote_token_id, + base_amount_raw, + quote_amount_raw, + price_quote_per_base, + source_kind, + source_endpoint_name, + payload_json, + created_at, + updated_at +FROM kb_trade_events +WHERE decoded_event_id = ? +LIMIT 1 + "#, + ) + .bind(decoded_event_id) + .fetch_optional(pool) + .await; + let entity_option = match query_result { + Ok(entity_option) => entity_option, + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot read kb_trade_events by decoded_event_id '{}' on sqlite: {}", + decoded_event_id, error + ))); + } + }; + match entity_option { + Some(entity) => crate::KbTradeEventDto::try_from(entity).map(Some), + None => Ok(None), + } + } + } +} + +/// Lists trade-event rows for one pair id ordered by creation time then id. +pub async fn list_trade_events_by_pair_id( + database: &crate::KbDatabase, + pair_id: i64, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + dex_id, + pool_id, + pair_id, + transaction_id, + decoded_event_id, + signature, + slot, + trade_side, + base_token_id, + quote_token_id, + base_amount_raw, + quote_amount_raw, + price_quote_per_base, + source_kind, + source_endpoint_name, + payload_json, + created_at, + updated_at +FROM kb_trade_events +WHERE pair_id = ? +ORDER BY created_at ASC, id ASC + "#, + ) + .bind(pair_id) + .fetch_all(pool) + .await; + let entities = match query_result { + Ok(entities) => entities, + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot list kb_trade_events by pair_id '{}' on sqlite: {}", + pair_id, error + ))); + } + }; + let mut dtos = std::vec::Vec::new(); + for entity in entities { + let dto_result = crate::KbTradeEventDto::try_from(entity); + let dto = match dto_result { + Ok(dto) => dto, + Err(error) => return Err(error), + }; + dtos.push(dto); + } + Ok(dtos) + } + } +} + +fn kb_trade_side_to_string(value: crate::KbSwapTradeSide) -> &'static str { + match value { + crate::KbSwapTradeSide::BuyBase => "BuyBase", + crate::KbSwapTradeSide::SellBase => "SellBase", + crate::KbSwapTradeSide::Unknown => "Unknown", + } +} diff --git a/kb_lib/src/db/schema.rs b/kb_lib/src/db/schema.rs index 09d62db..57f8609 100644 --- a/kb_lib/src/db/schema.rs +++ b/kb_lib/src/db/schema.rs @@ -294,6 +294,22 @@ pub(crate) async fn ensure_schema(database: &crate::KbDatabase) -> Result<(), cr if let Err(error) = result { return Err(error); } + let result = create_kb_trade_events_table(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_idx_trade_events_pair_id(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_idx_trade_events_pool_id(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_pair_metrics_table(pool).await; + if let Err(error) = result { + return Err(error); + } Ok(()) } } @@ -1608,3 +1624,92 @@ ON kb_wallet_participations(transaction_id) ) .await } + +async fn create_kb_trade_events_table(pool: &sqlx::SqlitePool) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_trade_events_table", + r#" +CREATE TABLE IF NOT EXISTS kb_trade_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + dex_id INTEGER NOT NULL, + pool_id INTEGER NOT NULL, + pair_id INTEGER NOT NULL, + transaction_id INTEGER NOT NULL, + decoded_event_id INTEGER NOT NULL UNIQUE, + signature TEXT NOT NULL, + slot INTEGER NULL, + trade_side TEXT NOT NULL, + base_token_id INTEGER NOT NULL, + quote_token_id INTEGER NOT NULL, + base_amount_raw TEXT NULL, + quote_amount_raw TEXT NULL, + price_quote_per_base REAL NULL, + source_kind INTEGER NOT NULL, + source_endpoint_name TEXT NULL, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY(dex_id) REFERENCES kb_dexes(id) ON DELETE CASCADE, + FOREIGN KEY(pool_id) REFERENCES kb_pools(id) ON DELETE CASCADE, + FOREIGN KEY(pair_id) REFERENCES kb_pairs(id) ON DELETE CASCADE, + FOREIGN KEY(transaction_id) REFERENCES kb_chain_transactions(id) ON DELETE CASCADE, + FOREIGN KEY(decoded_event_id) REFERENCES kb_dex_decoded_events(id) ON DELETE CASCADE, + FOREIGN KEY(base_token_id) REFERENCES kb_tokens(id) ON DELETE CASCADE, + FOREIGN KEY(quote_token_id) REFERENCES kb_tokens(id) ON DELETE CASCADE +) + "#, + ) + .await +} + +async fn create_kb_idx_trade_events_pair_id(pool: &sqlx::SqlitePool) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_idx_trade_events_pair_id", + r#" +CREATE INDEX IF NOT EXISTS kb_idx_trade_events_pair_id +ON kb_trade_events(pair_id) + "#, + ) + .await +} + +async fn create_kb_idx_trade_events_pool_id(pool: &sqlx::SqlitePool) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_idx_trade_events_pool_id", + r#" +CREATE INDEX IF NOT EXISTS kb_idx_trade_events_pool_id +ON kb_trade_events(pool_id) + "#, + ) + .await +} + +async fn create_kb_pair_metrics_table(pool: &sqlx::SqlitePool) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_pair_metrics_table", + r#" +CREATE TABLE IF NOT EXISTS kb_pair_metrics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + pair_id INTEGER NOT NULL UNIQUE, + first_slot INTEGER NULL, + last_slot INTEGER NULL, + first_signature TEXT NULL, + last_signature TEXT NULL, + trade_count INTEGER NOT NULL, + buy_count INTEGER NOT NULL, + sell_count INTEGER NOT NULL, + cumulative_base_amount_raw TEXT NULL, + cumulative_quote_amount_raw TEXT NULL, + last_price_quote_per_base REAL NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY(pair_id) REFERENCES kb_pairs(id) ON DELETE CASCADE +) + "#, + ) + .await +} diff --git a/kb_lib/src/lib.rs b/kb_lib/src/lib.rs index ee7b69a..f56069b 100644 --- a/kb_lib/src/lib.rs +++ b/kb_lib/src/lib.rs @@ -23,6 +23,7 @@ mod launch_origin; mod pool_origin; mod solana_pubsub_ws; mod tracing; +mod trade_aggregation; mod tx_model; mod tx_resolution; mod types; @@ -82,6 +83,8 @@ pub use db::KbOnchainObservationDto; pub use db::KbOnchainObservationEntity; pub use db::KbPairDto; pub use db::KbPairEntity; +pub use db::KbPairMetricDto; +pub use db::KbPairMetricEntity; pub use db::KbPoolDto; pub use db::KbPoolEntity; pub use db::KbPoolKind; @@ -102,6 +105,8 @@ pub use db::KbTokenDto; pub use db::KbTokenEntity; pub use db::KbTokenMintEventDto; pub use db::KbTokenMintEventEntity; +pub use db::KbTradeEventDto; +pub use db::KbTradeEventEntity; pub use db::KbWalletDto; pub use db::KbWalletEntity; pub use db::KbWalletParticipationDto; @@ -119,10 +124,12 @@ pub use db::get_launch_surface_by_code; pub use db::get_launch_surface_key_by_match; pub use db::get_observed_token_by_mint; pub use db::get_pair_by_pool_id; +pub use db::get_pair_metric_by_pair_id; pub use db::get_pool_by_address; pub use db::get_pool_listing_by_pool_id; pub use db::get_pool_origin_by_pool_id; pub use db::get_token_by_mint; +pub use db::get_trade_event_by_decoded_event_id; pub use db::get_wallet_by_address; pub use db::get_wallet_participation_by_unique_key; pub use db::insert_analysis_signal; @@ -139,6 +146,7 @@ pub use db::list_launch_attributions_by_pool_id; pub use db::list_launch_surface_keys_by_surface_id; pub use db::list_launch_surfaces; pub use db::list_observed_tokens; +pub use db::list_pair_metrics; pub use db::list_pairs; pub use db::list_pool_listings; pub use db::list_pool_origins; @@ -153,6 +161,7 @@ pub use db::list_recent_onchain_observations; pub use db::list_recent_swaps; pub use db::list_recent_token_burn_events; pub use db::list_recent_token_mint_events; +pub use db::list_trade_events_by_pair_id; pub use db::list_wallet_participations_by_pool_id; pub use db::list_wallet_participations_by_wallet_id; pub use db::list_wallets; @@ -169,6 +178,7 @@ pub use db::upsert_launch_surface_key; pub use db::upsert_liquidity_event; pub use db::upsert_observed_token; pub use db::upsert_pair; +pub use db::upsert_pair_metric; pub use db::upsert_pool; pub use db::upsert_pool_listing; pub use db::upsert_pool_origin; @@ -177,6 +187,7 @@ pub use db::upsert_swap; pub use db::upsert_token; pub use db::upsert_token_burn_event; pub use db::upsert_token_mint_event; +pub use db::upsert_trade_event; pub use db::upsert_wallet; pub use db::upsert_wallet_participation; pub use detect::KbDetectionObservationInput; @@ -268,6 +279,8 @@ pub use solana_pubsub_ws::parse_kb_solana_ws_typed_notification; pub use solana_pubsub_ws::parse_kb_solana_ws_typed_notification_from_event; pub use tracing::KbTracingGuard; pub use tracing::init_tracing; +pub use trade_aggregation::KbTradeAggregationResult; +pub use trade_aggregation::KbTradeAggregationService; pub use tx_model::KbTransactionModelService; pub use tx_resolution::KbTransactionResolutionOutcome; pub use tx_resolution::KbTransactionResolutionRequest; diff --git a/kb_lib/src/trade_aggregation.rs b/kb_lib/src/trade_aggregation.rs new file mode 100644 index 0000000..f4e5d9e --- /dev/null +++ b/kb_lib/src/trade_aggregation.rs @@ -0,0 +1,693 @@ +// file: kb_lib/src/trade_aggregation.rs + +//! Cross-DEX trade aggregation service. + +/// One trade-aggregation result. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbTradeAggregationResult { + /// Persisted trade-event id. + pub trade_event_id: i64, + /// Persisted pair-metric id. + pub pair_metric_id: i64, + /// Related pair id. + pub pair_id: i64, + /// Related pool id. + pub pool_id: i64, + /// Whether the trade-event row was newly created. + pub created_trade_event: bool, +} + +/// Trade-aggregation service. +#[derive(Debug, Clone)] +pub struct KbTradeAggregationService { + database: std::sync::Arc, + persistence: crate::KbDetectionPersistenceService, +} + +impl KbTradeAggregationService { + /// Creates a new trade-aggregation service. + pub fn new(database: std::sync::Arc) -> Self { + let persistence = crate::KbDetectionPersistenceService::new(database.clone()); + Self { + database, + persistence, + } + } + + /// Records normalized trade events and updates pair metrics for one transaction signature. + pub async fn record_transaction_by_signature( + &self, + signature: &str, + ) -> Result, crate::KbError> { + let transaction_result = + crate::get_chain_transaction_by_signature(self.database.as_ref(), signature).await; + let transaction_option = match transaction_result { + Ok(transaction_option) => transaction_option, + Err(error) => return Err(error), + }; + let transaction = match transaction_option { + Some(transaction) => transaction, + None => { + return Err(crate::KbError::InvalidState(format!( + "cannot aggregate trades for unknown transaction '{}'", + signature + ))); + } + }; + let transaction_id = match transaction.id { + Some(transaction_id) => transaction_id, + None => { + return Err(crate::KbError::InvalidState(format!( + "transaction '{}' has no internal id", + signature + ))); + } + }; + let decoded_events_result = crate::list_dex_decoded_events_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 results = std::vec::Vec::new(); + for decoded_event in &decoded_events { + if !decoded_event.event_kind.ends_with(".swap") { + continue; + } + let decoded_event_id = match decoded_event.id { + Some(decoded_event_id) => decoded_event_id, + None => { + return Err(crate::KbError::InvalidState( + "decoded event has no internal id".to_string(), + )); + } + }; + let existing_trade_result = crate::get_trade_event_by_decoded_event_id( + self.database.as_ref(), + decoded_event_id, + ) + .await; + let existing_trade_option = match existing_trade_result { + Ok(existing_trade_option) => existing_trade_option, + Err(error) => return Err(error), + }; + let pool_address = match decoded_event.pool_account.clone() { + Some(pool_address) => pool_address, + None => continue, + }; + let pool_result = + crate::get_pool_by_address(self.database.as_ref(), pool_address.as_str()).await; + let pool_option = match pool_result { + Ok(pool_option) => pool_option, + Err(error) => return Err(error), + }; + let pool = match pool_option { + Some(pool) => pool, + None => continue, + }; + let pool_id = match pool.id { + Some(pool_id) => pool_id, + None => { + return Err(crate::KbError::InvalidState(format!( + "pool '{}' has no internal id", + pool.address + ))); + } + }; + let pair_result = crate::get_pair_by_pool_id(self.database.as_ref(), pool_id).await; + let pair_option = match pair_result { + Ok(pair_option) => pair_option, + Err(error) => return Err(error), + }; + let pair = match pair_option { + Some(pair) => pair, + None => continue, + }; + let pair_id = match pair.id { + Some(pair_id) => pair_id, + None => { + return Err(crate::KbError::InvalidState(format!( + "pair for pool '{}' has no internal id", + pool_id + ))); + } + }; + let payload_result = + serde_json::from_str::(decoded_event.payload_json.as_str()); + let payload = match payload_result { + Ok(payload) => payload, + Err(error) => { + return Err(crate::KbError::Json(format!( + "cannot parse decoded_event payload_json '{}': {}", + decoded_event.payload_json, error + ))); + } + }; + let trade_side = kb_extract_trade_side(&payload); + let base_amount_raw = kb_extract_amount_string( + &payload, + &[ + "baseAmountRaw", + "base_amount_raw", + "baseAmount", + "amountBase", + "amountInBase", + ], + ); + let quote_amount_raw = kb_extract_amount_string( + &payload, + &[ + "quoteAmountRaw", + "quote_amount_raw", + "quoteAmount", + "amountQuote", + "amountOutQuote", + ], + ); + let price_quote_per_base = + kb_compute_price_quote_per_base(base_amount_raw.clone(), quote_amount_raw.clone()); + let slot_i64 = kb_convert_slot_to_i64(transaction.slot); + let created_trade_event = existing_trade_option.is_none(); + let trade_event_id = if let Some(existing_trade) = existing_trade_option { + match existing_trade.id { + Some(trade_event_id) => trade_event_id, + None => { + return Err(crate::KbError::InvalidState( + "trade event has no internal id".to_string(), + )); + } + } + } else { + let trade_event_dto = crate::KbTradeEventDto::new( + pool.dex_id, + pool_id, + pair_id, + transaction_id, + decoded_event_id, + transaction.signature.clone(), + slot_i64, + trade_side, + pair.base_token_id, + pair.quote_token_id, + base_amount_raw.clone(), + quote_amount_raw.clone(), + price_quote_per_base, + crate::KbObservationSourceKind::Dex, + transaction.source_endpoint_name.clone(), + decoded_event.payload_json.clone(), + ); + let upsert_result = + crate::upsert_trade_event(self.database.as_ref(), &trade_event_dto).await; + match upsert_result { + Ok(trade_event_id) => trade_event_id, + Err(error) => return Err(error), + } + }; + let pair_metric_result = + crate::get_pair_metric_by_pair_id(self.database.as_ref(), pair_id).await; + let pair_metric_option = match pair_metric_result { + Ok(pair_metric_option) => pair_metric_option, + Err(error) => return Err(error), + }; + let pair_metric_id = if let Some(existing_metric) = pair_metric_option { + let existing_metric_id = match existing_metric.id { + Some(existing_metric_id) => existing_metric_id, + None => { + return Err(crate::KbError::InvalidState( + "pair metric has no internal id".to_string(), + )); + } + }; + if created_trade_event { + let mut updated_metric = existing_metric.clone(); + kb_apply_trade_to_pair_metric( + &mut updated_metric, + slot_i64, + transaction.signature.clone(), + trade_side, + base_amount_raw.clone(), + quote_amount_raw.clone(), + price_quote_per_base, + ); + let upsert_result = + crate::upsert_pair_metric(self.database.as_ref(), &updated_metric).await; + if let Err(error) = upsert_result { + return Err(error); + } + } + existing_metric_id + } else { + let mut new_metric = crate::KbPairMetricDto::new(pair_id); + kb_apply_trade_to_pair_metric( + &mut new_metric, + slot_i64, + transaction.signature.clone(), + trade_side, + base_amount_raw.clone(), + quote_amount_raw.clone(), + price_quote_per_base, + ); + let upsert_result = + crate::upsert_pair_metric(self.database.as_ref(), &new_metric).await; + match upsert_result { + Ok(pair_metric_id) => pair_metric_id, + Err(error) => return Err(error), + } + }; + if created_trade_event { + let payload = serde_json::json!({ + "pairId": pair_id, + "poolId": pool_id, + "tradeEventId": trade_event_id, + "tradeSide": format!("{:?}", trade_side), + "baseAmountRaw": base_amount_raw, + "quoteAmountRaw": quote_amount_raw, + "priceQuotePerBase": price_quote_per_base, + "transactionSignature": transaction.signature + }); + let observation_result = self + .persistence + .record_observation(&crate::KbDetectionObservationInput::new( + "dex.trade_aggregation".to_string(), + crate::KbObservationSourceKind::Dex, + transaction.source_endpoint_name.clone(), + transaction.signature.clone(), + transaction.slot, + payload.clone(), + )) + .await; + let observation_id = match observation_result { + Ok(observation_id) => observation_id, + Err(error) => return Err(error), + }; + let signal_result = self + .persistence + .record_signal(&crate::KbDetectionSignalInput::new( + "signal.dex.trade_aggregation.recorded".to_string(), + crate::KbAnalysisSignalSeverity::Low, + transaction.signature.clone(), + Some(observation_id), + None, + payload, + )) + .await; + if let Err(error) = signal_result { + return Err(error); + } + } + results.push(crate::KbTradeAggregationResult { + trade_event_id, + pair_metric_id, + pair_id, + pool_id, + created_trade_event, + }); + } + Ok(results) + } +} + +fn kb_convert_slot_to_i64(slot: std::option::Option) -> std::option::Option { + match slot { + Some(slot) => match i64::try_from(slot) { + Ok(slot) => Some(slot), + Err(_) => None, + }, + None => None, + } +} + +fn kb_extract_trade_side(payload: &serde_json::Value) -> crate::KbSwapTradeSide { + let trade_side_option = kb_extract_string_by_candidate_keys(payload, &["tradeSide"]); + match trade_side_option.as_deref() { + Some("BuyBase") => crate::KbSwapTradeSide::BuyBase, + Some("SellBase") => crate::KbSwapTradeSide::SellBase, + _ => crate::KbSwapTradeSide::Unknown, + } +} + +fn kb_extract_amount_string( + payload: &serde_json::Value, + candidate_keys: &[&str], +) -> std::option::Option { + kb_extract_scalar_as_string_by_candidate_keys(payload, candidate_keys) +} + +fn kb_compute_price_quote_per_base( + base_amount_raw: std::option::Option, + quote_amount_raw: std::option::Option, +) -> std::option::Option { + let base_amount_text = match base_amount_raw { + Some(base_amount_text) => base_amount_text, + None => return None, + }; + let quote_amount_text = match quote_amount_raw { + Some(quote_amount_text) => quote_amount_text, + None => return None, + }; + let base_amount_result = base_amount_text.parse::(); + let base_amount = match base_amount_result { + Ok(base_amount) => base_amount, + Err(_) => return None, + }; + if base_amount <= 0.0 { + return None; + } + let quote_amount_result = quote_amount_text.parse::(); + let quote_amount = match quote_amount_result { + Ok(quote_amount) => quote_amount, + Err(_) => return None, + }; + Some(quote_amount / base_amount) +} + +fn kb_apply_trade_to_pair_metric( + metric: &mut crate::KbPairMetricDto, + slot: std::option::Option, + signature: std::string::String, + trade_side: crate::KbSwapTradeSide, + base_amount_raw: std::option::Option, + quote_amount_raw: std::option::Option, + price_quote_per_base: std::option::Option, +) { + metric.trade_count += 1; + if trade_side == crate::KbSwapTradeSide::BuyBase { + metric.buy_count += 1; + } + if trade_side == crate::KbSwapTradeSide::SellBase { + metric.sell_count += 1; + } + if metric.first_slot.is_none() { + metric.first_slot = slot; + } + if metric.first_signature.is_none() { + metric.first_signature = Some(signature.clone()); + } + metric.last_slot = slot; + metric.last_signature = Some(signature); + metric.cumulative_base_amount_raw = + kb_add_raw_amounts(metric.cumulative_base_amount_raw.clone(), base_amount_raw); + metric.cumulative_quote_amount_raw = + kb_add_raw_amounts(metric.cumulative_quote_amount_raw.clone(), quote_amount_raw); + if price_quote_per_base.is_some() { + metric.last_price_quote_per_base = price_quote_per_base; + } + metric.updated_at = chrono::Utc::now(); +} + +fn kb_add_raw_amounts( + left: std::option::Option, + right: std::option::Option, +) -> std::option::Option { + match (left, right) { + (None, None) => None, + (Some(left), None) => Some(left), + (None, Some(right)) => Some(right), + (Some(left), Some(right)) => { + let left_value_result = left.parse::(); + let left_value = match left_value_result { + Ok(left_value) => left_value, + Err(_) => return Some(left), + }; + let right_value_result = right.parse::(); + let right_value = match right_value_result { + Ok(right_value) => right_value, + Err(_) => return Some(left), + }; + Some((left_value + right_value).to_string()) + } + } +} + +fn kb_extract_string_by_candidate_keys( + value: &serde_json::Value, + candidate_keys: &[&str], +) -> std::option::Option { + if let Some(object) = value.as_object() { + for candidate_key in candidate_keys { + let direct_option = object.get(*candidate_key); + if let Some(direct) = direct_option { + let direct_text_option = direct.as_str(); + if let Some(direct_text) = direct_text_option { + return Some(direct_text.to_string()); + } + } + } + for nested_value in object.values() { + let nested_result = kb_extract_string_by_candidate_keys(nested_value, candidate_keys); + if nested_result.is_some() { + return nested_result; + } + } + return None; + } + if let Some(array) = value.as_array() { + for nested_value in array { + let nested_result = kb_extract_string_by_candidate_keys(nested_value, candidate_keys); + if nested_result.is_some() { + return nested_result; + } + } + } + None +} + +fn kb_extract_scalar_as_string_by_candidate_keys( + value: &serde_json::Value, + candidate_keys: &[&str], +) -> std::option::Option { + if let Some(object) = value.as_object() { + for candidate_key in candidate_keys { + let direct_option = object.get(*candidate_key); + if let Some(direct) = direct_option { + if let Some(text) = direct.as_str() { + return Some(text.to_string()); + } + if let Some(number) = direct.as_i64() { + return Some(number.to_string()); + } + if let Some(number) = direct.as_u64() { + return Some(number.to_string()); + } + if let Some(number) = direct.as_f64() { + return Some(number.to_string()); + } + } + } + for nested_value in object.values() { + let nested_result = + kb_extract_scalar_as_string_by_candidate_keys(nested_value, candidate_keys); + if nested_result.is_some() { + return nested_result; + } + } + return None; + } + if let Some(array) = value.as_array() { + for nested_value in array { + let nested_result = + kb_extract_scalar_as_string_by_candidate_keys(nested_value, candidate_keys); + if nested_result.is_some() { + return nested_result; + } + } + } + + None +} + +#[cfg(test)] +mod tests { + async fn make_database() -> std::sync::Arc { + let tempdir_result = tempfile::tempdir(); + let tempdir = match tempdir_result { + Ok(tempdir) => tempdir, + Err(error) => panic!("tempdir must succeed: {}", error), + }; + let database_path = tempdir.path().join("trade_aggregation.sqlite3"); + let config = crate::KbDatabaseConfig { + enabled: true, + backend: crate::KbDatabaseBackend::Sqlite, + sqlite: crate::KbSqliteDatabaseConfig { + path: database_path.to_string_lossy().to_string(), + create_if_missing: true, + busy_timeout_ms: 5000, + max_connections: 1, + auto_initialize_schema: true, + use_wal: true, + }, + }; + let database_result = crate::KbDatabase::connect_and_initialize(&config).await; + let database = match database_result { + Ok(database) => database, + Err(error) => panic!("database init must succeed: {}", error), + }; + std::sync::Arc::new(database) + } + + async fn seed_fluxbeam_swap_transaction( + database: std::sync::Arc, + signature: &str, + base_amount_raw: &str, + quote_amount_raw: &str, + ) { + let transaction_model = crate::KbTransactionModelService::new(database.clone()); + let dex_decode = crate::KbDexDecodeService::new(database.clone()); + let dex_detect = crate::KbDexDetectService::new(database); + let resolved_transaction = serde_json::json!({ + "slot": 940001, + "blockTime": 1779400001, + "version": 0, + "transaction": { + "message": { + "instructions": [ + { + "programId": crate::KB_FLUXBEAM_PROGRAM_ID, + "program": "fluxbeam", + "stackHeight": 1, + "accounts": [ + "TradeAggPool111", + "TradeAggLpMint111", + "TradeAggTokenA111", + "So11111111111111111111111111111111111111112" + ], + "parsed": { + "info": { + "instruction": "swap", + "pool": "TradeAggPool111", + "tokenA": "TradeAggTokenA111", + "tokenB": "So11111111111111111111111111111111111111112", + "baseAmountRaw": base_amount_raw, + "quoteAmountRaw": quote_amount_raw + } + }, + "data": "opaque" + } + ] + } + }, + "meta": { + "err": null, + "logMessages": [ + "Program log: Instruction: Swap", + "Program log: buy" + ] + } + }); + let project_result = transaction_model + .persist_resolved_transaction( + signature, + Some("helius_primary_http".to_string()), + &resolved_transaction, + ) + .await; + if let Err(error) = project_result { + panic!("projection must succeed: {}", error); + } + let decode_result = dex_decode.decode_transaction_by_signature(signature).await; + if let Err(error) = decode_result { + panic!("dex decode must succeed: {}", error); + } + let detect_result = dex_detect.detect_transaction_by_signature(signature).await; + if let Err(error) = detect_result { + panic!("dex detect must succeed: {}", error); + } + } + + #[tokio::test] + async fn record_transaction_by_signature_creates_trade_event_and_pair_metric() { + let database = make_database().await; + seed_fluxbeam_swap_transaction(database.clone(), "sig-trade-aggregation-1", "1000", "2500") + .await; + let service = crate::KbTradeAggregationService::new(database.clone()); + let record_result = service + .record_transaction_by_signature("sig-trade-aggregation-1") + .await; + let results = match record_result { + Ok(results) => results, + Err(error) => panic!("trade aggregation must succeed: {}", error), + }; + assert_eq!(results.len(), 1); + assert!(results[0].created_trade_event); + let trade_events_result = + crate::list_trade_events_by_pair_id(database.as_ref(), results[0].pair_id).await; + let trade_events = match trade_events_result { + Ok(trade_events) => trade_events, + Err(error) => panic!("trade event list must succeed: {}", error), + }; + assert_eq!(trade_events.len(), 1); + assert_eq!(trade_events[0].trade_side, crate::KbSwapTradeSide::BuyBase); + assert_eq!(trade_events[0].base_amount_raw, Some("1000".to_string())); + assert_eq!(trade_events[0].quote_amount_raw, Some("2500".to_string())); + assert_eq!(trade_events[0].price_quote_per_base, Some(2.5)); + let pair_metric_result = + crate::get_pair_metric_by_pair_id(database.as_ref(), results[0].pair_id).await; + let pair_metric_option = match pair_metric_result { + Ok(pair_metric_option) => pair_metric_option, + Err(error) => panic!("pair metric fetch must succeed: {}", error), + }; + let pair_metric = match pair_metric_option { + Some(pair_metric) => pair_metric, + None => panic!("pair metric must exist"), + }; + assert_eq!(pair_metric.trade_count, 1); + assert_eq!(pair_metric.buy_count, 1); + assert_eq!(pair_metric.sell_count, 0); + assert_eq!( + pair_metric.cumulative_base_amount_raw, + Some("1000".to_string()) + ); + assert_eq!( + pair_metric.cumulative_quote_amount_raw, + Some("2500".to_string()) + ); + assert_eq!(pair_metric.last_price_quote_per_base, Some(2.5)); + } + + #[tokio::test] + async fn record_transaction_by_signature_is_idempotent() { + let database = make_database().await; + seed_fluxbeam_swap_transaction(database.clone(), "sig-trade-aggregation-2", "1000", "2500") + .await; + let service = crate::KbTradeAggregationService::new(database.clone()); + let first_result = service + .record_transaction_by_signature("sig-trade-aggregation-2") + .await; + let first_results = match first_result { + Ok(first_results) => first_results, + Err(error) => panic!("first trade aggregation must succeed: {}", error), + }; + assert_eq!(first_results.len(), 1); + assert!(first_results[0].created_trade_event); + let second_result = service + .record_transaction_by_signature("sig-trade-aggregation-2") + .await; + let second_results = match second_result { + Ok(second_results) => second_results, + Err(error) => panic!("second trade aggregation must succeed: {}", error), + }; + assert_eq!(second_results.len(), 1); + assert!(!second_results[0].created_trade_event); + let trade_events_result = + crate::list_trade_events_by_pair_id(database.as_ref(), first_results[0].pair_id).await; + let trade_events = match trade_events_result { + Ok(trade_events) => trade_events, + Err(error) => panic!("trade event list must succeed: {}", error), + }; + assert_eq!(trade_events.len(), 1); + let pair_metric_result = + crate::get_pair_metric_by_pair_id(database.as_ref(), first_results[0].pair_id).await; + let pair_metric_option = match pair_metric_result { + Ok(pair_metric_option) => pair_metric_option, + Err(error) => panic!("pair metric fetch must succeed: {}", error), + }; + let pair_metric = match pair_metric_option { + Some(pair_metric) => pair_metric, + None => panic!("pair metric must exist"), + }; + assert_eq!(pair_metric.trade_count, 1); + } +} diff --git a/kb_lib/src/tx_resolution.rs b/kb_lib/src/tx_resolution.rs index 106d236..ec57805 100644 --- a/kb_lib/src/tx_resolution.rs +++ b/kb_lib/src/tx_resolution.rs @@ -105,6 +105,7 @@ pub struct KbTransactionResolutionService { launch_origin_service: crate::KbLaunchOriginService, pool_origin_service: crate::KbPoolOriginService, wallet_observation_service: crate::KbWalletObservationService, + trade_aggregation_service: crate::KbTradeAggregationService, resolved_signatures: std::sync::Arc>>, } @@ -123,6 +124,7 @@ impl KbTransactionResolutionService { let launch_origin_service = crate::KbLaunchOriginService::new(database.clone()); let pool_origin_service = crate::KbPoolOriginService::new(database.clone()); let wallet_observation_service = crate::KbWalletObservationService::new(database.clone()); + let trade_aggregation_service = crate::KbTradeAggregationService::new(database.clone()); Self { http_pool, persistence, @@ -133,6 +135,7 @@ impl KbTransactionResolutionService { launch_origin_service, pool_origin_service, wallet_observation_service, + trade_aggregation_service, resolved_signatures: std::sync::Arc::new(tokio::sync::Mutex::new( std::collections::HashSet::new(), )), @@ -318,6 +321,11 @@ impl KbTransactionResolutionService { .dex_detect_service .detect_transaction_by_signature(request.signature.as_str()) .await; + let detection_results = match detection_results_result { + Ok(detection_results) => detection_results, + Err(error) => return Err(error), + }; + let detected_object_count = detection_results.len(); let launch_attributions_result = self .launch_origin_service .attribute_transaction_by_signature(request.signature.as_str()) @@ -345,11 +353,15 @@ impl KbTransactionResolutionService { Err(error) => return Err(error), }; let wallet_participation_count = wallet_observations.len(); - let detection_results = match detection_results_result { - Ok(detection_results) => detection_results, + let trade_aggregations_result = self + .trade_aggregation_service + .record_transaction_by_signature(request.signature.as_str()) + .await; + let trade_aggregations = match trade_aggregations_result { + Ok(trade_aggregations) => trade_aggregations, Err(error) => return Err(error), }; - let detected_object_count = detection_results.len(); + let trade_event_count = trade_aggregations.len(); let payload = serde_json::json!({ "status": "resolved", "signature": request.signature.clone(), @@ -362,6 +374,7 @@ impl KbTransactionResolutionService { "launchAttributionCount": launch_attribution_count, "poolOriginCount": pool_origin_count, "walletParticipationCount": wallet_participation_count, + "tradeEventCount": trade_event_count, "triggerPayload": request.trigger_payload.clone(), "transaction": transaction_value });