ares/src/live.rs

use anyhow::{Context, Result};
use chrono::{DateTime, Timelike, Utc};
use domain::{Side, Timeframe};
use futures_util::StreamExt;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use std::{path::PathBuf, sync::Arc};
use tokio::time::{interval, Duration};

use crate::{
    detector,
    helpers::{d2f, fmt_price, rolling_ema, size_position},
    telegram::TelegramConfig,
};

const MAGIC:        u64 = 19730;
const CANDLE_FETCH: u32 = 100;

// ── config ────────────────────────────────────────────────────────────────────

#[derive(Debug, Clone)]
pub struct LiveConfig {
    pub symbol:             String,
    pub timeframe:          Timeframe,
    pub risk_pct:           Decimal,
    pub body_pct_min:       Decimal,
    pub close_pct_min:      Decimal,
    pub fvg_expiry_candles: usize,
    pub min_fvg_pips:       Decimal,
    pub min_sl_pips:        Decimal,
    pub sl_buffer:          Decimal,
    pub min_rr:             Decimal,
    pub slippage_points:    Decimal,
    pub spread_override:    Option<Decimal>,
    pub ema_period:         usize,
    pub poll_secs:            u64,
    pub mt5_base_url:         String,
    pub telegram:             Option<TelegramConfig>,
    // ── optimisations ────────────────────────────────────────────────────────
    pub session_from_utc:     Option<u32>,
    pub session_to_utc:       Option<u32>,
    pub breakeven_at_rr:      Decimal,
    pub daily_loss_limit_pct: Decimal,
}

// ── pending order state ───────────────────────────────────────────────────────

#[derive(Debug, Serialize, Deserialize)]
struct State {
    ticket:        u64,
    expires_at:    DateTime<Utc>,
    tg_message_id: Option<i64>,
    side:          String,
    entry:         f64,
    sl:            f64,
    tp:            f64,
    volume:        f64,
}

impl State {
    fn path(symbol: &str) -> PathBuf { PathBuf::from(format!(".ares_state_{symbol}.json")) }

    fn load(symbol: &str) -> Option<Self> {
        serde_json::from_str(&std::fs::read_to_string(Self::path(symbol)).ok()?).ok()
    }

    fn save(&self, symbol: &str) -> Result<()> {
        std::fs::write(Self::path(symbol), serde_json::to_string_pretty(self)?)?;
        Ok(())
    }

    fn clear(symbol: &str) { let _ = std::fs::remove_file(Self::path(symbol)); }
}

// ── open position state (persisted for SSE notifications after fill) ──────────

#[derive(Debug, Serialize, Deserialize)]
struct PosState {
    ticket:        u64,
    tg_message_id: Option<i64>,
    side:          String,
    entry:         f64,
    sl:            f64,
    tp:            f64,
    volume:        f64,
    #[serde(default)]
    be_set:        bool,
}

impl PosState {
    fn path(symbol: &str) -> PathBuf { PathBuf::from(format!(".ares_pos_{symbol}.json")) }

    fn load(symbol: &str) -> Option<Self> {
        serde_json::from_str(&std::fs::read_to_string(Self::path(symbol)).ok()?).ok()
    }

    fn save(&self, symbol: &str) -> Result<()> {
        std::fs::write(Self::path(symbol), serde_json::to_string_pretty(self)?)?;
        Ok(())
    }

    fn clear(symbol: &str) { let _ = std::fs::remove_file(Self::path(symbol)); }
}

// ── SSE event payload (unified for positions + orders) ────────────────────────

#[derive(Debug, Deserialize)]
struct SsePayload {
    ticket:        u64,
    symbol:        String,
    magic:         u64,
    #[serde(rename = "type")]
    kind:          Option<u32>,
    volume:        Option<f64>,
    price_open:    Option<f64>,   // position open price
    price:         Option<f64>,   // order limit price
    price_current: Option<f64>,
    sl:            Option<f64>,
    tp:            Option<f64>,
    profit:        Option<f64>,
}

