0.7.16

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  

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"

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 :
+- 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,
-- préparer des agrégations de prix/volume par paire,
+- prise en charge des compteurs `trade_count`, `buy_count`, `sell_count`,
-- introduire la base des futures candles et séries temporelles,
+- prise en charge optionnelle des volumes bruts `base` / `quote` et du dernier prix dérivé `quote_per_base`,
-- permettre plus tard le calcul d’OHLCV, volume, nombre de trades et liquidité par fenêtre,
+- branchement automatique dans le pipeline de résolution transactionnelle,
-- préparer le terrain pour la couche analytique `0.8.x`.
+- 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.

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;

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;

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<i64>,
+    /// Related pair id.
+    pub pair_id: i64,
+    /// Optional first observed slot.
+    pub first_slot: std::option::Option<i64>,
+    /// Optional last observed slot.
+    pub last_slot: std::option::Option<i64>,
+    /// Optional first observed signature.
+    pub first_signature: std::option::Option<std::string::String>,
+    /// Optional last observed signature.
+    pub last_signature: std::option::Option<std::string::String>,
+    /// 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<std::string::String>,
+    /// Optional cumulative raw quote volume.
+    pub cumulative_quote_amount_raw: std::option::Option<std::string::String>,
+    /// Optional last derived quote-per-base price.
+    pub last_price_quote_per_base: std::option::Option<f64>,
+    /// Creation timestamp.
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    /// Update timestamp.
+    pub updated_at: chrono::DateTime<chrono::Utc>,
+}
+
+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<crate::KbPairMetricEntity> for KbPairMetricDto {
+    type Error = crate::KbError;
+
+    fn try_from(entity: crate::KbPairMetricEntity) -> Result<Self, Self::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 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,
+        })
+    }
+}

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<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<i64>,
+    /// 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<std::string::String>,
+    /// Optional raw quote amount.
+    pub quote_amount_raw: std::option::Option<std::string::String>,
+    /// Optional derived quote-per-base price.
+    pub price_quote_per_base: std::option::Option<f64>,
+    /// Observation source kind.
+    pub source_kind: crate::KbObservationSourceKind,
+    /// Optional logical source endpoint name.
+    pub source_endpoint_name: std::option::Option<std::string::String>,
+    /// Persisted payload JSON.
+    pub payload_json: std::string::String,
+    /// Creation timestamp.
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    /// Update timestamp.
+    pub updated_at: chrono::DateTime<chrono::Utc>,
+}
+
+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<i64>,
+        trade_side: crate::KbSwapTradeSide,
+        base_token_id: i64,
+        quote_token_id: i64,
+        base_amount_raw: std::option::Option<std::string::String>,
+        quote_amount_raw: std::option::Option<std::string::String>,
+        price_quote_per_base: std::option::Option<f64>,
+        source_kind: crate::KbObservationSourceKind,
+        source_endpoint_name: std::option::Option<std::string::String>,
+        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<crate::KbTradeEventEntity> for KbTradeEventDto {
+    type Error = crate::KbError;
+
+    fn try_from(entity: crate::KbTradeEventEntity) -> Result<Self, Self::Error> {
+        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,
+    }
+}

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;

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<i64>,
+    /// Optional last observed slot.
+    pub last_slot: std::option::Option<i64>,
+    /// Optional first observed signature.
+    pub first_signature: std::option::Option<std::string::String>,
+    /// Optional last observed signature.
+    pub last_signature: std::option::Option<std::string::String>,
+    /// 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<std::string::String>,
+    /// Optional cumulative raw quote volume.
+    pub cumulative_quote_amount_raw: std::option::Option<std::string::String>,
+    /// Optional last derived quote-per-base price.
+    pub last_price_quote_per_base: std::option::Option<f64>,
+    /// 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,
+}

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<i64>,
+    /// 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<std::string::String>,
+    /// Optional raw quote amount.
+    pub quote_amount_raw: std::option::Option<std::string::String>,
+    /// Optional derived quote-per-base price.
+    pub price_quote_per_base: std::option::Option<f64>,
+    /// Observation source kind.
+    pub source_kind: i16,
+    /// Optional logical source endpoint name.
+    pub source_endpoint_name: std::option::Option<std::string::String>,
+    /// 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,
+}

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;

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<i64, crate::KbError> {
+    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::<sqlx::Sqlite, i64>(
+                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<std::option::Option<crate::KbPairMetricDto>, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbPairMetricEntity>(
+                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<std::vec::Vec<crate::KbPairMetricDto>, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbPairMetricEntity>(
+                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)
+        }
+    }
+}

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<i64, crate::KbError> {
+    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::<sqlx::Sqlite, i64>(
+                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<std::option::Option<crate::KbTradeEventDto>, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbTradeEventEntity>(
+                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<std::vec::Vec<crate::KbTradeEventDto>, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbTradeEventEntity>(
+                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",
+    }
+}

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
+}

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;

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<crate::KbDatabase>,
+    persistence: crate::KbDetectionPersistenceService,
+}
+
+impl KbTradeAggregationService {
+    /// Creates a new trade-aggregation service.
+    pub fn new(database: std::sync::Arc<crate::KbDatabase>) -> 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<std::vec::Vec<crate::KbTradeAggregationResult>, 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::<serde_json::Value>(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<u64>) -> std::option::Option<i64> {
+    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<std::string::String> {
+    kb_extract_scalar_as_string_by_candidate_keys(payload, candidate_keys)
+}
+
+fn kb_compute_price_quote_per_base(
+    base_amount_raw: std::option::Option<std::string::String>,
+    quote_amount_raw: std::option::Option<std::string::String>,
+) -> std::option::Option<f64> {
+    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::<f64>();
+    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::<f64>();
+    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<i64>,
+    signature: std::string::String,
+    trade_side: crate::KbSwapTradeSide,
+    base_amount_raw: std::option::Option<std::string::String>,
+    quote_amount_raw: std::option::Option<std::string::String>,
+    price_quote_per_base: std::option::Option<f64>,
+) {
+    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<std::string::String>,
+    right: std::option::Option<std::string::String>,
+) -> std::option::Option<std::string::String> {
+    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::<i128>();
+            let left_value = match left_value_result {
+                Ok(left_value) => left_value,
+                Err(_) => return Some(left),
+            };
+            let right_value_result = right.parse::<i128>();
+            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<std::string::String> {
+    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<std::string::String> {
+    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<crate::KbDatabase> {
+        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<crate::KbDatabase>,
+        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);
+    }
+}

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<tokio::sync::Mutex<std::collections::HashSet<std::string::String>>>,
 }
@@ -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 {
+        let trade_aggregations_result = self
-            Ok(detection_results) => detection_results,
+            .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
         });