594 lines
23 KiB
Rust
594 lines
23 KiB
Rust
// file: kb_lib/src/pair_candle_aggregation.rs
|
|
|
|
//! Pair-candle aggregation service.
|
|
|
|
/// One pair-candle aggregation result.
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub struct PairCandleAggregationResult {
|
|
/// Related pair id.
|
|
pub pair_id: i64,
|
|
/// Candle timeframe in seconds.
|
|
pub timeframe_seconds: i64,
|
|
/// Inclusive bucket start in unix seconds.
|
|
pub bucket_start_unix: i64,
|
|
/// Persisted candle id.
|
|
pub pair_candle_id: i64,
|
|
}
|
|
|
|
/// Pair-candle aggregation service.
|
|
///
|
|
/// This service materializes a small set of standard timeframes in base storage.
|
|
/// Arbitrary timeframes are rebuilt on demand through `PairCandleQueryService`.
|
|
#[derive(Debug, Clone)]
|
|
pub struct PairCandleAggregationService {
|
|
database: std::sync::Arc<crate::Database>,
|
|
persistence: crate::DetectionPersistenceService,
|
|
}
|
|
|
|
impl PairCandleAggregationService {
|
|
/// Creates a new pair-candle aggregation service.
|
|
pub fn new(database: std::sync::Arc<crate::Database>) -> Self {
|
|
let persistence = crate::DetectionPersistenceService::new(database.clone());
|
|
return Self { database, persistence };
|
|
}
|
|
|
|
/// Returns the list of materialized timeframes in seconds.
|
|
pub fn materialized_timeframes_seconds(&self) -> std::vec::Vec<i64> {
|
|
return vec![60, 300, 900, 3600];
|
|
}
|
|
|
|
/// Rebuilds all impacted materialized candles for one resolved transaction signature.
|
|
pub async fn record_transaction_by_signature(
|
|
&self,
|
|
signature: &str,
|
|
) -> Result<std::vec::Vec<crate::PairCandleAggregationResult>, crate::Error> {
|
|
let transaction_result =
|
|
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),
|
|
};
|
|
let transaction = match transaction_option {
|
|
Some(transaction) => transaction,
|
|
None => {
|
|
return Err(crate::Error::InvalidState(format!(
|
|
"cannot aggregate pair candles for unknown transaction '{}'",
|
|
signature
|
|
)));
|
|
},
|
|
};
|
|
let transaction_id = match transaction.id {
|
|
Some(transaction_id) => transaction_id,
|
|
None => {
|
|
return Err(crate::Error::InvalidState(format!(
|
|
"transaction '{}' has no internal id",
|
|
signature
|
|
)));
|
|
},
|
|
};
|
|
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),
|
|
};
|
|
let materialized_timeframes = self.materialized_timeframes_seconds();
|
|
let mut seen = std::collections::HashSet::<(i64, i64, i64)>::new();
|
|
let mut results = std::vec::Vec::new();
|
|
for trade_event in &trade_events {
|
|
let event_time_option_result =
|
|
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),
|
|
};
|
|
let event_time_unix = match event_time_option {
|
|
Some(event_time_unix) => event_time_unix,
|
|
None => continue,
|
|
};
|
|
for timeframe_seconds in &materialized_timeframes {
|
|
let bucket_start_unix_result =
|
|
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),
|
|
};
|
|
let dedupe_key = (trade_event.pair_id, *timeframe_seconds, bucket_start_unix);
|
|
if seen.contains(&dedupe_key) {
|
|
continue;
|
|
}
|
|
seen.insert(dedupe_key);
|
|
let rebuilt_result = self
|
|
.rebuild_one_candle(trade_event.pair_id, *timeframe_seconds, bucket_start_unix)
|
|
.await;
|
|
let rebuilt = match rebuilt_result {
|
|
Ok(rebuilt) => rebuilt,
|
|
Err(error) => return Err(error),
|
|
};
|
|
if let Some(rebuilt) = rebuilt {
|
|
results.push(rebuilt);
|
|
}
|
|
}
|
|
}
|
|
if !results.is_empty() {
|
|
let payload = serde_json::json!({
|
|
"transactionSignature": signature,
|
|
"pairCandleCount": results.len()
|
|
});
|
|
let observation_result = self
|
|
.persistence
|
|
.record_observation(&crate::DetectionObservationInput::new(
|
|
"pair.candle_aggregation".to_string(),
|
|
crate::ObservationSourceKind::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::DetectionSignalInput::new(
|
|
"signal.pair.candle_aggregation.recorded".to_string(),
|
|
crate::AnalysisSignalSeverity::Low,
|
|
transaction.signature.clone(),
|
|
Some(observation_id),
|
|
None,
|
|
payload,
|
|
))
|
|
.await;
|
|
if let Err(error) = signal_result {
|
|
return Err(error);
|
|
}
|
|
}
|
|
return Ok(results);
|
|
}
|
|
|
|
async fn rebuild_one_candle(
|
|
&self,
|
|
pair_id: i64,
|
|
timeframe_seconds: i64,
|
|
bucket_start_unix: i64,
|
|
) -> Result<std::option::Option<crate::PairCandleAggregationResult>, crate::Error> {
|
|
let trade_events_result =
|
|
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 = build_candle_from_trade_events(
|
|
self.database.as_ref(),
|
|
pair_id,
|
|
timeframe_seconds,
|
|
bucket_start_unix,
|
|
&trade_events,
|
|
)
|
|
.await;
|
|
let candle_option = match candle_option_result {
|
|
Ok(candle_option) => candle_option,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let candle = match candle_option {
|
|
Some(candle) => candle,
|
|
None => return Ok(None),
|
|
};
|
|
let pair_candle_id_result =
|
|
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::PairCandleAggregationResult {
|
|
pair_id,
|
|
timeframe_seconds,
|
|
bucket_start_unix,
|
|
pair_candle_id,
|
|
}));
|
|
}
|
|
}
|
|
|
|
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::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::<TradeEventForCandle>::new();
|
|
for trade_event in trade_events {
|
|
if trade_event.pair_id != pair_id {
|
|
continue;
|
|
}
|
|
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),
|
|
};
|
|
let event_time_unix = match event_time_option {
|
|
Some(event_time_unix) => event_time_unix,
|
|
None => continue,
|
|
};
|
|
if event_time_unix < bucket_start_unix || event_time_unix >= bucket_end_unix {
|
|
continue;
|
|
}
|
|
let price_quote_per_base = match trade_event.price_quote_per_base {
|
|
Some(price_quote_per_base) => price_quote_per_base,
|
|
None => continue,
|
|
};
|
|
rows.push(TradeEventForCandle {
|
|
event_time_unix,
|
|
decoded_event_id: trade_event.decoded_event_id,
|
|
signature: trade_event.signature.clone(),
|
|
trade_side: trade_event.trade_side,
|
|
price_quote_per_base,
|
|
base_amount_raw: trade_event.base_amount_raw.clone(),
|
|
quote_amount_raw: trade_event.quote_amount_raw.clone(),
|
|
});
|
|
}
|
|
if rows.is_empty() {
|
|
return Ok(None);
|
|
}
|
|
rows.sort_by(|left, right| {
|
|
let time_compare = left.event_time_unix.cmp(&right.event_time_unix);
|
|
if time_compare != std::cmp::Ordering::Equal {
|
|
return time_compare;
|
|
}
|
|
return left.decoded_event_id.cmp(&right.decoded_event_id);
|
|
});
|
|
let open_price_quote_per_base = rows[0].price_quote_per_base;
|
|
let close_price_quote_per_base = rows[rows.len() - 1].price_quote_per_base;
|
|
let mut high_price_quote_per_base = open_price_quote_per_base;
|
|
let mut low_price_quote_per_base = open_price_quote_per_base;
|
|
let mut trade_count = 0_i64;
|
|
let mut buy_count = 0_i64;
|
|
let mut sell_count = 0_i64;
|
|
let mut base_volume_raw = std::option::Option::<std::string::String>::None;
|
|
let mut quote_volume_raw = std::option::Option::<std::string::String>::None;
|
|
for row in &rows {
|
|
trade_count += 1;
|
|
if row.trade_side == crate::SwapTradeSide::BuyBase {
|
|
buy_count += 1;
|
|
}
|
|
if row.trade_side == crate::SwapTradeSide::SellBase {
|
|
sell_count += 1;
|
|
}
|
|
if row.price_quote_per_base > high_price_quote_per_base {
|
|
high_price_quote_per_base = row.price_quote_per_base;
|
|
}
|
|
if row.price_quote_per_base < low_price_quote_per_base {
|
|
low_price_quote_per_base = row.price_quote_per_base;
|
|
}
|
|
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::PairCandleDto::new(
|
|
pair_id,
|
|
timeframe_seconds,
|
|
bucket_start_unix,
|
|
bucket_end_unix,
|
|
open_price_quote_per_base,
|
|
high_price_quote_per_base,
|
|
low_price_quote_per_base,
|
|
close_price_quote_per_base,
|
|
trade_count,
|
|
buy_count,
|
|
sell_count,
|
|
base_volume_raw,
|
|
quote_volume_raw,
|
|
Some(rows[0].signature.clone()),
|
|
Some(rows[rows.len() - 1].signature.clone()),
|
|
)));
|
|
}
|
|
|
|
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::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),
|
|
};
|
|
let transaction = match transaction_option {
|
|
Some(transaction) => transaction,
|
|
None => return Ok(Some(trade_event.created_at.timestamp())),
|
|
};
|
|
match transaction.block_time_unix {
|
|
Some(block_time_unix) => return Ok(Some(block_time_unix)),
|
|
None => return Ok(Some(trade_event.created_at.timestamp())),
|
|
}
|
|
}
|
|
|
|
pub(crate) fn bucket_start_unix(
|
|
event_time_unix: i64,
|
|
timeframe_seconds: i64,
|
|
) -> Result<i64, crate::Error> {
|
|
if timeframe_seconds <= 0 {
|
|
return Err(crate::Error::InvalidState(format!(
|
|
"invalid timeframe_seconds '{}'",
|
|
timeframe_seconds
|
|
)));
|
|
}
|
|
return Ok((event_time_unix / timeframe_seconds) * timeframe_seconds);
|
|
}
|
|
|
|
fn 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) => return None,
|
|
(Some(left), None) => return Some(left),
|
|
(None, Some(right)) => return 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),
|
|
};
|
|
return Some((left_value + right_value).to_string());
|
|
},
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
struct TradeEventForCandle {
|
|
event_time_unix: i64,
|
|
decoded_event_id: i64,
|
|
signature: std::string::String,
|
|
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>,
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
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::DatabaseConfig {
|
|
enabled: true,
|
|
backend: crate::DatabaseBackend::Sqlite,
|
|
sqlite: crate::SqliteDatabaseConfig {
|
|
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::Database::connect_and_initialize(&config).await;
|
|
let database = match database_result {
|
|
Ok(database) => database,
|
|
Err(error) => panic!("database init must succeed: {}", error),
|
|
};
|
|
return std::sync::Arc::new(database);
|
|
}
|
|
|
|
async fn seed_fluxbeam_swap_transaction(
|
|
database: std::sync::Arc<crate::Database>,
|
|
signature: &str,
|
|
block_time_unix: i64,
|
|
base_amount_raw: &str,
|
|
quote_amount_raw: &str,
|
|
) {
|
|
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,
|
|
"version": 0,
|
|
"transaction": {
|
|
"message": {
|
|
"instructions": [
|
|
{
|
|
"programId": crate::FLUXBEAM_PROGRAM_ID,
|
|
"program": "fluxbeam",
|
|
"stackHeight": 1,
|
|
"accounts": [
|
|
"CandlePool111",
|
|
"CandleLpMint111",
|
|
"CandleTokenA111",
|
|
crate::WSOL_MINT_ID
|
|
],
|
|
"parsed": {
|
|
"info": {
|
|
"instruction": "swap",
|
|
"pool": "CandlePool111",
|
|
"tokenA": "CandleTokenA111",
|
|
"tokenB": crate::WSOL_MINT_ID,
|
|
"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);
|
|
}
|
|
let trade_result = trade_aggregation.record_transaction_by_signature(signature).await;
|
|
if let Err(error) = trade_result {
|
|
panic!("trade aggregation must succeed: {}", error);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn record_transaction_by_signature_creates_materialized_candles() {
|
|
let database = make_database().await;
|
|
seed_fluxbeam_swap_transaction(
|
|
database.clone(),
|
|
"sig-pair-candle-1",
|
|
1_700_000_000,
|
|
"1000",
|
|
"2000",
|
|
)
|
|
.await;
|
|
seed_fluxbeam_swap_transaction(
|
|
database.clone(),
|
|
"sig-pair-candle-2",
|
|
1_700_000_020,
|
|
"1000",
|
|
"3000",
|
|
)
|
|
.await;
|
|
seed_fluxbeam_swap_transaction(
|
|
database.clone(),
|
|
"sig-pair-candle-3",
|
|
1_700_000_070,
|
|
"1000",
|
|
"1500",
|
|
)
|
|
.await;
|
|
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);
|
|
}
|
|
let result_2 = service.record_transaction_by_signature("sig-pair-candle-2").await;
|
|
if let Err(error) = result_2 {
|
|
panic!("candle aggregation 2 must succeed: {}", error);
|
|
}
|
|
let result_3 = service.record_transaction_by_signature("sig-pair-candle-3").await;
|
|
if let Err(error) = result_3 {
|
|
panic!("candle aggregation 3 must succeed: {}", error);
|
|
}
|
|
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::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),
|
|
};
|
|
let pair = match pair_option {
|
|
Some(pair) => pair,
|
|
None => panic!("pair must exist"),
|
|
};
|
|
let pair_id = pair.id.unwrap_or_default();
|
|
let candles_result =
|
|
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),
|
|
};
|
|
assert_eq!(candles.len(), 2);
|
|
assert_eq!(candles[0].open_price_quote_per_base, 2.0);
|
|
assert_eq!(candles[0].high_price_quote_per_base, 3.0);
|
|
assert_eq!(candles[0].low_price_quote_per_base, 2.0);
|
|
assert_eq!(candles[0].close_price_quote_per_base, 3.0);
|
|
assert_eq!(candles[0].trade_count, 2);
|
|
assert_eq!(candles[0].base_volume_raw, Some("2000".to_string()));
|
|
assert_eq!(candles[0].quote_volume_raw, Some("5000".to_string()));
|
|
assert_eq!(candles[1].open_price_quote_per_base, 1.5);
|
|
assert_eq!(candles[1].close_price_quote_per_base, 1.5);
|
|
assert_eq!(candles[1].trade_count, 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn materialized_candle_rebuild_is_idempotent() {
|
|
let database = make_database().await;
|
|
seed_fluxbeam_swap_transaction(
|
|
database.clone(),
|
|
"sig-pair-candle-idempotent",
|
|
1_700_001_000,
|
|
"1000",
|
|
"2500",
|
|
)
|
|
.await;
|
|
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 {
|
|
Ok(first_results) => first_results,
|
|
Err(error) => panic!("first candle aggregation must succeed: {}", error),
|
|
};
|
|
assert!(!first_results.is_empty());
|
|
let second_result =
|
|
service.record_transaction_by_signature("sig-pair-candle-idempotent").await;
|
|
let second_results = match second_result {
|
|
Ok(second_results) => second_results,
|
|
Err(error) => panic!("second candle aggregation must succeed: {}", error),
|
|
};
|
|
assert!(!second_results.is_empty());
|
|
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::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),
|
|
};
|
|
let pair = match pair_option {
|
|
Some(pair) => pair,
|
|
None => panic!("pair must exist"),
|
|
};
|
|
let pair_id = pair.id.unwrap_or_default();
|
|
let candles_result =
|
|
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),
|
|
};
|
|
assert_eq!(candles.len(), 1);
|
|
assert_eq!(candles[0].trade_count, 1);
|
|
}
|
|
}
|