// ── entry point ───────────────────────────────────────────────────────────────

pub async fn run(mt5: &mt5_client::Mt5Client, cfg: &LiveConfig) -> Result<()> {
    tracing::info!(symbol = %cfg.symbol, tf = ?cfg.timeframe, "live mode starting");

    let sym_info = mt5.symbol(&cfg.symbol).await.context("fetch symbol info")?;
    let point         = sym_info.point;
    let prec          = sym_info.digits as usize;
    let contract_size = sym_info.trade_contract_size;
    let profit_is_usd = sym_info.currency_profit.eq_ignore_ascii_case("USD");
    let pip_size  = if sym_info.digits % 2 == 1 { point * Decimal::from(10u32) } else { point };
    let min_sl    = cfg.min_sl_pips  * pip_size;
    let min_zone  = cfg.min_fvg_pips * pip_size;
    let slip      = cfg.slippage_points * point;
    let spread    = cfg.spread_override
        .unwrap_or_else(|| Decimal::from(sym_info.spread) * point);

    let tf_mins    = timeframe_minutes(cfg.timeframe);
    let expiry_dur = chrono::Duration::minutes(tf_mins * cfg.fvg_expiry_candles as i64);

    if cfg.telegram.is_some() {
        let mt5_arc  = Arc::new(mt5_client::Mt5Client::new(cfg.mt5_base_url.clone()));
        let http_arc = Arc::new(reqwest::Client::new());

        // SSE position listener
        {
            let mt5_c    = Arc::clone(&mt5_arc);
            let http_c   = Arc::clone(&http_arc);
            let symbol   = cfg.symbol.clone();
            let tg       = cfg.telegram.clone();
            let base_url = cfg.mt5_base_url.clone();
            tokio::spawn(async move {
                sse_task(mt5_c, http_c, base_url, symbol, tg).await;
            });
        }

        // hourly PnL summary
        {
            let mt5_c  = Arc::clone(&mt5_arc);
            let http_c = Arc::clone(&http_arc);
            let symbol = cfg.symbol.clone();
            let tg     = cfg.telegram.clone().unwrap();
            tokio::spawn(async move {
                let mut ticker = interval(Duration::from_secs(3600));
                ticker.tick().await; // skip first immediate tick
                loop {
                    ticker.tick().await;
                    if let Err(e) = send_pnl_summary(&mt5_c, &http_c, &symbol, &tg).await {
                        tracing::warn!(symbol = %symbol, "PnL summary error: {e:#}");
                    }
                }
            });
        }
    }

    let http = reqwest::Client::new();

    // startup notification
    if let Some(tg) = &cfg.telegram {
        let bal = mt5.account().await.ok()
            .map(|a| d2f(a.balance))
            .unwrap_or(0.0);
        let session_str = match (cfg.session_from_utc, cfg.session_to_utc) {
            (Some(f), Some(t)) => format!("{f:02}:00–{t:02}:00 UTC"),
            _ => "All hours".to_string(),
        };
        let text = format!(
            "🤖 <b>ARES Started</b>\n{} · {:?}\n\nSession  <code>{}</code>\nRisk     <code>{:.1}%</code>\nEMA      <code>{}</code>\nBalance  <code>${:.2}</code>",
            cfg.symbol, cfg.timeframe,
            session_str,
            d2f(cfg.risk_pct) * 100.0,
            cfg.ema_period,
            bal,
        );
        let _ = tg.send(&http, &text).await;
    }

    let mut ticker = interval(Duration::from_secs(cfg.poll_secs));

    loop {
        ticker.tick().await;
        if let Err(e) = tick(
            mt5, cfg, &http, &sym_info, contract_size, point, prec,
            pip_size, min_sl, min_zone, slip, spread, profit_is_usd, expiry_dur,
        ).await {
            tracing::error!("tick error: {e:#}");
        }
    }
}

// ── single poll tick ──────────────────────────────────────────────────────────

