use bot_core::{now_ms, Fill, InstrumentId};
use reqwest::Client;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::time::Duration;
use crate::performance_metrics::PerformanceMetricsSnapshot;
pub use crate::account_syncer::SyncError;
#[derive(Debug, Clone)]
pub struct TradeSyncerConfig {
pub bot_id: String,
pub upstream_url: String,
pub sync_interval_ms: u64,
pub timeout_secs: u64,
pub max_retries: u32,
pub retry_delay_ms: u64,
pub instruments: Vec<InstrumentId>,
pub strategy_type: Option<String>,
pub sync_secret: Option<String>,
}
impl Default for TradeSyncerConfig {
fn default() -> Self {
Self {
bot_id: String::new(),
upstream_url: String::new(),
sync_interval_ms: 10_000,
timeout_secs: 10,
max_retries: 3,
retry_delay_ms: 1000,
instruments: Vec::new(),
strategy_type: None,
sync_secret: None,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UpstreamTrade {
pub trade_id: String,
pub client_order_id: String,
pub venue_order_id: String,
pub instrument_id: String,
pub side: String,
pub order_type: String,
pub qty: String,
pub price: String,
pub quote_notional: String,
pub fee: String,
pub fee_currency: String,
pub liquidity: String,
pub ts_event: i64,
}
#[derive(Debug, Clone, Serialize)]
pub struct SyncRequest {
pub trades: Vec<UpstreamTrade>,
pub ts: i64,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_price: Option<String>,
pub stop_bot: bool,
pub stop_reason: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub metadata: Option<serde_json::Value>,
}
#[derive(Debug, Clone, Deserialize)]
pub struct SyncResponse {
pub synced: SyncedInfo,
#[serde(default)]
pub pnl: f64,
}
#[derive(Debug, Clone, Deserialize)]
pub struct SyncedInfo {
pub trade_id: String,
pub ts: i64,
}
#[derive(Debug, Clone)]
pub struct SyncResult {
pub success: bool,
pub pnl: Option<f64>,
pub last_synced_trade_id: Option<String>,
pub trades_synced: usize,
}
pub struct TradeSyncer {
config: TradeSyncerConfig,
client: Client,
synced_trade_ids: HashSet<String>,
pending_fills: Vec<Fill>,
last_sync_ts: i64,
last_pnl: Option<f64>,
start_timestamp: i64,
metrics_snapshot: Option<PerformanceMetricsSnapshot>,
}
impl TradeSyncer {
pub fn new(config: TradeSyncerConfig) -> Result<Self, SyncError> {
if config.bot_id.is_empty() {
return Err(SyncError::Config("bot_id is required".to_string()));
}
if config.upstream_url.is_empty() {
return Err(SyncError::Config("upstream_url is required".to_string()));
}
let client = Client::builder()
.timeout(Duration::from_secs(config.timeout_secs))
.build()
.map_err(|e| SyncError::Http(e.to_string()))?;
let start_timestamp = now_ms();
tracing::info!(
"[TradeSyncer] Initialized with start_timestamp={} - only fills after this time will be synced",
start_timestamp
);
Ok(Self {
config,
client,
synced_trade_ids: HashSet::new(),
pending_fills: Vec::new(),
last_sync_ts: 0,
last_pnl: None,
start_timestamp,
metrics_snapshot: None,
})
}
pub fn add_fill(&mut self, fill: Fill) {
if fill.ts < self.start_timestamp {
tracing::debug!(
"Skipping historical fill: {} (ts={} < start_ts={})",
fill.trade_id,
fill.ts,
self.start_timestamp
);
return;
}
if self.synced_trade_ids.contains(&fill.trade_id.0) {
tracing::debug!("Skipping already-synced fill: {}", fill.trade_id);
return;
}
if !self.config.instruments.is_empty()
&& !self.config.instruments.contains(&fill.instrument)
{
tracing::debug!(
"Skipping fill for untracked instrument: {} (tracking: {:?})",
fill.instrument,
self.config.instruments
);
return;
}
tracing::info!(
"[TradeSyncer] Adding fill to pending queue: {} (ts={})",
fill.trade_id,
fill.ts
);
self.pending_fills.push(fill);
}
pub fn should_sync(&self) -> bool {
let now = now_ms();
now - self.last_sync_ts >= self.config.sync_interval_ms as i64
}
pub fn last_pnl(&self) -> Option<f64> {
self.last_pnl
}
pub fn pending_count(&self) -> usize {
self.pending_fills.len()
}
pub fn set_metrics_snapshot(&mut self, snapshot: Option<PerformanceMetricsSnapshot>) {
self.metrics_snapshot = snapshot;
}
pub async fn sync(
&mut self,
current_price: Option<Decimal>,
stop_bot: bool,
stop_reason: &str,
) -> Result<SyncResult, SyncError> {
let now = now_ms();
let trades: Vec<UpstreamTrade> = self
.pending_fills
.iter()
.filter(|f| !self.synced_trade_ids.contains(&f.trade_id.0))
.map(|f| self.fill_to_trade(f))
.collect();
let trades_count = trades.len();
tracing::info!(
"[TradeSyncer] Syncing {} trades to upstream (pending={}, synced={})",
trades_count,
self.pending_fills.len(),
self.synced_trade_ids.len()
);
let request = SyncRequest {
trades,
ts: now / 1000, current_price: current_price.map(|p| p.to_string()),
stop_bot,
stop_reason: stop_reason.to_string(),
metadata: self.metadata_payload(),
};
let response = self.execute_with_retry(&request).await?;
for fill in &self.pending_fills {
self.synced_trade_ids.insert(fill.trade_id.0.clone());
}
self.pending_fills.clear();
self.last_sync_ts = now;
self.last_pnl = Some(response.pnl);
tracing::info!(
"[TradeSyncer] Sync successful: pnl={:.4}, last_trade_id={}",
response.pnl,
response.synced.trade_id
);
Ok(SyncResult {
success: true,
pnl: Some(response.pnl),
last_synced_trade_id: Some(response.synced.trade_id),
trades_synced: trades_count,
})
}
fn metadata_payload(&self) -> Option<serde_json::Value> {
let mut metadata = serde_json::Map::new();
if let Some(strategy_type) = self
.config
.strategy_type
.as_deref()
.filter(|s| !s.is_empty())
{
metadata.insert(
"strategy_type".to_string(),
serde_json::json!(strategy_type),
);
}
if let Some(snapshot) = self.metrics_snapshot.as_ref() {
metadata.insert(
"performance_metrics".to_string(),
serde_json::json!(snapshot),
);
}
(!metadata.is_empty()).then(|| serde_json::Value::Object(metadata))
}
async fn execute_with_retry(&self, request: &SyncRequest) -> Result<SyncResponse, SyncError> {
let url = format!(
"{}/sync/{}",
self.config.upstream_url.trim_end_matches('/'),
self.config.bot_id
);
let mut last_error: Option<SyncError> = None;
let mut delay_ms = self.config.retry_delay_ms;
for attempt in 1..=self.config.max_retries {
tracing::debug!(
"[TradeSyncer] Sync attempt {}/{} to {}",
attempt,
self.config.max_retries,
url
);
match self.execute_request(&url, request).await {
Ok(response) => return Ok(response),
Err(e) => {
tracing::warn!(
"[TradeSyncer] Sync attempt {}/{} failed: {}",
attempt,
self.config.max_retries,
e
);
last_error = Some(e);
if attempt < self.config.max_retries {
tracing::debug!("[TradeSyncer] Retrying in {}ms...", delay_ms);
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
delay_ms *= 2; }
}
}
}
Err(last_error.unwrap_or(SyncError::MaxRetries))
}
async fn execute_request(
&self,
url: &str,
request: &SyncRequest,
) -> Result<SyncResponse, SyncError> {
let mut request_builder = self
.client
.post(url)
.header("Content-Type", "application/json");
if let Some(secret) = self.config.sync_secret.as_deref().filter(|s| !s.is_empty()) {
request_builder = request_builder.header("x-bot-sync-secret", secret);
}
let response = request_builder.json(request).send().await?;
let status = response.status();
if !status.is_success() {
let body = response.text().await.unwrap_or_default();
return Err(SyncError::Api {
status: status.as_u16(),
body,
});
}
let sync_response: SyncResponse = response
.json()
.await
.map_err(|e| SyncError::Parse(e.to_string()))?;
Ok(sync_response)
}
fn fill_to_trade(&self, fill: &Fill) -> UpstreamTrade {
let side = format!("{}", fill.side);
let quote_notional = fill.price.0 * fill.qty.0;
UpstreamTrade {
trade_id: fill.trade_id.0.clone(),
client_order_id: fill
.client_id
.as_ref()
.map(|c| c.0.clone())
.unwrap_or_default(),
venue_order_id: fill
.exchange_order_id
.as_ref()
.map(|e| e.0.clone())
.unwrap_or_default(),
instrument_id: fill.instrument.0.clone(),
side,
order_type: "LIMIT".to_string(),
qty: fill.qty.0.to_string(),
price: fill.price.0.to_string(),
quote_notional: quote_notional.to_string(),
fee: fill.fee.amount.to_string(),
fee_currency: fill.fee.asset.0.clone(),
liquidity: "UNKNOWN".to_string(), ts_event: fill.ts,
}
}
pub async fn shutdown_sync(
&mut self,
current_price: Option<Decimal>,
stop_reason: &str,
) -> Result<SyncResult, SyncError> {
tracing::info!(
"[TradeSyncer] Performing shutdown sync with reason: {}, price: {:?}",
stop_reason,
current_price
);
self.sync(current_price, true, stop_reason).await
}
}
#[async_trait::async_trait]
impl crate::sync_traits::TradeSync for TradeSyncer {
fn add_fill(&mut self, fill: Fill) {
TradeSyncer::add_fill(self, fill)
}
fn should_sync(&self) -> bool {
TradeSyncer::should_sync(self)
}
fn pending_count(&self) -> usize {
TradeSyncer::pending_count(self)
}
fn last_pnl(&self) -> Option<f64> {
TradeSyncer::last_pnl(self)
}
fn set_metrics_snapshot(
&mut self,
snapshot: Option<crate::performance_metrics::PerformanceMetricsSnapshot>,
) {
TradeSyncer::set_metrics_snapshot(self, snapshot)
}
async fn sync(
&mut self,
current_price: Option<Decimal>,
stop_bot: bool,
stop_reason: &str,
) -> Result<crate::sync_traits::TradeSyncResult, SyncError> {
let result = TradeSyncer::sync(self, current_price, stop_bot, stop_reason).await?;
Ok(crate::sync_traits::TradeSyncResult {
success: result.success,
pnl: result.pnl,
})
}
async fn shutdown_sync(
&mut self,
current_price: Option<Decimal>,
stop_reason: &str,
) -> Result<crate::sync_traits::TradeSyncResult, SyncError> {
let result = TradeSyncer::shutdown_sync(self, current_price, stop_reason).await?;
Ok(crate::sync_traits::TradeSyncResult {
success: result.success,
pnl: result.pnl,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::performance_metrics::{
PerformanceBenchmark, PerformanceMetrics, PerformanceMetricsSnapshot,
};
use bot_core::{AssetId, Fee, OrderSide, Price, Qty, TradeId};
fn make_metrics_snapshot() -> PerformanceMetricsSnapshot {
PerformanceMetricsSnapshot {
schema_version: 1,
mode: "backtest".to_string(),
scope: "backtest_window".to_string(),
run_started_at_ms: None,
metrics: PerformanceMetrics {
period_return_pct: Some(1.0),
apr_pct: Some(365.0),
sharpe: Some(1.5),
max_drawdown_pct: Some(0.5),
max_drawdown_usdc: "5".to_string(),
win_rate_pct: Some(100.0),
closed_trade_count: 1,
winning_trade_count: 1,
losing_trade_count: 0,
fill_count: 2,
total_fees: "0.2".to_string(),
total_volume: "200".to_string(),
net_pnl: "10".to_string(),
fee_drag_pct: Some(0.02),
},
benchmark: PerformanceBenchmark {
start_ts_ms: Some(1),
end_ts_ms: Some(2),
duration_ms: Some(1),
quote_count: 2,
starting_balance_usdc: Some("1000".to_string()),
ending_balance_usdc: Some("1010".to_string()),
instrument: Some("BTC-PERP".to_string()),
},
latest_equity: None,
}
}
fn make_test_fill(trade_id: &str, instrument: &str) -> Fill {
Fill {
trade_id: TradeId::new(trade_id),
client_id: None,
exchange_order_id: None,
instrument: InstrumentId::new(instrument),
side: OrderSide::Buy,
price: Price::new(Decimal::new(100, 0)),
qty: Qty::new(Decimal::new(1, 0)),
fee: Fee::new(Decimal::new(1, 2), AssetId::new("USDC")),
ts: now_ms() + 1000, }
}
fn make_test_fill_with_ts(trade_id: &str, instrument: &str, ts: i64) -> Fill {
Fill {
trade_id: TradeId::new(trade_id),
client_id: None,
exchange_order_id: None,
instrument: InstrumentId::new(instrument),
side: OrderSide::Buy,
price: Price::new(Decimal::new(100, 0)),
qty: Qty::new(Decimal::new(1, 0)),
fee: Fee::new(Decimal::new(1, 2), AssetId::new("USDC")),
ts,
}
}
#[test]
fn test_config_validation() {
let config = TradeSyncerConfig {
bot_id: String::new(),
upstream_url: "http://test.com".to_string(),
..Default::default()
};
assert!(TradeSyncer::new(config).is_err());
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: String::new(),
..Default::default()
};
assert!(TradeSyncer::new(config).is_err());
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
..Default::default()
};
assert!(TradeSyncer::new(config).is_ok());
}
#[test]
fn test_add_fill_deduplication() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
..Default::default()
};
let mut syncer = TradeSyncer::new(config).unwrap();
let fill = make_test_fill("trade-1", "BTC-PERP");
syncer.add_fill(fill.clone());
syncer.add_fill(fill);
assert_eq!(syncer.pending_count(), 2);
syncer.synced_trade_ids.insert("trade-1".to_string());
syncer.pending_fills.clear();
let fill = make_test_fill("trade-1", "BTC-PERP");
syncer.add_fill(fill);
assert_eq!(syncer.pending_count(), 0);
}
#[test]
fn test_start_timestamp_filter() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
..Default::default()
};
let mut syncer = TradeSyncer::new(config).unwrap();
let old_fill =
make_test_fill_with_ts("trade-old", "BTC-PERP", syncer.start_timestamp - 1000);
syncer.add_fill(old_fill);
assert_eq!(
syncer.pending_count(),
0,
"Historical fill should be filtered"
);
let new_fill =
make_test_fill_with_ts("trade-new", "BTC-PERP", syncer.start_timestamp + 1000);
syncer.add_fill(new_fill);
assert_eq!(syncer.pending_count(), 1, "New fill should be accepted");
}
#[test]
fn test_instrument_filter() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
instruments: vec![InstrumentId::new("BTC-PERP")],
..Default::default()
};
let mut syncer = TradeSyncer::new(config).unwrap();
syncer.add_fill(make_test_fill("trade-1", "BTC-PERP"));
assert_eq!(syncer.pending_count(), 1);
syncer.add_fill(make_test_fill("trade-2", "ETH-PERP"));
assert_eq!(syncer.pending_count(), 1); }
#[test]
fn test_multi_instrument_filter() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
instruments: vec![
InstrumentId::new("UBTC-SPOT"),
InstrumentId::new("BTC-PERP"),
],
..Default::default()
};
let mut syncer = TradeSyncer::new(config).unwrap();
syncer.add_fill(make_test_fill("trade-1", "UBTC-SPOT"));
syncer.add_fill(make_test_fill("trade-2", "BTC-PERP"));
assert_eq!(syncer.pending_count(), 2);
syncer.add_fill(make_test_fill("trade-3", "ETH-PERP"));
assert_eq!(syncer.pending_count(), 2); }
#[test]
fn test_fill_to_trade_conversion() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
..Default::default()
};
let syncer = TradeSyncer::new(config).unwrap();
let fill = Fill {
trade_id: TradeId::new("trade-123"),
client_id: Some(bot_core::ClientOrderId::new("client-456")),
exchange_order_id: Some(bot_core::ExchangeOrderId::new("exchange-789")),
instrument: InstrumentId::new("BTC-PERP"),
side: OrderSide::Sell,
price: Price::new(Decimal::new(50000, 0)),
qty: Qty::new(Decimal::new(5, 1)), fee: Fee::new(Decimal::new(25, 2), AssetId::new("USDC")), ts: 1700000000000,
};
let trade = syncer.fill_to_trade(&fill);
assert_eq!(trade.trade_id, "trade-123");
assert_eq!(trade.client_order_id, "client-456");
assert_eq!(trade.venue_order_id, "exchange-789");
assert_eq!(trade.instrument_id, "BTC-PERP");
assert_eq!(trade.side, "SELL");
assert_eq!(trade.price, "50000");
assert_eq!(trade.qty, "0.5");
assert!(
trade.quote_notional.starts_with("25000"),
"Expected quote_notional to start with 25000, got: {}",
trade.quote_notional
);
assert_eq!(trade.fee, "0.25");
assert_eq!(trade.fee_currency, "USDC");
assert_eq!(trade.ts_event, 1700000000000);
}
#[test]
fn test_metadata_payload_includes_performance_metrics() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
strategy_type: Some("orchestrator".to_string()),
..Default::default()
};
let mut syncer = TradeSyncer::new(config).unwrap();
syncer.set_metrics_snapshot(Some(make_metrics_snapshot()));
let metadata = syncer.metadata_payload().expect("metadata");
assert_eq!(metadata["strategy_type"], serde_json::json!("orchestrator"));
assert_eq!(
metadata["performance_metrics"]["metrics"]["net_pnl"],
serde_json::json!("10")
);
assert_eq!(
metadata["performance_metrics"]["mode"],
serde_json::json!("backtest")
);
}
#[test]
fn test_metadata_payload_includes_strategy_type_without_metrics() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
strategy_type: Some("grid".to_string()),
..Default::default()
};
let syncer = TradeSyncer::new(config).unwrap();
let metadata = syncer.metadata_payload().expect("metadata");
assert_eq!(metadata["strategy_type"], serde_json::json!("grid"));
assert!(metadata.get("performance_metrics").is_none());
}
#[test]
fn test_metadata_payload_omitted_without_strategy_type_or_metrics() {
let config = TradeSyncerConfig {
bot_id: "test-bot".to_string(),
upstream_url: "http://test.com".to_string(),
..Default::default()
};
let syncer = TradeSyncer::new(config).unwrap();
assert!(syncer.metadata_payload().is_none());
}
}