feat: Telegram notifications + SSE position stream + PnL summary

- Add telegram.rs: send/edit messages via Bot API
- Add SSE position listener: position_opened → edit "Filled",
  position_closed → edit TP/SL result + send daily+alltime PnL summary
- Extend State with tg_message_id and trade details for notifications
- Add PosState to persist position info across SSE events
- Add Deal domain type and history_deals() client method
- New env: TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, TELEGRAM_THREAD_ID

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.env.example

@@ -40,3 +40,8 @@ BACKTEST_CANDLES=50000
 # ── Live Trading ─────────────────────────────────────────────────────────────
 # LIVE=true
 # LIVE_POLL_SECS=30
+
+# ── Telegram Notifications ───────────────────────────────────────────────────
+# TELEGRAM_BOT_TOKEN=123456:ABC-xxxx
+# TELEGRAM_CHAT_ID=-100xxxxxxxxxx
+# TELEGRAM_THREAD_ID=123           # optional, for forum topics/threads

Cargo.lock

@@ -45,7 +45,9 @@ dependencies = [
  "chrono",
  "domain",
  "dotenvy",
+ "futures-util",
  "mt5-client",
+ "reqwest",
  "rust_decimal",
  "serde",
  "serde_json",
@@ -286,6 +288,29 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
 
+[[package]]
+name = "futures-io"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
+
+[[package]]
+name = "futures-macro"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+]
+
+[[package]]
+name = "futures-sink"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
+
 [[package]]
 name = "futures-task"
 version = "0.3.32"
@@ -299,7 +324,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
 dependencies = [
  "futures-core",
+ "futures-io",
+ "futures-macro",
+ "futures-sink",
  "futures-task",
+ "memchr",
  "pin-project-lite",
  "slab",
 ]
@@ -994,6 +1023,7 @@ dependencies = [
  "base64",
  "bytes",
  "futures-core",
+ "futures-util",
  "http",
  "http-body",
  "http-body-util",
@@ -1013,12 +1043,14 @@ dependencies = [
  "sync_wrapper",
  "tokio",
  "tokio-rustls",
+ "tokio-util",
  "tower",
  "tower-http",
  "tower-service",
  "url",
  "wasm-bindgen",
  "wasm-bindgen-futures",
+ "wasm-streams",
  "web-sys",
  "webpki-roots",
 ]
@@ -1414,6 +1446,19 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-util"
+version = "0.7.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "toml_datetime"
 version = "1.1.1+spec-1.1.0"
@@ -1688,6 +1733,19 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "wasm-streams"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65"
+dependencies = [
+ "futures-util",
+ "js-sys",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
 [[package]]
 name = "web-sys"
 version = "0.3.100"

Cargo.toml

@@ -15,6 +15,8 @@ path = "src/main.rs"
 anyhow             = "1"
 chrono             = { version = "0.4", features = ["serde"] }
 dotenvy            = "0.15"
+futures-util       = "0.3"
+reqwest            = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "stream"] }
 rust_decimal       = { version = "1", features = ["serde"] }
 serde              = { version = "1", features = ["derive"] }
 serde_json         = "1"

crates/domain/src/deal.rs

@@ -0,0 +1,26 @@
+use chrono::{DateTime, Utc};
+use rust_decimal::Decimal;
+use serde::Deserialize;
+
+use crate::serde_helpers;
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Deal {
+    pub ticket: u64,
+    #[serde(with = "serde_helpers::naive_utc_secs")]
+    pub time: DateTime<Utc>,
+    pub entry: u8,   // 0 = open, 1 = close
+    pub magic: u64,
+    #[serde(deserialize_with = "serde_helpers::de_decimal")]
+    pub volume: Decimal,
+    #[serde(deserialize_with = "serde_helpers::de_decimal")]
+    pub price: Decimal,
+    #[serde(deserialize_with = "serde_helpers::de_decimal")]
+    pub profit: Decimal,
+    #[serde(deserialize_with = "serde_helpers::de_decimal")]
+    pub commission: Decimal,
+    #[serde(deserialize_with = "serde_helpers::de_decimal")]
+    pub swap: Decimal,
+    pub symbol: String,
+    pub comment: String,
+}

crates/domain/src/lib.rs