#[allow(clippy::too_many_arguments)]
async fn tick(
    mt5:           &mt5_client::Mt5Client,
    cfg:           &LiveConfig,
    http:          &reqwest::Client,
    sym_info:      &domain::Symbol,
    contract_size: Decimal,
    _point:        Decimal,
    prec:          usize,
    _pip_size:     Decimal,
    min_sl:        Decimal,
    min_zone:      Decimal,
    _slip:         Decimal,
    _spread:       Decimal,
    profit_is_usd: bool,
    expiry_dur:    chrono::Duration,
) -> Result<()> {
    let symbol = &cfg.symbol;

    // ── 1. open position? ─────────────────────────────────────────────────────
    let positions = mt5.positions().await.context("fetch positions")?;
    let has_position = positions.iter().any(|p| p.symbol == *symbol && p.magic == MAGIC);

    if has_position {
        // ensure PosState exists so SSE task can find it on close
        let mut ps = if let Some(existing) = PosState::load(symbol) {
            existing
        } else if let Some(pos) = positions.iter().find(|p| p.symbol == *symbol && p.magic == MAGIC) {
            let tg_msg_id = State::load(symbol).and_then(|s| s.tg_message_id);
            let ps = PosState {
                ticket:        pos.ticket,
                tg_message_id: tg_msg_id,
                side:          format!("{:?}", pos.side),
                entry:         d2f(pos.price_open),
                sl:            d2f(pos.sl),
                tp:            d2f(pos.tp),
                volume:        d2f(pos.volume),
                be_set:        false,
            };
            let _ = ps.save(symbol);
            ps
        } else {
            tracing::debug!(%symbol, "position open — skip");
            return Ok(());
        };

        // ── breakeven SL management ───────────────────────────────────────────
        if !ps.be_set && cfg.breakeven_at_rr > Decimal::ZERO {
            if let Some(pos) = positions.iter().find(|p| p.ticket == ps.ticket) {
                let entry    = Decimal::try_from(ps.entry).unwrap_or_default();
                let sl       = Decimal::try_from(ps.sl).unwrap_or_default();
                let sl_dist  = (entry - sl).abs();
                if sl_dist > Decimal::ZERO {
                    let be_trigger = match pos.side {
                        domain::Side::Long  => entry + sl_dist * cfg.breakeven_at_rr,
                        domain::Side::Short => entry - sl_dist * cfg.breakeven_at_rr,
                    };
                    let reached = match pos.side {
                        domain::Side::Long  => pos.price_current >= be_trigger,
                        domain::Side::Short => pos.price_current <= be_trigger,
                    };
                    if reached {
                        let new_sl = ps.entry; // move SL to entry
                        match mt5.modify_position(pos.ticket, symbol, new_sl, ps.tp).await {
                            Ok(r) if r.retcode == 10009 => {
                                tracing::info!(%symbol, ticket = pos.ticket, "breakeven SL set");
                                ps.sl     = new_sl;
                                ps.be_set = true;
                                let _ = ps.save(symbol);
                                if let (Some(tg), Some(msg_id)) = (&cfg.telegram, ps.tg_message_id) {
                                    let text = format!(
                                        "🔒 <b>BREAKEVEN</b>\n{} · {}\n\nSL moved to entry <code>{}</code>\nTP <code>{}</code>",
                                        symbol, ps.side,
                                        fp(ps.entry), fp(ps.tp),
                                    );
                                    let _ = tg.edit(http, msg_id, &text).await;
                                }
                            }
                            Ok(r)  => tracing::warn!(retcode = r.retcode, "breakeven modify retcode unexpected"),
                            Err(e) => tracing::warn!("breakeven modify failed: {e:#}"),
                        }
                    }
                }
            }
        }

        tracing::debug!(%symbol, "position open — skip");
        return Ok(());
    }

    // ── 2. manage pending order ───────────────────────────────────────────────
    if let Some(state) = State::load(symbol) {
        let orders       = mt5.orders(symbol).await.context("fetch orders")?;
        let still_pending = orders.iter().any(|o| o.ticket == state.ticket && o.magic == MAGIC);

        if still_pending {
            if Utc::now() < state.expires_at {
                tracing::debug!(%symbol, ticket = state.ticket, "pending order alive — waiting");
                return Ok(());
            }
            // expired — cancel
            tracing::info!(%symbol, ticket = state.ticket, "FVG setup expired — cancelling");
            match mt5.cancel_order(state.ticket, symbol).await {
                Ok(r)  => tracing::info!(retcode = r.retcode, "cancel ok"),
                Err(e) => tracing::warn!("cancel failed: {e:#}"),
            }
            if let (Some(tg), Some(msg_id)) = (&cfg.telegram, state.tg_message_id) {
                let text = format!(
                    "⏱ <b>EXPIRED</b>\n{} · {}\n\nEntry <code>{}</code>\nUnfilled after {} candles",
                    symbol, state.side,
                    fp(state.entry), cfg.fvg_expiry_candles,
                );
                let _ = tg.edit(http, msg_id, &text).await;
            }
        } else {
            tracing::info!(%symbol, ticket = state.ticket, "order gone from MT5 (filled or cancelled)");
            // SSE task handles "Filled" notification when position opens
        }
        State::clear(symbol);
        return Ok(());
    }

    // ── 3. session filter ─────────────────────────────────────────────────────
    if let (Some(from), Some(to)) = (cfg.session_from_utc, cfg.session_to_utc) {
        let hour = Utc::now().time().hour();
        if hour < from || hour >= to {
            tracing::debug!(%symbol, hour, from, to, "outside session window — skip");
            return Ok(());
        }
    }

    // ── 4. daily loss limit ───────────────────────────────────────────────────
    if cfg.daily_loss_limit_pct > Decimal::ZERO {
        let now       = Utc::now();
        let today_str = now.format("%Y-%m-%dT00:00:00").to_string();
        let now_str   = now.format("%Y-%m-%dT%H:%M:%S").to_string();
        if let Ok(today_deals) = mt5.history_deals(&today_str, &now_str, Some(symbol)).await {
            let daily_pnl: Decimal = today_deals.iter()
                .filter(|d| d.entry == 1 && d.magic == MAGIC)
                .map(|d| d.profit + d.commission + d.swap)
                .sum();
            let acct      = mt5.account().await.context("fetch account for daily limit")?;
            let balance   = Decimal::try_from(acct.balance).context("balance")?;
            let limit     = -(balance * cfg.daily_loss_limit_pct);
            if daily_pnl <= limit {
                tracing::info!(%symbol, %daily_pnl, %limit, "daily loss limit hit — no new trades today");
                return Ok(());
            }
        }
    }

    // ── 5. fetch candles ──────────────────────────────────────────────────────
    let candles = mt5
        .rates_from_pos(symbol, cfg.timeframe, 0, CANDLE_FETCH)
        .await
        .context("fetch candles")?;

    if candles.len() < 5 { return Ok(()); }
    let n = candles.len();

    let pre     = &candles[n - 4];
    let impulse = &candles[n - 3];
    let post    = &candles[n - 2];

    // ── 6. EMA filter ─────────────────────────────────────────────────────────
    let ema_val: Option<Decimal> = if cfg.ema_period > 0 && candles.len() >= cfg.ema_period {
        let closes: Vec<Decimal> = candles.iter().map(|c| c.close).collect();
        rolling_ema(&closes, cfg.ema_period)[n - 2]
    } else {
        Some(Decimal::ZERO)
    };

    // ── 7. detect FVG ─────────────────────────────────────────────────────────
    let fvg = match detector::detect(
        pre, impulse, post,
        cfg.body_pct_min, cfg.close_pct_min, min_zone,
        n - 4, cfg.fvg_expiry_candles,
    ) {
        Some(f) => f,
        None    => return Ok(()),
    };

    let ema_ok = match ema_val {
        Some(ema) => match fvg.side {
            Side::Long  => post.close > ema,
            Side::Short => post.close < ema,
        },
        None => false,
    };
    if !ema_ok { return Ok(()); }

    // ── 8. SL / TP ────────────────────────────────────────────────────────────
    let sl = match fvg.side {
        Side::Long  => fvg.impulse_sl - cfg.sl_buffer,
        Side::Short => fvg.impulse_sl + cfg.sl_buffer,
    };
    let sl_dist = (fvg.entry - sl).abs();
    if sl_dist < min_sl { return Ok(()); }
    let tp = match fvg.side {
        Side::Long  => fvg.entry + sl_dist * cfg.min_rr,
        Side::Short => fvg.entry - sl_dist * cfg.min_rr,
    };

    // ── 9. position size ──────────────────────────────────────────────────────
    let acct    = mt5.account().await.context("fetch account")?;
    let balance = Decimal::try_from(acct.balance).context("balance")?;
    let value_per_lot = if profit_is_usd || post.close == Decimal::ZERO {
        contract_size
    } else {
        contract_size / post.close
    };
    let volume = match size_position(
        balance, cfg.risk_pct, sl_dist, value_per_lot,
        sym_info.volume_step, sym_info.volume_min, sym_info.volume_max,
    ) {
        Some(v) => v,
        None    => return Ok(()),
    };

    // ── 10. place order ───────────────────────────────────────────────────────
    let req = mt5_client::TradeRequest::limit(
        fvg.side, symbol.clone(), d2f(volume), d2f(fvg.entry), d2f(sl), d2f(tp),
        MAGIC, format!("ares-{}", post.time.format("%m%d-%H%M")),
    );

    tracing::info!(
        %symbol, side = ?fvg.side,
        entry = %fmt_price(fvg.entry, prec), sl = %fmt_price(sl, prec), tp = %fmt_price(tp, prec),
        vol = %volume, bal = %balance, "placing limit order",
    );

    let result = mt5.place_order(&req).await.context("place_order")?;
    if result.retcode != 10009 {
        tracing::error!(retcode = result.retcode, comment = %result.comment, "order rejected");
        return Ok(());
    }
    tracing::info!(ticket = result.order, "order placed");

    let side_str = format!("{:?}", fvg.side);
    let rr       = d2f(cfg.min_rr);

    // send Telegram "Pending" message
    let tg_msg_id = if let Some(tg) = &cfg.telegram {
        let text = format!(
            "🟡 <b>PENDING</b>\n{} · {}\n\nEntry  <code>{}</code>\nTP     <code>{}</code>\nSL     <code>{}</code>\nVol {} lot  ·  RR 1:{:.1}",
            symbol, side_str,
            fmt_price(fvg.entry, prec), fmt_price(tp, prec), fmt_price(sl, prec),
            volume, rr,
        );
        match tg.send(http, &text).await {
            Ok(id)  => { tracing::info!(msg_id = id, "Telegram pending sent"); Some(id) }
            Err(e)  => { tracing::warn!("Telegram send failed: {e:#}"); None }
        }
    } else { None };

    let state = State {
        ticket:        result.order,
        expires_at:    Utc::now() + expiry_dur,
        tg_message_id: tg_msg_id,
        side:          side_str,
        entry:         d2f(fvg.entry),
        sl:            d2f(sl),
        tp:            d2f(tp),
        volume:        d2f(volume),
    };
    state.save(symbol).context("save state")?;
    Ok(())
}

