0.7.24

kb_lib/src/dex_decode.rs

@@ -37,6 +37,42 @@ impl KbDexDecodeService {
         }
     }
 
+    async fn decode_and_persist_raydium_clmm_events(
+        &self,
+        transaction: &crate::KbChainTransactionDto,
+        instructions: &[crate::KbChainInstructionDto],
+    ) -> Result<std::vec::Vec<crate::KbDexDecodedEventDto>, crate::KbError> {
+        let mut persisted = std::vec::Vec::new();
+        for instruction in instructions {
+            let program_id = match instruction.program_id.as_ref() {
+                Some(program_id) => program_id,
+                None => continue,
+            };
+            if program_id.as_str() != crate::KB_RAYDIUM_CLMM_PROGRAM_ID {
+                continue;
+            }
+            let data_json = match instruction.data_json.as_ref() {
+                Some(data_json) => data_json,
+                None => continue,
+            };
+            let decoded_events = crate::kb_decode_raydium_clmm_instruction(
+                instruction.accounts_json.as_str(),
+                data_json.as_str(),
+            );
+            for decoded_event in &decoded_events {
+                let persist_result = self
+                    .persist_raydium_clmm_event(transaction, instruction, decoded_event)
+                    .await;
+                let persisted_event = match persist_result {
+                    Ok(persisted_event) => persisted_event,
+                    Err(error) => return Err(error),
+                };
+                persisted.push(persisted_event);
+            }
+        }
+        Ok(persisted)
+    }
+
     /// Decodes one projected transaction and persists the decoded events.
     pub async fn decode_transaction_by_signature(
         &self,
@@ -104,6 +140,16 @@ impl KbDexDecodeService {
         for persisted_event in raydium_cpmm_persisted {
             persisted.push(persisted_event);
         }
+        let raydium_clmm_persisted_result = self
+            .decode_and_persist_raydium_clmm_events(&transaction, &instructions)
+            .await;
+        let raydium_clmm_persisted = match raydium_clmm_persisted_result {
+            Ok(raydium_clmm_persisted) => raydium_clmm_persisted,
+            Err(error) => return Err(error),
+        };
+        for persisted_event in raydium_clmm_persisted {
+            persisted.push(persisted_event);
+        }
         let pump_fun_decoded_result = self
             .pump_fun_decoder
             .decode_transaction(&transaction, &instructions);
@@ -248,15 +294,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbDexlabDecodedEvent::CreatePool(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "dexlab",
+                    "dexlab.create_pool",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded dexlab payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -342,15 +387,14 @@ impl KbDexDecodeService {
                 Ok(fetched)
             }
             crate::KbDexlabDecodedEvent::Swap(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "dexlab",
+                    "dexlab.swap",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded dexlab payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -445,15 +489,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbFluxbeamDecodedEvent::CreatePool(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "fluxbeam",
+                    "fluxbeam.create_pool",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded fluxbeam payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -539,15 +582,14 @@ impl KbDexDecodeService {
                 Ok(fetched)
             }
             crate::KbFluxbeamDecodedEvent::Swap(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "fluxbeam",
+                    "fluxbeam.swap",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded fluxbeam payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -642,15 +684,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbOrcaWhirlpoolsDecodedEvent::CreatePool(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "orca_whirlpools",
+                    "orca_whirlpools.create_pool",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded orca whirlpools payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -736,15 +777,14 @@ impl KbDexDecodeService {
                 Ok(fetched)
             }
             crate::KbOrcaWhirlpoolsDecodedEvent::Swap(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "orca_whirlpools",
+                    "orca_whirlpools.swap",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded orca whirlpools payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -840,15 +880,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbMeteoraDammV1DecodedEvent::CreatePool(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "meteora_damm_v1",
+                    "meteora_damm_v1.create_pool",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded meteora damm v1 payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -934,15 +973,14 @@ impl KbDexDecodeService {
                 Ok(fetched)
             }
             crate::KbMeteoraDammV1DecodedEvent::Swap(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "meteora_damm_v1",
+                    "meteora_damm_v1.swap",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded meteora damm v1 payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -1037,15 +1075,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbMeteoraDammV2DecodedEvent::CreatePool(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "meteora_damm_v2",
+                    "meteora_damm_v2.create_pool",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded meteora damm v2 payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -1131,15 +1168,14 @@ impl KbDexDecodeService {
                 Ok(fetched)
             }
             crate::KbMeteoraDammV2DecodedEvent::Swap(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "meteora_damm_v2",
+                    "meteora_damm_v2.swap",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded meteora damm v2 payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -1234,15 +1270,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbMeteoraDbcDecodedEvent::CreatePool(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "meteora_dbc",
+                    "meteora_dbc.create_pool",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded meteora dbc payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -1328,15 +1363,14 @@ impl KbDexDecodeService {
                 Ok(fetched)
             }
             crate::KbMeteoraDbcDecodedEvent::Swap(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "meteora_dbc",
+                    "meteora_dbc.swap",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded meteora dbc payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -1431,15 +1465,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbRaydiumAmmV4DecodedEvent::Initialize2Pool(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "raydium_amm_v4",
+                    "raydium_amm_v4.initialize2_pool",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded raydium payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -1563,6 +1596,141 @@ impl KbDexDecodeService {
         Ok(persisted)
     }
 
+    async fn persist_raydium_clmm_event(
+        &self,
+        transaction: &crate::KbChainTransactionDto,
+        instruction: &crate::KbChainInstructionDto,
+        decoded_event: &crate::KbRaydiumClmmDecodedEvent,
+    ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
+        let transaction_id = match transaction.id {
+            Some(transaction_id) => transaction_id,
+            None => {
+                return Err(crate::KbError::InvalidState(format!(
+                    "transaction '{}' has no internal id",
+                    transaction.signature
+                )));
+            }
+        };
+        let instruction_id = match instruction.id {
+            Some(instruction_id) => instruction_id,
+            None => {
+                return Err(crate::KbError::InvalidState(format!(
+                    "raydium clmm instruction for transaction '{}' has no internal id",
+                    transaction.signature
+                )));
+            }
+        };
+        let event_kind = decoded_event.event_kind().to_string();
+        let raw_payload_json = match decoded_event.to_payload_json() {
+            Some(payload_json) => payload_json,
+            None => {
+                return Err(crate::KbError::Json(
+                    "cannot serialize decoded raydium clmm payload".to_string(),
+                ));
+            }
+        };
+        let payload_json_result = kb_enrich_serialized_dex_decoded_payload(
+            "raydium_clmm",
+            event_kind.as_str(),
+            raw_payload_json.as_str(),
+        );
+        let payload_json = match payload_json_result {
+            Ok(payload_json) => payload_json,
+            Err(error) => return Err(error),
+        };
+        let existing_result = crate::get_dex_decoded_event_by_key(
+            self.database.as_ref(),
+            transaction_id,
+            Some(instruction_id),
+            event_kind.as_str(),
+        )
+        .await;
+        let existing_option = match existing_result {
+            Ok(existing_option) => existing_option,
+            Err(error) => return Err(error),
+        };
+        let already_present = existing_option.is_some();
+        let dto = crate::KbDexDecodedEventDto::new(
+            transaction_id,
+            Some(instruction_id),
+            "raydium_clmm".to_string(),
+            crate::KB_RAYDIUM_CLMM_PROGRAM_ID.to_string(),
+            event_kind.clone(),
+            Some(decoded_event.pool_account().to_string()),
+            None,
+            Some(decoded_event.base_mint().to_string()),
+            Some(decoded_event.quote_mint().to_string()),
+            None,
+            payload_json.clone(),
+        );
+        let upsert_result = crate::upsert_dex_decoded_event(self.database.as_ref(), &dto).await;
+        if let Err(error) = upsert_result {
+            return Err(error);
+        }
+        let fetched_result = crate::get_dex_decoded_event_by_key(
+            self.database.as_ref(),
+            transaction_id,
+            Some(instruction_id),
+            event_kind.as_str(),
+        )
+        .await;
+        let fetched_option = match fetched_result {
+            Ok(fetched_option) => fetched_option,
+            Err(error) => return Err(error),
+        };
+        let fetched = match fetched_option {
+            Some(fetched) => fetched,
+            None => {
+                return Err(crate::KbError::InvalidState(
+                    "decoded raydium clmm event disappeared after upsert".to_string(),
+                ));
+            }
+        };
+        if !already_present {
+            let payload_value_result =
+                serde_json::from_str::<serde_json::Value>(payload_json.as_str());
+            let payload_value = match payload_value_result {
+                Ok(payload_value) => payload_value,
+                Err(error) => {
+                    return Err(crate::KbError::Json(format!(
+                        "cannot parse raydium clmm payload after serialization: {}",
+                        error
+                    )));
+                }
+            };
+            let observation_result = self
+                .persistence
+                .record_observation(&crate::KbDetectionObservationInput::new(
+                    format!("dex.{}", event_kind),
+                    crate::KbObservationSourceKind::HttpRpc,
+                    transaction.source_endpoint_name.clone(),
+                    transaction.signature.clone(),
+                    transaction.slot,
+                    payload_value.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(
+                    format!("signal.dex.{}", event_kind),
+                    crate::KbAnalysisSignalSeverity::Low,
+                    transaction.signature.clone(),
+                    Some(observation_id),
+                    None,
+                    payload_value,
+                ))
+                .await;
+            if let Err(error) = signal_result {
+                return Err(error);
+            }
+        }
+        Ok(fetched)
+    }
+
     async fn persist_raydium_cpmm_event(
         &self,
         transaction: &crate::KbChainTransactionDto,
@@ -1587,7 +1755,8 @@ impl KbDexDecodeService {
                 )));
             }
         };
-        let payload_json = match decoded_event.to_payload_json() {
+        let event_kind = decoded_event.event_kind().to_string();
+        let raw_payload_json = match decoded_event.to_payload_json() {
             Some(payload_json) => payload_json,
             None => {
                 return Err(crate::KbError::Json(
@@ -1595,7 +1764,15 @@ impl KbDexDecodeService {
                 ));
             }
         };
-        let event_kind = decoded_event.event_kind().to_string();
+        let payload_json_result = kb_enrich_serialized_dex_decoded_payload(
+            "raydium_cpmm",
+            event_kind.as_str(),
+            raw_payload_json.as_str(),
+        );
+        let payload_json = match payload_json_result {
+            Ok(payload_json) => payload_json,
+            Err(error) => return Err(error),
+        };
         let existing_result = crate::get_dex_decoded_event_by_key(
             self.database.as_ref(),
             transaction_id,
@@ -1696,15 +1873,14 @@ impl KbDexDecodeService {
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
         match decoded_event {
             crate::KbPumpFunDecodedEvent::CreateV2Token(event) => {
-                let payload_json_result = serde_json::to_string(&event.payload_json);
+                let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+                    "pump_fun",
+                    "pump_fun.create_v2_token",
+                    event.payload_json.clone(),
+                );
                 let payload_json = match payload_json_result {
                     Ok(payload_json) => payload_json,
-                    Err(error) => {
-                        return Err(crate::KbError::Json(format!(
-                            "cannot serialize decoded pump.fun payload: {}",
-                            error
-                        )));
-                    }
+                    Err(error) => return Err(error),
                 };
                 let existing_result = crate::get_dex_decoded_event_by_key(
                     self.database.as_ref(),
@@ -1820,15 +1996,14 @@ impl KbDexDecodeService {
         signal_kind: &str,
         observation_kind: &str,
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
-        let payload_json_result = serde_json::to_string(&event.payload_json);
+        let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+            "pump_fun",
+            event_kind,
+            event.payload_json.clone(),
+        );
         let payload_json = match payload_json_result {
             Ok(payload_json) => payload_json,
-            Err(error) => {
-                return Err(crate::KbError::Json(format!(
-                    "cannot serialize decoded pump.fun trade payload: {}",
-                    error
-                )));
-            }
+            Err(error) => return Err(error),
         };
         let existing_result = crate::get_dex_decoded_event_by_key(
             self.database.as_ref(),
@@ -1950,15 +2125,14 @@ impl KbDexDecodeService {
         signal_kind: &str,
         observation_kind: &str,
     ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
-        let payload_json_result = serde_json::to_string(&event.payload_json);
+        let payload_json_result = kb_enrich_and_serialize_dex_decoded_payload(
+            "pump_swap",
+            event_kind,
+            event.payload_json.clone(),
+        );
         let payload_json = match payload_json_result {
             Ok(payload_json) => payload_json,
-            Err(error) => {
-                return Err(crate::KbError::Json(format!(
-                    "cannot serialize decoded pump swap payload: {}",
-                    error
-                )));
-            }
+            Err(error) => return Err(error),
         };
         let existing_result = crate::get_dex_decoded_event_by_key(
             self.database.as_ref(),
@@ -2044,8 +2218,258 @@ impl KbDexDecodeService {
     }
 }
 
+// Classifies a DEX event kind into a stable business category.
+fn kb_classify_dex_event_category(event_kind: &str) -> &'static str {
+    if kb_is_dex_reward_event_kind(event_kind) {
+        return "reward";
+    }
+    if kb_is_dex_fee_event_kind(event_kind) {
+        return "fee";
+    }
+    if kb_is_dex_liquidity_event_kind(event_kind) {
+        return "liquidity";
+    }
+    if kb_is_dex_pool_lifecycle_event_kind(event_kind) {
+        return "pool_lifecycle";
+    }
+    if kb_is_dex_admin_event_kind(event_kind) {
+        return "admin";
+    }
+    if kb_is_dex_trade_event_kind(event_kind) {
+        return "trade";
+    }
+    "unknown"
+}
+
+// Returns true when the event kind represents a swap-like event.
+fn kb_is_dex_trade_event_kind(event_kind: &str) -> bool {
+    if event_kind.ends_with(".buy") {
+        return true;
+    }
+    if event_kind.ends_with(".sell") {
+        return true;
+    }
+    if event_kind.ends_with(".swap") {
+        return true;
+    }
+    if event_kind.contains(".swap_") {
+        return true;
+    }
+    false
+}
+
+// Returns true when the event kind can directly produce a candle candidate.
+fn kb_is_dex_candle_candidate_event_kind(event_kind: &str) -> bool {
+    if event_kind.contains("router") {
+        return false;
+    }
+    if event_kind.contains("route") {
+        return false;
+    }
+    kb_is_dex_trade_event_kind(event_kind)
+}
+
+// Returns true for liquidity lifecycle changes that must not become candles.
+fn kb_is_dex_liquidity_event_kind(event_kind: &str) -> bool {
+    if event_kind.contains(".deposit") {
+        return true;
+    }
+    if event_kind.contains(".withdraw") {
+        return true;
+    }
+    if event_kind.contains(".increase_liquidity") {
+        return true;
+    }
+    if event_kind.contains(".decrease_liquidity") {
+        return true;
+    }
+    if event_kind.contains(".open_position") {
+        return true;
+    }
+    if event_kind.contains(".close_position") {
+        return true;
+    }
+    false
+}
+
+// Returns true for fee collection events.
+fn kb_is_dex_fee_event_kind(event_kind: &str) -> bool {
+    if event_kind.contains("collect_creator_fee") {
+        return true;
+    }
+    if event_kind.contains("collect_protocol_fee") {
+        return true;
+    }
+    if event_kind.contains("collect_fund_fee") {
+        return true;
+    }
+    if event_kind.contains("collect_fee") {
+        return true;
+    }
+    false
+}
+
+// Returns true for reward/incentive events.
+fn kb_is_dex_reward_event_kind(event_kind: &str) -> bool {
+    if event_kind.contains("reward") {
+        return true;
+    }
+    if event_kind.contains("emission") {
+        return true;
+    }
+    false
+}
+
+// Returns true for pool creation / initialization / migration events.
+fn kb_is_dex_pool_lifecycle_event_kind(event_kind: &str) -> bool {
+    if event_kind.contains(".initialize") {
+        return true;
+    }
+    if event_kind.contains(".initialize_with_permission") {
+        return true;
+    }
+    if event_kind.contains(".create_pool") {
+        return true;
+    }
+    if event_kind.contains(".create_v2_token") {
+        return true;
+    }
+    if event_kind.contains(".migrate") {
+        return true;
+    }
+    false
+}
+
+// Returns true for admin/config/permission changes.
+fn kb_is_dex_admin_event_kind(event_kind: &str) -> bool {
+    if event_kind.contains("admin") {
+        return true;
+    }
+    if event_kind.contains("config") {
+        return true;
+    }
+    if event_kind.contains("permission") {
+        return true;
+    }
+    if event_kind.contains("set_") {
+        return true;
+    }
+    if event_kind.contains("update_") {
+        return true;
+    }
+    false
+}
+
+// Enriches a decoded payload with non-destructive classification metadata.
+fn kb_enrich_dex_decoded_payload(
+    protocol_name: &str,
+    event_kind: &str,
+    payload_json: serde_json::Value,
+) -> serde_json::Value {
+    let event_category = kb_classify_dex_event_category(event_kind);
+    let trade_candidate = kb_is_dex_trade_event_kind(event_kind);
+    let candle_candidate = kb_is_dex_candle_candidate_event_kind(event_kind);
+    let mut object = match payload_json {
+        serde_json::Value::Object(object) => object,
+        other => {
+            let mut object = serde_json::Map::new();
+            object.insert("rawPayload".to_owned(), other);
+            object
+        }
+    };
+    kb_json_insert_string_if_missing(&mut object, "protocolName", protocol_name);
+    kb_json_insert_string_if_missing(&mut object, "eventKind", event_kind);
+    kb_json_insert_string_if_missing(&mut object, "eventCategory", event_category);
+    kb_json_insert_bool_if_missing(&mut object, "tradeCandidate", trade_candidate);
+    kb_json_insert_bool_if_missing(&mut object, "candleCandidate", candle_candidate);
+    kb_json_insert_i64_if_missing(&mut object, "eventClassificationVersion", 1);
+    if !trade_candidate {
+        kb_json_insert_string_if_missing(&mut object, "skipTradeReason", "non_trade_event");
+    } else if !candle_candidate {
+        kb_json_insert_string_if_missing(
+            &mut object,
+            "skipCandleReason",
+            "route_or_multihop_event_requires_leg_resolution",
+        );
+    }
+    serde_json::Value::Object(object)
+}
+
+// Inserts a string JSON property without overriding existing decoded data.
+fn kb_json_insert_string_if_missing(
+    object: &mut serde_json::Map<String, serde_json::Value>,
+    key: &str,
+    value: &str,
+) {
+    if object.contains_key(key) {
+        return;
+    }
+    object.insert(key.to_owned(), serde_json::Value::String(value.to_owned()));
+}
+
+// Inserts a bool JSON property without overriding existing decoded data.
+fn kb_json_insert_bool_if_missing(
+    object: &mut serde_json::Map<String, serde_json::Value>,
+    key: &str,
+    value: bool,
+) {
+    if object.contains_key(key) {
+        return;
+    }
+    object.insert(key.to_owned(), serde_json::Value::Bool(value));
+}
+
+// Inserts an i64 JSON property without overriding existing decoded data.
+fn kb_json_insert_i64_if_missing(
+    object: &mut serde_json::Map<String, serde_json::Value>,
+    key: &str,
+    value: i64,
+) {
+    if object.contains_key(key) {
+        return;
+    }
+    object.insert(
+        key.to_owned(),
+        serde_json::Value::Number(serde_json::Number::from(value)),
+    );
+}
+fn kb_enrich_and_serialize_dex_decoded_payload(
+    protocol_name: &str,
+    event_kind: &str,
+    payload_json: serde_json::Value,
+) -> Result<String, crate::KbError> {
+    let enriched_payload = kb_enrich_dex_decoded_payload(protocol_name, event_kind, payload_json);
+    let payload_json_result = serde_json::to_string(&enriched_payload);
+    match payload_json_result {
+        Ok(payload_json) => Ok(payload_json),
+        Err(error) => Err(crate::KbError::Json(format!(
+            "cannot serialize enriched decoded payload for '{}': {}",
+            event_kind, error
+        ))),
+    }
+}
+
+fn kb_enrich_serialized_dex_decoded_payload(
+    protocol_name: &str,
+    event_kind: &str,
+    payload_json: &str,
+) -> Result<String, crate::KbError> {
+    let payload_value_result = serde_json::from_str::<serde_json::Value>(payload_json);
+    let payload_value = match payload_value_result {
+        Ok(payload_value) => payload_value,
+        Err(error) => {
+            return Err(crate::KbError::Json(format!(
+                "cannot parse decoded payload for '{}': {}",
+                event_kind, error
+            )));
+        }
+    };
+    kb_enrich_and_serialize_dex_decoded_payload(protocol_name, event_kind, payload_value)
+}
+
 #[cfg(test)]
 mod tests {
+
     async fn make_database() -> std::sync::Arc<crate::KbDatabase> {
         let tempdir_result = tempfile::tempdir();
         let tempdir = match tempdir_result {
@@ -2836,4 +3260,151 @@ mod tests {
             Some("So11111111111111111111111111111111111111112".to_string())
         );
     }
+
+    #[test]
+    fn classifies_swap_events_as_trade_candidates() {
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_cpmm.swap_base_input"),
+            "trade"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_cpmm.swap_base_output"),
+            "trade"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_clmm.swap"),
+            "trade"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_clmm.swap_v2"),
+            "trade"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("pump_fun.buy"),
+            "trade"
+        );
+        assert!(super::kb_is_dex_trade_event_kind(
+            "raydium_cpmm.swap_base_input"
+        ));
+        assert!(super::kb_is_dex_candle_candidate_event_kind(
+            "raydium_cpmm.swap_base_input"
+        ));
+    }
+
+    #[test]
+    fn classifies_router_swap_as_trade_but_not_direct_candle_candidate() {
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_clmm.swap_router_base_in"),
+            "trade"
+        );
+        assert!(super::kb_is_dex_trade_event_kind(
+            "raydium_clmm.swap_router_base_in"
+        ));
+        assert!(!super::kb_is_dex_candle_candidate_event_kind(
+            "raydium_clmm.swap_router_base_in"
+        ));
+    }
+
+    #[test]
+    fn classifies_fee_reward_liquidity_and_lifecycle_events() {
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_cpmm.collect_creator_fee"),
+            "fee"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_clmm.collect_protocol_fee"),
+            "fee"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_clmm.set_reward_params"),
+            "reward"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_clmm.increase_liquidity_v2"),
+            "liquidity"
+        );
+        assert_eq!(
+            super::kb_classify_dex_event_category("raydium_cpmm.initialize"),
+            "pool_lifecycle"
+        );
+    }
+
+    #[test]
+    fn enriches_payload_without_overriding_existing_fields() {
+        let payload_json = serde_json::json!({
+            "eventCategory": "custom",
+            "amountIn": "10"
+        });
+        let enriched_payload = super::kb_enrich_dex_decoded_payload(
+            "raydium_cpmm",
+            "raydium_cpmm.swap_base_input",
+            payload_json,
+        );
+        let object_option = enriched_payload.as_object();
+        let object = match object_option {
+            Some(object) => object,
+            None => {
+                panic!("expected enriched payload object");
+            }
+        };
+        assert_eq!(
+            object.get("eventCategory"),
+            Some(&serde_json::Value::String("custom".to_owned()))
+        );
+        assert_eq!(
+            object.get("protocolName"),
+            Some(&serde_json::Value::String("raydium_cpmm".to_owned()))
+        );
+        assert_eq!(
+            object.get("eventKind"),
+            Some(&serde_json::Value::String(
+                "raydium_cpmm.swap_base_input".to_owned()
+            ))
+        );
+        assert_eq!(
+            object.get("tradeCandidate"),
+            Some(&serde_json::Value::Bool(true))
+        );
+        assert_eq!(
+            object.get("candleCandidate"),
+            Some(&serde_json::Value::Bool(true))
+        );
+    }
+
+    #[test]
+    fn enriches_non_object_payload_as_raw_payload() {
+        let payload_json = serde_json::Value::String("raw".to_owned());
+        let enriched_payload = super::kb_enrich_dex_decoded_payload(
+            "raydium_clmm",
+            "raydium_clmm.collect_protocol_fee",
+            payload_json,
+        );
+        let object_option = enriched_payload.as_object();
+        let object = match object_option {
+            Some(object) => object,
+            None => {
+                panic!("expected enriched payload object");
+            }
+        };
+        assert_eq!(
+            object.get("rawPayload"),
+            Some(&serde_json::Value::String("raw".to_owned()))
+        );
+        assert_eq!(
+            object.get("eventCategory"),
+            Some(&serde_json::Value::String("fee".to_owned()))
+        );
+        assert_eq!(
+            object.get("tradeCandidate"),
+            Some(&serde_json::Value::Bool(false))
+        );
+        assert_eq!(
+            object.get("candleCandidate"),
+            Some(&serde_json::Value::Bool(false))
+        );
+        assert_eq!(
+            object.get("skipTradeReason"),
+            Some(&serde_json::Value::String("non_trade_event".to_owned()))
+        );
+    }
 }