Skip to main content

sandbox_quant/
order_manager.rs

1use std::collections::HashMap;
2use std::sync::Arc;
3use std::time::Instant;
4
5use anyhow::Result;
6use chrono::TimeZone;
7
8use crate::binance::rest::BinanceRestClient;
9use crate::binance::types::{BinanceMyTrade, BinanceOrderResponse};
10use crate::config::RiskConfig;
11use crate::model::order::{Fill, Order, OrderSide, OrderStatus, OrderType};
12use crate::model::position::Position;
13use crate::model::signal::Signal;
14use crate::order_store;
15use crate::risk_module::{
16    ApiEndpointGroup, EndpointRateLimits, OrderIntent, RateBudgetSnapshot, RejectionReasonCode,
17    RiskModule,
18};
19
20pub use crate::risk_module::MarketKind;
21
22#[derive(Debug, Clone)]
23pub enum OrderUpdate {
24    Submitted {
25        intent_id: String,
26        client_order_id: String,
27        server_order_id: u64,
28    },
29    Filled {
30        intent_id: String,
31        client_order_id: String,
32        side: OrderSide,
33        fills: Vec<Fill>,
34        avg_price: f64,
35    },
36    Rejected {
37        intent_id: String,
38        client_order_id: String,
39        reason_code: String,
40        reason: String,
41    },
42}
43
44#[derive(Debug, Clone, Default)]
45pub struct OrderHistoryStats {
46    pub trade_count: u32,
47    pub win_count: u32,
48    pub lose_count: u32,
49    pub realized_pnl: f64,
50}
51
52#[derive(Debug, Clone, Default)]
53pub struct OrderHistorySnapshot {
54    pub rows: Vec<String>,
55    pub stats: OrderHistoryStats,
56    pub strategy_stats: HashMap<String, OrderHistoryStats>,
57    pub fills: Vec<OrderHistoryFill>,
58    pub open_qty: f64,
59    pub open_entry_price: f64,
60    pub estimated_total_pnl_usdt: Option<f64>,
61    pub trade_data_complete: bool,
62    pub fetched_at_ms: u64,
63    pub fetch_latency_ms: u64,
64    pub latest_event_ms: Option<u64>,
65}
66
67#[derive(Debug, Clone)]
68pub struct OrderHistoryFill {
69    pub timestamp_ms: u64,
70    pub side: OrderSide,
71    pub price: f64,
72}
73
74pub struct OrderManager {
75    rest_client: Arc<BinanceRestClient>,
76    active_orders: HashMap<String, Order>,
77    position: Position,
78    symbol: String,
79    market: MarketKind,
80    order_amount_usdt: f64,
81    balances: HashMap<String, f64>,
82    last_price: f64,
83    risk_module: RiskModule,
84    default_strategy_cooldown_ms: u64,
85    default_strategy_max_active_orders: u32,
86    strategy_limits_by_tag: HashMap<String, StrategyExecutionLimit>,
87    last_strategy_submit_ms: HashMap<String, u64>,
88    default_symbol_max_exposure_usdt: f64,
89    symbol_exposure_limit_by_key: HashMap<String, f64>,
90}
91
92#[derive(Debug, Clone, Copy)]
93struct StrategyExecutionLimit {
94    cooldown_ms: u64,
95    max_active_orders: u32,
96}
97
98fn normalize_market_label(market: MarketKind) -> &'static str {
99    match market {
100        MarketKind::Spot => "spot",
101        MarketKind::Futures => "futures",
102    }
103}
104
105fn symbol_limit_key(symbol: &str, market: MarketKind) -> String {
106    format!(
107        "{}:{}",
108        symbol.trim().to_ascii_uppercase(),
109        normalize_market_label(market)
110    )
111}
112
113fn storage_symbol(symbol: &str, market: MarketKind) -> String {
114    match market {
115        MarketKind::Spot => symbol.to_string(),
116        MarketKind::Futures => format!("{}#FUT", symbol),
117    }
118}
119
120fn display_qty_for_history(status: &str, orig_qty: f64, executed_qty: f64) -> f64 {
121    match status {
122        "FILLED" | "PARTIALLY_FILLED" => executed_qty,
123        _ => orig_qty,
124    }
125}
126
127fn format_history_time(timestamp_ms: u64) -> String {
128    chrono::Utc
129        .timestamp_millis_opt(timestamp_ms as i64)
130        .single()
131        .map(|dt| {
132            dt.with_timezone(&chrono::Local)
133                .format("%H:%M:%S")
134                .to_string()
135        })
136        .unwrap_or_else(|| "--:--:--".to_string())
137}
138
139fn format_order_history_row(
140    timestamp_ms: u64,
141    status: &str,
142    side: &str,
143    qty: f64,
144    avg_price: f64,
145    client_order_id: &str,
146) -> String {
147    format!(
148        "{} {:<10} {:<4} {:.5} @ {:.2}  {}",
149        format_history_time(timestamp_ms),
150        status,
151        side,
152        qty,
153        avg_price,
154        client_order_id
155    )
156}
157
158fn source_label_from_client_order_id(client_order_id: &str) -> String {
159    if client_order_id.contains("-mnl-") {
160        "MANUAL".to_string()
161    } else if client_order_id.contains("-cfg-") {
162        "MA(Config)".to_string()
163    } else if client_order_id.contains("-fst-") {
164        "MA(Fast 5/20)".to_string()
165    } else if client_order_id.contains("-slw-") {
166        "MA(Slow 20/60)".to_string()
167    } else if let Some(source_tag) = parse_source_tag_from_client_order_id(client_order_id) {
168        source_tag.to_ascii_lowercase()
169    } else {
170        "UNKNOWN".to_string()
171    }
172}
173
174fn parse_source_tag_from_client_order_id(client_order_id: &str) -> Option<&str> {
175    let body = client_order_id.strip_prefix("sq-")?;
176    let (source_tag, _) = body.split_once('-')?;
177    if source_tag.is_empty() {
178        None
179    } else {
180        Some(source_tag)
181    }
182}
183
184fn format_trade_history_row(t: &BinanceMyTrade, source: &str) -> String {
185    let side = if t.is_buyer { "BUY" } else { "SELL" };
186    format_order_history_row(
187        t.time,
188        "FILLED",
189        side,
190        t.qty,
191        t.price,
192        &format!("order#{}#T{} [{}]", t.order_id, t.id, source),
193    )
194}
195
196fn split_symbol_assets(symbol: &str) -> (String, String) {
197    const QUOTE_SUFFIXES: [&str; 10] = [
198        "USDT", "USDC", "FDUSD", "BUSD", "TUSD", "TRY", "EUR", "BTC", "ETH", "BNB",
199    ];
200    for q in QUOTE_SUFFIXES {
201        if let Some(base) = symbol.strip_suffix(q) {
202            if !base.is_empty() {
203                return (base.to_string(), q.to_string());
204            }
205        }
206    }
207    (symbol.to_string(), String::new())
208}
209
210#[derive(Clone, Copy, Default)]
211struct LongPos {
212    qty: f64,
213    cost_quote: f64,
214}
215
216fn apply_spot_trade_with_fee(
217    pos: &mut LongPos,
218    stats: &mut OrderHistoryStats,
219    t: &BinanceMyTrade,
220    base_asset: &str,
221    quote_asset: &str,
222) {
223    let qty = t.qty.max(0.0);
224    if qty <= f64::EPSILON {
225        return;
226    }
227    let fee_asset = t.commission_asset.as_str();
228    let fee_is_base = !base_asset.is_empty() && fee_asset.eq_ignore_ascii_case(base_asset);
229    let fee_is_quote = !quote_asset.is_empty() && fee_asset.eq_ignore_ascii_case(quote_asset);
230
231    if t.is_buyer {
232        let net_qty = (qty
233            - if fee_is_base {
234                t.commission.max(0.0)
235            } else {
236                0.0
237            })
238        .max(0.0);
239        if net_qty <= f64::EPSILON {
240            return;
241        }
242        let fee_quote = if fee_is_quote {
243            t.commission.max(0.0)
244        } else {
245            0.0
246        };
247        pos.qty += net_qty;
248        pos.cost_quote += qty * t.price + fee_quote;
249        return;
250    }
251
252    // Spot sell: close against existing long inventory.
253    if pos.qty <= f64::EPSILON {
254        return;
255    }
256    let close_qty = qty.min(pos.qty);
257    if close_qty <= f64::EPSILON {
258        return;
259    }
260    let avg_cost = pos.cost_quote / pos.qty.max(f64::EPSILON);
261    let fee_quote_total = if fee_is_quote {
262        t.commission.max(0.0)
263    } else if fee_is_base {
264        // If fee is charged in base on sell, approximate its quote impact at fill price.
265        t.commission.max(0.0) * t.price
266    } else {
267        0.0
268    };
269    let fee_quote = fee_quote_total * (close_qty / qty.max(f64::EPSILON));
270    let pnl_delta = (close_qty * t.price - fee_quote) - (avg_cost * close_qty);
271    if pnl_delta > 0.0 {
272        stats.win_count += 1;
273        stats.trade_count += 1;
274    } else if pnl_delta < 0.0 {
275        stats.lose_count += 1;
276        stats.trade_count += 1;
277    }
278    stats.realized_pnl += pnl_delta;
279
280    pos.qty -= close_qty;
281    pos.cost_quote -= avg_cost * close_qty;
282    if pos.qty <= f64::EPSILON {
283        pos.qty = 0.0;
284        pos.cost_quote = 0.0;
285    }
286}
287
288fn compute_trade_state(
289    mut trades: Vec<BinanceMyTrade>,
290    symbol: &str,
291) -> (OrderHistoryStats, LongPos) {
292    trades.sort_by_key(|t| (t.time, t.id));
293    let (base_asset, quote_asset) = split_symbol_assets(symbol);
294    let mut pos = LongPos::default();
295    let mut stats = OrderHistoryStats::default();
296    for t in trades {
297        apply_spot_trade_with_fee(&mut pos, &mut stats, &t, &base_asset, &quote_asset);
298    }
299    (stats, pos)
300}
301
302fn compute_futures_open_state(mut trades: Vec<BinanceMyTrade>) -> LongPos {
303    trades.sort_by_key(|t| (t.time, t.id));
304    let mut pos = LongPos::default();
305    for t in trades {
306        let qty = t.qty.max(0.0);
307        if qty <= f64::EPSILON {
308            continue;
309        }
310        if t.is_buyer {
311            pos.qty += qty;
312            pos.cost_quote += qty * t.price;
313            continue;
314        }
315        if pos.qty <= f64::EPSILON {
316            continue;
317        }
318        let close_qty = qty.min(pos.qty);
319        let avg_cost = pos.cost_quote / pos.qty.max(f64::EPSILON);
320        pos.qty -= close_qty;
321        pos.cost_quote -= avg_cost * close_qty;
322        if pos.qty <= f64::EPSILON {
323            pos.qty = 0.0;
324            pos.cost_quote = 0.0;
325        }
326    }
327    pos
328}
329
330fn compute_trade_stats_by_source(
331    mut trades: Vec<BinanceMyTrade>,
332    order_source_by_id: &HashMap<u64, String>,
333    symbol: &str,
334) -> HashMap<String, OrderHistoryStats> {
335    trades.sort_by_key(|t| (t.time, t.id));
336
337    // Futures: realized_pnl is exchange-provided per fill.
338    if symbol.ends_with("#FUT") {
339        let mut stats_by_source: HashMap<String, OrderHistoryStats> = HashMap::new();
340        for t in trades {
341            let source = order_source_by_id
342                .get(&t.order_id)
343                .cloned()
344                .unwrap_or_else(|| "UNKNOWN".to_string());
345            let stats = stats_by_source.entry(source).or_default();
346            if t.realized_pnl > 0.0 {
347                stats.win_count += 1;
348                stats.trade_count += 1;
349            } else if t.realized_pnl < 0.0 {
350                stats.lose_count += 1;
351                stats.trade_count += 1;
352            }
353            stats.realized_pnl += t.realized_pnl;
354        }
355        return stats_by_source;
356    }
357
358    let (base_asset, quote_asset) = split_symbol_assets(symbol);
359    let mut pos_by_source: HashMap<String, LongPos> = HashMap::new();
360    let mut stats_by_source: HashMap<String, OrderHistoryStats> = HashMap::new();
361
362    for t in trades {
363        let source = order_source_by_id
364            .get(&t.order_id)
365            .cloned()
366            .unwrap_or_else(|| "UNKNOWN".to_string());
367        let pos = pos_by_source.entry(source.clone()).or_default();
368        let stats = stats_by_source.entry(source).or_default();
369        apply_spot_trade_with_fee(pos, stats, &t, &base_asset, &quote_asset);
370    }
371
372    stats_by_source
373}
374
375fn to_persistable_stats_map(
376    strategy_stats: &HashMap<String, OrderHistoryStats>,
377) -> HashMap<String, order_store::StrategyScopedStats> {
378    strategy_stats
379        .iter()
380        .map(|(k, v)| {
381            (
382                k.clone(),
383                order_store::StrategyScopedStats {
384                    trade_count: v.trade_count,
385                    win_count: v.win_count,
386                    lose_count: v.lose_count,
387                    realized_pnl: v.realized_pnl,
388                },
389            )
390        })
391        .collect()
392}
393
394fn from_persisted_stats_map(
395    persisted: HashMap<String, order_store::StrategyScopedStats>,
396) -> HashMap<String, OrderHistoryStats> {
397    persisted
398        .into_iter()
399        .map(|(k, v)| {
400            (
401                k,
402                OrderHistoryStats {
403                    trade_count: v.trade_count,
404                    win_count: v.win_count,
405                    lose_count: v.lose_count,
406                    realized_pnl: v.realized_pnl,
407                },
408            )
409        })
410        .collect()
411}
412
413impl OrderManager {
414    /// Create a new order manager bound to a single symbol/market context.
415    ///
416    /// The instance keeps in-memory position, cached balances, and an embedded
417    /// `RiskModule` that enforces pre-trade checks and global rate budget.
418    ///
419    /// # Caution
420    /// This manager is stateful (`last_price`, balances, active orders). Reuse
421    /// the same instance for a symbol stream instead of recreating per tick.
422    pub fn new(
423        rest_client: Arc<BinanceRestClient>,
424        symbol: &str,
425        market: MarketKind,
426        order_amount_usdt: f64,
427        risk_config: &RiskConfig,
428    ) -> Self {
429        let mut strategy_limits_by_tag = HashMap::new();
430        let mut symbol_exposure_limit_by_key = HashMap::new();
431        let default_strategy_cooldown_ms = risk_config.default_strategy_cooldown_ms;
432        let default_strategy_max_active_orders =
433            risk_config.default_strategy_max_active_orders.max(1);
434        let default_symbol_max_exposure_usdt =
435            risk_config.default_symbol_max_exposure_usdt.max(0.0);
436        for profile in &risk_config.strategy_limits {
437            let source_tag = profile.source_tag.trim().to_ascii_lowercase();
438            if source_tag.is_empty() {
439                continue;
440            }
441            strategy_limits_by_tag.insert(
442                source_tag,
443                StrategyExecutionLimit {
444                    cooldown_ms: profile.cooldown_ms.unwrap_or(default_strategy_cooldown_ms),
445                    max_active_orders: profile
446                        .max_active_orders
447                        .unwrap_or(default_strategy_max_active_orders)
448                        .max(1),
449                },
450            );
451        }
452        for limit in &risk_config.symbol_exposure_limits {
453            let symbol = limit.symbol.trim().to_ascii_uppercase();
454            if symbol.is_empty() {
455                continue;
456            }
457            let market = match limit
458                .market
459                .as_deref()
460                .unwrap_or("spot")
461                .trim()
462                .to_ascii_lowercase()
463                .as_str()
464            {
465                "spot" => MarketKind::Spot,
466                "futures" | "future" | "fut" => MarketKind::Futures,
467                _ => continue,
468            };
469            symbol_exposure_limit_by_key.insert(
470                symbol_limit_key(&symbol, market),
471                limit.max_exposure_usdt.max(0.0),
472            );
473        }
474        Self {
475            rest_client: rest_client.clone(),
476            active_orders: HashMap::new(),
477            position: Position::new(symbol.to_string()),
478            symbol: symbol.to_string(),
479            market,
480            order_amount_usdt,
481            balances: HashMap::new(),
482            last_price: 0.0,
483            risk_module: RiskModule::new(
484                rest_client.clone(),
485                risk_config.global_rate_limit_per_minute,
486                EndpointRateLimits {
487                    orders_per_minute: risk_config.endpoint_rate_limits.orders_per_minute,
488                    account_per_minute: risk_config.endpoint_rate_limits.account_per_minute,
489                    market_data_per_minute: risk_config.endpoint_rate_limits.market_data_per_minute,
490                },
491            ),
492            default_strategy_cooldown_ms,
493            default_strategy_max_active_orders,
494            strategy_limits_by_tag,
495            last_strategy_submit_ms: HashMap::new(),
496            default_symbol_max_exposure_usdt,
497            symbol_exposure_limit_by_key,
498        }
499    }
500
501    /// Return current in-memory position snapshot.
502    ///
503    /// Values reflect fills processed by this process. They are not a full
504    /// exchange reconciliation snapshot.
505    pub fn position(&self) -> &Position {
506        &self.position
507    }
508
509    pub fn market_kind(&self) -> MarketKind {
510        self.market
511    }
512
513    /// Return latest cached free balances.
514    ///
515    /// Cache is updated by `refresh_balances`. Missing assets should be treated
516    /// as zero balance.
517    pub fn balances(&self) -> &HashMap<String, f64> {
518        &self.balances
519    }
520
521    /// Update last price and recompute unrealized PnL.
522    ///
523    /// # Usage
524    /// Call on every market data tick before `submit_order`, so risk checks use
525    /// a valid `last_price`.
526    pub fn update_unrealized_pnl(&mut self, current_price: f64) {
527        self.last_price = current_price;
528        self.position.update_unrealized_pnl(current_price);
529    }
530
531    pub fn last_price(&self) -> Option<f64> {
532        (self.last_price > f64::EPSILON).then_some(self.last_price)
533    }
534
535    /// Return current global rate-budget snapshot from the risk module.
536    ///
537    /// Intended for UI display and observability.
538    pub fn rate_budget_snapshot(&self) -> RateBudgetSnapshot {
539        self.risk_module.rate_budget_snapshot()
540    }
541
542    pub fn orders_rate_budget_snapshot(&self) -> RateBudgetSnapshot {
543        self.risk_module
544            .endpoint_budget_snapshot(ApiEndpointGroup::Orders)
545    }
546
547    pub fn account_rate_budget_snapshot(&self) -> RateBudgetSnapshot {
548        self.risk_module
549            .endpoint_budget_snapshot(ApiEndpointGroup::Account)
550    }
551
552    pub fn market_data_rate_budget_snapshot(&self) -> RateBudgetSnapshot {
553        self.risk_module
554            .endpoint_budget_snapshot(ApiEndpointGroup::MarketData)
555    }
556
557    fn strategy_limits_for(&self, source_tag: &str) -> StrategyExecutionLimit {
558        self.strategy_limits_by_tag
559            .get(source_tag)
560            .copied()
561            .unwrap_or(StrategyExecutionLimit {
562                cooldown_ms: self.default_strategy_cooldown_ms,
563                max_active_orders: self.default_strategy_max_active_orders,
564            })
565    }
566
567    fn active_order_count_for_source(&self, source_tag: &str) -> u32 {
568        let prefix = format!("sq-{}-", source_tag);
569        self.active_orders
570            .values()
571            .filter(|o| !o.status.is_terminal() && o.client_order_id.starts_with(&prefix))
572            .count() as u32
573    }
574
575    fn evaluate_strategy_limits(
576        &self,
577        source_tag: &str,
578        created_at_ms: u64,
579    ) -> Option<(String, String)> {
580        let limits = self.strategy_limits_for(source_tag);
581        let active_count = self.active_order_count_for_source(source_tag);
582        if active_count >= limits.max_active_orders {
583            return Some((
584                RejectionReasonCode::RiskStrategyMaxActiveOrdersExceeded
585                    .as_str()
586                    .to_string(),
587                format!(
588                    "Strategy '{}' active order limit exceeded (active {}, limit {})",
589                    source_tag, active_count, limits.max_active_orders
590                ),
591            ));
592        }
593
594        if limits.cooldown_ms > 0 {
595            if let Some(last_submit_ms) = self.last_strategy_submit_ms.get(source_tag) {
596                let elapsed = created_at_ms.saturating_sub(*last_submit_ms);
597                if elapsed < limits.cooldown_ms {
598                    let remaining = limits.cooldown_ms - elapsed;
599                    return Some((
600                        RejectionReasonCode::RiskStrategyCooldownActive
601                            .as_str()
602                            .to_string(),
603                        format!(
604                            "Strategy '{}' cooldown active ({}ms remaining)",
605                            source_tag, remaining
606                        ),
607                    ));
608                }
609            }
610        }
611
612        None
613    }
614
615    fn mark_strategy_submit(&mut self, source_tag: &str, created_at_ms: u64) {
616        self.last_strategy_submit_ms
617            .insert(source_tag.to_string(), created_at_ms);
618    }
619
620    fn max_symbol_exposure_usdt(&self) -> f64 {
621        self.symbol_exposure_limit_by_key
622            .get(&symbol_limit_key(&self.symbol, self.market))
623            .copied()
624            .unwrap_or(self.default_symbol_max_exposure_usdt)
625    }
626
627    fn projected_notional_after_fill(&self, side: OrderSide, qty: f64) -> (f64, f64) {
628        let price = self.last_price.max(0.0);
629        if price <= f64::EPSILON {
630            return (0.0, 0.0);
631        }
632        let current_qty_signed = match self.position.side {
633            Some(OrderSide::Buy) => self.position.qty,
634            Some(OrderSide::Sell) => -self.position.qty,
635            None => 0.0,
636        };
637        let delta = match side {
638            OrderSide::Buy => qty,
639            OrderSide::Sell => -qty,
640        };
641        let projected_qty_signed = current_qty_signed + delta;
642        (
643            current_qty_signed.abs() * price,
644            projected_qty_signed.abs() * price,
645        )
646    }
647
648    fn evaluate_symbol_exposure_limit(
649        &self,
650        side: OrderSide,
651        qty: f64,
652    ) -> Option<(String, String)> {
653        let max_exposure = self.max_symbol_exposure_usdt();
654        if max_exposure <= f64::EPSILON {
655            return None;
656        }
657        let (current_notional, projected_notional) = self.projected_notional_after_fill(side, qty);
658        if projected_notional > max_exposure && projected_notional > current_notional + f64::EPSILON
659        {
660            return Some((
661                RejectionReasonCode::RiskSymbolExposureLimitExceeded
662                    .as_str()
663                    .to_string(),
664                format!(
665                    "Symbol exposure limit exceeded for {} ({:?}): projected {:.2} USDT > limit {:.2} USDT",
666                    self.symbol, self.market, projected_notional, max_exposure
667                ),
668            ));
669        }
670        None
671    }
672
673    /// Return whether a hypothetical fill would exceed symbol exposure limit.
674    ///
675    /// This is intended for validation and tests; it does not mutate state.
676    pub fn would_exceed_symbol_exposure_limit(&self, side: OrderSide, qty: f64) -> bool {
677        self.evaluate_symbol_exposure_limit(side, qty).is_some()
678    }
679
680    /// Fetch account balances from Binance and update internal state.
681    ///
682    /// Returns the map `asset -> free` for assets with non-zero total (spot) or
683    /// non-trivial wallet balance (futures).
684    ///
685    /// # Usage
686    /// Refresh before order submission cycles to reduce false "insufficient
687    /// balance" rejections from stale cache.
688    ///
689    /// # Caution
690    /// Network/API failures return `Err(_)` and leave previous cache untouched.
691    pub async fn refresh_balances(&mut self) -> Result<HashMap<String, f64>> {
692        if !self
693            .risk_module
694            .reserve_endpoint_budget(ApiEndpointGroup::Account)
695        {
696            return Err(anyhow::anyhow!(
697                "Account endpoint budget exceeded; try again after reset"
698            ));
699        }
700        if self.market == MarketKind::Futures {
701            let account = self.rest_client.get_futures_account().await?;
702            self.balances.clear();
703            for a in &account.assets {
704                if a.wallet_balance.abs() > f64::EPSILON {
705                    self.balances.insert(a.asset.clone(), a.available_balance);
706                }
707            }
708            return Ok(self.balances.clone());
709        }
710        let account = self.rest_client.get_account().await?;
711        self.balances.clear();
712        for b in &account.balances {
713            let total = b.free + b.locked;
714            if total > 0.0 {
715                self.balances.insert(b.asset.clone(), b.free);
716            }
717        }
718        tracing::info!(balances = ?self.balances, "Balances refreshed");
719        Ok(self.balances.clone())
720    }
721
722    /// Fetch order history from exchange and format rows for UI display.
723    ///
724    /// This method combines order and trade endpoints, persists snapshots to
725    /// local sqlite, and emits a best-effort history view even if one endpoint
726    /// fails.
727    ///
728    /// # Caution
729    /// `trade_data_complete = false` means derived PnL may be partial.
730    pub async fn refresh_order_history(&mut self, limit: usize) -> Result<OrderHistorySnapshot> {
731        if !self
732            .risk_module
733            .reserve_endpoint_budget(ApiEndpointGroup::Orders)
734        {
735            return Err(anyhow::anyhow!(
736                "Orders endpoint budget exceeded; try again after reset"
737            ));
738        }
739        if self.market == MarketKind::Futures {
740            let fetch_started = Instant::now();
741            let fetched_at_ms = chrono::Utc::now().timestamp_millis() as u64;
742            let orders_result = self
743                .rest_client
744                .get_futures_all_orders(&self.symbol, limit)
745                .await;
746            let trades_result = self
747                .rest_client
748                .get_futures_my_trades_history(&self.symbol, limit.max(1))
749                .await;
750            let fetch_latency_ms = fetch_started.elapsed().as_millis() as u64;
751
752            if orders_result.is_err() && trades_result.is_err() {
753                let oe = orders_result.err().unwrap();
754                let te = trades_result.err().unwrap();
755                return Err(anyhow::anyhow!(
756                    "futures order history fetch failed: allOrders={} | userTrades={}",
757                    oe,
758                    te
759                ));
760            }
761
762            let mut orders = orders_result.unwrap_or_default();
763            let trades = trades_result.unwrap_or_default();
764            orders.sort_by_key(|o| o.update_time.max(o.time));
765
766            let storage_key = storage_symbol(&self.symbol, self.market);
767            if let Err(e) = order_store::persist_order_snapshot(&storage_key, &orders, &trades) {
768                tracing::warn!(error = %e, "Failed to persist futures order snapshot to sqlite");
769            }
770
771            let mut order_source_by_id = HashMap::new();
772            for o in &orders {
773                order_source_by_id.insert(
774                    o.order_id,
775                    source_label_from_client_order_id(&o.client_order_id),
776                );
777            }
778            let mut trades_for_stats = trades.clone();
779            match order_store::load_persisted_trades(&storage_key) {
780                Ok(saved) if !saved.is_empty() => {
781                    trades_for_stats = saved.iter().map(|r| r.trade.clone()).collect();
782                    for row in saved {
783                        order_source_by_id.entry(row.trade.order_id).or_insert(row.source);
784                    }
785                }
786                Ok(_) => {}
787                Err(e) => {
788                    tracing::warn!(
789                        error = %e,
790                        "Failed to load persisted futures trades; using API trades"
791                    );
792                }
793            }
794
795            let mut history = Vec::new();
796            let mut fills = Vec::new();
797            for t in &trades {
798                let side = if t.is_buyer { "BUY" } else { "SELL" };
799                let source = order_source_by_id
800                    .get(&t.order_id)
801                    .cloned()
802                    .unwrap_or_else(|| "UNKNOWN".to_string());
803                fills.push(OrderHistoryFill {
804                    timestamp_ms: t.time,
805                    side: if t.is_buyer {
806                        OrderSide::Buy
807                    } else {
808                        OrderSide::Sell
809                    },
810                    price: t.price,
811                });
812                history.push(format_order_history_row(
813                    t.time,
814                    "FILLED",
815                    side,
816                    t.qty,
817                    t.price,
818                    &format!("order#{}#T{} [{}]", t.order_id, t.id, source),
819                ));
820            }
821            for o in &orders {
822                if o.executed_qty <= 0.0 {
823                    history.push(format_order_history_row(
824                        o.update_time.max(o.time),
825                        &o.status,
826                        &o.side,
827                        display_qty_for_history(&o.status, o.orig_qty, o.executed_qty),
828                        if o.executed_qty > 0.0 {
829                            o.cummulative_quote_qty / o.executed_qty
830                        } else {
831                            o.price
832                        },
833                        &o.client_order_id,
834                    ));
835                }
836            }
837
838            let mut stats = OrderHistoryStats::default();
839            for t in &trades {
840                if t.realized_pnl > 0.0 {
841                    stats.win_count += 1;
842                    stats.trade_count += 1;
843                } else if t.realized_pnl < 0.0 {
844                    stats.lose_count += 1;
845                    stats.trade_count += 1;
846                }
847                stats.realized_pnl += t.realized_pnl;
848            }
849            let open_pos = compute_futures_open_state(trades_for_stats.clone());
850            let open_entry_price = if open_pos.qty > f64::EPSILON {
851                open_pos.cost_quote / open_pos.qty
852            } else {
853                0.0
854            };
855            self.position.side = if open_pos.qty > f64::EPSILON {
856                Some(OrderSide::Buy)
857            } else {
858                None
859            };
860            self.position.qty = open_pos.qty;
861            self.position.entry_price = open_entry_price;
862            self.position.realized_pnl = stats.realized_pnl;
863            if self.last_price > 0.0 {
864                self.position.update_unrealized_pnl(self.last_price);
865            } else {
866                self.position.unrealized_pnl = 0.0;
867            }
868            let estimated_total_pnl_usdt = if self.last_price > 0.0 && open_pos.qty > f64::EPSILON {
869                Some(stats.realized_pnl + (self.last_price - open_entry_price) * open_pos.qty)
870            } else {
871                Some(stats.realized_pnl)
872            };
873            let latest_order_event = orders.iter().map(|o| o.update_time.max(o.time)).max();
874            let latest_trade_event = trades.iter().map(|t| t.time).max();
875            let mut strategy_stats =
876                compute_trade_stats_by_source(trades_for_stats, &order_source_by_id, &storage_key);
877            let persisted_stats = to_persistable_stats_map(&strategy_stats);
878            if let Err(e) = order_store::persist_strategy_symbol_stats(&storage_key, &persisted_stats)
879            {
880                tracing::warn!(error = %e, "Failed to persist strategy stats (futures)");
881            }
882            if strategy_stats.is_empty() {
883                match order_store::load_strategy_symbol_stats(&storage_key) {
884                    Ok(persisted) => {
885                        strategy_stats = from_persisted_stats_map(persisted);
886                    }
887                    Err(e) => {
888                        tracing::warn!(
889                            error = %e,
890                            "Failed to load persisted strategy stats (futures)"
891                        );
892                    }
893                }
894            }
895            return Ok(OrderHistorySnapshot {
896                rows: history,
897                stats,
898                strategy_stats,
899                fills,
900                open_qty: open_pos.qty,
901                open_entry_price,
902                estimated_total_pnl_usdt,
903                trade_data_complete: true,
904                fetched_at_ms,
905                fetch_latency_ms,
906                latest_event_ms: latest_order_event.max(latest_trade_event),
907            });
908        }
909
910        let fetch_started = Instant::now();
911        let fetched_at_ms = chrono::Utc::now().timestamp_millis() as u64;
912        let orders_result = self.rest_client.get_all_orders(&self.symbol, limit).await;
913        let storage_key = storage_symbol(&self.symbol, self.market);
914        let last_trade_id = order_store::load_last_trade_id(&storage_key).ok().flatten();
915        let persisted_trade_count = order_store::load_trade_count(&storage_key).unwrap_or(0);
916        let need_backfill = persisted_trade_count < limit;
917        let trades_result = match (need_backfill, last_trade_id) {
918            (true, _) => {
919                self.rest_client
920                    .get_my_trades_history(&self.symbol, limit.max(1))
921                    .await
922            }
923            (false, Some(last_id)) => {
924                self.rest_client
925                    .get_my_trades_since(&self.symbol, last_id.saturating_add(1), 10)
926                    .await
927            }
928            (false, None) => {
929                self.rest_client
930                    .get_my_trades_history(&self.symbol, limit.max(1))
931                    .await
932            }
933        };
934        let fetch_latency_ms = fetch_started.elapsed().as_millis() as u64;
935        let trade_data_complete = trades_result.is_ok();
936
937        if orders_result.is_err() && trades_result.is_err() {
938            let oe = orders_result.err().unwrap();
939            let te = trades_result.err().unwrap();
940            return Err(anyhow::anyhow!(
941                "order history fetch failed: allOrders={} | myTrades={}",
942                oe,
943                te
944            ));
945        }
946
947        let mut orders = match orders_result {
948            Ok(v) => v,
949            Err(e) => {
950                tracing::warn!(error = %e, "Failed to fetch allOrders; falling back to trade-only history");
951                Vec::new()
952            }
953        };
954        let recent_trades = match trades_result {
955            Ok(t) => t,
956            Err(e) => {
957                tracing::warn!(error = %e, "Failed to fetch myTrades; falling back to order-only history");
958                Vec::new()
959            }
960        };
961        let mut trades = recent_trades.clone();
962        orders.sort_by_key(|o| o.update_time.max(o.time));
963
964        if let Err(e) = order_store::persist_order_snapshot(&storage_key, &orders, &recent_trades) {
965            tracing::warn!(error = %e, "Failed to persist order snapshot to sqlite");
966        }
967        let mut persisted_source_by_order_id: HashMap<u64, String> = HashMap::new();
968        match order_store::load_persisted_trades(&storage_key) {
969            Ok(saved) => {
970                if !saved.is_empty() {
971                    trades = saved.iter().map(|r| r.trade.clone()).collect();
972                    for row in saved {
973                        persisted_source_by_order_id
974                            .entry(row.trade.order_id)
975                            .or_insert(row.source);
976                    }
977                }
978            }
979            Err(e) => {
980                tracing::warn!(error = %e, "Failed to load persisted trades; using recent API trades");
981            }
982        }
983
984        let (stats, open_pos) = compute_trade_state(trades.clone(), &self.symbol);
985        self.position.side = if open_pos.qty > f64::EPSILON {
986            Some(OrderSide::Buy)
987        } else {
988            None
989        };
990        self.position.qty = open_pos.qty;
991        self.position.entry_price = if open_pos.qty > f64::EPSILON {
992            open_pos.cost_quote / open_pos.qty
993        } else {
994            0.0
995        };
996        self.position.realized_pnl = stats.realized_pnl;
997        if self.last_price > 0.0 {
998            self.position.update_unrealized_pnl(self.last_price);
999        } else {
1000            self.position.unrealized_pnl = 0.0;
1001        }
1002        let estimated_total_pnl_usdt = if self.last_price > 0.0 {
1003            Some(stats.realized_pnl + (open_pos.qty * self.last_price - open_pos.cost_quote))
1004        } else {
1005            Some(stats.realized_pnl)
1006        };
1007        let latest_order_event = orders.iter().map(|o| o.update_time.max(o.time)).max();
1008        let latest_trade_event = trades.iter().map(|t| t.time).max();
1009        let latest_event_ms = latest_order_event.max(latest_trade_event);
1010
1011        let mut trades_by_order_id: HashMap<u64, Vec<BinanceMyTrade>> = HashMap::new();
1012        for trade in &trades {
1013            trades_by_order_id
1014                .entry(trade.order_id)
1015                .or_default()
1016                .push(trade.clone());
1017        }
1018        for bucket in trades_by_order_id.values_mut() {
1019            bucket.sort_by_key(|t| t.time);
1020        }
1021
1022        let mut order_source_by_id = HashMap::new();
1023        for o in &orders {
1024            order_source_by_id.insert(
1025                o.order_id,
1026                source_label_from_client_order_id(&o.client_order_id),
1027            );
1028        }
1029        for (order_id, source) in persisted_source_by_order_id {
1030            order_source_by_id.entry(order_id).or_insert(source);
1031        }
1032        let mut strategy_stats =
1033            compute_trade_stats_by_source(trades.clone(), &order_source_by_id, &self.symbol);
1034        let persisted_stats = to_persistable_stats_map(&strategy_stats);
1035        if let Err(e) = order_store::persist_strategy_symbol_stats(&storage_key, &persisted_stats) {
1036            tracing::warn!(error = %e, "Failed to persist strategy+symbol scoped stats");
1037        }
1038        if strategy_stats.is_empty() {
1039            match order_store::load_strategy_symbol_stats(&storage_key) {
1040                Ok(persisted) => {
1041                    strategy_stats = from_persisted_stats_map(persisted);
1042                }
1043                Err(e) => {
1044                    tracing::warn!(error = %e, "Failed to load persisted strategy+symbol stats");
1045                }
1046            }
1047        }
1048
1049        let mut history = Vec::new();
1050        let mut fills = Vec::new();
1051        let mut used_trade_ids = std::collections::HashSet::new();
1052
1053        if orders.is_empty() && !trades.is_empty() {
1054            let mut sorted = trades;
1055            sorted.sort_by_key(|t| (t.time, t.id));
1056            history.extend(sorted.iter().map(|t| {
1057                fills.push(OrderHistoryFill {
1058                    timestamp_ms: t.time,
1059                    side: if t.is_buyer {
1060                        OrderSide::Buy
1061                    } else {
1062                        OrderSide::Sell
1063                    },
1064                    price: t.price,
1065                });
1066                format_trade_history_row(
1067                    t,
1068                    order_source_by_id
1069                        .get(&t.order_id)
1070                        .map(String::as_str)
1071                        .unwrap_or("UNKNOWN"),
1072                )
1073            }));
1074            return Ok(OrderHistorySnapshot {
1075                rows: history,
1076                stats,
1077                strategy_stats,
1078                fills,
1079                open_qty: open_pos.qty,
1080                open_entry_price: if open_pos.qty > f64::EPSILON {
1081                    open_pos.cost_quote / open_pos.qty
1082                } else {
1083                    0.0
1084                },
1085                estimated_total_pnl_usdt,
1086                trade_data_complete,
1087                fetched_at_ms,
1088                fetch_latency_ms,
1089                latest_event_ms,
1090            });
1091        }
1092
1093        for o in orders {
1094            if o.executed_qty > 0.0 {
1095                if let Some(order_trades) = trades_by_order_id.get(&o.order_id) {
1096                    for t in order_trades {
1097                        used_trade_ids.insert(t.id);
1098                        let side = if t.is_buyer { "BUY" } else { "SELL" };
1099                        fills.push(OrderHistoryFill {
1100                            timestamp_ms: t.time,
1101                            side: if t.is_buyer {
1102                                OrderSide::Buy
1103                            } else {
1104                                OrderSide::Sell
1105                            },
1106                            price: t.price,
1107                        });
1108                        history.push(format_order_history_row(
1109                            t.time,
1110                            "FILLED",
1111                            side,
1112                            t.qty,
1113                            t.price,
1114                            &format!(
1115                                "{}#T{} [{}]",
1116                                o.client_order_id,
1117                                t.id,
1118                                source_label_from_client_order_id(&o.client_order_id)
1119                            ),
1120                        ));
1121                    }
1122                    continue;
1123                }
1124            }
1125
1126            let avg_price = if o.executed_qty > 0.0 {
1127                o.cummulative_quote_qty / o.executed_qty
1128            } else {
1129                o.price
1130            };
1131            history.push(format_order_history_row(
1132                o.update_time.max(o.time),
1133                &o.status,
1134                &o.side,
1135                display_qty_for_history(&o.status, o.orig_qty, o.executed_qty),
1136                avg_price,
1137                &o.client_order_id,
1138            ));
1139        }
1140
1141        // Include trades that did not match fetched order pages.
1142        for bucket in trades_by_order_id.values() {
1143            for t in bucket {
1144                if !used_trade_ids.contains(&t.id) {
1145                    fills.push(OrderHistoryFill {
1146                        timestamp_ms: t.time,
1147                        side: if t.is_buyer {
1148                            OrderSide::Buy
1149                        } else {
1150                            OrderSide::Sell
1151                        },
1152                        price: t.price,
1153                    });
1154                    history.push(format_trade_history_row(
1155                        t,
1156                        order_source_by_id
1157                            .get(&t.order_id)
1158                            .map(String::as_str)
1159                            .unwrap_or("UNKNOWN"),
1160                    ));
1161                }
1162            }
1163        }
1164        Ok(OrderHistorySnapshot {
1165            rows: history,
1166            stats,
1167            strategy_stats,
1168            fills,
1169            open_qty: open_pos.qty,
1170            open_entry_price: if open_pos.qty > f64::EPSILON {
1171                open_pos.cost_quote / open_pos.qty
1172            } else {
1173                0.0
1174            },
1175            estimated_total_pnl_usdt,
1176            trade_data_complete,
1177            fetched_at_ms,
1178            fetch_latency_ms,
1179            latest_event_ms,
1180        })
1181    }
1182
1183    /// Build an order intent, run risk checks, and submit to broker when approved.
1184    ///
1185    /// # Behavior
1186    /// - `Signal::Hold` returns `Ok(None)`.
1187    /// - For buy/sell signals, this method:
1188    ///   1. Builds `OrderIntent`.
1189    ///   2. Calls `RiskModule::evaluate_intent`.
1190    ///   3. Reserves one global rate token via `reserve_rate_budget`.
1191    ///   4. Submits market order to spot/futures broker endpoint.
1192    /// - Rejections are returned as `Ok(Some(OrderUpdate::Rejected { .. }))`
1193    ///   with structured `reason_code`.
1194    ///
1195    /// # Usage
1196    /// Recommended sequence:
1197    /// 1. `update_unrealized_pnl(last_price)`
1198    /// 2. `refresh_balances()` (periodic or before trading loop)
1199    /// 3. `submit_order(signal, source_tag)`
1200    ///
1201    /// # Caution
1202    /// - Spot sell requires base-asset balance (e.g. `ETH` for `ETHUSDT`).
1203    /// - If balances are stale, you may see "No position to sell" or
1204    ///   `"Insufficient <asset>"` even though exchange state changed recently.
1205    /// - This method returns transport/runtime errors as `Err(_)`; business
1206    ///   rejections are encoded in `OrderUpdate::Rejected`.
1207    pub async fn submit_order(
1208        &mut self,
1209        signal: Signal,
1210        source_tag: &str,
1211    ) -> Result<Option<OrderUpdate>> {
1212        let side = match &signal {
1213            Signal::Buy => OrderSide::Buy,
1214            Signal::Sell => OrderSide::Sell,
1215            Signal::Hold => return Ok(None),
1216        };
1217        let source_tag = source_tag.to_ascii_lowercase();
1218        let intent = OrderIntent {
1219            intent_id: format!("intent-{}", &uuid::Uuid::new_v4().to_string()[..8]),
1220            source_tag: source_tag.clone(),
1221            symbol: self.symbol.clone(),
1222            market: self.market,
1223            side,
1224            order_amount_usdt: self.order_amount_usdt,
1225            last_price: self.last_price,
1226            created_at_ms: chrono::Utc::now().timestamp_millis() as u64,
1227        };
1228        if let Some((reason_code, reason)) =
1229            self.evaluate_strategy_limits(&intent.source_tag, intent.created_at_ms)
1230        {
1231            return Ok(Some(OrderUpdate::Rejected {
1232                intent_id: intent.intent_id.clone(),
1233                client_order_id: "n/a".to_string(),
1234                reason_code,
1235                reason,
1236            }));
1237        }
1238        let decision = self
1239            .risk_module
1240            .evaluate_intent(&intent, &self.balances)
1241            .await?;
1242        if !decision.approved {
1243            return Ok(Some(OrderUpdate::Rejected {
1244                intent_id: intent.intent_id.clone(),
1245                client_order_id: "n/a".to_string(),
1246                reason_code: decision
1247                    .reason_code
1248                    .unwrap_or_else(|| RejectionReasonCode::RiskUnknown.as_str().to_string()),
1249                reason: decision
1250                    .reason
1251                    .unwrap_or_else(|| "Rejected by RiskModule".to_string()),
1252            }));
1253        }
1254        if !self.risk_module.reserve_rate_budget() {
1255            return Ok(Some(OrderUpdate::Rejected {
1256                intent_id: intent.intent_id.clone(),
1257                client_order_id: "n/a".to_string(),
1258                reason_code: RejectionReasonCode::RateGlobalBudgetExceeded
1259                    .as_str()
1260                    .to_string(),
1261                reason: "Global rate budget exceeded; try again after reset".to_string(),
1262            }));
1263        }
1264        if !self
1265            .risk_module
1266            .reserve_endpoint_budget(ApiEndpointGroup::Orders)
1267        {
1268            return Ok(Some(OrderUpdate::Rejected {
1269                intent_id: intent.intent_id.clone(),
1270                client_order_id: "n/a".to_string(),
1271                reason_code: RejectionReasonCode::RateEndpointBudgetExceeded
1272                    .as_str()
1273                    .to_string(),
1274                reason: "Orders endpoint budget exceeded; try again after reset".to_string(),
1275            }));
1276        }
1277        let qty = decision.normalized_qty;
1278        if let Some((reason_code, reason)) = self.evaluate_symbol_exposure_limit(side, qty) {
1279            return Ok(Some(OrderUpdate::Rejected {
1280                intent_id: intent.intent_id.clone(),
1281                client_order_id: "n/a".to_string(),
1282                reason_code,
1283                reason,
1284            }));
1285        }
1286        self.mark_strategy_submit(&intent.source_tag, intent.created_at_ms);
1287
1288        let client_order_id = format!(
1289            "sq-{}-{}",
1290            intent.source_tag,
1291            &uuid::Uuid::new_v4().to_string()[..8]
1292        );
1293
1294        let order = Order {
1295            client_order_id: client_order_id.clone(),
1296            server_order_id: None,
1297            symbol: self.symbol.clone(),
1298            side,
1299            order_type: OrderType::Market,
1300            quantity: qty,
1301            price: None,
1302            status: OrderStatus::PendingSubmit,
1303            created_at: chrono::Utc::now(),
1304            updated_at: chrono::Utc::now(),
1305            fills: vec![],
1306        };
1307
1308        self.active_orders.insert(client_order_id.clone(), order);
1309
1310        tracing::info!(
1311            side = %side,
1312            qty,
1313            usdt_amount = intent.order_amount_usdt,
1314            price = intent.last_price,
1315            intent_id = %intent.intent_id,
1316            created_at_ms = intent.created_at_ms,
1317            "Submitting order"
1318        );
1319
1320        let submit_res = if self.market == MarketKind::Futures {
1321            self.rest_client
1322                .place_futures_market_order(&self.symbol, side, qty, &client_order_id)
1323                .await
1324        } else {
1325            self.rest_client
1326                .place_market_order(&self.symbol, side, qty, &client_order_id)
1327                .await
1328        };
1329
1330        match submit_res {
1331            Ok(response) => {
1332                let update = self.process_order_response(
1333                    &intent.intent_id,
1334                    &client_order_id,
1335                    side,
1336                    &response,
1337                );
1338
1339                // Refresh balances after fill
1340                if matches!(update, OrderUpdate::Filled { .. }) {
1341                    if let Err(e) = self.refresh_balances().await {
1342                        tracing::warn!(error = %e, "Failed to refresh balances after fill");
1343                    }
1344                }
1345
1346                Ok(Some(update))
1347            }
1348            Err(e) => {
1349                tracing::error!(
1350                    client_order_id,
1351                    error = %e,
1352                    "Order rejected"
1353                );
1354                if let Some(order) = self.active_orders.get_mut(&client_order_id) {
1355                    order.status = OrderStatus::Rejected;
1356                    order.updated_at = chrono::Utc::now();
1357                }
1358                Ok(Some(OrderUpdate::Rejected {
1359                    intent_id: intent.intent_id.clone(),
1360                    client_order_id,
1361                    reason_code: RejectionReasonCode::BrokerSubmitFailed.as_str().to_string(),
1362                    reason: e.to_string(),
1363                }))
1364            }
1365        }
1366    }
1367
1368    /// Attempt to place a protective stop for the currently open position.
1369    ///
1370    /// Futures: submits `STOP_MARKET` reduce-only order.
1371    /// Spot: currently returns `Ok(None)` (spot stop order path is not yet wired).
1372    pub async fn place_protective_stop_for_open_position(
1373        &mut self,
1374        source_tag: &str,
1375        stop_price: f64,
1376    ) -> Result<Option<String>> {
1377        if self.position.is_flat() {
1378            return Ok(None);
1379        }
1380        let stop_side = match self.position.side {
1381            Some(OrderSide::Buy) => OrderSide::Sell,
1382            Some(OrderSide::Sell) => OrderSide::Buy,
1383            None => return Ok(None),
1384        };
1385        let qty = self.position.qty.max(0.0);
1386        if qty <= f64::EPSILON {
1387            return Ok(None);
1388        }
1389
1390        if self.market != MarketKind::Futures {
1391            tracing::warn!(
1392                symbol = %self.symbol,
1393                market = %normalize_market_label(self.market),
1394                source_tag = %source_tag,
1395                stop_price,
1396                "Spot protective stop placement is not implemented yet; skipping"
1397            );
1398            return Ok(None);
1399        }
1400
1401        let client_order_id = format!(
1402            "sq-{}-stp-{}",
1403            source_tag.to_ascii_lowercase(),
1404            &uuid::Uuid::new_v4().to_string()[..8]
1405        );
1406        let res = self
1407            .rest_client
1408            .place_futures_stop_market_order(
1409                &self.symbol,
1410                stop_side,
1411                qty,
1412                stop_price,
1413                &client_order_id,
1414            )
1415            .await?;
1416        tracing::info!(
1417            symbol = %self.symbol,
1418            stop_order_id = res.order_id,
1419            stop_side = %stop_side,
1420            stop_price,
1421            qty,
1422            source_tag = %source_tag,
1423            "Protective stop order submitted"
1424        );
1425        Ok(Some(res.order_id.to_string()))
1426    }
1427
1428    /// Ensure a protective stop exists for open position.
1429    ///
1430    /// Returns:
1431    /// - `Ok(true)` when flat (no protection needed) or placement succeeded
1432    /// - `Ok(false)` when protection was needed but could not be placed
1433    pub async fn ensure_protective_stop(
1434        &mut self,
1435        source_tag: &str,
1436        fallback_stop_price: f64,
1437    ) -> Result<bool> {
1438        if self.position.is_flat() {
1439            return Ok(true);
1440        }
1441        Ok(self
1442            .place_protective_stop_for_open_position(source_tag, fallback_stop_price)
1443            .await?
1444            .is_some())
1445    }
1446
1447    /// Emergency close helper for runtime/system-triggered liquidation paths.
1448    ///
1449    /// If position is already flat, returns `Ok(None)` without broker call.
1450    pub async fn emergency_close_position(
1451        &mut self,
1452        source_tag: &str,
1453        reason_code: &str,
1454    ) -> Result<Option<OrderUpdate>> {
1455        if self.position.is_flat() {
1456            return Ok(None);
1457        }
1458        // Emergency close should prefer best-effort execution over stale local cache.
1459        // Refresh balances first and ensure we have a fallback price for risk normalization.
1460        let _ = self.refresh_balances().await;
1461        if self.last_price <= f64::EPSILON && self.position.entry_price > f64::EPSILON {
1462            self.last_price = self.position.entry_price;
1463            self.position.update_unrealized_pnl(self.last_price);
1464        }
1465        tracing::warn!(
1466            symbol = %self.symbol,
1467            market = %normalize_market_label(self.market),
1468            source_tag = %source_tag,
1469            reason_code = %reason_code,
1470            "Emergency close triggered"
1471        );
1472        self.submit_order(Signal::Sell, source_tag).await
1473    }
1474
1475    fn process_order_response(
1476        &mut self,
1477        intent_id: &str,
1478        client_order_id: &str,
1479        side: OrderSide,
1480        response: &BinanceOrderResponse,
1481    ) -> OrderUpdate {
1482        let fills: Vec<Fill> = response
1483            .fills
1484            .iter()
1485            .map(|f| Fill {
1486                price: f.price,
1487                qty: f.qty,
1488                commission: f.commission,
1489                commission_asset: f.commission_asset.clone(),
1490            })
1491            .collect();
1492
1493        let status = OrderStatus::from_binance_str(&response.status);
1494
1495        if let Some(order) = self.active_orders.get_mut(client_order_id) {
1496            order.server_order_id = Some(response.order_id);
1497            order.status = status;
1498            order.fills = fills.clone();
1499            order.updated_at = chrono::Utc::now();
1500        }
1501
1502        if status == OrderStatus::Filled || status == OrderStatus::PartiallyFilled {
1503            self.position.apply_fill(side, &fills);
1504
1505            let avg_price = if fills.is_empty() {
1506                0.0
1507            } else {
1508                let total_value: f64 = fills.iter().map(|f| f.price * f.qty).sum();
1509                let total_qty: f64 = fills.iter().map(|f| f.qty).sum();
1510                total_value / total_qty
1511            };
1512
1513            tracing::info!(
1514                client_order_id,
1515                order_id = response.order_id,
1516                side = %side,
1517                avg_price,
1518                filled_qty = response.executed_qty,
1519                "Order filled"
1520            );
1521
1522            OrderUpdate::Filled {
1523                intent_id: intent_id.to_string(),
1524                client_order_id: client_order_id.to_string(),
1525                side,
1526                fills,
1527                avg_price,
1528            }
1529        } else {
1530            OrderUpdate::Submitted {
1531                intent_id: intent_id.to_string(),
1532                client_order_id: client_order_id.to_string(),
1533                server_order_id: response.order_id,
1534            }
1535        }
1536    }
1537}
1538
1539#[cfg(test)]
1540mod tests {
1541    use super::{
1542        compute_trade_stats_by_source, display_qty_for_history, split_symbol_assets, OrderManager,
1543    };
1544    use crate::binance::types::BinanceMyTrade;
1545    use crate::binance::rest::BinanceRestClient;
1546    use crate::config::{EndpointRateLimitConfig, RiskConfig, SymbolExposureLimitConfig};
1547    use crate::model::order::{Order, OrderSide, OrderStatus, OrderType};
1548    use std::sync::Arc;
1549
1550    fn build_test_order_manager() -> OrderManager {
1551        let rest = Arc::new(BinanceRestClient::new(
1552            "https://demo-api.binance.com",
1553            "https://demo-fapi.binance.com",
1554            "k",
1555            "s",
1556            "fk",
1557            "fs",
1558            5000,
1559        ));
1560        let risk = RiskConfig {
1561            global_rate_limit_per_minute: 600,
1562            default_strategy_cooldown_ms: 3_000,
1563            default_strategy_max_active_orders: 1,
1564            default_symbol_max_exposure_usdt: 200.0,
1565            strategy_limits: vec![],
1566            symbol_exposure_limits: vec![SymbolExposureLimitConfig {
1567                symbol: "BTCUSDT".to_string(),
1568                market: Some("spot".to_string()),
1569                max_exposure_usdt: 150.0,
1570            }],
1571            endpoint_rate_limits: EndpointRateLimitConfig {
1572                orders_per_minute: 240,
1573                account_per_minute: 180,
1574                market_data_per_minute: 360,
1575            },
1576        };
1577        OrderManager::new(
1578            rest,
1579            "BTCUSDT",
1580            crate::order_manager::MarketKind::Spot,
1581            10.0,
1582            &risk,
1583        )
1584    }
1585
1586    #[test]
1587    fn valid_state_transitions() {
1588        // PendingSubmit -> Submitted
1589        let from = OrderStatus::PendingSubmit;
1590        let to = OrderStatus::Submitted;
1591        assert!(!from.is_terminal());
1592        assert!(!to.is_terminal());
1593
1594        // Submitted -> Filled
1595        let to = OrderStatus::Filled;
1596        assert!(to.is_terminal());
1597
1598        // Submitted -> Rejected
1599        let to = OrderStatus::Rejected;
1600        assert!(to.is_terminal());
1601
1602        // Submitted -> Cancelled
1603        let to = OrderStatus::Cancelled;
1604        assert!(to.is_terminal());
1605    }
1606
1607    #[test]
1608    fn from_binance_str_mapping() {
1609        assert_eq!(OrderStatus::from_binance_str("NEW"), OrderStatus::Submitted);
1610        assert_eq!(OrderStatus::from_binance_str("FILLED"), OrderStatus::Filled);
1611        assert_eq!(
1612            OrderStatus::from_binance_str("CANCELED"),
1613            OrderStatus::Cancelled
1614        );
1615        assert_eq!(
1616            OrderStatus::from_binance_str("REJECTED"),
1617            OrderStatus::Rejected
1618        );
1619        assert_eq!(
1620            OrderStatus::from_binance_str("EXPIRED"),
1621            OrderStatus::Expired
1622        );
1623        assert_eq!(
1624            OrderStatus::from_binance_str("PARTIALLY_FILLED"),
1625            OrderStatus::PartiallyFilled
1626        );
1627    }
1628
1629    #[test]
1630    fn order_history_uses_executed_qty_for_filled_states() {
1631        assert!((display_qty_for_history("FILLED", 1.0, 0.4) - 0.4).abs() < f64::EPSILON);
1632        assert!((display_qty_for_history("PARTIALLY_FILLED", 1.0, 0.4) - 0.4).abs() < f64::EPSILON);
1633    }
1634
1635    #[test]
1636    fn order_history_uses_orig_qty_for_non_filled_states() {
1637        assert!((display_qty_for_history("NEW", 1.0, 0.4) - 1.0).abs() < f64::EPSILON);
1638        assert!((display_qty_for_history("CANCELED", 1.0, 0.4) - 1.0).abs() < f64::EPSILON);
1639        assert!((display_qty_for_history("REJECTED", 1.0, 0.0) - 1.0).abs() < f64::EPSILON);
1640    }
1641
1642    #[test]
1643    fn split_symbol_assets_parses_known_quote_suffixes() {
1644        assert_eq!(
1645            split_symbol_assets("ETHUSDT"),
1646            ("ETH".to_string(), "USDT".to_string())
1647        );
1648        assert_eq!(
1649            split_symbol_assets("ETHBTC"),
1650            ("ETH".to_string(), "BTC".to_string())
1651        );
1652    }
1653
1654    #[test]
1655    fn split_symbol_assets_falls_back_when_quote_unknown() {
1656        assert_eq!(
1657            split_symbol_assets("FOOBAR"),
1658            ("FOOBAR".to_string(), String::new())
1659        );
1660    }
1661
1662    #[test]
1663    fn strategy_limit_rejects_when_active_orders_reach_limit() {
1664        let mut mgr = build_test_order_manager();
1665        let client_order_id = "sq-cfg-abcdef12".to_string();
1666        mgr.active_orders.insert(
1667            client_order_id.clone(),
1668            Order {
1669                client_order_id,
1670                server_order_id: None,
1671                symbol: "BTCUSDT".to_string(),
1672                side: OrderSide::Buy,
1673                order_type: OrderType::Market,
1674                quantity: 0.1,
1675                price: None,
1676                status: OrderStatus::Submitted,
1677                created_at: chrono::Utc::now(),
1678                updated_at: chrono::Utc::now(),
1679                fills: vec![],
1680            },
1681        );
1682
1683        let rejected = mgr
1684            .evaluate_strategy_limits("cfg", chrono::Utc::now().timestamp_millis() as u64)
1685            .expect("must be rejected");
1686        assert_eq!(
1687            rejected.0,
1688            "risk.strategy_max_active_orders_exceeded".to_string()
1689        );
1690    }
1691
1692    #[test]
1693    fn strategy_limit_rejects_during_cooldown_window() {
1694        let mut mgr = build_test_order_manager();
1695        let now = chrono::Utc::now().timestamp_millis() as u64;
1696        mgr.mark_strategy_submit("cfg", now);
1697
1698        let rejected = mgr
1699            .evaluate_strategy_limits("cfg", now + 500)
1700            .expect("must be rejected");
1701        assert_eq!(rejected.0, "risk.strategy_cooldown_active".to_string());
1702    }
1703
1704    #[test]
1705    fn symbol_exposure_limit_rejects_when_projected_notional_exceeds_limit() {
1706        let mut mgr = build_test_order_manager();
1707        mgr.last_price = 100.0;
1708        // Buy 2.0 -> projected notional 200, but configured spot BTCUSDT limit is 150.
1709        let rejected = mgr
1710            .evaluate_symbol_exposure_limit(OrderSide::Buy, 2.0)
1711            .expect("must be rejected");
1712        assert_eq!(
1713            rejected.0,
1714            "risk.symbol_exposure_limit_exceeded".to_string()
1715        );
1716    }
1717
1718    #[test]
1719    fn symbol_exposure_limit_allows_risk_reducing_order() {
1720        let mut mgr = build_test_order_manager();
1721        mgr.last_price = 100.0;
1722        mgr.position.side = Some(OrderSide::Buy);
1723        mgr.position.qty = 2.0; // current notional 200 > limit 150
1724
1725        // Sell reduces exposure to 100; should be allowed.
1726        let rejected = mgr.evaluate_symbol_exposure_limit(OrderSide::Sell, 1.0);
1727        assert!(rejected.is_none());
1728    }
1729
1730    #[test]
1731    fn futures_trade_stats_by_source_use_realized_pnl() {
1732        let trades = vec![
1733            BinanceMyTrade {
1734                symbol: "XRPUSDT".to_string(),
1735                id: 1,
1736                order_id: 1001,
1737                price: 1.0,
1738                qty: 100.0,
1739                commission: 0.0,
1740                commission_asset: "USDT".to_string(),
1741                time: 1,
1742                is_buyer: false,
1743                is_maker: false,
1744                realized_pnl: 5.0,
1745            },
1746            BinanceMyTrade {
1747                symbol: "XRPUSDT".to_string(),
1748                id: 2,
1749                order_id: 1002,
1750                price: 1.0,
1751                qty: 100.0,
1752                commission: 0.0,
1753                commission_asset: "USDT".to_string(),
1754                time: 2,
1755                is_buyer: false,
1756                is_maker: false,
1757                realized_pnl: -2.5,
1758            },
1759        ];
1760        let mut source_by_order = std::collections::HashMap::new();
1761        source_by_order.insert(1001, "c20".to_string());
1762        source_by_order.insert(1002, "c20".to_string());
1763
1764        let stats = compute_trade_stats_by_source(trades, &source_by_order, "XRPUSDT#FUT");
1765        let c20 = stats.get("c20").expect("source tag must exist");
1766        assert_eq!(c20.trade_count, 2);
1767        assert_eq!(c20.win_count, 1);
1768        assert_eq!(c20.lose_count, 1);
1769        assert!((c20.realized_pnl - 2.5).abs() < f64::EPSILON);
1770    }
1771}