// ── SSE position listener ─────────────────────────────────────────────────────

async fn sse_task(
    mt5:      Arc<mt5_client::Mt5Client>,
    http:     Arc<reqwest::Client>,
    base_url: String,
    symbol:   String,
    tg:       Option<TelegramConfig>,
) {
    loop {
        tracing::debug!(%symbol, "SSE connecting");
        if let Err(e) = sse_loop(&mt5, &http, &base_url, &symbol, &tg).await {
            tracing::warn!(%symbol, "SSE error: {e:#}");
        }
        tokio::time::sleep(Duration::from_secs(5)).await;
    }
}

async fn sse_loop(
    mt5:      &mt5_client::Mt5Client,
    http:     &reqwest::Client,
    base_url: &str,
    symbol:   &str,
    tg:       &Option<TelegramConfig>,
) -> Result<()> {
    let url = format!("{}/positions/stream?symbol={}", base_url, symbol);
    let resp = http.get(&url).send().await?;
    if !resp.status().is_success() {
        anyhow::bail!("SSE stream status {}", resp.status());
    }

    let mut stream     = resp.bytes_stream();
    let mut buf        = String::new();
    let mut event_type = String::new();

    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        buf.push_str(&String::from_utf8_lossy(&chunk));

        loop {
            match buf.find('\n') {
                None => break,
                Some(pos) => {
                    let line = buf[..pos].trim_end_matches('\r').to_string();
                    buf.drain(..=pos);

                    if line.starts_with("event:") {
                        event_type = line[6..].trim().to_string();
                    } else if line.starts_with("data:") && !event_type.is_empty() {
                        let data = line[5..].trim().to_string();
                        handle_sse_event(mt5, http, symbol, &event_type, &data, tg).await;
                        event_type.clear();
                    } else if line.is_empty() {
                        event_type.clear();
                    }
                }
            }
        }
    }
    Ok(())
}

