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
784                            .entry(row.trade.order_id)
785                            .or_insert(row.source);
786                    }
787                }
788                Ok(_) => {}
789                Err(e) => {
790                    tracing::warn!(
791                        error = %e,
792                        "Failed to load persisted futures trades; using API trades"
793                    );
794                }
795            }
796
797            let mut history = Vec::new();
798            let mut fills = Vec::new();
799            for t in &trades {
800                let side = if t.is_buyer { "BUY" } else { "SELL" };
801                let source = order_source_by_id
802                    .get(&t.order_id)
803                    .cloned()
804                    .unwrap_or_else(|| "UNKNOWN".to_string());
805                fills.push(OrderHistoryFill {
806                    timestamp_ms: t.time,
807                    side: if t.is_buyer {
808                        OrderSide::Buy
809                    } else {
810                        OrderSide::Sell
811                    },
812                    price: t.price,
813                });
814                history.push(format_order_history_row(
815                    t.time,
816                    "FILLED",
817                    side,
818                    t.qty,
819                    t.price,
820                    &format!("order#{}#T{} [{}]", t.order_id, t.id, source),
821                ));
822            }
823            for o in &orders {
824                if o.executed_qty <= 0.0 {
825                    history.push(format_order_history_row(
826                        o.update_time.max(o.time),
827                        &o.status,
828                        &o.side,
829                        display_qty_for_history(&o.status, o.orig_qty, o.executed_qty),
830                        if o.executed_qty > 0.0 {
831                            o.cummulative_quote_qty / o.executed_qty
832                        } else {
833                            o.price
834                        },
835                        &o.client_order_id,
836                    ));
837                }
838            }
839
840            let mut stats = OrderHistoryStats::default();
841            for t in &trades {
842                if t.realized_pnl > 0.0 {
843                    stats.win_count += 1;
844                    stats.trade_count += 1;
845                } else if t.realized_pnl < 0.0 {
846                    stats.lose_count += 1;
847                    stats.trade_count += 1;
848                }
849                stats.realized_pnl += t.realized_pnl;
850            }
851            let open_pos = compute_futures_open_state(trades_for_stats.clone());
852            let open_entry_price = if open_pos.qty > f64::EPSILON {
853                open_pos.cost_quote / open_pos.qty
854            } else {
855                0.0
856            };
857            self.position.side = if open_pos.qty > f64::EPSILON {
858                Some(OrderSide::Buy)
859            } else {
860                None
861            };
862            self.position.qty = open_pos.qty;
863            self.position.entry_price = open_entry_price;
864            self.position.realized_pnl = stats.realized_pnl;
865            if self.last_price > 0.0 {
866                self.position.update_unrealized_pnl(self.last_price);
867            } else {
868                self.position.unrealized_pnl = 0.0;
869            }
870            let estimated_total_pnl_usdt = if self.last_price > 0.0 && open_pos.qty > f64::EPSILON {
871                Some(stats.realized_pnl + (self.last_price - open_entry_price) * open_pos.qty)
872            } else {
873                Some(stats.realized_pnl)
874            };
875            let latest_order_event = orders.iter().map(|o| o.update_time.max(o.time)).max();
876            let latest_trade_event = trades.iter().map(|t| t.time).max();
877            let mut strategy_stats =
878                compute_trade_stats_by_source(trades_for_stats, &order_source_by_id, &storage_key);
879            let persisted_stats = to_persistable_stats_map(&strategy_stats);
880            if let Err(e) =
881                order_store::persist_strategy_symbol_stats(&storage_key, &persisted_stats)
882            {
883                tracing::warn!(error = %e, "Failed to persist strategy stats (futures)");
884            }
885            if strategy_stats.is_empty() {
886                match order_store::load_strategy_symbol_stats(&storage_key) {
887                    Ok(persisted) => {
888                        strategy_stats = from_persisted_stats_map(persisted);
889                    }
890                    Err(e) => {
891                        tracing::warn!(
892                            error = %e,
893                            "Failed to load persisted strategy stats (futures)"
894                        );
895                    }
896                }
897            }
898            return Ok(OrderHistorySnapshot {
899                rows: history,
900                stats,
901                strategy_stats,
902                fills,
903                open_qty: open_pos.qty,
904                open_entry_price,
905                estimated_total_pnl_usdt,
906                trade_data_complete: true,
907                fetched_at_ms,
908                fetch_latency_ms,
909                latest_event_ms: latest_order_event.max(latest_trade_event),
910            });
911        }
912
913        let fetch_started = Instant::now();
914        let fetched_at_ms = chrono::Utc::now().timestamp_millis() as u64;
915        let orders_result = self.rest_client.get_all_orders(&self.symbol, limit).await;
916        let storage_key = storage_symbol(&self.symbol, self.market);
917        let last_trade_id = order_store::load_last_trade_id(&storage_key).ok().flatten();
918        let persisted_trade_count = order_store::load_trade_count(&storage_key).unwrap_or(0);
919        let need_backfill = persisted_trade_count < limit;
920        let trades_result = match (need_backfill, last_trade_id) {
921            (true, _) => {
922                self.rest_client
923                    .get_my_trades_history(&self.symbol, limit.max(1))
924                    .await
925            }
926            (false, Some(last_id)) => {
927                self.rest_client
928                    .get_my_trades_since(&self.symbol, last_id.saturating_add(1), 10)
929                    .await
930            }
931            (false, None) => {
932                self.rest_client
933                    .get_my_trades_history(&self.symbol, limit.max(1))
934                    .await
935            }
936        };
937        let fetch_latency_ms = fetch_started.elapsed().as_millis() as u64;
938        let trade_data_complete = trades_result.is_ok();
939
940        if orders_result.is_err() && trades_result.is_err() {
941            let oe = orders_result.err().unwrap();
942            let te = trades_result.err().unwrap();
943            return Err(anyhow::anyhow!(
944                "order history fetch failed: allOrders={} | myTrades={}",
945                oe,
946                te
947            ));
948        }
949
950        let mut orders = match orders_result {
951            Ok(v) => v,
952            Err(e) => {
953                tracing::warn!(error = %e, "Failed to fetch allOrders; falling back to trade-only history");
954                Vec::new()
955            }
956        };
957        let recent_trades = match trades_result {
958            Ok(t) => t,
959            Err(e) => {
960                tracing::warn!(error = %e, "Failed to fetch myTrades; falling back to order-only history");
961                Vec::new()
962            }
963        };
964        let mut trades = recent_trades.clone();
965        orders.sort_by_key(|o| o.update_time.max(o.time));
966
967        if let Err(e) = order_store::persist_order_snapshot(&storage_key, &orders, &recent_trades) {
968            tracing::warn!(error = %e, "Failed to persist order snapshot to sqlite");
969        }
970        let mut persisted_source_by_order_id: HashMap<u64, String> = HashMap::new();
971        match order_store::load_persisted_trades(&storage_key) {
972            Ok(saved) => {
973                if !saved.is_empty() {
974                    trades = saved.iter().map(|r| r.trade.clone()).collect();
975                    for row in saved {
976                        persisted_source_by_order_id
977                            .entry(row.trade.order_id)
978                            .or_insert(row.source);
979                    }
980                }
981            }
982            Err(e) => {
983                tracing::warn!(error = %e, "Failed to load persisted trades; using recent API trades");
984            }
985        }
986
987        let (stats, open_pos) = compute_trade_state(trades.clone(), &self.symbol);
988        self.position.side = if open_pos.qty > f64::EPSILON {
989            Some(OrderSide::Buy)
990        } else {
991            None
992        };
993        self.position.qty = open_pos.qty;
994        self.position.entry_price = if open_pos.qty > f64::EPSILON {
995            open_pos.cost_quote / open_pos.qty
996        } else {
997            0.0
998        };
999        self.position.realized_pnl = stats.realized_pnl;
1000        if self.last_price > 0.0 {
1001            self.position.update_unrealized_pnl(self.last_price);
1002        } else {
1003            self.position.unrealized_pnl = 0.0;
1004        }
1005        let estimated_total_pnl_usdt = if self.last_price > 0.0 {
1006            Some(stats.realized_pnl + (open_pos.qty * self.last_price - open_pos.cost_quote))
1007        } else {
1008            Some(stats.realized_pnl)
1009        };
1010        let latest_order_event = orders.iter().map(|o| o.update_time.max(o.time)).max();
1011        let latest_trade_event = trades.iter().map(|t| t.time).max();
1012        let latest_event_ms = latest_order_event.max(latest_trade_event);
1013
1014        let mut trades_by_order_id: HashMap<u64, Vec<BinanceMyTrade>> = HashMap::new();
1015        for trade in &trades {
1016            trades_by_order_id
1017                .entry(trade.order_id)
1018                .or_default()
1019                .push(trade.clone());
1020        }
1021        for bucket in trades_by_order_id.values_mut() {
1022            bucket.sort_by_key(|t| t.time);
1023        }
1024
1025        let mut order_source_by_id = HashMap::new();
1026        for o in &orders {
1027            order_source_by_id.insert(
1028                o.order_id,
1029                source_label_from_client_order_id(&o.client_order_id),
1030            );
1031        }
1032        for (order_id, source) in persisted_source_by_order_id {
1033            order_source_by_id.entry(order_id).or_insert(source);
1034        }
1035        let mut strategy_stats =
1036            compute_trade_stats_by_source(trades.clone(), &order_source_by_id, &self.symbol);
1037        let persisted_stats = to_persistable_stats_map(&strategy_stats);
1038        if let Err(e) = order_store::persist_strategy_symbol_stats(&storage_key, &persisted_stats) {
1039            tracing::warn!(error = %e, "Failed to persist strategy+symbol scoped stats");
1040        }
1041        if strategy_stats.is_empty() {
1042            match order_store::load_strategy_symbol_stats(&storage_key) {
1043                Ok(persisted) => {
1044                    strategy_stats = from_persisted_stats_map(persisted);
1045                }
1046                Err(e) => {
1047                    tracing::warn!(error = %e, "Failed to load persisted strategy+symbol stats");
1048                }
1049            }
1050        }
1051
1052        let mut history = Vec::new();
1053        let mut fills = Vec::new();
1054        let mut used_trade_ids = std::collections::HashSet::new();
1055
1056        if orders.is_empty() && !trades.is_empty() {
1057            let mut sorted = trades;
1058            sorted.sort_by_key(|t| (t.time, t.id));
1059            history.extend(sorted.iter().map(|t| {
1060                fills.push(OrderHistoryFill {
1061                    timestamp_ms: t.time,
1062                    side: if t.is_buyer {
1063                        OrderSide::Buy
1064                    } else {
1065                        OrderSide::Sell
1066                    },
1067                    price: t.price,
1068                });
1069                format_trade_history_row(
1070                    t,
1071                    order_source_by_id
1072                        .get(&t.order_id)
1073                        .map(String::as_str)
1074                        .unwrap_or("UNKNOWN"),
1075                )
1076            }));
1077            return Ok(OrderHistorySnapshot {
1078                rows: history,
1079                stats,
1080                strategy_stats,
1081                fills,
1082                open_qty: open_pos.qty,
1083                open_entry_price: if open_pos.qty > f64::EPSILON {
1084                    open_pos.cost_quote / open_pos.qty
1085                } else {
1086                    0.0
1087                },
1088                estimated_total_pnl_usdt,
1089                trade_data_complete,
1090                fetched_at_ms,
1091                fetch_latency_ms,
1092                latest_event_ms,
1093            });
1094        }
1095
1096        for o in orders {
1097            if o.executed_qty > 0.0 {
1098                if let Some(order_trades) = trades_by_order_id.get(&o.order_id) {
1099                    for t in order_trades {
1100                        used_trade_ids.insert(t.id);
1101                        let side = if t.is_buyer { "BUY" } else { "SELL" };
1102                        fills.push(OrderHistoryFill {
1103                            timestamp_ms: t.time,
1104                            side: if t.is_buyer {
1105                                OrderSide::Buy
1106                            } else {
1107                                OrderSide::Sell
1108                            },
1109                            price: t.price,
1110                        });
1111                        history.push(format_order_history_row(
1112                            t.time,
1113                            "FILLED",
1114                            side,
1115                            t.qty,
1116                            t.price,
1117                            &format!(
1118                                "{}#T{} [{}]",
1119                                o.client_order_id,
1120                                t.id,
1121                                source_label_from_client_order_id(&o.client_order_id)
1122                            ),
1123                        ));
1124                    }
1125                    continue;
1126                }
1127            }
1128
1129            let avg_price = if o.executed_qty > 0.0 {
1130                o.cummulative_quote_qty / o.executed_qty
1131            } else {
1132                o.price
1133            };
1134            history.push(format_order_history_row(
1135                o.update_time.max(o.time),
1136                &o.status,
1137                &o.side,
1138                display_qty_for_history(&o.status, o.orig_qty, o.executed_qty),
1139                avg_price,
1140                &o.client_order_id,
1141            ));
1142        }
1143
1144        // Include trades that did not match fetched order pages.
1145        for bucket in trades_by_order_id.values() {
1146            for t in bucket {
1147                if !used_trade_ids.contains(&t.id) {
1148                    fills.push(OrderHistoryFill {
1149                        timestamp_ms: t.time,
1150                        side: if t.is_buyer {
1151                            OrderSide::Buy
1152                        } else {
1153                            OrderSide::Sell
1154                        },
1155                        price: t.price,
1156                    });
1157                    history.push(format_trade_history_row(
1158                        t,
1159                        order_source_by_id
1160                            .get(&t.order_id)
1161                            .map(String::as_str)
1162                            .unwrap_or("UNKNOWN"),
1163                    ));
1164                }
1165            }
1166        }
1167        Ok(OrderHistorySnapshot {
1168            rows: history,
1169            stats,
1170            strategy_stats,
1171            fills,
1172            open_qty: open_pos.qty,
1173            open_entry_price: if open_pos.qty > f64::EPSILON {
1174                open_pos.cost_quote / open_pos.qty
1175            } else {
1176                0.0
1177            },
1178            estimated_total_pnl_usdt,
1179            trade_data_complete,
1180            fetched_at_ms,
1181            fetch_latency_ms,
1182            latest_event_ms,
1183        })
1184    }
1185
1186    /// Build an order intent, run risk checks, and submit to broker when approved.
1187    ///
1188    /// # Behavior
1189    /// - `Signal::Hold` returns `Ok(None)`.
1190    /// - For buy/sell signals, this method:
1191    ///   1. Builds `OrderIntent`.
1192    ///   2. Calls `RiskModule::evaluate_intent`.
1193    ///   3. Reserves one global rate token via `reserve_rate_budget`.
1194    ///   4. Submits market order to spot/futures broker endpoint.
1195    /// - Rejections are returned as `Ok(Some(OrderUpdate::Rejected { .. }))`
1196    ///   with structured `reason_code`.
1197    ///
1198    /// # Usage
1199    /// Recommended sequence:
1200    /// 1. `update_unrealized_pnl(last_price)`
1201    /// 2. `refresh_balances()` (periodic or before trading loop)
1202    /// 3. `submit_order(signal, source_tag)`
1203    ///
1204    /// # Caution
1205    /// - Spot sell requires base-asset balance (e.g. `ETH` for `ETHUSDT`).
1206    /// - If balances are stale, you may see "No position to sell" or
1207    ///   `"Insufficient <asset>"` even though exchange state changed recently.
1208    /// - This method returns transport/runtime errors as `Err(_)`; business
1209    ///   rejections are encoded in `OrderUpdate::Rejected`.
1210    pub async fn submit_order(
1211        &mut self,
1212        signal: Signal,
1213        source_tag: &str,
1214    ) -> Result<Option<OrderUpdate>> {
1215        let side = match &signal {
1216            Signal::Buy => OrderSide::Buy,
1217            Signal::Sell => OrderSide::Sell,
1218            Signal::Hold => return Ok(None),
1219        };
1220        let source_tag = source_tag.to_ascii_lowercase();
1221        let intent = OrderIntent {
1222            intent_id: format!("intent-{}", &uuid::Uuid::new_v4().to_string()[..8]),
1223            source_tag: source_tag.clone(),
1224            symbol: self.symbol.clone(),
1225            market: self.market,
1226            side,
1227            order_amount_usdt: self.order_amount_usdt,
1228            last_price: self.last_price,
1229            created_at_ms: chrono::Utc::now().timestamp_millis() as u64,
1230        };
1231        if let Some((reason_code, reason)) =
1232            self.evaluate_strategy_limits(&intent.source_tag, intent.created_at_ms)
1233        {
1234            return Ok(Some(OrderUpdate::Rejected {
1235                intent_id: intent.intent_id.clone(),
1236                client_order_id: "n/a".to_string(),
1237                reason_code,
1238                reason,
1239            }));
1240        }
1241        let decision = self
1242            .risk_module
1243            .evaluate_intent(&intent, &self.balances)
1244            .await?;
1245        if !decision.approved {
1246            return Ok(Some(OrderUpdate::Rejected {
1247                intent_id: intent.intent_id.clone(),
1248                client_order_id: "n/a".to_string(),
1249                reason_code: decision
1250                    .reason_code
1251                    .unwrap_or_else(|| RejectionReasonCode::RiskUnknown.as_str().to_string()),
1252                reason: decision
1253                    .reason
1254                    .unwrap_or_else(|| "Rejected by RiskModule".to_string()),
1255            }));
1256        }
1257        if !self.risk_module.reserve_rate_budget() {
1258            return Ok(Some(OrderUpdate::Rejected {
1259                intent_id: intent.intent_id.clone(),
1260                client_order_id: "n/a".to_string(),
1261                reason_code: RejectionReasonCode::RateGlobalBudgetExceeded
1262                    .as_str()
1263                    .to_string(),
1264                reason: "Global rate budget exceeded; try again after reset".to_string(),
1265            }));
1266        }
1267        if !self
1268            .risk_module
1269            .reserve_endpoint_budget(ApiEndpointGroup::Orders)
1270        {
1271            return Ok(Some(OrderUpdate::Rejected {
1272                intent_id: intent.intent_id.clone(),
1273                client_order_id: "n/a".to_string(),
1274                reason_code: RejectionReasonCode::RateEndpointBudgetExceeded
1275                    .as_str()
1276                    .to_string(),
1277                reason: "Orders endpoint budget exceeded; try again after reset".to_string(),
1278            }));
1279        }
1280        let qty = decision.normalized_qty;
1281        if let Some((reason_code, reason)) = self.evaluate_symbol_exposure_limit(side, qty) {
1282            return Ok(Some(OrderUpdate::Rejected {
1283                intent_id: intent.intent_id.clone(),
1284                client_order_id: "n/a".to_string(),
1285                reason_code,
1286                reason,
1287            }));
1288        }
1289        self.mark_strategy_submit(&intent.source_tag, intent.created_at_ms);
1290
1291        let client_order_id = format!(
1292            "sq-{}-{}",
1293            intent.source_tag,
1294            &uuid::Uuid::new_v4().to_string()[..8]
1295        );
1296
1297        let order = Order {
1298            client_order_id: client_order_id.clone(),
1299            server_order_id: None,
1300            symbol: self.symbol.clone(),
1301            side,
1302            order_type: OrderType::Market,
1303            quantity: qty,
1304            price: None,
1305            status: OrderStatus::PendingSubmit,
1306            created_at: chrono::Utc::now(),
1307            updated_at: chrono::Utc::now(),
1308            fills: vec![],
1309        };
1310
1311        self.active_orders.insert(client_order_id.clone(), order);
1312
1313        tracing::info!(
1314            side = %side,
1315            qty,
1316            usdt_amount = intent.order_amount_usdt,
1317            price = intent.last_price,
1318            intent_id = %intent.intent_id,
1319            created_at_ms = intent.created_at_ms,
1320            "Submitting order"
1321        );
1322
1323        let submit_res = if self.market == MarketKind::Futures {
1324            self.rest_client
1325                .place_futures_market_order(&self.symbol, side, qty, &client_order_id)
1326                .await
1327        } else {
1328            self.rest_client
1329                .place_market_order(&self.symbol, side, qty, &client_order_id)
1330                .await
1331        };
1332
1333        match submit_res {
1334            Ok(response) => {
1335                let update = self.process_order_response(
1336                    &intent.intent_id,
1337                    &client_order_id,
1338                    side,
1339                    &response,
1340                );
1341
1342                // Refresh balances after fill
1343                if matches!(update, OrderUpdate::Filled { .. }) {
1344                    if let Err(e) = self.refresh_balances().await {
1345                        tracing::warn!(error = %e, "Failed to refresh balances after fill");
1346                    }
1347                }
1348
1349                Ok(Some(update))
1350            }
1351            Err(e) => {
1352                tracing::error!(
1353                    client_order_id,
1354                    error = %e,
1355                    "Order rejected"
1356                );
1357                if let Some(order) = self.active_orders.get_mut(&client_order_id) {
1358                    order.status = OrderStatus::Rejected;
1359                    order.updated_at = chrono::Utc::now();
1360                }
1361                Ok(Some(OrderUpdate::Rejected {
1362                    intent_id: intent.intent_id.clone(),
1363                    client_order_id,
1364                    reason_code: RejectionReasonCode::BrokerSubmitFailed.as_str().to_string(),
1365                    reason: e.to_string(),
1366                }))
1367            }
1368        }
1369    }
1370
1371    /// Attempt to place a protective stop for the currently open position.
1372    ///
1373    /// Futures: submits `STOP_MARKET` reduce-only order.
1374    /// Spot: currently returns `Ok(None)` (spot stop order path is not yet wired).
1375    pub async fn place_protective_stop_for_open_position(
1376        &mut self,
1377        source_tag: &str,
1378        stop_price: f64,
1379    ) -> Result<Option<String>> {
1380        if self.position.is_flat() {
1381            return Ok(None);
1382        }
1383        let stop_side = match self.position.side {
1384            Some(OrderSide::Buy) => OrderSide::Sell,
1385            Some(OrderSide::Sell) => OrderSide::Buy,
1386            None => return Ok(None),
1387        };
1388        let qty = self.position.qty.max(0.0);
1389        if qty <= f64::EPSILON {
1390            return Ok(None);
1391        }
1392
1393        if self.market != MarketKind::Futures {
1394            tracing::warn!(
1395                symbol = %self.symbol,
1396                market = %normalize_market_label(self.market),
1397                source_tag = %source_tag,
1398                stop_price,
1399                "Spot protective stop placement is not implemented yet; skipping"
1400            );
1401            return Ok(None);
1402        }
1403
1404        let client_order_id = format!(
1405            "sq-{}-stp-{}",
1406            source_tag.to_ascii_lowercase(),
1407            &uuid::Uuid::new_v4().to_string()[..8]
1408        );
1409        let res = self
1410            .rest_client
1411            .place_futures_stop_market_order(
1412                &self.symbol,
1413                stop_side,
1414                qty,
1415                stop_price,
1416                &client_order_id,
1417            )
1418            .await?;
1419        tracing::info!(
1420            symbol = %self.symbol,
1421            stop_order_id = res.order_id,
1422            stop_side = %stop_side,
1423            stop_price,
1424            qty,
1425            source_tag = %source_tag,
1426            "Protective stop order submitted"
1427        );
1428        Ok(Some(res.order_id.to_string()))
1429    }
1430
1431    /// Ensure a protective stop exists for open position.
1432    ///
1433    /// Returns:
1434    /// - `Ok(true)` when flat (no protection needed) or placement succeeded
1435    /// - `Ok(false)` when protection was needed but could not be placed
1436    pub async fn ensure_protective_stop(
1437        &mut self,
1438        source_tag: &str,
1439        fallback_stop_price: f64,
1440    ) -> Result<bool> {
1441        if self.position.is_flat() {
1442            return Ok(true);
1443        }
1444        Ok(self
1445            .place_protective_stop_for_open_position(source_tag, fallback_stop_price)
1446            .await?
1447            .is_some())
1448    }
1449
1450    /// Emergency close helper for runtime/system-triggered liquidation paths.
1451    ///
1452    /// If position is already flat, returns `Ok(None)` without broker call.
1453    pub async fn emergency_close_position(
1454        &mut self,
1455        source_tag: &str,
1456        reason_code: &str,
1457    ) -> Result<Option<OrderUpdate>> {
1458        if self.position.is_flat() {
1459            return Ok(None);
1460        }
1461        // Emergency close should prefer best-effort execution over stale local cache.
1462        // Refresh balances first and ensure we have a fallback price for risk normalization.
1463        let _ = self.refresh_balances().await;
1464        if self.last_price <= f64::EPSILON && self.position.entry_price > f64::EPSILON {
1465            self.last_price = self.position.entry_price;
1466            self.position.update_unrealized_pnl(self.last_price);
1467        }
1468        tracing::warn!(
1469            symbol = %self.symbol,
1470            market = %normalize_market_label(self.market),
1471            source_tag = %source_tag,
1472            reason_code = %reason_code,
1473            "Emergency close triggered"
1474        );
1475        self.submit_order(Signal::Sell, source_tag).await
1476    }
1477
1478    fn process_order_response(
1479        &mut self,
1480        intent_id: &str,
1481        client_order_id: &str,
1482        side: OrderSide,
1483        response: &BinanceOrderResponse,
1484    ) -> OrderUpdate {
1485        let fills: Vec<Fill> = response
1486            .fills
1487            .iter()
1488            .map(|f| Fill {
1489                price: f.price,
1490                qty: f.qty,
1491                commission: f.commission,
1492                commission_asset: f.commission_asset.clone(),
1493            })
1494            .collect();
1495
1496        let status = OrderStatus::from_binance_str(&response.status);
1497
1498        if let Some(order) = self.active_orders.get_mut(client_order_id) {
1499            order.server_order_id = Some(response.order_id);
1500            order.status = status;
1501            order.fills = fills.clone();
1502            order.updated_at = chrono::Utc::now();
1503        }
1504
1505        if status == OrderStatus::Filled || status == OrderStatus::PartiallyFilled {
1506            self.position.apply_fill(side, &fills);
1507
1508            let avg_price = if fills.is_empty() {
1509                0.0
1510            } else {
1511                let total_value: f64 = fills.iter().map(|f| f.price * f.qty).sum();
1512                let total_qty: f64 = fills.iter().map(|f| f.qty).sum();
1513                total_value / total_qty
1514            };
1515
1516            tracing::info!(
1517                client_order_id,
1518                order_id = response.order_id,
1519                side = %side,
1520                avg_price,
1521                filled_qty = response.executed_qty,
1522                "Order filled"
1523            );
1524
1525            OrderUpdate::Filled {
1526                intent_id: intent_id.to_string(),
1527                client_order_id: client_order_id.to_string(),
1528                side,
1529                fills,
1530                avg_price,
1531            }
1532        } else {
1533            OrderUpdate::Submitted {
1534                intent_id: intent_id.to_string(),
1535                client_order_id: client_order_id.to_string(),
1536                server_order_id: response.order_id,
1537            }
1538        }
1539    }
1540}
1541
1542#[cfg(test)]
1543mod tests {
1544    use super::{
1545        compute_trade_stats_by_source, display_qty_for_history, split_symbol_assets, OrderManager,
1546    };
1547    use crate::binance::rest::BinanceRestClient;
1548    use crate::binance::types::BinanceMyTrade;
1549    use crate::config::{EndpointRateLimitConfig, RiskConfig, SymbolExposureLimitConfig};
1550    use crate::model::order::{Order, OrderSide, OrderStatus, OrderType};
1551    use std::sync::Arc;
1552
1553    fn build_test_order_manager() -> OrderManager {
1554        let rest = Arc::new(BinanceRestClient::new(
1555            "https://demo-api.binance.com",
1556            "https://demo-fapi.binance.com",
1557            "k",
1558            "s",
1559            "fk",
1560            "fs",
1561            5000,
1562        ));
1563        let risk = RiskConfig {
1564            global_rate_limit_per_minute: 600,
1565            default_strategy_cooldown_ms: 3_000,
1566            default_strategy_max_active_orders: 1,
1567            default_symbol_max_exposure_usdt: 200.0,
1568            strategy_limits: vec![],
1569            symbol_exposure_limits: vec![SymbolExposureLimitConfig {
1570                symbol: "BTCUSDT".to_string(),
1571                market: Some("spot".to_string()),
1572                max_exposure_usdt: 150.0,
1573            }],
1574            endpoint_rate_limits: EndpointRateLimitConfig {
1575                orders_per_minute: 240,
1576                account_per_minute: 180,
1577                market_data_per_minute: 360,
1578            },
1579        };
1580        OrderManager::new(
1581            rest,
1582            "BTCUSDT",
1583            crate::order_manager::MarketKind::Spot,
1584            10.0,
1585            &risk,
1586        )
1587    }
1588
1589    #[test]
1590    fn valid_state_transitions() {
1591        // PendingSubmit -> Submitted
1592        let from = OrderStatus::PendingSubmit;
1593        let to = OrderStatus::Submitted;
1594        assert!(!from.is_terminal());
1595        assert!(!to.is_terminal());
1596
1597        // Submitted -> Filled
1598        let to = OrderStatus::Filled;
1599        assert!(to.is_terminal());
1600
1601        // Submitted -> Rejected
1602        let to = OrderStatus::Rejected;
1603        assert!(to.is_terminal());
1604
1605        // Submitted -> Cancelled
1606        let to = OrderStatus::Cancelled;
1607        assert!(to.is_terminal());
1608    }
1609
1610    #[test]
1611    fn from_binance_str_mapping() {
1612        assert_eq!(OrderStatus::from_binance_str("NEW"), OrderStatus::Submitted);
1613        assert_eq!(OrderStatus::from_binance_str("FILLED"), OrderStatus::Filled);
1614        assert_eq!(
1615            OrderStatus::from_binance_str("CANCELED"),
1616            OrderStatus::Cancelled
1617        );
1618        assert_eq!(
1619            OrderStatus::from_binance_str("REJECTED"),
1620            OrderStatus::Rejected
1621        );
1622        assert_eq!(
1623            OrderStatus::from_binance_str("EXPIRED"),
1624            OrderStatus::Expired
1625        );
1626        assert_eq!(
1627            OrderStatus::from_binance_str("PARTIALLY_FILLED"),
1628            OrderStatus::PartiallyFilled
1629        );
1630    }
1631
1632    #[test]
1633    fn order_history_uses_executed_qty_for_filled_states() {
1634        assert!((display_qty_for_history("FILLED", 1.0, 0.4) - 0.4).abs() < f64::EPSILON);
1635        assert!((display_qty_for_history("PARTIALLY_FILLED", 1.0, 0.4) - 0.4).abs() < f64::EPSILON);
1636    }
1637
1638    #[test]
1639    fn order_history_uses_orig_qty_for_non_filled_states() {
1640        assert!((display_qty_for_history("NEW", 1.0, 0.4) - 1.0).abs() < f64::EPSILON);
1641        assert!((display_qty_for_history("CANCELED", 1.0, 0.4) - 1.0).abs() < f64::EPSILON);
1642        assert!((display_qty_for_history("REJECTED", 1.0, 0.0) - 1.0).abs() < f64::EPSILON);
1643    }
1644
1645    #[test]
1646    fn split_symbol_assets_parses_known_quote_suffixes() {
1647        assert_eq!(
1648            split_symbol_assets("ETHUSDT"),
1649            ("ETH".to_string(), "USDT".to_string())
1650        );
1651        assert_eq!(
1652            split_symbol_assets("ETHBTC"),
1653            ("ETH".to_string(), "BTC".to_string())
1654        );
1655    }
1656
1657    #[test]
1658    fn split_symbol_assets_falls_back_when_quote_unknown() {
1659        assert_eq!(
1660            split_symbol_assets("FOOBAR"),
1661            ("FOOBAR".to_string(), String::new())
1662        );
1663    }
1664
1665    #[test]
1666    fn strategy_limit_rejects_when_active_orders_reach_limit() {
1667        let mut mgr = build_test_order_manager();
1668        let client_order_id = "sq-cfg-abcdef12".to_string();
1669        mgr.active_orders.insert(
1670            client_order_id.clone(),
1671            Order {
1672                client_order_id,
1673                server_order_id: None,
1674                symbol: "BTCUSDT".to_string(),
1675                side: OrderSide::Buy,
1676                order_type: OrderType::Market,
1677                quantity: 0.1,
1678                price: None,
1679                status: OrderStatus::Submitted,
1680                created_at: chrono::Utc::now(),
1681                updated_at: chrono::Utc::now(),
1682                fills: vec![],
1683            },
1684        );
1685
1686        let rejected = mgr
1687            .evaluate_strategy_limits("cfg", chrono::Utc::now().timestamp_millis() as u64)
1688            .expect("must be rejected");
1689        assert_eq!(
1690            rejected.0,
1691            "risk.strategy_max_active_orders_exceeded".to_string()
1692        );
1693    }
1694
1695    #[test]
1696    fn strategy_limit_rejects_during_cooldown_window() {
1697        let mut mgr = build_test_order_manager();
1698        let now = chrono::Utc::now().timestamp_millis() as u64;
1699        mgr.mark_strategy_submit("cfg", now);
1700
1701        let rejected = mgr
1702            .evaluate_strategy_limits("cfg", now + 500)
1703            .expect("must be rejected");
1704        assert_eq!(rejected.0, "risk.strategy_cooldown_active".to_string());
1705    }
1706
1707    #[test]
1708    fn symbol_exposure_limit_rejects_when_projected_notional_exceeds_limit() {
1709        let mut mgr = build_test_order_manager();
1710        mgr.last_price = 100.0;
1711        // Buy 2.0 -> projected notional 200, but configured spot BTCUSDT limit is 150.
1712        let rejected = mgr
1713            .evaluate_symbol_exposure_limit(OrderSide::Buy, 2.0)
1714            .expect("must be rejected");
1715        assert_eq!(
1716            rejected.0,
1717            "risk.symbol_exposure_limit_exceeded".to_string()
1718        );
1719    }
1720
1721    #[test]
1722    fn symbol_exposure_limit_allows_risk_reducing_order() {
1723        let mut mgr = build_test_order_manager();
1724        mgr.last_price = 100.0;
1725        mgr.position.side = Some(OrderSide::Buy);
1726        mgr.position.qty = 2.0; // current notional 200 > limit 150
1727
1728        // Sell reduces exposure to 100; should be allowed.
1729        let rejected = mgr.evaluate_symbol_exposure_limit(OrderSide::Sell, 1.0);
1730        assert!(rejected.is_none());
1731    }
1732
1733    #[test]
1734    fn futures_trade_stats_by_source_use_realized_pnl() {
1735        let trades = vec![
1736            BinanceMyTrade {
1737                symbol: "XRPUSDT".to_string(),
1738                id: 1,
1739                order_id: 1001,
1740                price: 1.0,
1741                qty: 100.0,
1742                commission: 0.0,
1743                commission_asset: "USDT".to_string(),
1744                time: 1,
1745                is_buyer: false,
1746                is_maker: false,
1747                realized_pnl: 5.0,
1748            },
1749            BinanceMyTrade {
1750                symbol: "XRPUSDT".to_string(),
1751                id: 2,
1752                order_id: 1002,
1753                price: 1.0,
1754                qty: 100.0,
1755                commission: 0.0,
1756                commission_asset: "USDT".to_string(),
1757                time: 2,
1758                is_buyer: false,
1759                is_maker: false,
1760                realized_pnl: -2.5,
1761            },
1762        ];
1763        let mut source_by_order = std::collections::HashMap::new();
1764        source_by_order.insert(1001, "c20".to_string());
1765        source_by_order.insert(1002, "c20".to_string());
1766
1767        let stats = compute_trade_stats_by_source(trades, &source_by_order, "XRPUSDT#FUT");
1768        let c20 = stats.get("c20").expect("source tag must exist");
1769        assert_eq!(c20.trade_count, 2);
1770        assert_eq!(c20.win_count, 1);
1771        assert_eq!(c20.lose_count, 1);
1772        assert!((c20.realized_pnl - 2.5).abs() < f64::EPSILON);
1773    }
1774}