@@ -2,6 +2,7 @@ mod serde_helpers;
 
 pub mod account;
 pub mod candle;
+pub mod deal;
 pub mod position;
 pub mod symbol;
 pub mod tick;
@@ -9,6 +10,7 @@ pub mod timeframe;
 
 pub use account::AccountInfo;
 pub use candle::Candle;
+pub use deal::Deal;
 pub use position::{Position, Side};
 pub use symbol::Symbol;
 pub use tick::Tick;

crates/mt5-client/src/client.rs

@@ -122,6 +122,23 @@ impl Mt5Client {
         Ok(w.data)
     }
 
+    pub async fn history_deals(
+        &self,
+        date_from: &str,
+        date_to: &str,
+        symbol: Option<&str>,
+    ) -> Result<Vec<domain::Deal>, Mt5Error> {
+        let url = format!("{}/history/deals", self.base_url);
+        let mut req = self.http.get(&url)
+            .query(&[("date_from", date_from), ("date_to", date_to)]);
+        if let Some(sym) = symbol {
+            req = req.query(&[("symbol", sym)]);
+        }
+        let text = self.fetch_text(req).await?;
+        let w: DataVec<domain::Deal> = serde_json::from_str(&text)?;
+        Ok(w.data)
+    }
+
     pub async fn place_order(
         &self,
         request: &TradeRequest,

crates/mt5-client/src/lib.rs

@@ -4,4 +4,5 @@ mod types;
 
 pub use client::Mt5Client;
 pub use error::Mt5Error;
+pub use domain::Deal;
 pub use types::{HealthStatus, OrderCheckResult, PendingOrder, TradeRequest, TradeResult};

src/live.rs

@@ -1,55 +1,62 @@
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use domain::{Side, Timeframe};
+use futures_util::StreamExt;
 use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::Arc};
 use tokio::time::{interval, Duration};
 
-use crate::{detector, helpers::{d2f, fmt_price, rolling_ema, size_position}};
+use crate::{
+    detector,
+    helpers::{d2f, fmt_price, rolling_ema, size_position},
+    telegram::TelegramConfig,
+};
 
-// Magic number that identifies all Ares orders/positions in MT5.
+const MAGIC:        u64 = 19730;
-const MAGIC: u64 = 19730;
-
-// How many recent candles to fetch per tick (must cover EMA warm-up + 3 for FVG).
 const CANDLE_FETCH: u32 = 100;
 
 // ── config ────────────────────────────────────────────────────────────────────
 
 #[derive(Debug, Clone)]
 pub struct LiveConfig {
-    pub symbol: String,
+    pub symbol:             String,
-    pub timeframe: Timeframe,
+    pub timeframe:          Timeframe,
-    pub risk_pct: Decimal,
+    pub risk_pct:           Decimal,
-    pub body_pct_min: Decimal,
+    pub body_pct_min:       Decimal,
-    pub close_pct_min: Decimal,
+    pub close_pct_min:      Decimal,
     pub fvg_expiry_candles: usize,
-    pub min_fvg_pips: Decimal,
+    pub min_fvg_pips:       Decimal,
-    pub min_sl_pips: Decimal,
+    pub min_sl_pips:        Decimal,
-    pub sl_buffer: Decimal,
+    pub sl_buffer:          Decimal,
-    pub min_rr: Decimal,
+    pub min_rr:             Decimal,
-    pub slippage_points: Decimal,
+    pub slippage_points:    Decimal,
-    pub spread_override: Option<Decimal>,
+    pub spread_override:    Option<Decimal>,
-    pub ema_period: usize,
+    pub ema_period:         usize,
-    pub poll_secs: u64,
+    pub poll_secs:          u64,
+    pub mt5_base_url:       String,
+    pub telegram:           Option<TelegramConfig>,
 }
 