async fn handle_sse_event(
    mt5:        &mt5_client::Mt5Client,
    http:       &reqwest::Client,
    symbol:     &str,
    event_type: &str,
    data:       &str,
    tg:         &Option<TelegramConfig>,
) {
    let payload: SsePayload = match serde_json::from_str(data) {
        Ok(p)  => p,
        Err(e) => { tracing::warn!(%symbol, event_type, "SSE parse error: {e:#}"); return; }
    };

    if payload.magic != MAGIC || payload.symbol != symbol { return; }

    match event_type {
        "position_opened"   => on_position_opened(http, symbol, &payload, tg).await,
        "position_closed"   => on_position_closed(mt5, http, symbol, &payload, tg).await,
        "position_modified" => on_position_modified(http, symbol, &payload, tg).await,
        "order_placed"      => tracing::info!(%symbol, ticket = payload.ticket, "order placed (SSE)"),
        "order_cancelled"   => on_order_cancelled(http, symbol, &payload, tg).await,
        "order_modified"    => on_order_modified(http, symbol, &payload, tg).await,
        other               => tracing::debug!(%symbol, "unknown SSE event: {other}"),
    }
}

async fn on_position_opened(
    http:    &reqwest::Client,
    symbol:  &str,
    payload: &SsePayload,
    tg:      &Option<TelegramConfig>,
) {
    tracing::info!(%symbol, ticket = payload.ticket, "position opened (SSE)");

    let state     = State::load(symbol);
    let tg_msg_id = state.as_ref().and_then(|s| s.tg_message_id);
    let kind      = payload.kind.unwrap_or(0);
    let entry     = payload.price_open.unwrap_or(0.0);
    let sl        = payload.sl.unwrap_or(0.0);
    let tp        = payload.tp.unwrap_or(0.0);
    let volume    = payload.volume.unwrap_or(0.0);

    let ps = PosState {
        ticket:        payload.ticket,
        tg_message_id: tg_msg_id,
        side:          if kind == 0 { "Long".to_string() } else { "Short".to_string() },
        entry, sl, tp, volume,
        be_set:        false,
    };
    let _ = ps.save(symbol);
    State::clear(symbol);

    if let (Some(tg), Some(msg_id)) = (tg, tg_msg_id) {
        let text = format!(
            "⚡ <b>FILLED</b>\n{} · {}\n\nEntry  <code>{}</code>\nTP     <code>{}</code>\nSL     <code>{}</code>\nVol {:.2} lot",
            symbol, ps.side,
            fp(entry), fp(tp), fp(sl), volume,
        );
        let _ = tg.edit(http, msg_id, &text).await;
    }
}

