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