-// ── persisted state ───────────────────────────────────────────────────────────
+// ── pending order state ───────────────────────────────────────────────────────
 
 #[derive(Debug, Serialize, Deserialize)]
 struct State {
-    ticket: u64,
+    ticket:        u64,
-    expires_at: DateTime<Utc>,
+    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 {
+    fn path(symbol: &str) -> PathBuf { PathBuf::from(format!(".ares_state_{symbol}.json")) }
-        PathBuf::from(format!(".ares_state_{symbol}.json"))
-    }
 
     fn load(symbol: &str) -> Option<Self> {
-        let content = std::fs::read_to_string(Self::path(symbol)).ok()?;
+        serde_json::from_str(&std::fs::read_to_string(Self::path(symbol)).ok()?).ok()
-        serde_json::from_str(&content).ok()
     }
 
     fn save(&self, symbol: &str) -> Result<()> {
@@ -57,9 +64,52 @@ impl State {
         Ok(())
     }
 
-    fn clear(symbol: &str) {
+    fn clear(symbol: &str) { let _ = std::fs::remove_file(Self::path(symbol)); }
-        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,
+}
+
+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 position event ────────────────────────────────────────────────────────
+
+#[derive(Debug, Deserialize)]
+struct SsePosition {
+    ticket:        u64,
+    symbol:        String,
+    #[serde(rename = "type")]
+    pos_type:      u32,
+    volume:        f64,
+    price_open:    f64,
+    price_current: Option<f64>,
+    sl:            f64,
+    tp:            f64,
+    profit:        f64,
+    magic:         u64,
 }
 
 // ── entry point ───────────────────────────────────────────────────────────────
@@ -79,20 +129,30 @@ pub async fn run(mt5: &mt5_client::Mt5Client, cfg: &LiveConfig) -> Result<()> {
     let spread    = cfg.spread_override
         .unwrap_or_else(|| Decimal::from(sym_info.spread) * point);
 
-    let tf_mins = timeframe_minutes(cfg.timeframe);
+    let tf_mins    = timeframe_minutes(cfg.timeframe);
     let expiry_dur = chrono::Duration::minutes(tf_mins * cfg.fvg_expiry_candles as i64);
 
+    // spawn SSE position listener
+    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());
+        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_arc, http_arc, base_url, symbol, tg).await;
+        });
+    }
+
+    let http = reqwest::Client::new();
     let mut ticker = interval(Duration::from_secs(cfg.poll_secs));
 
     loop {
         ticker.tick().await;
-
         if let Err(e) = tick(
-            mt5, cfg, &sym_info, contract_size, point, prec, pip_size, min_sl, min_zone,
+            mt5, cfg, &http, &sym_info, contract_size, point, prec,
-            slip, spread, profit_is_usd, expiry_dur,
+            pip_size, min_sl, min_zone, slip, spread, profit_is_usd, expiry_dur,
-        )
+        ).await {
-        .await
-        {
             tracing::error!("tick error: {e:#}");
         }
     }
@@ -100,9 +160,11 @@ pub async fn run(mt5: &mt5_client::Mt5Client, cfg: &LiveConfig) -> Result<()> {
 
 // ── 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,
@@ -117,20 +179,34 @@ async fn tick(
 ) -> Result<()> {
     let symbol = &cfg.symbol;
 
-    // ── 1. check for open positions by this bot ───────────────────────────────
+    // ── 1. open position? ─────────────────────────────────────────────────────
     let positions = mt5.positions().await.context("fetch positions")?;
-    let has_position = positions
+    let has_position = positions.iter().any(|p| p.symbol == *symbol && p.magic == MAGIC);
-        .iter()
-        .any(|p| p.symbol == *symbol && p.magic == MAGIC);
 
     if has_position {
-        tracing::debug!(%symbol, "position already open — skip");
+        // ensure PosState exists so SSE task can find it on close
+        if PosState::load(symbol).is_none() {
+            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),
+                };
+                let _ = ps.save(symbol);
+            }
+        }
+        tracing::debug!(%symbol, "position open — skip");
         return Ok(());
     }
 