async fn on_position_closed(
    mt5:     &mt5_client::Mt5Client,
    http:    &reqwest::Client,
    symbol:  &str,
    payload: &SsePayload,
    tg:      &Option<TelegramConfig>,
) {
    let profit = payload.profit.unwrap_or(0.0);
    tracing::info!(%symbol, ticket = payload.ticket, profit, "position closed (SSE)");

    let ps = PosState::load(symbol);
    PosState::clear(symbol);

    let Some(tg) = tg else { return };
    let Some(msg_id) = ps.as_ref().and_then(|s| s.tg_message_id) else { return };

    let (icon, label) = if profit >= 0.0 { ("✅", "TAKE PROFIT") } else { ("❌", "STOP LOSS") };
    let exit     = payload.price_current.unwrap_or(0.0);
    let entry    = ps.as_ref().map(|s| s.entry).unwrap_or(payload.price_open.unwrap_or(0.0));
    let volume   = payload.volume.unwrap_or(ps.as_ref().map(|s| s.volume).unwrap_or(0.0));
    let side_str = ps.as_ref().map(|s| s.side.as_str()).unwrap_or("?");

    let acct_bal = mt5.account().await.ok().map(|a| a.balance).unwrap_or(Decimal::ZERO);

    let text = format!(
        "{} <b>{}  {:+.2}</b>\n{} · {}\n\n<code>{}</code> → <code>{}</code>\nVol {:.2} lot  ·  Bal ${:.2}",
        icon, label, profit,
        symbol, side_str,
        fp(entry), fp(exit),
        volume, d2f(acct_bal),
    );
    let _ = tg.edit(http, msg_id, &text).await;
}

