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

0.7.27 +Refactor

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-05-10 00:33:01 +02:00
parent cb2e8e7096
commit 1f0137b9de
261 changed files with 12308 additions and 8928 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
kb_lib/src/pair_candle_aggregation.rs
View File

@@ -4,7 +4,7 @@
/// One pair-candle aggregation result.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct KbPairCandleAggregationResult {
pub struct PairCandleAggregationResult {
/// Related pair id.
pub pair_id: i64,
/// Candle timeframe in seconds.
@@ -18,17 +18,17 @@ pub struct KbPairCandleAggregationResult {
/// Pair-candle aggregation service.
///
/// This service materializes a small set of standard timeframes in base storage.
/// Arbitrary timeframes are rebuilt on demand through `KbPairCandleQueryService`.
/// Arbitrary timeframes are rebuilt on demand through `PairCandleQueryService`.
#[derive(Debug, Clone)]
pub struct KbPairCandleAggregationService {
database: std::sync::Arc<crate::KbDatabase>,
persistence: crate::KbDetectionPersistenceService,
pub struct PairCandleAggregationService {
database: std::sync::Arc<crate::Database>,
persistence: crate::DetectionPersistenceService,
}
impl KbPairCandleAggregationService {
impl PairCandleAggregationService {
/// Creates a new pair-candle aggregation service.
pub fn new(database: std::sync::Arc<crate::KbDatabase>) -> Self {
let persistence = crate::KbDetectionPersistenceService::new(database.clone());
pub fn new(database: std::sync::Arc<crate::Database>) -> Self {
let persistence = crate::DetectionPersistenceService::new(database.clone());
return Self { database, persistence };
}
@@ -41,9 +41,10 @@ impl KbPairCandleAggregationService {
pub async fn record_transaction_by_signature(
&self,
signature: &str,
) -> Result<std::vec::Vec<crate::KbPairCandleAggregationResult>, crate::KbError> {
) -> Result<std::vec::Vec<crate::PairCandleAggregationResult>, crate::Error> {
let transaction_result =
crate::get_chain_transaction_by_signature(self.database.as_ref(), signature).await;
crate::query_chain_transactions_get_by_signature(self.database.as_ref(), signature)
.await;
let transaction_option = match transaction_result {
Ok(transaction_option) => transaction_option,
Err(error) => return Err(error),
@@ -51,7 +52,7 @@ impl KbPairCandleAggregationService {
let transaction = match transaction_option {
Some(transaction) => transaction,
None => {
return Err(crate::KbError::InvalidState(format!(
return Err(crate::Error::InvalidState(format!(
"cannot aggregate pair candles for unknown transaction '{}'",
signature
)));
@@ -60,15 +61,17 @@ impl KbPairCandleAggregationService {
let transaction_id = match transaction.id {
Some(transaction_id) => transaction_id,
None => {
return Err(crate::KbError::InvalidState(format!(
return Err(crate::Error::InvalidState(format!(
"transaction '{}' has no internal id",
signature
)));
},
};
let trade_events_result =
crate::list_trade_events_by_transaction_id(self.database.as_ref(), transaction_id)
.await;
let trade_events_result = crate::query_trade_events_list_by_transaction_id(
self.database.as_ref(),
transaction_id,
)
.await;
let trade_events = match trade_events_result {
Ok(trade_events) => trade_events,
Err(error) => return Err(error),
@@ -78,7 +81,7 @@ impl KbPairCandleAggregationService {
let mut results = std::vec::Vec::new();
for trade_event in &trade_events {
let event_time_option_result =
kb_extract_trade_event_unix_time(self.database.as_ref(), trade_event).await;
extract_trade_event_unix_time(self.database.as_ref(), trade_event).await;
let event_time_option = match event_time_option_result {
Ok(event_time_option) => event_time_option,
Err(error) => return Err(error),
@@ -89,7 +92,7 @@ impl KbPairCandleAggregationService {
};
for timeframe_seconds in &materialized_timeframes {
let bucket_start_unix_result =
kb_bucket_start_unix(event_time_unix, *timeframe_seconds);
bucket_start_unix(event_time_unix, *timeframe_seconds);
let bucket_start_unix = match bucket_start_unix_result {
Ok(bucket_start_unix) => bucket_start_unix,
Err(error) => return Err(error),
@@ -118,9 +121,9 @@ impl KbPairCandleAggregationService {
});
let observation_result = self
.persistence
.record_observation(&crate::KbDetectionObservationInput::new(
.record_observation(&crate::DetectionObservationInput::new(
"pair.candle_aggregation".to_string(),
crate::KbObservationSourceKind::Dex,
crate::ObservationSourceKind::Dex,
transaction.source_endpoint_name.clone(),
transaction.signature.clone(),
transaction.slot,
@@ -133,9 +136,9 @@ impl KbPairCandleAggregationService {
};
let signal_result = self
.persistence
.record_signal(&crate::KbDetectionSignalInput::new(
.record_signal(&crate::DetectionSignalInput::new(
"signal.pair.candle_aggregation.recorded".to_string(),
crate::KbAnalysisSignalSeverity::Low,
crate::AnalysisSignalSeverity::Low,
transaction.signature.clone(),
Some(observation_id),
None,
@@ -154,14 +157,14 @@ impl KbPairCandleAggregationService {
pair_id: i64,
timeframe_seconds: i64,
bucket_start_unix: i64,
) -> Result<std::option::Option<crate::KbPairCandleAggregationResult>, crate::KbError> {
) -> Result<std::option::Option<crate::PairCandleAggregationResult>, crate::Error> {
let trade_events_result =
crate::list_trade_events_by_pair_id(self.database.as_ref(), pair_id).await;
crate::query_trade_events_list_by_pair_id(self.database.as_ref(), pair_id).await;
let trade_events = match trade_events_result {
Ok(trade_events) => trade_events,
Err(error) => return Err(error),
};
let candle_option_result = kb_build_candle_from_trade_events(
let candle_option_result = build_candle_from_trade_events(
self.database.as_ref(),
pair_id,
timeframe_seconds,
@@ -178,12 +181,12 @@ impl KbPairCandleAggregationService {
None => return Ok(None),
};
let pair_candle_id_result =
crate::upsert_pair_candle(self.database.as_ref(), &candle).await;
crate::query_pair_candles_upsert(self.database.as_ref(), &candle).await;
let pair_candle_id = match pair_candle_id_result {
Ok(pair_candle_id) => pair_candle_id,
Err(error) => return Err(error),
};
return Ok(Some(crate::KbPairCandleAggregationResult {
return Ok(Some(crate::PairCandleAggregationResult {
pair_id,
timeframe_seconds,
bucket_start_unix,
@@ -192,21 +195,20 @@ impl KbPairCandleAggregationService {
}
}
pub(crate) async fn kb_build_candle_from_trade_events(
database: &crate::KbDatabase,
pub(crate) async fn build_candle_from_trade_events(
database: &crate::Database,
pair_id: i64,
timeframe_seconds: i64,
bucket_start_unix: i64,
trade_events: &[crate::KbTradeEventDto],
) -> Result<std::option::Option<crate::KbPairCandleDto>, crate::KbError> {
trade_events: &[crate::TradeEventDto],
) -> Result<std::option::Option<crate::PairCandleDto>, crate::Error> {
let bucket_end_unix = bucket_start_unix.saturating_add(timeframe_seconds);
let mut rows = std::vec::Vec::<KbTradeEventForCandle>::new();
let mut rows = std::vec::Vec::<TradeEventForCandle>::new();
for trade_event in trade_events {
if trade_event.pair_id != pair_id {
continue;
}
let event_time_option_result =
kb_extract_trade_event_unix_time(database, trade_event).await;
let event_time_option_result = extract_trade_event_unix_time(database, trade_event).await;
let event_time_option = match event_time_option_result {
Ok(event_time_option) => event_time_option,
Err(error) => return Err(error),
@@ -222,7 +224,7 @@ pub(crate) async fn kb_build_candle_from_trade_events(
Some(price_quote_per_base) => price_quote_per_base,
None => continue,
};
rows.push(KbTradeEventForCandle {
rows.push(TradeEventForCandle {
event_time_unix,
decoded_event_id: trade_event.decoded_event_id,
signature: trade_event.signature.clone(),
@@ -253,10 +255,10 @@ pub(crate) async fn kb_build_candle_from_trade_events(
let mut quote_volume_raw = std::option::Option::<std::string::String>::None;
for row in &rows {
trade_count += 1;
if row.trade_side == crate::KbSwapTradeSide::BuyBase {
if row.trade_side == crate::SwapTradeSide::BuyBase {
buy_count += 1;
}
if row.trade_side == crate::KbSwapTradeSide::SellBase {
if row.trade_side == crate::SwapTradeSide::SellBase {
sell_count += 1;
}
if row.price_quote_per_base > high_price_quote_per_base {
@@ -265,10 +267,10 @@ pub(crate) async fn kb_build_candle_from_trade_events(
if row.price_quote_per_base < low_price_quote_per_base {
low_price_quote_per_base = row.price_quote_per_base;
}
base_volume_raw = kb_add_raw_amounts(base_volume_raw, row.base_amount_raw.clone());
quote_volume_raw = kb_add_raw_amounts(quote_volume_raw, row.quote_amount_raw.clone());
base_volume_raw = add_raw_amounts(base_volume_raw, row.base_amount_raw.clone());
quote_volume_raw = add_raw_amounts(quote_volume_raw, row.quote_amount_raw.clone());
}
return Ok(Some(crate::KbPairCandleDto::new(
return Ok(Some(crate::PairCandleDto::new(
pair_id,
timeframe_seconds,
bucket_start_unix,
@@ -287,12 +289,13 @@ pub(crate) async fn kb_build_candle_from_trade_events(
)));
}
pub(crate) async fn kb_extract_trade_event_unix_time(
database: &crate::KbDatabase,
trade_event: &crate::KbTradeEventDto,
) -> Result<std::option::Option<i64>, crate::KbError> {
pub(crate) async fn extract_trade_event_unix_time(
database: &crate::Database,
trade_event: &crate::TradeEventDto,
) -> Result<std::option::Option<i64>, crate::Error> {
let transaction_result =
crate::get_chain_transaction_by_signature(database, trade_event.signature.as_str()).await;
crate::query_chain_transactions_get_by_signature(database, trade_event.signature.as_str())
.await;
let transaction_option = match transaction_result {
Ok(transaction_option) => transaction_option,
Err(error) => return Err(error),
@@ -307,12 +310,12 @@ pub(crate) async fn kb_extract_trade_event_unix_time(
}
}
pub(crate) fn kb_bucket_start_unix(
pub(crate) fn bucket_start_unix(
event_time_unix: i64,
timeframe_seconds: i64,
) -> Result<i64, crate::KbError> {
) -> Result<i64, crate::Error> {
if timeframe_seconds <= 0 {
return Err(crate::KbError::InvalidState(format!(
return Err(crate::Error::InvalidState(format!(
"invalid timeframe_seconds '{}'",
timeframe_seconds
)));
@@ -320,7 +323,7 @@ pub(crate) fn kb_bucket_start_unix(
return Ok((event_time_unix / timeframe_seconds) * timeframe_seconds);
}
fn kb_add_raw_amounts(
fn add_raw_amounts(
left: std::option::Option<std::string::String>,
right: std::option::Option<std::string::String>,
) -> std::option::Option<std::string::String> {
@@ -345,11 +348,11 @@ fn kb_add_raw_amounts(
}
#[derive(Debug, Clone)]
struct KbTradeEventForCandle {
struct TradeEventForCandle {
event_time_unix: i64,
decoded_event_id: i64,
signature: std::string::String,
trade_side: crate::KbSwapTradeSide,
trade_side: crate::SwapTradeSide,
price_quote_per_base: f64,
base_amount_raw: std::option::Option<std::string::String>,
quote_amount_raw: std::option::Option<std::string::String>,
@@ -357,17 +360,17 @@ struct KbTradeEventForCandle {
#[cfg(test)]
mod tests {
async fn make_database() -> std::sync::Arc<crate::KbDatabase> {
async fn make_database() -> std::sync::Arc<crate::Database> {
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("pair_candle_aggregation.sqlite3");
let config = crate::KbDatabaseConfig {
let config = crate::DatabaseConfig {
enabled: true,
backend: crate::KbDatabaseBackend::Sqlite,
sqlite: crate::KbSqliteDatabaseConfig {
backend: crate::DatabaseBackend::Sqlite,
sqlite: crate::SqliteDatabaseConfig {
path: database_path.to_string_lossy().to_string(),
create_if_missing: true,
busy_timeout_ms: 5000,
@@ -376,7 +379,7 @@ mod tests {
use_wal: true,
},
};
let database_result = crate::KbDatabase::connect_and_initialize(&config).await;
let database_result = crate::Database::connect_and_initialize(&config).await;
let database = match database_result {
Ok(database) => database,
Err(error) => panic!("database init must succeed: {}", error),
@@ -385,16 +388,16 @@ mod tests {
}
async fn seed_fluxbeam_swap_transaction(
database: std::sync::Arc<crate::KbDatabase>,
database: std::sync::Arc<crate::Database>,
signature: &str,
block_time_unix: i64,
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.clone());
let trade_aggregation = crate::KbTradeAggregationService::new(database.clone());
let transaction_model = crate::TransactionModelService::new(database.clone());
let dex_decode = crate::DexDecodeService::new(database.clone());
let dex_detect = crate::DexDetectService::new(database.clone());
let trade_aggregation = crate::TradeAggregationService::new(database.clone());
let resolved_transaction = serde_json::json!({
"slot": 960001,
"blockTime": block_time_unix,
@@ -403,21 +406,21 @@ mod tests {
"message": {
"instructions": [
{
"programId": crate::KB_FLUXBEAM_PROGRAM_ID,
"programId": crate::FLUXBEAM_PROGRAM_ID,
"program": "fluxbeam",
"stackHeight": 1,
"accounts": [
"CandlePool111",
"CandleLpMint111",
"CandleTokenA111",
"So11111111111111111111111111111111111111112"
crate::WSOL_MINT_ID
],
"parsed": {
"info": {
"instruction": "swap",
"pool": "CandlePool111",
"tokenA": "CandleTokenA111",
"tokenB": "So11111111111111111111111111111111111111112",
"tokenB": crate::WSOL_MINT_ID,
"baseAmountRaw": base_amount_raw,
"quoteAmountRaw": quote_amount_raw
}
@@ -486,7 +489,7 @@ mod tests {
"1500",
)
.await;
let service = crate::KbPairCandleAggregationService::new(database.clone());
let service = crate::PairCandleAggregationService::new(database.clone());
let result_1 = service.record_transaction_by_signature("sig-pair-candle-1").await;
if let Err(error) = result_1 {
panic!("candle aggregation 1 must succeed: {}", error);
@@ -499,13 +502,13 @@ mod tests {
if let Err(error) = result_3 {
panic!("candle aggregation 3 must succeed: {}", error);
}
let pools_result = crate::list_pools(database.as_ref()).await;
let pools_result = crate::query_pools_list(database.as_ref()).await;
let pools = match pools_result {
Ok(pools) => pools,
Err(error) => panic!("pool list must succeed: {}", error),
};
let pool_id = pools[0].id.unwrap_or_default();
let pair_result = crate::get_pair_by_pool_id(database.as_ref(), pool_id).await;
let pair_result = crate::query_pairs_get_by_pool_id(database.as_ref(), pool_id).await;
let pair_option = match pair_result {
Ok(pair_option) => pair_option,
Err(error) => panic!("pair fetch must succeed: {}", error),
@@ -516,7 +519,8 @@ mod tests {
};
let pair_id = pair.id.unwrap_or_default();
let candles_result =
crate::list_pair_candles_by_pair_and_timeframe(database.as_ref(), pair_id, 60).await;
crate::query_pair_candles_list_by_pair_and_timeframe(database.as_ref(), pair_id, 60)
.await;
let candles = match candles_result {
Ok(candles) => candles,
Err(error) => panic!("candle list must succeed: {}", error),
@@ -545,7 +549,7 @@ mod tests {
"2500",
)
.await;
let service = crate::KbPairCandleAggregationService::new(database.clone());
let service = crate::PairCandleAggregationService::new(database.clone());
let first_result =
service.record_transaction_by_signature("sig-pair-candle-idempotent").await;
let first_results = match first_result {
@@ -560,13 +564,13 @@ mod tests {
Err(error) => panic!("second candle aggregation must succeed: {}", error),
};
assert!(!second_results.is_empty());
let pools_result = crate::list_pools(database.as_ref()).await;
let pools_result = crate::query_pools_list(database.as_ref()).await;
let pools = match pools_result {
Ok(pools) => pools,
Err(error) => panic!("pool list must succeed: {}", error),
};
let pool_id = pools[0].id.unwrap_or_default();
let pair_result = crate::get_pair_by_pool_id(database.as_ref(), pool_id).await;
let pair_result = crate::query_pairs_get_by_pool_id(database.as_ref(), pool_id).await;
let pair_option = match pair_result {
Ok(pair_option) => pair_option,
Err(error) => panic!("pair fetch must succeed: {}", error),
@@ -577,7 +581,8 @@ mod tests {
};
let pair_id = pair.id.unwrap_or_default();
let candles_result =
crate::list_pair_candles_by_pair_and_timeframe(database.as_ref(), pair_id, 60).await;
crate::query_pair_candles_list_by_pair_and_timeframe(database.as_ref(), pair_id, 60)
.await;
let candles = match candles_result {
Ok(candles) => candles,
Err(error) => panic!("candle list must succeed: {}", error),