-    // ── 2. manage pending order state ────────────────────────────────────────
+    // ── 2. manage pending order ───────────────────────────────────────────────
     if let Some(state) = State::load(symbol) {
-        let orders = mt5.orders(symbol).await.context("fetch orders")?;
+        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 {
@@ -139,65 +215,58 @@ async fn tick(
                 return Ok(());
             }
             // expired — cancel
-            tracing::info!(%symbol, ticket = state.ticket, "FVG setup expired — cancelling order");
+            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"),
+                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{} {}\nEntry: {:.5}\nNo fill after {} candles",
+                    symbol, state.side, state.entry,
+                    cfg.fvg_expiry_candles,
+                );
+                let _ = tg.edit(http, msg_id, &text).await;
+            }
         } else {
-            tracing::info!(%symbol, ticket = state.ticket, "pending order no longer in MT5 (filled/cancelled externally)");
+            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. fetch recent candles ───────────────────────────────────────────────
+    // ── 3. fetch candles ──────────────────────────────────────────────────────
     let candles = mt5
         .rates_from_pos(symbol, cfg.timeframe, 0, CANDLE_FETCH)
         .await
         .context("fetch candles")?;
 
-    if candles.len() < 5 {
+    if candles.len() < 5 { return Ok(()); }
-        tracing::warn!(%symbol, "too few candles");
-        return Ok(());
-    }
-
     let n = candles.len();
 
-    // Use last 3 fully-closed bars: [n-4], [n-3], [n-2] — skip [n-1] which may
+    let pre     = &candles[n - 4];
-    // still be forming at poll time.
+    let impulse = &candles[n - 3];
-    let pre      = &candles[n - 4];
+    let post    = &candles[n - 2];
-    let impulse  = &candles[n - 3];
-    let post     = &candles[n - 2];
-    let last_idx = n - 4; // detector uses absolute index only for expiry, we don't need it
 
-    // ── 4. EMA trend filter ───────────────────────────────────────────────────
+    // ── 4. 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();
-        let emas = rolling_ema(&closes, cfg.ema_period);
+        rolling_ema(&closes, cfg.ema_period)[n - 2]
-        emas[n - 2]
     } else {
         Some(Decimal::ZERO)
     };
 