async fn on_position_modified(
    http:    &reqwest::Client,
    symbol:  &str,
    payload: &SsePayload,
    tg:      &Option<TelegramConfig>,
) {
    tracing::info!(%symbol, ticket = payload.ticket, "position modified (SSE)");

    if let Some(mut ps) = PosState::load(symbol) {
        if let Some(sl) = payload.sl     { ps.sl     = sl; }
        if let Some(tp) = payload.tp     { ps.tp     = tp; }
        if let Some(v)  = payload.volume { ps.volume = v;  }
        let _ = ps.save(symbol);

        if let (Some(tg), Some(msg_id)) = (tg, ps.tg_message_id) {
            let text = format!(
                "🔄 <b>MODIFIED</b>\n{} · {}\n\nEntry  <code>{}</code>\nTP     <code>{}</code>\nSL     <code>{}</code>\nVol {:.2} lot",
                symbol, ps.side,
                fp(ps.entry), fp(ps.tp), fp(ps.sl), ps.volume,
            );
            let _ = tg.edit(http, msg_id, &text).await;
        }
    }
}

async fn on_order_cancelled(
    http:    &reqwest::Client,
    symbol:  &str,
    payload: &SsePayload,
    tg:      &Option<TelegramConfig>,
) {
    tracing::info!(%symbol, ticket = payload.ticket, "order cancelled (SSE)");

    let state = State::load(symbol);
    if state.as_ref().map(|s| s.ticket) != Some(payload.ticket) { return; }

    if let (Some(tg), Some(msg_id)) = (tg, state.as_ref().and_then(|s| s.tg_message_id)) {
        let side_str = state.as_ref().map(|s| s.side.as_str()).unwrap_or("?");
        let entry    = state.as_ref().map(|s| s.entry).unwrap_or(0.0);
        let text = format!(
            "🚫 <b>CANCELLED</b>\n{} · {}\n\nEntry <code>{}</code>",
            symbol, side_str, fp(entry),
        );
        let _ = tg.edit(http, msg_id, &text).await;
    }
    State::clear(symbol);
}

async fn on_order_modified(
    http:    &reqwest::Client,
    symbol:  &str,
    payload: &SsePayload,
    tg:      &Option<TelegramConfig>,
) {
    tracing::info!(%symbol, ticket = payload.ticket, "order modified (SSE)");

    if let Some(mut state) = State::load(symbol) {
        if state.ticket != payload.ticket { return; }
        if let Some(p)  = payload.price { state.entry = p; }
        if let Some(sl) = payload.sl    { state.sl    = sl; }
        if let Some(tp) = payload.tp    { state.tp    = tp; }
        let _ = state.save(symbol);

        if let (Some(tg), Some(msg_id)) = (tg, state.tg_message_id) {
            let text = format!(
                "🔄 <b>ORDER MODIFIED</b>\n{} · {}\n\nEntry  <code>{}</code>\nTP     <code>{}</code>\nSL     <code>{}</code>",
                symbol, state.side,
                fp(state.entry), fp(state.tp), fp(state.sl),
            );
            let _ = tg.edit(http, msg_id, &text).await;
        }
    }
}

// ── PnL summary ───────────────────────────────────────────────────────────────

async fn send_pnl_summary(
    mt5:    &mt5_client::Mt5Client,
    http:   &reqwest::Client,
    symbol: &str,
    tg:     &TelegramConfig,
) -> Result<()> {
    let now       = Utc::now();
    let today_str = now.format("%Y-%m-%dT00:00:00").to_string();
    let now_str   = now.format("%Y-%m-%dT%H:%M:%S").to_string();
    let epoch_str = "2020-01-01T00:00:00".to_string();

    let (today_deals, all_deals) = tokio::try_join!(
        mt5.history_deals(&today_str, &now_str, Some(symbol)),
        mt5.history_deals(&epoch_str, &now_str, Some(symbol)),
    )?;

    let summary_text = |deals: &[domain::Deal]| {
        let closing: Vec<_> = deals.iter()
            .filter(|d| d.entry == 1 && d.magic == MAGIC)
            .collect();
        let total  = closing.len();
        let wins   = closing.iter().filter(|d| d.profit > Decimal::ZERO).count();
        let losses = total - wins;
        let net: Decimal = closing.iter().map(|d| d.profit + d.commission + d.swap).sum();
        let wr = if total > 0 { wins as f64 / total as f64 * 100.0 } else { 0.0 };
        format!(
            "{} trades  ·  {}W {}L  ·  WR {:.0}%\nNet  <b>{:+.2}</b>",
            total, wins, losses, wr, net,
        )
    };

    let today_date = now.format("%Y-%m-%d").to_string();
    let text = format!(
        "📊 <b>PnL Summary</b>\n\n<b>Today</b>  ·  {}  ·  {}\n{}\n\n<b>All-time</b>\n{}",
        symbol, today_date,
        summary_text(&today_deals),
        summary_text(&all_deals),
    );

    tg.send(http, &text).await?;
    Ok(())
}

// ── helpers ───────────────────────────────────────────────────────────────────

fn fp(v: f64) -> String {
    let s = format!("{v:.5}");
    let s = s.trim_end_matches('0');
    s.trim_end_matches('.').to_string()
}

fn timeframe_minutes(tf: Timeframe) -> i64 {
    match tf {
        Timeframe::M1  => 1,
        Timeframe::M2  => 2,
        Timeframe::M3  => 3,
        Timeframe::M4  => 4,
        Timeframe::M5  => 5,
        Timeframe::M6  => 6,
        Timeframe::M10 => 10,
        Timeframe::M12 => 12,
        Timeframe::M15 => 15,
        Timeframe::M20 => 20,
        Timeframe::M30 => 30,
        Timeframe::H1  => 60,
        Timeframe::H2  => 120,
        Timeframe::H3  => 180,
        Timeframe::H4  => 240,
        Timeframe::H6  => 360,
        Timeframe::H8  => 480,
        Timeframe::H12 => 720,
        Timeframe::D1  => 1440,
        Timeframe::W1  => 10080,
        Timeframe::Mn1 => 43200,
    }
}