-    // ── 5. detect momentum FVG ────────────────────────────────────────────────
+    // ── 5. detect FVG ─────────────────────────────────────────────────────────
-    let fvg = detector::detect(
+    let fvg = match detector::detect(
-        pre,
+        pre, impulse, post,
-        impulse,
+        cfg.body_pct_min, cfg.close_pct_min, min_zone,
-        post,
+        n - 4, cfg.fvg_expiry_candles,
-        cfg.body_pct_min,
+    ) {
-        cfg.close_pct_min,
-        min_zone,
-        last_idx,
-        cfg.fvg_expiry_candles,
-    );
-
-    let fvg = match fvg {
         Some(f) => f,
         None    => return Ok(()),
     };
 
-    // EMA filter
     let ema_ok = match ema_val {
         Some(ema) => match fvg.side {
             Side::Long  => post.close > ema,
@@ -205,63 +274,46 @@ async fn tick(
         },
         None => false,
     };
-    if !ema_ok {
+    if !ema_ok { return Ok(()); }
-        tracing::debug!(%symbol, ?fvg.side, "EMA filter rejected FVG");
-        return Ok(());
-    }
 
-    // ── 6. compute SL / TP ───────────────────────────────────────────────────
+    // ── 6. 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 {
+    if sl_dist < min_sl { return Ok(()); }
-        tracing::debug!(%symbol, %sl_dist, "SL too tight — skip");
-        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,
     };
 
-    // ── 7. size position ──────────────────────────────────────────────────────
+    // ── 7. position size ──────────────────────────────────────────────────────
-    let acct = mt5.account().await.context("fetch account")?;
+    let acct    = mt5.account().await.context("fetch account")?;
-    let balance = Decimal::try_from(acct.balance).context("balance conversion")?;
+    let balance = Decimal::try_from(acct.balance).context("balance")?;
-
+    let value_per_lot = if profit_is_usd || post.close == Decimal::ZERO {
-    let ref_price = post.close;
-    let value_per_lot = if profit_is_usd || ref_price == Decimal::ZERO {
         contract_size
     } else {
-        contract_size / ref_price
+        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    => {
+        None    => return Ok(()),
-            tracing::warn!(%symbol, "position sizing returned None (SL=0 or too small)");
-            return Ok(());
-        }
     };
 
-    // ── 8. place pending limit order ──────────────────────────────────────────
+    // ── 8. place order ────────────────────────────────────────────────────────
-    let entry_price = fvg.entry;
     let req = mt5_client::TradeRequest::limit(
-        fvg.side, symbol.clone(), d2f(volume), d2f(entry_price), d2f(sl), d2f(tp),
+        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(entry_price, prec),
+        entry = %fmt_price(fvg.entry, prec), sl = %fmt_price(sl, prec), tp = %fmt_price(tp, prec),
-        sl    = %fmt_price(sl, prec),
+        vol = %volume, bal = %balance, "placing limit order",
-        tp    = %fmt_price(tp, prec),
-        vol   = %volume,
-        bal   = %balance,
-        "placing limit order",
     );
 
     let result = mt5.place_order(&req).await.context("place_order")?;
@@ -269,15 +321,234 @@ async fn tick(
         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{} {}\nEntry: {}\nSL: {}   TP: {}\nVol: {} lot   RR: {:.1}",
+            symbol, side_str,
+            fmt_price(fvg.entry, prec), fmt_price(sl, prec), fmt_price(tp, 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,
+        ticket:        result.order,
-        expires_at: Utc::now() + expiry_dur,
+        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 pos: SsePosition = match serde_json::from_str(data) {
+        Ok(p)  => p,
+        Err(e) => { tracing::warn!("SSE parse error: {e:#}"); return; }
+    };
+
+    if pos.magic != MAGIC || pos.symbol != symbol { return; }
+
+    match event_type {
+        "position_opened" => on_position_opened(http, symbol, &pos, tg).await,
+        "position_closed" => on_position_closed(mt5, http, symbol, &pos, tg).await,
+        _ => {}
+    }
+}
+
+async fn on_position_opened(
+    http:   &reqwest::Client,
+    symbol: &str,
+    pos:    &SsePosition,
+    tg:     &Option<TelegramConfig>,
+) {
+    tracing::info!(%symbol, ticket = pos.ticket, "position opened (SSE)");
+
+    // promote pending state → position state
+    let state = State::load(symbol);
+    let tg_msg_id = state.as_ref().and_then(|s| s.tg_message_id);
+
+    let ps = PosState {
+        ticket:        pos.ticket,
+        tg_message_id: tg_msg_id,
+        side:          if pos.pos_type == 0 { "Long".to_string() } else { "Short".to_string() },
+        entry:         pos.price_open,
+        sl:            pos.sl,
+        tp:            pos.tp,
+        volume:        pos.volume,
+    };
+    let _ = ps.save(symbol);
+    State::clear(symbol);
+
+    if let (Some(tg), Some(msg_id)) = (tg, tg_msg_id) {
+        let side_str = &ps.side;
+        let text = format!(
+            "⚡ <b>FILLED</b>\n{} {}\nEntry: {:.5}\nSL: {:.5}   TP: {:.5}\nVol: {:.2} lot",
+            symbol, side_str, pos.price_open, pos.sl, pos.tp, pos.volume,
+        );
+        let _ = tg.edit(http, msg_id, &text).await;
+    }
+}
+
+async fn on_position_closed(
+    mt5:    &mt5_client::Mt5Client,
+    http:   &reqwest::Client,
+    symbol: &str,
+    pos:    &SsePosition,
+    tg:     &Option<TelegramConfig>,
+) {
+    tracing::info!(%symbol, ticket = pos.ticket, profit = pos.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 pos.profit >= 0.0 { ("✅", "TP HIT") } else { ("❌", "SL HIT") };
+    let exit = pos.price_current.unwrap_or(0.0);
+    let entry = ps.as_ref().map(|s| s.entry).unwrap_or(pos.price_open);
+    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{:.5} → {:.5}\nVol: {:.2} lot\nBal: ${}",
+        icon, label, pos.profit,
+        symbol, side_str,
+        entry, exit,
+        pos.volume, acct_bal,
+    );
+    let _ = tg.edit(http, msg_id, &text).await;
+
+    // send PnL summary after close
+    let _ = send_pnl_summary(mt5, http, symbol, tg).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 = |label: &str, 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 profit: 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!(
+            "{}\nTrades: {} ({}W / {}L)   WR: {:.0}%\nProfit: ${:+.2}",
+            label, total, wins, losses, wr, profit,
+        )
+    };
+
+    let today_date = now.format("%Y-%m-%d").to_string();
+    let text = format!(
+        "📊 <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(())
 }
 
@@ -308,4 +579,3 @@ fn timeframe_minutes(tf: Timeframe) -> i64 {
         Timeframe::Mn1 => 43200,
     }
 }
-

src/main.rs

@@ -2,6 +2,7 @@ mod backtest;
 mod detector;
 mod helpers;
 mod live;
+mod telegram;
 
 use anyhow::Context;
 use chrono::NaiveDate;
@@ -90,12 +91,26 @@ async fn main() -> anyhow::Result<()> {
     let spread_override  = env_spread_override()?;
     let ema_period       = env_usize("EMA_PERIOD",       "20")?;
 
-    let mt5 = Arc::new(mt5_client::Mt5Client::new(mt5_base_url));
+    let mt5 = Arc::new(mt5_client::Mt5Client::new(mt5_base_url.clone()));
 
     // ── live mode ─────────────────────────────────────────────────────────────
     let live_mode = std::env::var("LIVE").map(|v| v == "true" || v == "1").unwrap_or(false);
     if live_mode {
         let poll_secs = env_u64("LIVE_POLL_SECS", "30")?;
+        let tg_token   = std::env::var("TELEGRAM_BOT_TOKEN").ok().filter(|s| !s.is_empty());
+        let tg_chat_id = std::env::var("TELEGRAM_CHAT_ID").ok()
+            .and_then(|s| s.parse::<i64>().ok());
+        let tg_thread_id = std::env::var("TELEGRAM_THREAD_ID").ok()
+            .and_then(|s| s.parse::<i64>().ok());
+        let telegram = match (tg_token, tg_chat_id) {
+            (Some(token), Some(chat_id)) => Some(telegram::TelegramConfig {
+                token,
+                chat_id,
+                thread_id: tg_thread_id,
+            }),
+            _ => None,
+        };
+
         let base_cfg = live::LiveConfig {
             symbol:            String::new(), // filled per-spawn
             timeframe,
@@ -111,6 +126,8 @@ async fn main() -> anyhow::Result<()> {
             spread_override,
             ema_period,
             poll_secs,
+            mt5_base_url: mt5_base_url.clone(),
+            telegram,
         };
 
         let mut handles = Vec::new();

src/telegram.rs

@@ -0,0 +1,67 @@
+use anyhow::{Context, Result};
+use serde::Deserialize;
+
+#[derive(Debug, Clone)]
+pub struct TelegramConfig {
+    pub token:     String,
+    pub chat_id:   i64,
+    pub thread_id: Option<i64>,
+}
+
+#[derive(Deserialize)]
+struct TgResp<T> {
+    ok:          bool,
+    result:      Option<T>,
+    description: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct TgMsg {
+    message_id: i64,
+}
+
+impl TelegramConfig {
+    fn url(&self, method: &str) -> String {
+        format!("https://api.telegram.org/bot{}/{}", self.token, method)
+    }
+
+    pub async fn send(&self, http: &reqwest::Client, text: &str) -> Result<i64> {
+        let mut body = serde_json::json!({
+            "chat_id":                  self.chat_id,
+            "text":                     text,
+            "parse_mode":               "HTML",
+            "disable_web_page_preview": true,
+        });
+        if let Some(tid) = self.thread_id {
+            body["message_thread_id"] = serde_json::json!(tid);
+        }
+        let resp: TgResp<TgMsg> = http
+            .post(self.url("sendMessage"))
+            .json(&body)
+            .send().await?
+            .json().await?;
+        if !resp.ok {
+            anyhow::bail!("sendMessage: {}", resp.description.unwrap_or_default());
+        }
+        Ok(resp.result.context("no result")?.message_id)
+    }
+
+    pub async fn edit(&self, http: &reqwest::Client, msg_id: i64, text: &str) -> Result<()> {
+        let body = serde_json::json!({
+            "chat_id":                  self.chat_id,
+            "message_id":               msg_id,
+            "text":                     text,
+            "parse_mode":               "HTML",
+            "disable_web_page_preview": true,
+        });
+        let resp: TgResp<serde_json::Value> = http
+            .post(self.url("editMessageText"))
+            .json(&body)
+            .send().await?
+            .json().await?;
+        if !resp.ok {
+            tracing::warn!("Telegram editMessageText: {}", resp.description.unwrap_or_default());
+        }
+        Ok(())
+    }
+}